ECI日志采集对接Kafka

前提条件

• ECI Pod和Kafka之间网络互通。

  如果您在阿里云上部署Kafka，建议将Kafka和ECI Pod部署在同一VPC下。

• 已完成ECI日志采集的相关配置。具体操作，请参见通过阿里云日志服务采集日志。

通过日志服务控制台配置

Logtail日志采集支持文件和标准输出两种方式，您可以在日志服务控制台配置flushers插件，将日志导入Kafka。配置文本如下：

"flushers": [
    {
      "detail": {
        "Brokers": [
          "Kafka Broker地址"
        ],
        "Topic": "log"
      },
      "type": "flusher_kafka"
    }
  ]

• 对于标准输出，您可以直接修改Logtail配置，在插件配置中新增flushers插件。如下图所示。

  

• 对于文件采集，您需要在高级选项中启用插件处理开关，然后在插件配置中新增flushers插件。如下图所示。

  

通过SLS CRD配置

如果您使用的Kubernetes集群已经安装了日志服务的CRD，您可以通过CRD的方式配置flushers插件，将日志导入Kafka。

请参考以下配置示例修改SLS CRD（资源类型为AliyunLogConfig）的YAML，然后执行kubectl apply命令更新配置。

• 标准输出YAML配置示例

  apiVersion: log.alibabacloud.com/v1alpha1     
  kind: AliyunLogConfig                        
  metadata:
    name: test-stdout                     #资源名，在集群内唯一
  spec:
    project: k8s-log-xxx                  #（可选）Project名称，默认为安装时设置的Project，若指定Project请确保该Project未被使用
    logstore: test-stdout                 #Logstore名称，不存在时自动创建
    shardCount: 2                         #（可选）Shard数量，默认为2，支持1-10
    lifeCycle: 90                         #（可选）Logstore中数据的存储时间，默认为90，支持1-7300，7300天为永久存储
    logtailConfig:                        
      inputType: plugin                   #采集的数据源类型，file（文本文件）或plugin（标准输出）
      configName: test-stdout             #采集配置的名称，与资源名（metadata.name）保持一致
      inputDetail:
        plugin:
          inputs:
            - type: service_docker_stdout
              detail:
                Stdout: true
                Stderr: true
                IncludeEnv:
                  aliyun_logs_test-stdout: "stdout"
          flushers:
            - type: flusher_kafka         #flusher类型为Kafka
              detail:
                Brokers:
                - 192.XX.XX.XX:9092       #配置Kafka Broker地址
                - 192.XX.XX.XX:9092
                - 192.XX.XX.XX:9092
                Topic: log

• 文件采集YAML配置示例

  apiVersion: log.alibabacloud.com/v1alpha1    
  kind: AliyunLogConfig                         
  metadata:
    name: test-file                          
  spec:
    project: k8s-log-xxx                  #（可选）Project名称，默认为安装时设置的Project，若指定Project请确保该Project未被使用
    logstore: test-file                   #Logstore名称，不存在时自动创建
    shardCount: 2                         #（可选）Shard数量，默认为2，支持1-10
    lifeCycle: 90                         #（可选）Logstore中数据的存储时间，默认为90，支持1-7300，7300天为永久存储
    logtailConfig:                    
      inputType: file                     #采集的数据源类型，file（文本文件）或plugin（标准输出）
      configName: test-file               #采集配置的名称，与资源名（metadata.name）保持一致
      inputDetail:
          logType: common_reg_log        #对于分隔符类型的日志，logType可以设置为json_log
          logPath: /log/                 #日志文件夹
          filePattern: "*.log"           #文件名, 支持通配符，例如log_*.log
          dockerFile: true               #采集容器内的文件
          plugin:        
            flushers:
              - type: flusher_kafka        #flusher类型为Kafka
                detail:
                  Brokers:
                   - 192.XX.XX.XX:9092           ## 配置Kafka Broker地址
                   - 192.XX.XX.XX:9092
                   - 192.XX.XX.XX:9092
                  Topic: log

结果验证

配置完成后，您可以看到ECI日志已经被采集到Kafka。例如：如果使用的是阿里云消息队列Kafka版，可以在消息查询页面查询到ECI的日志，如下图所示。