容器运行完成时能否自动发送Kafka事件并将标准输出作为消息内容？

以下两种方案均可直接替换现有轮询ES的逻辑，实现事件驱动的下游流程触发，彻底解决性能损耗问题：

方案1：复用现有Fluentd链路改造（改造成本最低） #

直接在现有Fluentd配置中新增Kafka输出规则，不需要新增组件、不需要修改业务Pod逻辑：

• 确认Fluentd已安装fluent-plugin-kafka输出插件，官方标准镜像默认自带，缺失可直接执行gem install fluent-plugin-kafka安装
• 修改Fluentd配置，新增过滤规则仅匹配运行完成（状态为Succeeded/Failed）的Pod日志，同时保留原有ES输出链路，并行将Pod stdout发送到指定Kafka Topic，配置参考如下：

# 过滤已运行完成的Pod日志
<filter kubernetes.**>
  @type grep
  <regexp>
    key $.kubernetes.pod_status.phase
    pattern /^(Succeeded|Failed)$/
  </regexp>
</filter>

# 并行输出到ES和Kafka
<match kubernetes.**>
  @type copy
  <store>
    @type elasticsearch
    # 原有ES配置保持不变
    host elasticsearch.default.svc
    port 9200
    logstash_format true
  </store>
  <store>
    @type kafka2
    brokers kafka.default.svc:9092
    default_topic pod_completed_events
    <format>
      @type json
      # 可自定义保留字段，仅下发下游需要的内容
      include_fields $.kubernetes.pod_name, $.kubernetes.namespace_name, $.log, $.time
    </format>
  </store>
</match>

• 下游流程直接消费Kafka Topic即可获取Pod完成事件及stdout内容，端到端延迟可控制在百毫秒级。

方案2：Sidecar容器上报（可靠性最高） #

若对事件上报准确率要求极高，担心日志采集链路丢数，可以给业务Pod新增轻量上报Sidecar：

• 业务容器和Sidecar通过emptyDir共享存储卷，业务容器将stdout重定向到共享目录
• Sidecar监听业务容器进程状态，检测到业务进程退出后，读取共享目录中的stdout文件直接发往Kafka，配置参考如下：

apiVersion: v1
kind: Pod
metadata:
  name: business-job-pod
spec:
  volumes:
  - name: log-share
    emptyDir: {}
  containers:
  # 业务容器
  - name: business-container
    image: your-business-image:v1
    command: ["/bin/sh", "-c"]
    args: ["your-business-command > /var/log/business/stdout.log 2>&1"]
    volumeMounts:
    - name: log-share
      mountPath: /var/log/business
  # Kafka上报Sidecar，镜像仅10M左右
  - name: kafka-reporter
    image: edenhill/kafkacat:1.7.0
    command: ["/bin/sh", "-c"]
    args: |
      while true; do
        if ! pgrep -f your-business-command > /dev/null; then
          cat /var/log/business/stdout.log | kafkacat -b kafka.default.svc:9092 -t pod_completed_events -k $(hostname)
          exit 0
        fi
        sleep 1
      done
    volumeMounts:
    - name: log-share
      mountPath: /var/log/business
  restartPolicy: Never

该方案和日志采集链路完全解耦，不会受Fluentd采集延迟、资源不足等问题影响，事件上报准确率可达99.99%以上。

额外优化建议 #

• 可在Kafka Topic中按业务工作流ID做分区，下游消费者可直接过滤匹配对应流程的事件，不需要全量消费
• 若对消息可靠性要求高，可配置Kafka Producer的ACK=all参数，保证消息持久化后再返回
• 事件消息中可自定义注入业务唯一标识，不需要下游额外解析stdout提取信息

内容的提问来源于stack exchange，提问作者Augusto WYZ