您可以通过Kafka协议上报日志到日志服务，目前支持各类Kafka Producer SDK或采集工具，仅依赖于Kafka协议。支持以下场景：

• 场景1：已有基于开源采集的自建系统，仅修改配置文件便可以将日志上报到LTS，例如Logstash。

• 场景2：希望通过Kafka producer SDK来采集日志并上报，不必再安装采集ICAgent。

前提条件

• 确认云日志服务的区域，请用户根据所在区域，获取regionid。

• 获取需要上报到LTS的日志组ID、日志流ID。

• 当前仅支持内网上报，需要在ECS主机上使用。

相关限制

• 当前仅支持内网上报，端口固定为，IP根据所在局点进行配置。

• 支持 Kafka 协议版本为：1.0.X，2.X.X，3.X.X。

• 支持压缩方式：gzip，snappy，lz4。

• KAFKA认证方式为 SASL_PLAINTEXT 认证。

• KAFKA协议的ACKS参数必须设置为0。

配置方式

• 使用Kafka协议上报日志时，需要使用到的通用参数如下。

通用参数

• 使用Kafka协议上报日志时，需要配置以下参数。

配置参数

• ${message}日志格式

  仅当headers中添加了key为LTS_LOG_TYPE，value为FORMAT的header时，日志需要符合该格式规范。

日志参数

日志示例

{ 
"tenant_project_id": "${projectId}", 
"tenant_group_id": "${logGroupId}", 
"tenant_stream_id": "${logStreamId}", 
"log_time_ns": "XXXXXXXXXXXXXXXXXXX", 
"contents": [ 
"This is a log 1", 
"This is a log 2" 
], 
"labels": { 
"type": "kafka" 
} 
}

调用示例

1. Beat系列软件调用（FileBeat等）。以FileBeat为例，配置参数如下：

output.kafka: 
hosts: ["${ip}:${port}"] 
partition.round_robin: 
reachable_only: false 
username: "${projectId}" 
password: "${accessKey}#${accessSecret}" 
topic: "${logGroupId}_${logStreamId}" 
sasl.mechanism: "PLAIN" 
security.protocol: "SASL_PLAINTEXT" 
acks: "0" 
compression: gzip

2. 通过Logstash软件上报日志。

input { 
stdin {} 
} 
output { 
kafka { 
# 配置地址
bootstrap_servers => "${ip}:${port}" 
# 配置topic 
topic_id => "${logGroupId}_${logStreamId}" 
# 配置消息确认机制
acks => "0" 
# 配置压缩方式
compression_type => "gzip" 
# 配置认证方式
security_protocol => "SASL_PLAINTEXT" 
sasl_mechanism => "PLAIN" 
# 用户名 projectId 密码 accessKey#accessSecret 
sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='${projectId}' password='${accessKey}#${accessSecret}';" 
} 
}

3. 通过Flume软件上报日志。

#Name 
a1.sources = r1 
a1.channels = c1 
a1.sinks = k1 
#Source 
a1.sources.r1.type = TAILDIR 
a1.sources.r1.channels = c1 
a1.sources.r1.filegroups = f1 
a1.sources.r1.filegroups.f1 = /tmp/test.txt 
a1.sources.r1.fileHeader = true 
a1.sources.r1.maxBatchCount = 1000 
#Channel 
a1.channels.c1.type = memory 
a1.channels.c1.capacity = 10000 
a1.channels.c1.transactionCapacity = 100 
#Sink 
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink 
a1.sinks.k1.kafka.topic = ${logGroupId}_${logStreamId} 
a1.sinks.k1.kafka.bootstrap.servers = ${ip}:${port} 
a1.sinks.k1.kafka.producer.acks = 0 
a1.sinks.k1.kafka.producer.security.protocol = SASL_PLAINTEXT 
a1.sinks.k1.kafka.producer.sasl.mechanism = PLAIN 
a1.sinks.k1.kafka.producer.compression.type = gzip 
a1.sinks.k1.kafka.producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="${projectId}" password="${accessKey}#${accessSecret}"; 
#Bind 
a1.sources.r1.channels = c1 
a1.sinks.k1.channel = c1

SDK 调用示例

1. Java SDK调用示例。

maven依赖（示例kafka协议版本为2.7.1）：

<dependencies> 
<dependency> 
<groupId>org.apache.kafka</groupId> 
<artifactId>kafka-clients</artifactId> 
<version>2.7.1</version> 
</dependency> 
</dependencies>

代码示例：

package org.example; 
import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.Producer; 
import org.apache.kafka.clients.producer.ProducerRecord; 
import java.util.Properties; 
public class ProducerDemo { 
public static void main(String[] args) { 
Properties props = new Properties(); 
// 配置地址
props.put("bootstrap.servers", "${ip}:${port}"); 
// 配置消息确认机制
props.put("acks", "0"); 
// 配置认证方式
props.put("security.protocol", "SASL_PLAINTEXT"); 
props.put("sasl.mechanism", "PLAIN"); 
// 用户名 projectId 密码 accessKey#accessSecret 
props.put("sasl.jaas.config", 
"org.apache.kafka.common.security.plain.PlainLoginModule required username='${projectId}' password='${accessKey}#${accessSecret}';"); 
// 配置压缩方式
props.put("compression.type", "${compress_type}"); 
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
// 1.创建一个生产者对象
Producer<String, String> producer = new KafkaProducer<>(props); 
// 2.调用send方法
for (int i = 0; i < 1; i++) { 
ProducerRecord record = new ProducerRecord<>("${logGroupId}_${logStreamId}", "${message}"); 
// 配置recordHeader 
// record.headers().add(new RecordHeader("LTS_LOG_TYPE","FORMAT".getBytes())); 
producer.send(record); 
} 
// 3.关闭生产者
producer.close(); 
} 
}

2. Python SDK调用示例。

from kafka import KafkaProducer 
producer = KafkaProducer( 
# 配置地址
bootstrap_servers="${ip}:${port}", 
# 配置消息确认机制
acks="0", 
# 配置压缩方式
compression_type ="${compression_type}" 
# 配置认证方式
sasl_mechanism="PLAIN", 
security_protocol="SASL_PLAINTEXT", 
# 用户名 projectId 密码 accessKey#accessSecret 
sasl_plain_username="${projectId}", 
sasl_plain_password="${accessKey}#${accessSecret}" 
) 
print('start producer') 
for i in range(0, 3): 
data = bytes("${message}", encoding="utf-8") 
future = producer.send("${logGroupId}_{logStreamId}", data) 
result = future.get(timeout=10) 
print(result) 
print('end producer')

报错说明

当参数错误或不匹配时，会有相应的报错提示。

报错说明