kubelet 事件机制

对于 kubernetes 来说，apiserver 是整个集群的交互中心，客户端主要和它打交道，kubelet 是各个节点上的 worker，负责执行具体的任务。kubelet 需要把关键步骤中的执行事件发送到 apiserver，这样客户端就能通过查询知道整个流程发生了哪些事情，不需要登录到 kubelet 所在的节点查看日志的内容或者容器的运行状态。

kubernetes 是以 pod 为核心概念的，不管是 deployment、statefulSet、replicaSet，最终都会创建出来 pod。因此事件机制也是围绕 pod 进行的，在 pod 生命周期的关键步骤都会产生事件消息。比如 Controller Manager 会记录节点注册和销毁的事件、Deployment 扩容和升级的事件；kubelet 会记录镜像回收事件、volume 无法挂载事件等；Scheduler 会记录调度事件等。

kl.recorder.Eventf(kl.nodeRef, eventType, event, "Node %s status is now: %s", kl.nodeName, event)

kubelet 代码中有很多类似这样的事件。

事件的定义

recorder 的定义在 client-go/tools/record/event.go

EventRecorder interface {

        Event(object runtime.Object, eventtype, reason, message string)

        Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{})

        PastEventf(object runtime.Object, timestamp metav1.Time, eventtype, reason, messageFmt string, args ...interface{})

}

这里的三个方法都是记录事件用的，Eventf 就是封装了类似 Printf 的信息打印机制，内部也会调用 Event，而 PastEventf 允许用户传进来自定义的时间戳，因此可以设置事件产生的时间。

在 cmd/kubelet/app/server.go RunKubelet 方法中

eventBroadcaster 是个事件广播器，StartLogging 和 StartRecordingToSink 创建了两个不同的事件处理函数，分别把事件记录到日志和发送给 apiserver。而 NewRecorder 新建了一个 Recoder 对象，通过它的 Event、Eventf 和 PastEventf 方法，用户可以往里面发送事件，eventBroadcaster 会把接收到的事件发送个多个处理函数，比如这里提到的写日志和发送到 apiserver。

先看一下 EventBroadcaster ，定义在 client-go/tools/record/event.go

EventBroadcaster interface {

        StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface

        StartRecordingToSink(sink EventSink) watch.Interface

        StartLogging(logf func(format string, args ...interface{})) watch.Interface

        NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder

}

EventBroadcaster 是个接口类型，NewRecorder 新建一个 EventRecoder 对象，它就像一个事件记录仪，用户可以通过它记录事件，它在内部会把事件发送给 EventBroadcaster。

此外，EventBroadcaster 定义了三个 Start 开头的方法，它们用来添加事件处理 handler 。其中核心方法是 StartEventWatcher，它会在后台启动一个 goroutine，不断从 EventBroadcaster 提供的管道中接收事件，然后调用 eventHandler 处理函数对事件进行处理。StartRecordingToSink 和 StartLogging 是对 StartEventWatcher 的封装，分别实现了不同的处理函数（发送给 apiserver 和写日志）。

watcher := eventBroadcaster.Watch() 意思是注册了一个 watcher 到 broadcaster 里面。

EventBroadcaster 通过 EventRecorder 提供接口供用户写事件，内部把接收到的事件发送给处理函数。处理函数是可以扩展的，用户可以通过 StartEventWatcher 来编写自己的事件处理逻辑，kubelet 默认会使用 StartRecordingToSink和 StartLogging，也就是说任何一个事件会同时发送给 apiserver，并打印到日志中。

发送事件的过程

事件是通过 EventRecorder 对象发送出来的，具体实现在 client-go/tools/record/event.go：

