# 使用Logstash将OBS数据导入Elasticsearch
通过CSS服务的Logstash将OBS数据导入Elasticsearch，可实现高效的数据检索与分析能力，满足数据迁移场景。
#### 应用场景
对象存储服务OBS用于存储海量数据。当需要对这些数据进行快速搜索和分析时，可以使用CSS服务的Logstash将其导入到Elasticsearch集群中。常见场景包括：
- 数据检索与分析： 将存储在OBS中的数据（如日志、业务数据等）定期或实时同步到Elasticsearch，以便快速执行全文检索、聚合分析和可视化。
- OBS桶操作日志分析： 将OBS桶的操作日志（访问日志）同步到Elasticsearch，便于审计追踪、行为分析和异常检测。
 
#### 方案架构
图1架构图   
![](https://support.huaweicloud.com/bestpractice-css/figure/zh-cn_image_0000002405348005.png "点击放大")
Logstash支持从多种数据源采集数据。对于存储在OBS中的数据，可以使用Logstash的[logstash-input-s3](https://www.elastic.co/docs/reference/logstash/plugins/plugins-inputs-s3)插件来读取OBS上的对象，并通过[logstash-output-elasticsearch](https://www.elastic.co/docs/reference/logstash/plugins/plugins-outputs-elasticsearch)插件将处理后的数据同步到目标Elasticsearch集群。
1. 输入 (Input)：通过logstash-input-s3插件配置连接到OBS桶，监控指定的文件并读取内容。
2. 处理 (Processing)：Logstash配置文件中可选的filter部分用于对读取的数据进行清洗、转换和结构化处理（例如，使用grok过滤器解析日志格式）。
3. 输出 (Output)：logstash-output-elasticsearch插件配置连接到目标Elasticsearch集群，将数据索引化。
 
#### 方案优势
- 兼容性强：支持多种数据格式，如JSON、CSV、OBS日志。
- 高扩展性：可结合Filter插件对数据进行清洗、字段提取或格式转换。
- 灵活性高： 支持多种目标存储，如CSS服务的Elasticsearch、自建Elasticsearch或第三方Elasticsearch。
 
#### 前提条件
- 目标数据已上传至OBS桶中，且OBS桶必须和Elasticsearch集群在同一区域。并获取OBS桶的如下信息：桶名称、终端节点（Endpoint）和区域（Region），获取方式请参见[OBS访问规则](https://support.huaweicloud.com/usermanual-obs/obs_41_0003.html)。
- 已创建CSS Elasticsearch集群，并获取Elasticsearch集群的访问地址（Host），如果是安全模式的集群还需要获取用户名和密码。
- 已创建CSS Logstash集群，并确认Logstash与Elasticsearch部署在同一VPC下，确保网络连通。
- 已获取账号的AK/SK，获取方式请参见[如何获取访问密钥AK/SK](https://support.huaweicloud.com/iam_faq/iam_01_0618.html)。
 
#### 操作步骤
1. 访问CSS服务的Logstash集群。 
   1. 登录[云搜索服务管理控制台](https://console.huaweicloud.com/elasticsearch/)。
   
   2. 在左侧导航栏，选择"集群管理 \> Logstash"。
   
   
   
   
2. 网络连通性测试。当Logstash和Elasticsearch部署在相同VPC下时，可跳过该步骤。 
   1. 在Logstash集群列表，选择目标集群，单击操作列的"配置中心"。
   
   2. 在配置中心页面，单击"连通性测试"。
   
   3. 在连通性测试弹窗中，输入Elasticsearch的访问地址和端口号，单击"测试"。 当显示"可用"时，表示集群间网络连通。如果显示"不可用"可以参考[Logstash连通性测试的结果为不可用时怎么办？](https://support.huaweicloud.com/css_faq/css_02_0090.html)解决。
      
   
   
   
   
3. 准备Logstash配置文件。 
   1. 在Logstash集群列表，选择目标集群，单击操作列的"配置中心"。
   
   2. 在配置中心页面，单击右上角"创建"，编辑配置文件并保存。
      配置文件内容（以下仅为配置示例，需基于业务需求修改）：
      ```
      input { 
        s3 { 
          access_key_id => "YOUR_AK"         # 账号的Access Key ID 
          secret_access_key => "YOUR_SK"     # 账号的Secret Access Key 
          bucket => "log_obs_access"         # OBS桶名称 
          prefix => "test/access_log"        # （可选）包含桶路径的文件名前缀，如果不指定路径则表示桶下所有匹配文件名前缀的文件，例如此样例中，获取的是log_obs_access桶的test路径下所有以access_log开头的文件, 不支持正则匹配格式
          endpoint => "https://OBS_Endpoint"  # OBS访问地址（此处域名是OBS终端节点） 
          region => "REGION"                  # OBS所属区域 
          watch_for_new_files => false        # （可选）是否监控新文件变化，用于控制是否重复读取数据，默认值是true 
          backup_to_bucket => "backup_log_obs_access"  # （可选）备份桶，用于备份已经读取过的文件，不设置则不备份 
          backup_add_prefix => "backup/"      # （可选）备份文件的存储路径前缀，和prefix组合使用，例如此样例中，实际备份路径是/backup/test/access_log
        } 
      } 
      filter { 
        # 移除一些Logstash事件添加的元数据字段
        mutate { 
          remove_field => ["@version"]
        } 
      } 
      output { 
        elasticsearch { 
          # 目的端集群的节点访问地址，无需添加协议
          hosts => ["xxx.xxx.xxx.xxx:9200", "xxx.xxx.xxx.xxx:9200"]
          # 事件写入的索引名称
          index  => "delivery_events_log_alias"
          manage_template => false              # 不管理索引模板 
          ilm_enabled => false                  # 关闭ILM策略（CSS不支持开启）
          # 安全集群必配 (非安全集群请删除以下配置)
          # user => "xxx"           # 访问集群的用户名
          # password => "xxx"       # 用户名对应的密码
          # 目的端集群启用SSL时，则需额外配置以下信息
          # ssl => true
          # cacert => "/opt/logstash/extend/certs"       # 用于认证目的端集群的CA证书文件路径
          # ssl_certificate_verification => false        # 目的端集群是否启用安全证书认证
        } 
      }
      ```
      ![](https://support.huaweicloud.com/bestpractice-css/public_sys-resources/note_3.0-zh-cn.png)
      - filter配置可以自定义pattern切割目标文件的数据，例如分析OBS桶的访问日志时，可以使用grok匹配模式，配置示例请参见[常见问题：如何查看OBS桶访问日志？]除了grok匹配模式还支持dissect、kv、mutate、split等其他方式，更多详细使用指导请参见[Filter plugins](https://www.elastic.co/docs/reference/logstash/plugins/filter-plugins)。
      
      - 如果需要多次使用Logstash读取OBS数据，可以通过联合配置watch_for_new_files、delete和backup_to_bucket参数避免数据重复。配置示例请参见[常见问题：读取的OBS数据有重复怎么办？]
        
   
   
   
   
4. 启动Logstash管道任务。 
   1. 在配置中心页面，选择新建的配置文件，单击左上角的"启动"。
   
   2. 在"启动Logstash服务"对话框中，勾选"是否保持常驻"，开启保持常驻，确保Logstash配置文件持续运行，不会因为服务重启或其他原因导致数据处理中断。 保持常驻会在每个节点上面配置一个守护进程，当Logstash服务出现故障的时候，会主动拉起并修复，从而保证Logstash管道稳定运行。
      
   
   3. 单击"确定"，启动配置文件。 在管道列表中可以查看启动的配置文件状态，监控数据迁移情况。
      
   
   
   
   
5. 验证数据同步。 
   1. 登录[云搜索服务管理控制台](https://console.huaweicloud.com/elasticsearch/)。
   
   2. 在左侧导航栏，选择"集群管理 \> Elasticsearch"。
   
   3. 在集群列表，选择目标集群，单击操作列的"Kibana"，登录Kibana。
   
   4. 在Kibana左侧导航栏选择"Dev Tools"，进入操作页面。
   
   5. 执行索引查询命令，检查目标索引是否返回了预期的数据记录数。
      ```
      GET delivery_events_log_alias/_count
      {
        "query": {"match_all": {}}
      }
      ```
      如果返回结果中"count"值不为0，表示数据已成功同步。
      
   
   
   
   
 
 #### 常见问题：如何查看OBS桶访问日志？
当OBS桶开启了日志记录功能，则可以通过Logstash读取OBS桶的访问日志，将日志数据写入Elasticsearch，查看并分析OBS桶的操作行为。
OBS桶日志格式如下所示：
```
15e02840b2784ffb9dcac293afc01a75 genean-hot-test [02/Feb/2024:07:51:17 +0000] 58.250.177.72 15e02840b2784ffb9dcac293afc01a75 0000018D68CCEF1AD3296F98BB7B4255 REST.GET.OBJECT 00002c9d-172a-495d-b0a3-4aca8c1f86cc "GET /genean-hot-test/00002c9d-172a-495d-b0a3-4aca8c1f86cc?AWSAccessKeyId=H0N1CQY4D1ZBSHABB3NF&Expires=1706860576&response-content-disposition=attachment&response-content-type=application/octet-stream&x-amz-security-token=*****&Signature=DeC2lkZVFB4CjDjKe147ZOl8sKY%3D HTTP/1.1" 200 - 82509 82509 84 84 "https://console.example.com/" "Mozilla/5.0 (xx.xx.xx; xx; xx) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0" - - STANDARD - "-" efe6d963779e4475bb81c5cb1f5f5368
```
OBS日志字段的详细定义请参见[使用桶日志记录OBS日志信息](https://support.huaweicloud.com/usermanual-obs/obs_03_0329.html)。
在Logstash配置文件中增加filter配置可以自定义切割OBS桶的日志数据，配置方式参考如下，此示例使用grok匹配模式。
```
filter {
 grok {
   match => {
     "message" => '(?<BucketOwner>[^ ]+) (?<Bucket>[^ ]+) \[%{HTTPDATE:Time}\] (?<RemoteIP>[^ ]+) (?<Requester>[^ ]+) (?<RequestID>[^ ]+) (?<Operation>[^ ]+) (?<Key>[^ ]+) \"(?<RequestURI>[^"]+)\" (?<HTTPStatus>[^ ]+) (?<ErrorCode>[^ ]+) (?<BytesSent>[^ ]+) (?<ObjectSize>[^ ]+) (?<TotalTime>[^ ]+) (?<Turn-AroundTime>[^ ]+) \"(?<Referer>[^"]+)\" \"(?<User-Agent>[^"]+)\" (?<VersionID>[^ ]+) (?<STSLogUrn>[^ ]+) (?<StorageClass>[^ ]+) (?<TargetStorageClass>[^ ]+) \"(?<DentryName>[^ ]+)\" (?<IAMUserID>[^ ]+)'
    }
    timeout_millis => 3000
    timeout_scope => "event"
  }
   mutate {
    remove_field => ["@version","message"]
  }
}
```
OBS桶的日志数据输出到Elasticsearch后，即可搜索分析。
 #### 常见问题：读取的OBS数据有重复怎么办？
logstash-input-s3插件通过watch_for_new_files参数判断是否需要监控文件变化。
- 当watch_for_new_files为false，任务第一次运行时会读取配置中桶路径下的数据，读取完成后会记录文件的最后修改时间信息到sincedb文件中，任务退出。此时，再次重启Logstash任务会将和最后修改时间一样的文件重复消费，导致数据重复。为避免数据重复消费，建议不要重启Logstash消费任务。
- 当watch_for_new_files为true，任务持续运行时就会定期读取配置中桶路径下的数据，读取完成后会记录文件修改时间信息到sincedb文件中，下一次会基于已经记录的时间向后读取数据，如果桶中的文件没有更新则读取和最后修改时间一样的文件，导致数据重复消费。为了避免数据重复消费，可以结合delete和backup_to_bucket参数，将已消费的文件备份到新的OBS桶中，并删除原始桶中已消费过的文件。配置示例如下：
  ```
  input {
    s3 {
      access_key_id => "YOUR_AK"
      secret_access_key => "YOUR_SK"
      bucket => "log_obs_access"
      prefix => "logstash"
      endpoint => "https://OBS_Endpoint"
      region => "REGION"
      delete => true                           #  文件读取完成后是否删除原始桶中的文件，默认是false，设置为true会删除原始文件
      watch_for_new_files => true              #  是否监控新文件变化，默认值是true，当前有文件添加到桶中会自动读取数据
      backup_to_bucket => "backup_log_obs_access"  # 备份桶名称，将读取完成的文件备份到该OBS桶中，默认不备份
      backup_add_prefix => "backup/"            # 备份文件的存储路径前缀，和prefix组合使用，例如此样例中，实际备份路径地址是/backup/logstash
    }
  }
  ```