4. 日志采集与结构化

Confluent 平台的可观测性并不只依赖 JMX 指标。Broker 的副本状态迁移、Connect 的 connector 重试、ksqlDB 的查询失败，往往只在日志里留下完整因果链。本节的目标是把三类组件的文本日志，变成带 Kafka 元数据字段的结构化 JSON 事件，并送入统一的可观测性后端。

一、先明确三类日志的差异

Broker 日志由 Log4j 输出，典型行包含时间、级别、线程、logger 和消息，例如 INFO [ReplicaFetcherThread-0-2] [Partition topic=orders-3] Incremental fetch ...。Connect 日志同样走 Log4j，但消息里常带 connector、task 上下文。ksqlDB 日志则包含 query ID、statement 类型和状态迁移。三者格式不同，但都共享一个关键点：日志行本身携带了可提取的 Kafka 实体标识，这正是关联元数据的入口。

采集前先确认日志落盘位置。以 Confluent Platform 默认布局为例：Broker 在 /var/log/kafka/server.log，Connect 在 /var/log/kafka-connect/connect.log，ksqlDB 在 /var/log/ksqldb/ksqldb.log。若使用容器部署，则对应到挂载卷路径。采集器必须对这些路径有读权限，且要处理日志轮转（server.log.1、.gz）。

二、采集器选型：Filebeat、Fluentd、Vector 的定位

三者都能完成采集与解析，但侧重点不同。Filebeat 轻量、与 Elastic 生态集成好，适合已有 ELK 的团队；Fluentd 插件生态成熟，适合多源聚合与复杂路由；Vector 用 Rust 编写，单进程即可完成采集、转换、聚合、路由，配置以 TOML 表达，资源占用低，适合在 Broker 节点上与 Kafka 共部署。本节以 Vector 为主示例，因为它对“解析为 JSON + 关联元数据”的表达最直接，同时给出 Filebeat 与 Fluentd 的等价要点。

日志采集与结构化数据流 Broker server.log · Log4j Connect connect.log · connector/task ksqlDB ksqldb.log · query/statement Vector Agent file source grok / regex parser remap (VRL) enrichment json encoder → sink: kafka / loki Kafka 元数据关联 cluster_id topic / partition connector / task 结构化 JSON 事件 @timestamp level / logger message / kafka.* 可观测性后端 Loki / ES Kafka topic Grafana 查询 关键字段映射（示例） Broker：logger → kafka.logger，thread → kafka.thread，topic/partition 从消息体提取 Connect：connector → kafka.connect.connector，task → kafka.connect.task，state → kafka.connect.state ksqlDB：queryId → ksql.query.id，statement → ksql.statement.type，status → ksql.query.status 公共字段：cluster_id、node_id、host.name、service.name、@timestamp、log.level 说明：以上字段名为示例约定，实际以团队日志规范为准，需保证查询侧字段一致。

三、用 Vector 完成解析与关联

Vector 的配置分为 source、transform、sink 三段。下面是一份可直接参考的示例，采集三类日志并输出 JSON 到 Kafka topic platform.logs。

[sources.broker_log]
type = "file"
include = ["/var/log/kafka/server.log*"]
read_from = "beginning"

[sources.connect_log]
type = "file"
include = ["/var/log/kafka-connect/connect.log*"]

[sources.ksqldb_log]
type = "file"
include = ["/var/log/ksqldb/ksqldb.log*"]

[transforms.parse_broker]
type = "remap"
inputs = ["broker_log"]
source = '''
  . = parse_regex!(.message, r'^(?P<ts>\S+ \S+) (?P<level>\w+) \[(?P<thread>[^\]]+)\] (?P<logger>\S+) (?P<msg>.*)$')
  .kafka.logger = .logger
  .kafka.thread = .thread
  .log.level = .level
  .service.name = "kafka-broker"
  .cluster_id = "prod-c1"
  .node_id = 1
  if match(.msg, r'topic=(?P<t>[A-Za-z0-9._-]+)-(?P<p>\d+)') {
    .kafka.topic = t
    .kafka.partition = to_int(p)
  }
'''