func (eventBroadcaster *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder {

        return &recorderImpl{scheme, source, eventBroadcaster.Broadcaster, clock.RealClock{}}

}

recorderImpl struct {

        scheme *runtime.Scheme

        source v1.EventSource

        *watch.Broadcaster

        clock clock.Clock

}

func (recorder *recorderImpl) Event(object runtime.Object, eventtype, reason, message string) {

        recorder.generateEvent(object, metav1.Now(), eventtype, reason, message)

}

func (recorder *recorderImpl) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {

        recorder.Event(object, eventtype, reason, fmt.Sprintf(messageFmt, args...))

}

func (recorder *recorderImpl) PastEventf(object runtime.Object, timestamp metav1.Time, eventtype, reason, messageFmt string, args ...interface{}) {

        recorder.generateEvent(object, timestamp, eventtype, reason, fmt.Sprintf(messageFmt, args...))

}

recorderImpl 是具体的实现，eventBroadcaster.NewRecorder 会创建一个指定 EventSource 的 EventRecorder，EventSource 指明了哪个节点的哪个组件。

recorder 对外暴露了三个方法：Event、Eventf 和 PastEventf，它们的内部最终都是调用 generateEvent 方法：

generateEvent 就是根据传入的参数，生成一个 api.Event 对象，并发送出去。它各个参数的意思是：

object：哪个组件/对象发出的事件，比如 kubelet 产生的事件会使用 node 对象

timestamp：事件产生的时间

eventtype：事件类型，目前有两种：Normal 和 Warning，分别代表正常的事件和可能有问题的事件，定义在 pkg/api/types.go 文件中，未来可能有其他类型的事件扩展

reason：事件产生的原因，可以在 pkg/kubelet/events/event.go 看到 kubelet 定义的所有事件类型

message：事件的具体内容，用户可以理解的语句

makeEvent 就是根据参数构建 api.Event 对象，自动填充时间戳和 namespace：

注意 Event 事件的名字的构成，它有两部分：事件关联对象的名字和当前的时间，中间用点隔开。

Event struct {

        metav1.TypeMeta `json:",inline"`

        metav1.ObjectMeta `json:"metadata" protobuf:"bytes,1,opt,name=metadata"`

        InvolvedObject ObjectReference `json:"involvedObject" protobuf:"bytes,2,opt,name=involvedObject"`

        Reason string `json:"reason,omitempty" protobuf:"bytes,3,opt,name=reason"`

        Message string `json:"message,omitempty" protobuf:"bytes,4,opt,name=message"`

        Source EventSource `json:"source,omitempty" protobuf:"bytes,5,opt,name=source"`

        FirstTimestamp metav1.Time `json:"firstTimestamp,omitempty" protobuf:"bytes,6,opt,name=firstTimestamp"`

        LastTimestamp metav1.Time `json:"lastTimestamp,omitempty" protobuf:"bytes,7,opt,name=lastTimestamp"`

        Count int32 `json:"count,omitempty" protobuf:"varint,8,opt,name=count"`

        Type string `json:"type,omitempty" protobuf:"bytes,9,opt,name=type"`

        EventTime metav1.MicroTime `json:"eventTime,omitempty" protobuf:"bytes,10,opt,name=eventTime"`

        Series *EventSeries `json:"series,omitempty" protobuf:"bytes,11,opt,name=series"`

        Action string `json:"action,omitempty" protobuf:"bytes,12,opt,name=action"`

        Related *ObjectReference `json:"related,omitempty" protobuf:"bytes,13,opt,name=related"`

        ReportingController string `json:"reportingComponent" protobuf:"bytes,14,opt,name=reportingComponent"`

        ReportingInstance string `json:"reportingInstance" protobuf:"bytes,15,opt,name=reportingInstance"`

}

除了所有的 kubernetes 资源都有的 unversioned.TypeMeta（资源的类型和版本，对应了 yaml 文件的 Kind 和 apiVersion 字段）和 ObjectMera 字段（资源的元数据，比如 name、nemspace、labels、uuid、创建时间等）之外，还有和事件本身息息相关的字段，比如事件消息、来源、类型，以及数量（kubernetes 会把多个相同的事件汇聚到一起）和第一个事件的发生的时间等。

中间有个 InvolvedObject 字段，它其实指向了和事件关联的对象，如果是启动容器的事件，这个对象就是 Pod。

事件的发送是通过 recorder.Action() 实现的。代码在apimachinery/pkg/watch/mux.go

仅仅把对象封装并放入 channel 里面。Broadcaster 是 Recoder 内部的对象，调用 NewRecoder 的时候 EventBroadcaster 传给它的。

EventBroadcaster 也在 pkg/event/record/event.go 文件中

它的核心组件是 watch.Broadcaster，Broadcaster 就是广播的意思，主要功能就是把发给它的消息，广播给所有的监听者（watcher）。

简单来说，watch.Broadcaster 是一个分发器，内部保存了一个消息队列，可以通过 Watch创建监听它内部的 watcher。当有消息发送到队列中，watch.Broadcaster 后台运行的 goroutine 会接收消息并通过 distribute() 发送给所有的 watcher。而每个 watcher 都有一个接收消息的 channel，用户可以通过它的 ResultChan() 获取这个 channel 从中读取数据进行处理。

StartRecordingToSink 是一个事件处理函数

StartLogging 也是一个事件处理函数

其实这两个函数都只是对 StartEventWatcher 函数的封装

它启动一个 goroutine，不断从 watcher.ResultChan() 中读取消息，然后调用 eventHandler(event) 对事件进行处理。

事件的处理过程

recordToSink 负责把事件发送到 apiserver，这里的 sink 其实就是和 apiserver 交互的 restclient， event 是要发送的事件，eventCorrelator 在发送事件之前先对事件进行预处理。

eventCorrelator.EventCorrelate 会对事件做预处理，主要是聚合相同的事件（避免产生的事件过多，增加 etcd 和 apiserver 的压力，也会导致查看 pod 事件很不清晰）

result, err := eventCorrelator.EventCorrelate(event)

recordEvent 负责最终把事件发送到 apiserver，它会重试很多次（默认是 12 次），并且每次重试都有一定时间间隔（默认是 10 秒钟）

recordEvent(sink, result.Event, result.Patch, result.Event.Count > 1, eventCorrelator)

它根据 eventCorrelator 的结果来决定是新建一个事件还是更新已经存在的事件，并根据请求的结果决定是否需要重试（返回值为 false 说明需要重试，返回值为 true 表明已经操作成功或者忽略请求错误）。sink.Create 和 sink.Patch 是自动生成的 apiserver 的 client，对应的代码在： pkg/client/clientset_generated/internalclientset/typed/core/internalversion/event_expansion.go 。

EventCorrelator 内部有三个对象：filterFunc、aggregator 和 logger，它们分别对事件进行过滤、把相似的事件汇聚在一起、把相同的事件记录到一起。使用 NewEventCorrelator 初始化的时候内部会自动创建各个对象的默认值，EventCorrelate 会以此调用三个对象的方法，并返回最终的结果。现在它们的逻辑是这样的：

filterFunc：目前不做过滤，也就是说所有的事件都要经过后续处理，后面可能会做扩展

aggregator：如果在最近 10 分钟出现过 10 个相似的事件（除了 message 和时间戳之外其他关键字段都相同的事件），aggregator 会把它们的 message 设置为 events with common reason combined，这样它们就完全一样了

logger：这个变量的名字有点奇怪，其实它会把相同的事件（除了时间戳之外其他字段都相同）变成同一个事件，通过增加事件的 Count 字段来记录该事件发生了多少次。经过 aggregator 的事件会在这里变成同一个事件

aggregator 和 logger 都会在内部维护一个缓存（默认长度是 4096），事件的相似性和相同性比较是和缓存中的事件进行的，也就是说它并在乎 kubelet 启动之前的事件，而且如果事件超过 4096 的长度，最近没有被访问的事件也会被从缓存中移除。这也是这个文件中带有 cache 的原因。

总结

1. kubelet 通过 recorder 对象提供的 Event、Eventf 和 PastEventf 方法产生特性的事件

2. recorder 根据传递过来的参数新建一个 Event 对象，并把它发送给 EventBroadcaster 的管道

3. EventBroadcaster 后台运行的 goroutine 从管道中读取事件消息，把它广播给之前注册的 handler 进行处理

4. kubelet 有两个 handler，它们分别把事件记录到日志和发送给 apiserver。记录到日志很简单，直接打印就行

5. 发送给 apiserver 的 handler 叫做 EventSink，它在发送事件给 apiserver 之前会先做预处理

6. 预处理操作是 EventCorrelator 完成的，它会对事件做过滤、汇聚和去重操作，返回处理后的事件（可能是原来的事件，也可能是新创建的事件）

7. 最后通过 restclient （eventClient）调用对应的方法，给 apiserver 发送请求，这个过程如果出错会进行重试

8. apiserver 接收到事件的请求把数据更新到 etcd

事件一般用于调试，用户可以通过 kubectl 命令获取整个集群或者某个 pod 的事件信息。kubectl get events 可以看到所有的事件，kubectl describe pod PODNAME 能看到关于某个 pod 的事件。对于前者很好理解，kubectl 会直接访问 apiserver 的 event 资源，而对于后者 kubectl 还根据 pod 的名字进行搜索，匹配 InvolvedObject 名称和 pod 名称匹配的事件。