日志服务使用 Kafka 协议上传日志-操作指南-文档中心-腾讯云

日志服务（Cloud Log Service，CLS）目前已支持使用 Kafka Producer SDK 和其他 Kafka 相关 agent 上传日志到 CLS。
使用场景
日志应用中使用 Kafka 作为消息管道是非常普遍的场景。如通过机器上的开源采集客户端或者使用 producer 直接写入日志，再通过消费管道提供给下游如 spark、flink 等进行消费。CLS 具备完整的 Kafka 数据管道上下行能力，以下主要介绍哪些场景适合您使用 Kafka 协议上传日志，更多 Kafka 协议消费场景请参见 Kafka 协议实时消费。
场景1：您已有基于开源采集的自建系统，不希望有复杂的二次改造，您可以通过修改配置文件将日志上传到 CLS。
例如，您之前使用 ELK 搭建日志系统的客户，现在只需要通过修改 Filebeat 或者 Logstash 的配置文件，将 Output 配置（详情请参见 filebeat 配置）到 CLS，即可非常方便简洁地将日志上传。
场景2：您希望通过 Kafka producer SDK 来采集日志并上传，不必再安装采集 Agent。
CLS 支持您使用各类 Kafka producer SDK 采集日志，并通过 Kafka 协议上传到 CLS。（详情请参见本文提供的 SDK 调用示例 ）
相关限制
支持 Kafka 协议版本为：0.11.0.X，1.0.X，1.1.X，2.0.X，2.1.X，2.2.X，2.3.X，2.4.X，2.5.X，2.6.X，2.7.X，2.8.X。
支持压缩方式：gzip，snappy，lz4。
当前使用 SASL_PLAINTEXT 认证。
使用 Kafka 协议上传需要配置 RealtimeProducer 权限，详情请参见 CLS 访问策略模板。
配置方式
使用 Kafka 协议上传日志至 CLS Kafka 生产端时，需要配置下 CLS Kafka 访问信息：
参数
说明
鉴权机制
当前支持 SASL_PLAINTEXT
hosts
CLS Kafka 地址，根据目标写入日志主题所在地域配置。请参见 CLS Kafka 地址。
topic
CLS Kafka topic 名称，配置为日志主题 ID。例如：76c63473-c496-466b-XXXX-XXXXXXXXXXXX
username
CLS Kafka 访问用户名，配置为日志集 ID。例如：0f8e4b82-8adb-47b1-XXXX-XXXXXXXXXXXX
password
CLS Kafka 访问密码，格式为 ${SecurityId}#${SecurityKey}。例如：XXXXXXXXXXXXXX#YYYYYYYY。密钥信息获取请前往 密钥获取。请确保密钥关联的账号具有相应的 Kafka 协议上传日志权限。
若要匿名写入，格式为 topic_id#${日志主题 ID}。例如：topic_id#76c63473-c496-466b-XXXX-XXXX。 
注意：
若使用匿名写入，目标日志主题需开启匿名访问，并在匿名操作选择 Kafka 协议上传日志。详情请参考 管理日志主题。
﻿
﻿
﻿
header
定义在通过 Kafka 协议上传时日志的解析行为。
json_remove_escape：对日志进行 JSON 解析时是否进行去转义，取值为 true、false，若不指定默认为 false。
time_key：日志中的指定时间字段，表示选择日志中的指定字段作为日志采集时间。
time_format：当配置了 time_key，需额外配置指定字段的时间解析格式，详情请参考 配置时间格式。
CLS Kafka 地址
通过 Kafka 协议上传日志时，基于访问方式（内网/外网）以及目标日志主题所在地域，CLS Kafka 地址配置为如下：
说明：
1. 地域缩写可参考 可用域名 - Kafka上传日志。
2. 访问方式：
内网访问：数据发起端为腾讯云服务器且服务器所在地域与目标日志主题一致
外网访问：数据发起端为非腾讯云服务器或者服务器所在地域与目标日志主题不一致
访问方式
CLS Kafka 地址
内网
${地域缩写}-producer.cls.tencentyun.com:9095
外网
${地域缩写}-producer.cls.tencentcs.com:9096
示例
Beat 调用示例
filebeat/winlogbeat 配置
output.kafka:
  enabled: true
  hosts: ["${region}-producer.cls.tencentyun.com:9095"] # TODO 服务地址；外网端口9096，内网端口9095
  topic: "${ClsTopicID}" # TODO 日志主题ID
  version: "0.11.0.2"
  compression: "${compress}"   # TODO 配置压缩方式，支持gzip，snappy，lz4；例如"lz4"
  username: "${ClslogsetID}" # TODO 日志集ID
  # 若要匿名写入，password: "topic_id#${日志主题 ID}"
  password: "${SecurityId}#${SecurityKey}" 
