使用Kafka协议上传日志

相关限制

• 支持的Kafka协议版本为：0.8.0到2.1.1V2。
• 为保证日志传输安全性，必须使用SASL_SSL连接协议。
• 如果您的Logstore有多个Shard，请使用负载均衡模式上传日志。
• 目前只支持将Kafka Producer SDK或采集Agent采集到的日志使用Kafka协议上传日志到日志服务。

配置方式

示例1：通过Beats系列软件采集日志到日志服务

Beats系列软件（MetricBeat、PacketBeat、Winlogbeat、Auditbeat、Filebeat、Heartbeat等）采集到日志后，支持通过Kafka协议将日志上传到日志服务，详情请参见Beats-Kafka-Output。

• 配置示例

  output.kafka: 
    # initial brokers for reading cluster metadata 
    hosts: ["cn-hangzhou.log.aliyuncs.com:10012"] 
    username: "<yourusername>" 
    password: "<yourpassword>" 
    ssl.certificate_authorities: 
    # message topic selection + partitioning 
    topic: 'test-logstore-1' 
    partition.round_robin: 
      reachable_only: false 
  
    required_acks: 1 
    compression: gzip 
    max_message_bytes: 1000000

• 日志样例

  Beats系列软件默认输出的日志为JSON类型，您可以给content字段创建JSON类型的索引，详情请参见JSON类型。

  

示例2：通过Collectd采集日志到日志服务

Collectd是一个守护（daemon）进程，用于定期采集系统和应用程序的性能指标，并支持通过Kafka协议上传到日志服务，详情请参见Write Kafka Plugin。

将Collectd采集到日志上传到日志服务时，还需安装Kafka插件以及相关依赖。例如：在linux Centos中，可以使用yum安装Kafka插件，命令为sudo yum install collectd-write_kafka，安装RPM请参见Collectd-write_kafka。

• 配置示例

  示例中将日志输出格式（Format）设置为JSON，除此之外，还支持Command、Graphite类型，详情请参见Collectd配置文档。

  <Plugin write_kafka>
    Property "metadata.broker.list" "cn-hangzhou.log.aliyuncs.com:10012" 
    Property "security.protocol" "sasl_ssl" 
    Property "sasl.mechanism" "PLAIN" 
    Property "sasl.username" "<yourusername>" 
    Property "sasl.password" "<yourpassword>" 
    Property "broker.address.family" "v4"  
    <Topic "test-logstore-1">
      Format JSON 
      Key "content"  
    </Topic>
  </Plugin>
                      

• 日志样例

  使用JSON模式输出日志后，您可以给content字段创建JSON类型的索引，详情请参见JSON类型。

  

使用Telegraf采集日志到日志服务

Telegraf是由Go语言编写的代理程序，内存占用小，用于收集、处理、汇总数据指标。Telegraf具有丰富的插件及具备集成功能，可从其运行的系统中获取各种指标、从第三方API中获取指标以及通过statsd和Kafka消费者服务监听指标。

将Telegraf采集到的日志通过kafka协议上传到日志服务前，您需要先修改配置文件。

• 配置示例
  示例中将日志输出格式（Format）设置为JSON，除此之外还支持Graphite、Carbon2等类型，详情请参见Telegraf输出格式。

  说明 Telegraf必须配置一个合法的tls_ca路径，使用服务器自带的根证书的路径即可。Linux环境中，根证书CA路径一般为/etc/ssl/certs/ca-bundle.crt。

  [[outputs.kafka]] 
    ## URLs of kafka brokers 
    brokers = ["cn-hangzhou.log.aliyuncs.com:10012"] 
    ## Kafka topic for producer messages 
    topic = "test-logstore-1" 
    routing_key = "content" 
    ## CompressionCodec represents the various compression codecs recognized by 
    ## Kafka in messages. 
    ## 0 : No compression 
    ## 1 : Gzip compression 
    ## 2 : Snappy compression 
    ## 3 : LZ4 compression 
    compression_codec = 1 
    ## Optional TLS Config tls_ca = "/etc/ssl/certs/ca-bundle.crt" 
    # tls_cert = "/etc/telegraf/cert.pem" # tls_key = "/etc/telegraf/key.pem" 
    ## Use TLS but skip chain & host verification 
    # insecure_skip_verify = false 
    ## Optional SASL Config 
    sasl_username = "<yourusername>" 
    sasl_password = "<yourpassword>" 
    ## Data format to output. 
    ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md 
    data_format = "json"

• 日志样例

  使用JSON模式输出日志后，您可以给content字段创建JSON类型的索引，详情请参见JSON类型。

  

使用Fluentd采集日志到日志服务

Fluentd是一个开源的日志收集器，是云端原生计算基金会（CNCF）的成员项目之一，遵循Apache 2 License协议。

Fluentd支持众多输入、处理、输出插件，支持通过Kafka插件将日志上传到日志服务，您只需安装并配置Kafka插件即可，详情请参见fluent-plugin-kafka。

• 配置示例
  示例中将日志输出格式（Format）设置为JSON，除此之外还支持数十种Format类型，详情请参见Fluentd Formatter。

  <match **>
    @type kafka 
    # Brokers: you can choose either brokers or zookeeper. 
    brokers      cn-hangzhou.log.aliyuncs.com:10012 
    default_topic test-logstore-1 
    default_message_key content 
    output_data_type json 
    output_include_tag true 
    output_include_time true 
    sasl_over_ssl true 
    username <yourusername> 
    password <yourpassword> 
    ssl_ca_certs_from_system true 
    # ruby-kafka producer options 
    max_send_retries 10000 
    required_acks 1 
    compression_codec gzip 
  </match>

• 日志样例
  使用JSON模式输出日志后，您可以给content字段创建JSON类型的索引，详情请参见JSON类型。

示例5：使用Logstash采集日志到日志服务

Logstash是一个具备实时处理能力、开源的日志采集引擎，可以动态采集不同来源的日志。

错误信息