将Kafka集群审计日志接入GCP Logging的不同实现方式有哪些？

核心实现路径 #

• 路径1：使用Fluent Bit作为日志采集代理（轻量低耗，适配绝大多数场景，推荐优先选用）
• 路径2：通过GCP官方Cloud Logging Agent直接采集（生态原生，配置简单）
• 路径3：Kafka Connect + GCP Logging Sink Connector（适合已经在使用Kafka Connect生态的集群，无需在每个Broker节点部署采集组件）
• 路径4：自研程序直接调用Cloud Logging API写入（适合有强定制化需求的场景，灵活度最高）

各路径具体操作方法 #

路径1：Fluent Bit采集写入 #

1. 开启Kafka审计日志输出，编辑Kafka的server.properties配置文件，添加如下规则后重启Broker：

audit.log.enable=true
audit.log.files=/var/log/kafka/kafka-audit.log
audit.log.level=INFO

重启后校验/var/log/kafka/kafka-audit.log路径有正常日志生成即可。
2. 在所有Kafka Broker节点安装Fluent Bit，以Debian系系统为例：

curl https://packages.fluentbit.io/fluentbit.key | gpg --dearmor > /usr/share/keyrings/fluentbit-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/fluentbit-keyring.gpg] https://packages.fluentbit.io/debian/bullseye bullseye main" > /etc/apt/sources.list.d/fluentbit.list
apt update && apt install fluent-bit

3. 编辑Fluent Bit配置文件/etc/fluent-bit/fluent-bit.conf，添加采集和输出规则：

# 输入段：采集本地审计日志文件
[INPUT]
    name tail
    path /var/log/kafka/kafka-audit.log
    tag kafka.audit
    parser json # 若审计日志为非JSON格式，替换为对应解析器即可
# 输出段：写入GCP Cloud Logging
[OUTPUT]
    name stackdriver
    match kafka.audit
    resource generic_node
    node_id ${HOSTNAME}
    location 替换为你的GCP区域，例如us-central1
    log_name kafka-audit-log

4. 配置GCP权限：给运行Fluent Bit的服务账号绑定roles/logging.logWriter权限，确保有日志写入权限。
5. 执行systemctl restart fluent-bit重启服务生效，到GCP Cloud Logging控制台搜索日志名kafka-audit-log即可验证上报结果。

路径2：GCP Cloud Logging Agent采集 #

1. 同路径1第一步，先开启Kafka审计日志并输出到本地文件。
2. 在所有Broker节点安装Cloud Logging Agent：

curl -sSO https://dl.google.com/cloudagents/add-logging-agent-repo.sh
sudo bash add-logging-agent-repo.sh
sudo apt update && sudo apt install google-fluentd

3. 新增采集配置文件/etc/google-fluentd/config.d/kafka-audit.conf：

<source>
  @type tail
  format json
  path /var/log/kafka/kafka-audit.log
  pos_file /var/lib/google-fluentd/pos/kafka-audit.pos
  read_from_head true
  tag kafka.audit
</source>
<filter kafka.audit>
  @type record_transformer
  <record>
    cluster_name 替换为你的Kafka集群名
    broker_id ${HOSTNAME}
  </record>
</filter>
<match kafka.audit>
  @type google_cloud
  log_name kafka-audit-log
</match>

4. 执行systemctl restart google-fluentd重启Agent，到Cloud Logging控制台验证日志即可。

路径3：Kafka Connect + GCP Logging Sink Connector #

适合已经将审计日志输出到Kafka Topic的场景，无需在每个节点部署采集组件。

1. 配置Kafka审计日志输出到指定Topic，编辑server.properties添加如下规则后重启Broker：

audit.log.enable=true
audit.log.topic=_kafka-audit-logs
audit.log.producer.bootstrap.servers=替换为你的Kafka集群地址，例如broker1:9092,broker2:9092

重启后校验_kafka-audit-logs Topic有正常的审计日志写入即可。
2. 将GCP Logging Sink Connector安装包放到Kafka Connect集群的插件目录下。
3. 编写Connector配置文件kafka-audit-sink.json：

{
  "name": "kafka-audit-to-gcp-logging",
  "config": {
    "connector.class": "com.google.cloud.gcl.sink.GoogleCloudLoggingSinkConnector",
    "topics": "_kafka-audit-logs",
    "gcp.project.id": "替换为你的GCP项目ID",
    "log.name": "kafka-audit-log",
    "gcp.credentials.file.path": "替换为服务账号密钥文件路径",
    "tasks.max": 3
  }
}

4. 提交配置到Kafka Connect集群：

curl -X POST -H "Content-Type: application/json" http://替换为你的Kafka Connect地址:8083/connectors -d @kafka-audit-sink.json

验证Connector运行状态正常后，即可到Cloud Logging控制台查看上报的审计日志。

注意事项 #

• 所有方案都需要为对应服务账号配置roles/logging.logWriter的IAM权限，避免日志写入被拒绝。
• 如果Kafka审计日志为非结构化的自定义格式，需要提前配置对应解析规则，确保上报后可以正常做结构化查询。
• 高并发场景下建议配置日志批量上报和缓存策略，避免频繁调用GCP API产生额外成本，也能降低组件资源消耗。

内容的提问来源于stack exchange，提问作者Vishal Dhanani