当前位置：首页 > news >正文

如何借助Kafka持久化存储K8S事件数据？

news 2025/7/8 11:26:01

大家应该对 Kubernetes Events 并不陌生，特别是当你使用 kubectl describe 命令或 Event API 资源来了解集群中的故障时。

$ kubectl get events15m         Warning   FailedCreate                                                                                                      replicaset/ml-pipeline-visualizationserver-865c7865bc    Error creating: pods "ml-pipeline-visualizationserver-865c7865bc-" is forbidden: error looking up service account default/default-editor: serviceaccount "default-editor" not found

尽管这些信息十分有用，但它只是临时的，保留时间最长为30天。如果出于审计或是故障诊断等目的，你可能想要把这些信息保留得更久，比如保存在像 Kafka 这样更持久、高效的存储中。然后你可以借助其他工具（如 Argo Events）或自己的应用程序订阅 Kafka 主题来对某些事件做出响应。

构建K8s事件处理链路

我们将构建一整套 Kubernetes 事件处理链路，其主要构成为：

Eventrouter，开源的 Kubernetes event 处理器，它可以将所有集群事件整合汇总到某个 Kafka 主题中。
Strimzi Operator，在 Kubernetes 中轻松管理 Kafka broker。
自定义 Go 二进制文件以将事件分发到相应的 Kafka 主题中。

为什么要把事件分发到不同的主题中？比方说，在集群的每个命名空间中存在与特定客户相关的 Kubernetes 资产，那么在使用这些资产之前你当然希望将相关事件隔离开。

本示例中所有的配置、源代码和详细设置指示都已经放在以下代码仓库中：
https://github.com/esys/kube-events-kafka

创建 Kafka broker 和主题

我选择使用 Strimzi（strimzi.io/）将 Kafka 部署到 Kubernetes 中。简而言之，它是用于创建和更新 Kafka broker 和主题的。你可以在官方文档中找到如何安装该 Operator 的详细说明：
https://strimzi.io/docs/operators/latest/overview.html

首先，创建一个新的 Kafka 集群：

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:name: kube-events
spec:entityOperator:topicOperator: {}userOperator: {}kafka:config:default.replication.factor: 3log.message.format.version: "2.6"offsets.topic.replication.factor: 3transaction.state.log.min.isr: 2transaction.state.log.replication.factor: 3listeners:- name: plainport: 9092tls: falsetype: internal- name: tlsport: 9093tls: truetype: internalreplicas: 3storage:type: jbodvolumes:- deleteClaim: falseid: 0size: 10Gitype: persistent-claimversion: 2.6.0zookeeper:replicas: 3storage:deleteClaim: falsesize: 10Gitype: persistent-claim

然后创建 Kafka 主题来接收我们的事件：

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaTopic
metadata:name: cluster-events
spec:config:retention.ms: 7200000segment.bytes: 1073741824partitions: 1replicas: 1

设置 EventRouter

在本教程中使用 kubectl apply 命令即可，我们需要编辑 router 的配置，以指明我们的 Kafka 端点和要使用的主题：

apiVersion: v1
data:config.json: |-{"sink": "kafka","kafkaBrokers": "kube-events-kafka-bootstrap.kube-events.svc.cluster.local:9092","kafkaTopic": "cluster-events"}
kind: ConfigMap
metadata:name: eventrouter-cm

验证设置是否正常工作

我们的 cluster-events Kafka 的主题现在应该收到所有的事件。最简单的方法是在主题上运行一个 consumer 来检验是否如此。为了方便期间，我们使用我们的一个 Kafka broker pods，它已经有了所有必要的工具，你可以看到事件流：

kubectl -n kube-events exec kube-events-kafka-0 -- bin/kafka-console-consumer.sh \--bootstrap-server kube-events-kafka-bootstrap:9092 \--topic kube-events \--from-beginning
{"verb":"ADDED","event":{...}}
{"verb":"ADDED","event":{...}}
...

编写 Golang 消费者

现在我们想将我们的 Kubernetes 事件依据其所在的命名空间分发到多个主题中。我们将编写一个 Golang 消费者和生产者来实现这一逻辑：

消费者部分在 cluster-events 主题上监听传入的集群事件
生产者部分写入与事件的命名空间相匹配的 Kafka 主题中

如果为Kafka配置了适当的选项（默认情况），就不需要特地创建新的主题，因为 Kafka 会默认为你创建主题。这是 Kafka 客户端 API 的一个非常酷的功能。

p, err := kafka.NewProducer(cfg.Endpoint)
if err != nil {sugar.Fatal("cannot create producer")
}
defer p.Close()c, err := kafka.NewConsumer(cfg.Endpoint, cfg.Topic)
if err != nil {sugar.Fatal("cannot create consumer")
}
defer c.Close()run := true
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
go func() {sig := <-sigssugar.Infof("signal %s received, terminating", sig)run = false
}()var wg sync.WaitGroup
go func() {wg.Add(1)for run {data, err := c.Read()if err != nil {sugar.Errorf("read event error: %v", err)time.Sleep(5 * time.Second)continue}if data == nil {continue}msg, err := event.CreateDestinationMessage(data)if err != nil {sugar.Errorf("cannot create destination event: %v", err)}p.Write(msg.Topic, msg.Message)}sugar.Info("worker thread done")wg.Done()
}()wg.Wait()

完整代码在此处：
https://github.com/esys/kube-events-kafka/blob/master/events-fanout/cmd/main.go

当然还有更高性能的选择，这取决于预计的事件量和扇出（fanout）逻辑的复杂性。对于一个更强大的实现，使用 Spark Structured Streaming 的消费者将是一个很好的选择。

部署消费者

构建并将二进制文件推送到 Docker 镜像之后，我们将它封装为 Kubernetes deployment：

apiVersion: apps/v1
kind: Deployment
metadata:labels:app: events-fanoutname: events-fanout
spec:replicas: 1selector:matchLabels:app: events-fanouttemplate:metadata:labels:app: events-fanoutspec:containers:- image: emmsys/events-fanout:latestname: events-fanoutcommand: [ "./events-fanout"]args:- -logLevel=infoenv:- name: ENDPOINTvalue: kube-events-kafka-bootstrap:9092- name: TOPICvalue: cluster-events

检查目标主题是否创建

现在，新的主题已经创建完成：

kubectl -n kube-events get kafkatopics.kafka.strimzi.io -o namekafkatopic.kafka.strimzi.io/cluster-events
kafkatopic.kafka.strimzi.io/kube-system
kafkatopic.kafka.strimzi.io/default
kafkatopic.kafka.strimzi.io/kafka
kafkatopic.kafka.strimzi.io/kube-events

你会发现你的事件根据其命名空间整齐地存储在这些主题中。