Logstash 调用示例
logstash 配置
output {
  kafka {
    topic_id => "${ClstopicID}" 
    bootstrap_servers => "${region}-producer.cls.tencentyun.com:${port}"
    sasl_mechanism => "PLAIN"
    security_protocol => "SASL_PLAINTEXT"
    compression_type => "${compress}"
    # 若要匿名写入，password='topic_id#${日志主题 ID}'
    sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='${ClslogsetID}' password='${securityID}#${securityKEY}';"
    codec => json
  }
}
FluentTD 调用示例
注意：
该 demo 基于 fluentd-1.15.3 版本验证；依赖的 ruby 版本 ruby=2.7.6。
FluentTD 配置
<match *>
  @type rdkafka2
﻿
  # brokers setting
  # TODO 域名参考 https://cloud.tencent.com/document/product/614/18940，注意内网端口9095，公网端口9096
  brokers "${domain}:${port}" # 例如 gz-producer.cls.tencentyun.com:9095
﻿
  # topic settings
  # TODO 替换日志主题ID
  topic "${topic_id}"
﻿
  # sasl
  rdkafka_options {
    "sasl.mechanism": "PLAIN",
    "security.protocol": "sasl_plaintext",
    # TODO  topic所属日志集ID
    "sasl.username": "${logset_id}", 
    # TODO  topic所属uin的密钥，格式为 ${secret_id}#${secret_key}；若要匿名写入，格式为 topic_id#${日志主题 ID}
    "sasl.password": "${secret_id}#${secret_key}" 
  }
﻿
  required_acks 1
  compression_codec gzip
﻿
  <format>
    @type json
  </format>
﻿
  <buffer tag>
    flush_at_shutdown true
    flush_mode interval
    flush_interval 1s
    chunk_limit_size 3MB
    chunk_full_threshold 1
    total_limit_size 1024MB
    overflow_action block
  </buffer>
</match>
FluentBit 调用示例
FluentBit 配置
[OUTPUT]
    Name kafka
    Match *
    # TODO 域名参考 https://cloud.tencent.com/document/product/614/18940，注意内网端口9095，公网端口9096
    Brokers ${domain}:${port} # 例如 gz-producer.cls.tencentyun.com:9095
    # TODO 替换日志主题ID
    Topics ${topic_id}
    # TODO 请求消息的最大大小，最大不能超过5M
    rdkafka.message.max.bytes 5242880
    rdkafka.sasl.mechanisms PLAIN
    rdkafka.security.protocol sasl_plaintext
    # TODO 根据使用场景选择acks的值
    rdkafka.acks 1
    # TODO 配置压缩方式
    rdkafka.compression.codec lz4
    # TODO topic所属日志集ID
    rdkafka.sasl.username ${logset_id}
    # TODO topic所属uin的密钥，格式为 ${secret_id}#${secret_key}；若要匿名写入，格式为 topic_id#${日志主题 ID}
    rdkafka.sasl.password ${secret_id}#${secret_key}