[transforms.parse_connect]
type = "remap"
inputs = ["connect_log"]
source = '''
  . = parse_regex!(.message, r'^(?P<ts>\S+ \S+) (?P<level>\w+) (?P<msg>.*)$')
  .log.level = .level
  .service.name = "kafka-connect"
  .cluster_id = "prod-c1"
  if match(.msg, r'connector=(?P<c>[A-Za-z0-9._-]+)') { .kafka.connect.connector = c }
  if match(.msg, r'task=(?P<t>\d+)') { .kafka.connect.task = to_int(t) }
'''

[transforms.parse_ksqldb]
type = "remap"
inputs = ["ksqldb_log"]
source = '''
  . = parse_regex!(.message, r'^(?P<ts>\S+ \S+) (?P<level>\w+) (?P<msg>.*)$')
  .log.level = .level
  .service.name = "ksqldb"
  .cluster_id = "prod-c1"
  if match(.msg, r'queryId=(?P<q>[a-f0-9-]+)') { .ksql.query.id = q }
  if match(.msg, r'statement=(?P<s>\w+)') { .ksql.statement.type = s }
'''

[sinks.platform_logs]
type = "kafka"
inputs = ["parse_broker", "parse_connect", "parse_ksqldb"]
bootstrap_servers = "broker1:9092,broker2:9092"
topic = "platform.logs"
encoding.codec = "json"

这段配置的关键点有三处。第一，parse_regex! 用命名捕获组把日志行拆成字段，失败会直接报错，便于早期发现格式漂移。第二，match 条件提取 topic、partition、connector、queryId，这些就是后续关联 Kafka 元数据的锚点。第三，cluster_id、node_id、service.name 是静态注入的公共字段，保证多节点日志可区分。

四、Filebeat 与 Fluentd 的等价做法

Filebeat 侧，用 filestream input 加 dissect 或 grok processor 完成同样的拆分，再用 add_fields 注入 cluster_id 与 service.name。Fluentd 侧，用 in_tail 采集，filter_record_transformer 或 filter_grok 解析，record_transformer 注入公共字段，最后 out_kafka2 输出 JSON。三者的差异主要在配置语法与资源占用，字段命名应保持一致，否则查询侧要写多套规则。

建议在采集器侧统一字段命名规范：Kafka 相关字段统一放在 kafka.* 命名空间下，ksqlDB 放在 ksql.* 下，公共字段放在顶层。这样 Grafana 或 Kibana 的查询模板可以跨组件复用。

五、验证与排障

配置完成后，先做最小验证：向 Broker 日志写入一行测试日志，观察 Kafka topic 中是否出现对应 JSON。可用 kafka-console-consumer --topic platform.logs --from-beginning 查看。重点检查三件事：@timestamp 是否被正确解析为时间类型而非字符串；kafka.topic 与 kafka.partition 是否被提取；cluster_id 是否与目标集群一致。

常见问题包括：日志轮转后新文件未被采集（检查 include 通配符与 read_from）；正则不匹配导致 parse_regex! 报错（先用单行日志在本地验证正则）；字段名大小写不一致导致查询为空（统一用 snake_case 或点分命名）。排障时优先看采集器自身日志，Vector 可用 vector top 观察事件吞吐与错误计数。

不要把 cluster_id 写成硬编码后忘记在多集群环境修改。建议通过环境变量注入，例如 cluster_id = "${CLUSTER_ID}"，避免日志串集群。

六、关联 Kafka 元数据的实际价值

当日志事件带上 kafka.topic、kafka.partition、kafka.connect.connector 后，可观测性后端就能把日志与指标、链路对齐。例如，Grafana 中可同时展示某 topic 的 ISR 收缩指标与该 topic 的 Broker 日志错误；Connect 的 connector 重试日志可与 connector 状态指标联动；ksqlDB 的 query 失败日志可与查询延迟指标关联。这正是“日志结构化 + 元数据关联”带来的核心收益：从孤立文本变成可聚合、可过滤、可关联的可观测性事件。

落地时建议先覆盖 Broker 与 Connect 两类日志，验证字段规范与查询模板，再扩展到 ksqlDB 与 Schema Registry。每新增一类日志，只需在采集器中增加一个 source 与 transform，公共字段与 sink 保持不变，整体维护成本可控。

← 上一章 📖 返回目录 下一章 →