Kafka 业务日志采集最佳实践

字数 13815阅读 381评论 0赞 0

内容摘要：

文章介绍了Kafka作为消息中间件在业务日志采集中的应用，结合观测云的采集器Datakit实现快速集成。包括准备Kafka环境、安装DataKit、开启Kafka采集器等步骤，最后通过Pipeline对业务日志进行字段提取。

当前摘要为AI生成

## 简介

Apache Kafka 是一个分布式流处理平台，主要用于构建实时数据流管道和应用程序。在收集业务日志的场景中，Kafka 可以作为一个消息中间件，用于接收、存储和转发大量的日志数据。将 Kafka 与其他系统（如 Elasticsearch、Flume、Spark Streaming 等）集成，以提供更丰富的日志处理和分析功能。本文提到的是和观测云集成，即通过观测云的采集器 Datakit 采集 Kafka 中的业务日志，下面通过一些例子了解下观测云的快速集成效果。

## 实践环境

### 前置条件

* [注册观测云](https://auth.guance.com/register)

### 软件和中间件

* Kafka3.2.0
* Datakit采集器
* JDK 8

### 硬件

* 云服务器 CentOS7.9 64位 4vCPU，8GB 内存，100GB 云盘一台。

## 接入方案

### 准备 Kafka 环境

#### 安装 Kafka

下载 3.2.0 版本，解压即可使用。

wget [https://archive.apache.org/dist/kafka/3.2.0/kafka_2.13-3.2.0.tgz](https://archive.apache.org/dist/kafka/3.2.0/kafka_2.13-3.2.0.tgz)

_注：目前 Datakit 支持的 Kafka 版本有[version:0.8.2 ~ 3.2.0]_

#### 启动 Zookeeper 服务

$ bin/zookeeper-server-start.sh config/zookeeper.properties
    
#### 启动 KafkaServer

$ bin/kafka-server-start.sh config/server.properties
    
#### 创建 Topic

创建名为 testlog 的 Topic 。

$ bin/kafka-topics.sh --create --topic testlog --bootstrap-server localhost:9092
    
#### 启动 Producer

$ bin/kafka-console-producer.sh --topic testlog --bootstrap-server localhost:9092
    
### 安装 DataKit

参考官网文档[安装 DataKit 采集器](https://docs.guance.com/datakit/datakit-install/)

TOKEN 依据你的观测云工作空间来填写
    DK_DATAWAY=https://openway.guance.com?token= bash -c "$(curl -L https://static.guance.com/datakit/install.sh)"
    
### 开启 Kafka 采集器

进入 DataKit 安装目录下 (默认是 /usr/local/datakit/conf.d/ ) 的 conf.d/kafkamq 目录，复制 kafkamq.conf.sample 并命名为 kafkamq.conf 。

类似如下：

-rwxr-xr-x 1 root root 2574 Apr 30 23:52 kafkamq.conf
    -rwxr-xr-x 1 root root 2579 May  1 00:40 kafkamq.conf.sample
    
调制 kafka 采集器配置如下：
* addrs = ["localhost:9092"]，该文采集器 DataKit 和 Kafka 安装到同一台操作系统中，localhost 即可。
* kafka_version = "3.2.0"，该文使用 Kafka 的版本。
* [inputs.kafkamq.custom]，删除注释符号“#”。
* [inputs.kafkamq.custom.log_topic_map]，删除注释符号“#”。
* "testlog"="log.p"，testlog 为 Topic 的名字，log.p 为观测云 [Pipeline 可编程数据处理器](https://docs.guance.com/developers/pipeline/)的日志字段提取规则配置。涉及的业务日志和 log.p 的内容详细见下面的《使用 Pipeline》。

# {"version": "1.28.1", "desc": "do NOT edit this line"}
    
    [[inputs.kafkamq]]
      addrs = ["localhost:9092"]
      # your kafka version:0.8.2 ~ 3.2.0
      kafka_version = "3.2.0"
      group_id = "datakit-group"
      # consumer group partition assignment strategy (range, roundrobin, sticky)
      assignor = "roundrobin"
    
      ## rate limit.
      #limit_sec = 100
      ## sample
      # sampling_rate = 1.0
    
      ## kafka tls config
      # tls_enable = true
      # tls_security_protocol = "SASL_PLAINTEXT"
      # tls_sasl_mechanism = "PLAIN"
      # tls_sasl_plain_username = "user"
      # tls_sasl_plain_password = "pw"
    
      ## -1:Offset Newest, -2:Offset Oldest
      offsets=-1
    
      ## skywalking custom
      #[inputs.kafkamq.skywalking]
        ## Required！send to datakit skywalking input.
        #dk_endpoint="http://localhost:9529"
        #thread = 8 
        #topics = [
        #  "skywalking-metrics",
        #  "skywalking-profilings",
        #  "skywalking-segments",
        #  "skywalking-managements",
        #  "skywalking-meters",
        #  "skywalking-logging",
        #]
        #namespace = ""
    
      ## Jaeger from kafka. Please make sure your Datakit Jaeger collector is open ！！！
      #[inputs.kafkamq.jaeger]
        ## Required！ ipv6 is "[::1]:9529"
        #dk_endpoint="http://localhost:9529"
        #thread = 8 
        #source: agent,otel,others...
        #source = "agent"
        ## Required！ topics
        #topics=["jaeger-spans","jaeger-my-spans"]
    
      ## user custom message with PL script.
      [inputs.kafkamq.custom]
        #spilt_json_body = true
        #thread = 8 
        ## spilt_topic_map determines whether to enable log splitting for specific topic based on the values in the spilt_topic_map[topic].
        #[inputs.kafkamq.custom.spilt_topic_map]
        #  "log_topic"=true
        #  "log01"=false
        [inputs.kafkamq.custom.log_topic_map]
          "testlog"="log.p"
        #  "log01"="log_01.p"
        #[inputs.kafkamq.custom.metric_topic_map]
        #  "metric_topic"="metric.p"
        #  "metric01"="rum_apm.p"
        #[inputs.kafkamq.custom.rum_topic_map]
        #  "rum_topic"="rum_01.p"
        #  "rum_02"="rum_02.p"
    
      #[inputs.kafkamq.remote_handle]
        ## Required！
        #endpoint="http://localhost:8080"
        ## Required！ topics
        #topics=["spans","my-spans"]
        # send_message_count = 100
        # debug = false
        # is_response_point = true
        # header_check = false
      
      ## Receive and consume OTEL data from kafka.
      #[inputs.kafkamq.otel]
        #dk_endpoint="http://localhost:9529"
        #trace_api="/otel/v1/trace"
        #metric_api="/otel/v1/metric"
        #trace_topics=["trace1","trace2"]
        #metric_topics=["otel-metric","otel-metric1"]
        #thread = 8 
    
      ## todo: add other input-mq
    
注意：开启或调整 DataKit 的配置，需重启采集器（shell 下执行 datakit service -R）。

### 使用 Pipeline

log.p 规则内容

data = load_json(message)
    protocol = data["protocol"]
    response_code = data["response_code"]
    set_tag(protocol,protocol)
    set_tag(response_code,response_code)
    group_between(response_code,[200,300],"info","status")
    group_between(response_code,[400,499],"warning","status")
    group_between(response_code,[500,599],"error","status")
    
    time = data["start_time"]
    set_tag(time,time)
    default_time(time)
    
## 效果展示

### 发送业务日志样例

业务日志样例文件如下：

#info
    {"protocol":"HTTP/1.1","upstream_local_address":"172.20.32.97:33878","response_flags":"-","istio_policy_status":null,"trace_id":"5532224c1013b9ad6da1efe88778dd64","authority":"server:1338","method":"PUT","response_code":204,"duration":83,"upstream_service_time":"83","user_agent":"Jakarta Commons-HttpClient/3.1","bytes_received":103,"downstream_local_address":"172.21.2.130:1338","start_time":"2024-05-01T00:37:11.230Z","upstream_transport_failure_reason":null,"requested_server_name":null,"bytes_sent":0,"route_name":"routes","x_forwarded_for":"10.0.0.69,10.23.0.31","upstream_cluster":"outbound|1338|svc.cluster.local","request_id":"80ac7d31-a598-4dc8-bb74-1850593f61e4","downstream_remote_address":"10.23.0.31:0","path":"/api/dimensions/items","upstream_host":"172.20.9.101:1338"}
    #error
    {"protocol":"HTTP/1.1","upstream_local_address":"172.20.32.97:33878","response_flags":"-","istio_policy_status":null,"trace_id":"5532224c1013b9ad6da1efe88778dd64","authority":"server:1338","method":"PUT","response_code":504,"duration":83,"upstream_service_time":"83","user_agent":"Jakarta Commons-HttpClient/3.1","bytes_received":103,"downstream_local_address":"172.21.2.130:1338","start_time":"2024-05-01T00:39:11.230Z","upstream_transport_failure_reason":null,"requested_server_name":null,"bytes_sent":0,"route_name":"routes","x_forwarded_for":"10.0.0.69,10.23.0.31","upstream_cluster":"outbound|1338|svc.cluster.local","request_id":"80ac7d31-a598-4dc8-bb74-1850593f61e4","downstream_remote_address":"10.23.0.31:0","path":"/api/dimensions/items","upstream_host":"172.20.9.101:1338"}
    #warn
    {"protocol":"HTTP/1.1","upstream_local_address":"172.20.32.97:33878","response_flags":"-","istio_policy_status":null,"trace_id":"5532224c1013b9ad6da1efe88778dd64","authority":"server:1338","method":"PUT","response_code":404,"duration":83,"upstream_service_time":"83","user_agent":"Jakarta Commons-HttpClient/3.1","bytes_received":103,"downstream_local_address":"172.21.2.130:1338","start_time":"2024-05-01T00:38:11.230Z","upstream_transport_failure_reason":null,"requested_server_name":null,"bytes_sent":0,"route_name":"routes","x_forwarded_for":"10.0.0.69,10.23.0.31","upstream_cluster":"outbound|1338|svc.cluster.local","request_id":"80ac7d31-a598-4dc8-bb74-1850593f61e4","downstream_remote_address":"10.23.0.31:0","path":"/api/dimensions/items","upstream_host":"172.20.9.101:1338"}
    
### 日志发送命令

在 Producer 启动后，分别发送如下三条日志内容，三条日志一条为 info 级别（"response_code":204），另一条为 error 级别（"response_code":504)，最后一条为 warn 级别日志（"response_code":404）。

>{"protocol":"HTTP/1.1","upstream_local_address":"172.20.32.97:33878","response_flags":"-","istio_policy_status":null,"trace_id":"5532224c1013b9ad6da1efe88778dd64","authority":"server:1338","method":"PUT","response_code":204,"duration":83,"upstream_service_time":"83","user_agent":"Jakarta Commons-HttpClient/3.1","bytes_received":103,"downstream_local_address":"172.21.2.130:1338","start_time":"2024-04-30T08:47:11.230Z","upstream_transport_failure_reason":null,"requested_server_name":null,"bytes_sent":0,"route_name":"routes","x_forwarded_for":"10.0.0.69,10.23.0.31","upstream_cluster":"outbound|1338|svc.cluster.local","request_id":"80ac7d31-a598-4dc8-bb74-1850593f61e4","downstream_remote_address":"10.23.0.31:0","path":"/api/dimensions/items","upstream_host":"172.20.9.101:1338"}
    >
    >
    >
    >
    >{"protocol":"HTTP/1.1","upstream_local_address":"172.20.32.97:33878","response_flags":"-","istio_policy_status":null,"trace_id":"5532224c1013b9ad6da1efe88778dd64","authority":"server:1338","method":"PUT","response_code":504,"duration":83,"upstream_service_time":"83","user_agent":"Jakarta Commons-HttpClient/3.1","bytes_received":103,"downstream_local_address":"172.21.2.130:1338","start_time":"2024-04-30T08:47:11.230Z","upstream_transport_failure_reason":null,"requested_server_name":null,"bytes_sent":0,"route_name":"routes","x_forwarded_for":"10.0.0.69,10.23.0.31","upstream_cluster":"outbound|1338|svc.cluster.local","request_id":"80ac7d31-a598-4dc8-bb74-1850593f61e4","downstream_remote_address":"10.23.0.31:0","path":"/api/dimensions/items","upstream_host":"172.20.9.101:1338"}
    >
    >
    >
    >{"protocol":"HTTP/1.1","upstream_local_address":"172.20.32.97:33878","response_flags":"-","istio_policy_status":null,"trace_id":"5532224c1013b9ad6da1efe88778dd64","authority":"server:1338","method":"PUT","response_code":404,"duration":83,"upstream_service_time":"83","user_agent":"Jakarta Commons-HttpClient/3.1","bytes_received":103,"downstream_local_address":"172.21.2.130:1338","start_time":"2024-04-30T08:47:11.230Z","upstream_transport_failure_reason":null,"requested_server_name":null,"bytes_sent":0,"route_name":"routes","x_forwarded_for":"10.0.0.69,10.23.0.31","upstream_cluster":"outbound|1338|svc.cluster.local","request_id":"80ac7d31-a598-4dc8-bb74-1850593f61e4","downstream_remote_address":"10.23.0.31:0","path":"/api/dimensions/items","upstream_host":"172.20.9.101:1338"}
    
### 通过 DataKit 采集到 Kafka 的三条业务日志

![](https://learning.guance.com/uploads/1_5030c08dc9.png)

### 使用 Pipeline 对业务日志进行字段提取

下图 protocol、response_code 以及 time 都是使用 Pipeline 提取后的效果。

![](https://learning.guance.com/uploads/2_d9c31218aa.png)

kafka

著作权归作者所有

如果觉得我的文章对您有用，请点赞。您的支持将鼓励我继续创作！

添加新评论0 条评论

Ctrl+Enter 发表

匿名评论

Kafka 业务日志采集最佳实践

添加新评论0 条评论

文章目录展开目录

作者其他文章

相关文章

相关问题

相关资料