SDK 调用示例
Golang SDK 调用示例
以下以 sarama.V1_1_0_0为例，其他版本按照类似规则配置：
import (
    "fmt"
    "github.com/Shopify/sarama"
)
﻿
func main() {
    config := sarama.NewConfig()
﻿
    config.Net.SASL.Mechanism = "PLAIN"
    config.Net.SASL.Version = int16(1)
    config.Net.SASL.Enable = true
    // TODO 日志集 ID
    config.Net.SASL.User = "${logsetID}"                        
    // TODO 格式为 ${SecurityId}#${SecurityKey}，若要匿名写入，格式为 topic_id#${日志主题 ID}
    config.Net.SASL.Password = "${SecurityId}#${SecurityKey}"   
    config.Producer.Return.Successes = true
    // TODO 根据使用场景选择acks的值
    config.Producer.RequiredAcks = ${acks}                      
    config.Version = sarama.V1_1_0_0
    // TODO 配置压缩方式
    config.Producer.Compression = ${compress}                   
﻿
    // TODO 服务地址；外网端口9096，内网端口9095
    producer, err := sarama.NewSyncProducer([]string{"${region}-producer.cls.tencentyun.com:9095"}, config)
    if err != nil {
        panic(err)
    }
﻿
    msg := &sarama.ProducerMessage{
        Topic: "${topicID}", // TODO 日志主题ID
        Value: sarama.StringEncoder("goland sdk sender demo"),
    }
    // 发送消息
    for i := 0; i <= 5; i++ {
        partition, offset, err := producer.SendMessage(msg)
        if err != nil {
            panic(err)
        }
        fmt.Printf("send response; partition:%d, offset:%d\\n", partition, offset)
    }
﻿
    _ = producer.Close()
﻿
}
Python SDK 调用示例
from kafka import KafkaProducer
﻿
if __name__ == '__main__':
    produce = KafkaProducer(
        # TODO 服务地址；外网端口9096，内网端口9095
        bootstrap_servers=["${region}-producer.cls.tencentyun.com:9095"],
        security_protocol='SASL_PLAINTEXT',
        sasl_mechanism='PLAIN',
        # TODO 日志集 ID
        sasl_plain_username='${logsetID}',
        # TODO 格式为 ${SecurityId}#${SecurityKey}, 若要匿名写入，格式为 topic_id#${日志主题 ID} 
        sasl_plain_password='${SecurityId}#${SecurityKey}',
        api_version=(0, 11, 0),
        # TODO 配置压缩方式
        compression_type="${compress_type}",
    )
﻿
    for i in range(0, 5):
        # 发送消息 TODO 日志主题ID
        future = produce.send(topic="${topicID}", value=b'python sdk sender demo')
        result = future.get(timeout=10)
        print(result)
Java SDK 调用示例
maven 依赖：
<dependencies>
  <!--https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients-->
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.11.0.2</version>
  </dependency>
</dependencies>
代码示例：
import org.apache.kafka.clients.producer.*;
﻿
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
﻿
public class ProducerDemo {
    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        // 0.配置一系列参数
        Properties props = new Properties();
        // TODO 使用时
        props.put("bootstrap.servers", "${region}-producer.cls.tencentyun.com:9095");
        // TODO 以下值根据业务场景设置 
        props.put("acks", ${acks});
        props.put("retries", ${retries});
        props.put("batch.size", ${batch.size});
        props.put("linger.ms", ${linger.ms});
        props.put("buffer.memory", ${buffer.memory});
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "${compress_type}"); // TODO 配置压缩方式
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
﻿
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.mechanism", "PLAIN");
        // TODO 用户名为logsetID；密码为securityID和securityKEY的组合securityID#securityKEY,格式为 ${SecurityId}#${SecurityKey}, 
        // 若要匿名写入，密码为 topic_id#${日志主题 ID} 
        props.put("sasl.jaas.config",
                "org.apache.kafka.common.security.plain.PlainLoginModule required username='${logsetID}' password='${SecurityId}#${SecurityKey}';");
﻿
        // 1.创建一个生产者对象
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
﻿
        // 2.调用send方法 TODO 日志主题ID
        Future<RecordMetadata> meta = producer.send(new ProducerRecord<String, String>("${topicID}", ${message}));
        RecordMetadata recordMetadata = meta.get(${timeout}, TimeUnit.MILLISECONDS);
        System.out.println("offset = " + recordMetadata.offset());
