Use the Kafka protocol to upload logs

Limits

• Only Kafka 0.8.0 to Kafka 2.1.1 (message format version 2) are supported.
• You must use the SASL_SSL protocol to ensure the security of log transfer.
• If your Logstore contains multiple shards, you must upload log data in load balancing mode.
• You can use only the Kafka protocol to upload logs that are collected by using Kafka Producer SDKs or collection agents.

Configurations

Example 1: Use Beats to upload logs to Log Service

You can use Beats such as MetricBeat, PacketBeat, Winlogbeat, Auditbeat, Filebeat, and Heartbeat to collect logs. After the logs are collected, you can use the Kafka protocol to upload the logs to Log Service. For more information, visit Beats-Kafka-Output.

• Configuration example

  output.kafka: 
    # initial brokers for reading cluster metadata 
    hosts: ["test-project-1.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

• Sample log entry

  By default, Beats provide JSON-formatted logs in the content field. You can create a JSON index for the content field. For more information, visit JSON.

  

Example 2: Use collectd to upload logs to Log Service

collectd is a daemon that is used to collect the performance metrics of systems and applications at a regular interval. The collectd daemon allows you to upload logs to Log Service by using the Kafka protocol. For more information, visit Write Kafka Plug-in.

Before you upload logs to Log Service, you must install the collectd-write_kafka plug-in and relevant dependencies. If you are using CentOS, you can run the sudo yum install collectd-write_kafka command to install the collectd-write_kafka plug-in. For more information, visit RPM resource collectd-write_kafka.

• Configuration example

  In the following configuration example, the output format of logs is set to JSON. You can also set the output format to Command and Graphite. For more information, visit Manpage collectd.conf.

  <Plugin write_kafka>
    Property "metadata.broker.list" "test-project-1.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>
                      

• Sample log entry

  Logs are sent to the Log Service console in the JSON format. The log content is included in the content field. You can create a JSON index for the content field. For more information, see JSON.

  

Example 3: Use Telegraf to upload logs to Log Service

Telegraf is an agent in the Go programming language and is used to collect, process, and analyze metrics. It consumes only a small number of memory resources. In addition, Telegraf supports integration with multiple plug-ins. You can use Telegraf to retrieve metrics from the system where it runs, or from third-party APIs. You can also use Telegraf to monitor metrics by using StatsD and Kafka consumers.

Before you use the Kafka protocol to upload collected logs to Log Service, you must modify the configuration file of Telegraf.

• Configuration example
  In the following configuration example, the output format of logs is set to JSON. You can also set the output format to Graphite and Carbon2. For more information, visit Telegraf output formats.

  Note You must set a valid tls_ca path for Telegraf. You can use the default directory of the root certificate of the server. In a Linux-based server, the default directory of the root certificate is /etc/ssl/certs/ca-bundle.crt.

  [[outputs.kafka]] 
    ## URLs of kafka brokers 
    brokers = ["test-project-1.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"

• Sample log entry

  Logs are sent to the Log Service console in the JSON format. The log content is included in the content field. You can create a JSON index for the content field. For more information, see JSON.

  

Example 4: Use Fluentd to upload logs to Log Service

Fluentd is an open source log data collector. It is a Cloud Native Computing Foundation (CNCF) project and complies with the Apache 2 License protocol.

Fluentd is compatible with multiple input plug-ins, processing plug-ins, and output plug-ins. You can use the fluent-plugin-kafka plug-in to upload logs to Log Service. For more information about how to install and configure this plug-in, visit fluent-plugin-kafka.

• Configuration example
  In the following configuration example, the output format of logs is set to JSON. Fluentd also supports other output formats. For more information, visit Fluentd Formatter.

  <match **>
    @type kafka 
    # Brokers: you can choose either brokers or zookeeper. 
    brokers      test-project-1.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>

• Sample log entry
  Logs are sent to the Log Service console in the JSON format. The log content is included in the content field. You can create a JSON index for the content field. For more information, see JSON.

Example 5: Use Logstash to upload logs to Log Service

Logstash is an open source log collection engine that provides real-time processing capabilities. You can use Logstash to dynamically collect logs from different sources.

Error messages