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

$ kubectl get events

15m         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

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: kube-events
spec:
  entityOperator:
    topicOperator: {}
    userOperator: {}
  kafka:
    config:
      default.replication.factor: 3
      log.message.format.version: "2.6"
      offsets.topic.replication.factor: 3
      transaction.state.log.min.isr: 2
      transaction.state.log.replication.factor: 3
    listeners:
    - name: plain
      port: 9092
      tls: false
      type: internal
    - name: tls
      port: 9093
      tls: true
      type: internal
    replicas: 3
    storage:
      type: jbod
      volumes:
      - deleteClaim: false
        id: 0
        size: 10Gi
        type: persistent-claim
    version: 2.6.0
  zookeeper:
    replicas: 3
    storage:
      deleteClaim: false
      size: 10Gi
      type: persistent-claim

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

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

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":{...}}
...

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 := <-sigs
        sugar.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()

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: events-fanout
  name: events-fanout
spec:
  replicas: 1
  selector:
    matchLabels:
      app: events-fanout
  template:
    metadata:
      labels:
        app: events-fanout
    spec:
      containers:
        - image: emmsys/events-fanout:latest
          name: events-fanout
          command: [ "./events-fanout"]
          args:
            - -logLevel=info
          env:
            - name: ENDPOINT
              value: kube-events-kafka-bootstrap:9092
            - name: TOPIC
              value: cluster-events

kubectl -n kube-events get kafkatopics.kafka.strimzi.io -o name

kafkatopic.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