﻿
        // 3.关闭生产者
        producer.close();
    }
}
C SDK 调用示例
// https://github.com/edenhill/librdkafka - master
#include <iostream>
#include <librdkafka/rdkafka.h>
#include <string>
#include <unistd.h>
﻿
#define BOOTSTRAP_SERVER "${region}-producer.cls.tencentyun.com:${port}"
// USERNAME 为日志集 ID
#define USERNAME "${logsetID}"
// PASSWORD 格式为 ${SecurityId}#${SecurityKey}, 若要匿名写入，格式为 topic_id#${日志主题 ID} 
#define PASSWORD "${SecurityId}#${SecurityKey}"
// 日志主题ID
#define TOPIC "${topicID}"
#define ACKS "${acks}"
// 配置压缩方式
#define COMPRESS_TYPE "${compress_type}"
﻿
static void dr_msg_cb(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *opaque) {
    if (rkmessage->err) {
        fprintf(stdout, "%% Message delivery failed : %s\\n", rd_kafka_err2str(rkmessage->err));
    } else {
        fprintf(stdout, "%% Message delivery successful %zu:%d\\n", rkmessage->len, rkmessage->partition);
    }
}
﻿
int main(int argc, char **argv) {
    // 1. 初始化配置
    rd_kafka_conf_t *conf = rd_kafka_conf_new();
﻿
    rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb);
﻿
    char errstr[512];
    if (rd_kafka_conf_set(conf, "bootstrap.servers", BOOTSTRAP_SERVER, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "%s\\n", errstr);
        return -1;
    }
﻿
    if (rd_kafka_conf_set(conf, "acks", ACKS, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "%s\\n", errstr);
        return -1;
    }
﻿
    if (rd_kafka_conf_set(conf, "compression.codec", COMPRESS_TYPE, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "%s\\n", errstr);
        return -1;
    }
﻿
    // 设置认证方式
    if (rd_kafka_conf_set(conf, "security.protocol", "sasl_plaintext", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "%s\\n", errstr);
        return -1;
    }
    if (rd_kafka_conf_set(conf, "sasl.mechanisms", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "%s\\n", errstr);
        return -1;
    }
    if (rd_kafka_conf_set(conf, "sasl.username", USERNAME, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "%s\\n", errstr);
        return -1;
﻿
    }
    if (rd_kafka_conf_set(conf, "sasl.password", PASSWORD, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "%s\\n", errstr);
        return -1;
    }
﻿
    // 2. 创建 handler
    rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    if (!rk) {
        rd_kafka_conf_destroy(conf);
        fprintf(stdout, "create produce handler failed: %s\\n", errstr);
        return -1;
    }
﻿
    // 3. 发送数据
    std::string value = "test lib kafka ---- ";
    for (int i = 0; i < 100; ++i) {
        retry:
        rd_kafka_resp_err_t err = rd_kafka_producev(
                rk, RD_KAFKA_V_TOPIC(TOPIC),
                RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
                RD_KAFKA_V_VALUE((void *) value.c_str(), value.size()),
                RD_KAFKA_V_OPAQUE(nullptr), RD_KAFKA_V_END);
﻿
        if (err) {
            fprintf(stdout, "Failed to produce to topic : %s, error : %s", TOPIC, rd_kafka_err2str(err));
            if (err == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
                rd_kafka_poll(rk, 1000);
                goto retry;
            }
        } else {
            fprintf(stdout, "send message to topic successful : %s\\n", TOPIC);
        }
﻿
        rd_kafka_poll(rk, 0);
    }
﻿
    std::cout << "message flush final" << std::endl;
    rd_kafka_flush(rk, 10 * 1000);
﻿
    if (rd_kafka_outq_len(rk) > 0) {
        fprintf(stdout, "%d message were not deliverer\\n", rd_kafka_outq_len(rk));
    }
﻿
    rd_kafka_destroy(rk);
﻿
    return 0;
}
C# SDK 调用示例
/*
 * 该demo只提供了最简单的使用方法，具体生产还需要结合调用放来实现
 * 在使用过程中，demo中留的todo项需要替换使用
 *
 * 注意：
 *  1. 该Demo基于Confluent.Kafka/1.8.2版本验证通过
 *  2. MessageMaxBytes最大值不能超过5M
 *  3. 该demo使用同步的方式生产，在使用时也可根据业务场景调整为异步的方式
 *  4. 其他参数在使用过程中可以根据业务参考文档自己调整：https://docs.confluent.io/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.ProducerConfig.html
 *
 * Confluent.Kafka 参考文档：https://docs.confluent.io/platform/current/clients/confluent-kafka-dotnet/_site/api/Confluent.Kafka.html
 */
﻿
﻿
using Confluent.Kafka;
﻿
namespace Producer
{
    class Producer
    {
        private static void Main(string[] args)
        {
            var config = new ProducerConfig
            {
                // TODO 域名参考 https://cloud.tencent.com/document/product/614/18940
               //  Kafka 填写，注意内网端口9095，公网端口9096
                BootstrapServers = "${domain}:${port}", 
                SaslMechanism = SaslMechanism.Plain,
                // TODO topic所属日志集ID
                SaslUsername = "${logsetID}", 
                // TODO topic所属uin的密钥，格式为 ${SecurityId}#${SecurityKey}, 
                // 若要匿名写入，格式为 topic_id#${日志主题 ID} 
                SaslPassword = "${SecurityId}#${SecurityKey}",
                SecurityProtocol = SecurityProtocol.SaslPlaintext,
                // TODO 根据实际使用场景赋值。可取值: Acks.None、Acks.Leader、Acks.All
                Acks         = Acks.None, 
                // TODO 请求消息的最大大小，最大不能超过5M
                MessageMaxBytes = 5242880 
            };
﻿
            // deliveryHandler
            Action<DeliveryReport<Null, string>> handler =
                r => Console.WriteLine(!r.Error.IsError ? $"Delivered message to {r.TopicPartitionOffset}" : $"Delivery Error: {r.Error.Reason}");
﻿
﻿
            using (var produce = new ProducerBuilder<Null, string>(config).Build())
            {
                try
                {
                    // TODO 测试验证代码
                    for (var i = 0; i < 100; i++)
                    {
                        // TODO 替换日志主题ID
                        produce.Produce("${topicID}", new Message<Null, string> { Value = "C# demo value" }, handler);
                    }
                    produce.Flush(TimeSpan.FromSeconds(10));
﻿
                }
                catch (ProduceException<Null, string> pe)
                {
                    Console.WriteLine($"send message receiver error : {pe.Error.Reason}");
                }
            }
        }
    }
}
﻿
﻿
参数	说明
鉴权机制	当前支持 SASL_PLAINTEXT
hosts	CLS Kafka 地址，根据目标写入日志主题所在地域配置。请参见 CLS Kafka 地址。
topic	CLS Kafka topic 名称，配置为日志主题 ID。例如：76c63473-c496-466b-XXXX-XXXXXXXXXXXX
username	CLS Kafka 访问用户名，配置为日志集 ID。例如：0f8e4b82-8adb-47b1-XXXX-XXXXXXXXXXXX
password	CLS Kafka 访问密码，格式为 `${SecurityId}#${SecurityKey}`。例如：XXXXXXXXXXXXXX#YYYYYYYY。密钥信息获取请前往密钥获取。请确保密钥关联的账号具有相应的 Kafka 协议上传日志权限。若要匿名写入，格式为 topic_id#${日志主题 ID}。例如：topic_id#76c63473-c496-466b-XXXX-XXXX。注意：若使用匿名写入，目标日志主题需开启匿名访问，并在匿名操作选择 Kafka 协议上传日志。详情请参考管理日志主题。
header	定义在通过 Kafka 协议上传时日志的解析行为。 json_remove_escape：对日志进行 JSON 解析时是否进行去转义，取值为 true、false，若不指定默认为 false。 time_key：日志中的指定时间字段，表示选择日志中的指定字段作为日志采集时间。 time_format：当配置了 time_key，需额外配置指定字段的时间解析格式，详情请参考配置时间格式。
访问方式	CLS Kafka 地址
内网	${地域缩写}-producer.cls.tencentyun.com:9095
外网	${地域缩写}-producer.cls.tencentcs.com:9096
使用 Kafka 协议上传日志

本页目录：

使用场景

相关限制

配置方式

CLS Kafka 地址

示例

Beat 调用示例

filebeat/winlogbeat 配置

Logstash 调用示例

logstash 配置

FluentTD 调用示例

FluentTD 配置

FluentBit 调用示例

FluentBit 配置

SDK 调用示例

Golang SDK 调用示例

Python SDK 调用示例

Java SDK 调用示例

C SDK 调用示例

C# SDK 调用示例
