当集群中的 node 或 pod 异常时，大部分用户会使用 kubectl 查看对应的 events，那么 events 是从何而来的？其实 k8s 中的各个组件会将运行时产生的各种事件汇报到 apiserver，对于 k8s 中的可描述资源，使用 kubectl describe 都可以看到其相关的 events，那 k8s 中又有哪几个组件都上报 events 呢？
只要在 k8s.io/kubernetes/cmd 目录下暴力搜索一下就能知道哪些组件会产生 events：1
$ grep -R -n -i "EventRecorder" .
可以看出，controller-manage、kube-proxy、kube-scheduler、kubelet 都使用了 EventRecorder，本文只讲述 kubelet 中对 Events 的使用。
1、Events 的定义
events 在 k8s.io/api/core/v1/types.go 中进行定义,结构体如下所示：
1 | type Event struct { |
其中 InvolvedObject 代表和事件关联的对象，source 代表事件源，使用 kubectl 看到的事件一般包含 Type、Reason、Age、From、Message 几个字段。
k8s 中 events 目前只有两种类型：”Normal” 和 “Warning”：
2、EventBroadcaster 的初始化
events 的整个生命周期都与 EventBroadcaster 有关，kubelet 中对 EventBroadcaster 的初始化在k8s.io/kubernetes/cmd/kubelet/app/server.go中：
1 | func RunKubelet(kubeServer *options.KubeletServer, kubeDeps *kubelet.Dependencies, runOnce bool) error { |
Kubelet 在启动的时候会初始化一个 EventBroadcaster，它主要是对接收到的 events 做一些后续的处理(保存、上报等），EventBroadcaster 也会被 kubelet 中的其他模块使用，以下是相关的定义，对 events 生成和处理的函数都定义在 k8s.io/client-go/tools/record/event.go 中：
1 | type eventBroadcasterImpl struct { |
EventBroadcaster 是个接口类型，该接口有以下四个方法：
- StartEventWatcher() ： EventBroadcaster 中的核心方法，接收各模块产生的 events，参数为一个处理 events 的函数，用户可以使用 StartEventWatcher() 接收 events 然后使用自定义的 handle 进行处理
- StartRecordingToSink() ： 调用 StartEventWatcher() 接收 events，并将收到的 events 发送到 apiserver
- StartLogging() ：也是调用 StartEventWatcher() 接收 events，然后保存 events 到日志
- NewRecorder() ：会创建一个指定 EventSource 的 EventRecorder，EventSource 指明了哪个节点的哪个组件
eventBroadcasterImpl 是 eventBroadcaster 实际的对象，初始化 EventBroadcaster 对象的时候会初始化一个 Broadcaster，Broadcaster 会启动一个 goroutine 接收各组件产生的 events 并广播到每一个 watcher。
1 | func NewBroadcaster() EventBroadcaster { |
可以看到，kubelet 在初始化完 EventBroadcaster 后会调用 StartRecordingToSink() 和 StartLogging() 两个方法，StartRecordingToSink() 处理函数会将收到的 events 进行缓存、过滤、聚合而后发送到 apiserver，StartLogging() 仅将 events 保存到 kubelet 的日志中。
3、Events 的生成
从初始化 EventBroadcaster 的代码中可以看到 kubelet 在初始化完 EventBroadcaster 后紧接着初始化了 EventRecorder，并将已经初始化的 Broadcaster 对象作为参数传给了 EventRecorder，至此，EventBroadcaster、EventRecorder、Broadcaster 三个对象产生了关联。EventRecorder 的主要功能是生成指定格式的 events，以下是相关的定义：
1 | type recorderImpl struct { |
EventRecorder 中包含的几个方法都是产生指定格式的 events，Event() 和 Eventf() 的功能类似 fmt.Println() 和 fmt.Printf()，kubelet 中的各个模块会调用 EventRecorder 生成 events。recorderImpl 是 EventRecorder 实际的对象。EventRecorder 的每个方法会调用 generateEvent，在 generateEvent 中初始化 events 。
以下是生成 events 的函数：
1 | func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations map[string]string, timestamp metav1.Time, eventtype, reason, message string) { |
初始化完 events 后会调用 recorder.Action() 将 events 发送到 Broadcaster 的事件接收队列中, Action() 是 Broadcaster 中的方法。
以下是 Action() 方法的实现：
1 | func (m *Broadcaster) Action(action EventType, obj runtime.Object) { |
4、Events 的广播
上面已经说了，EventBroadcaster 初始化时会初始化一个 Broadcaster，Broadcaster 的作用就是接收所有的 events 并进行广播，Broadcaster 的实现在 k8s.io/apimachinery/pkg/watch/mux.go 中，Broadcaster 初始化完成后会在后台启动一个 goroutine，然后接收所有从 EventRecorder 发送过来的 events，Broadcaster 中有一个 map 会保存每一个注册的 watcher， 接着将 events 广播给所有的 watcher，每个 watcher 都有一个接收消息的 channel，watcher 可以通过它的 ResultChan() 方法从 channel 中读取数据进行消费。
以下是 Broadcaster 广播 events 的实现：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33func (m *Broadcaster) loop() {
for event := range m.incoming {
if event.Type == internalRunFunctionMarker {
event.Object.(functionFakeRuntimeObject)()
continue
}
m.distribute(event)
}
m.closeAll()
m.distributing.Done()
}
// distribute sends event to all watchers. Blocking.
func (m *Broadcaster) distribute(event Event) {
m.lock.Lock()
defer m.lock.Unlock()
if m.fullChannelBehavior == DropIfChannelFull {
for _, w := range m.watchers {
select {
case w.result <- event:
case <-w.stopped:
default: // Don't block if the event can't be queued.
}
}
} else {
for _, w := range m.watchers {
select {
case w.result <- event:
case <-w.stopped:
}
}
}
}
5、Events 的处理
那么 watcher 是从何而来呢？每一个要处理 events 的 client 都需要初始化一个 watcher，处理 events 的方法是在 EventBroadcaster 中定义的，以下是 EventBroadcaster 中对 events 处理的三个函数：
1 | func (eventBroadcaster *eventBroadcasterImpl) StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface { |
StartEventWatcher() 首先实例化一个 watcher，每个 watcher 都会被塞入到 Broadcaster 的 watcher 列表中，watcher 从 Broadcaster 提供的 channel 中读取 events，然后再调用 eventHandler 进行处理，StartLogging() 和 StartRecordingToSink() 都是对 StartEventWatcher() 的封装，都会传入自己的处理函数。
1 | func (eventBroadcaster *eventBroadcasterImpl) StartLogging(logf func(format string, args ...interface{})) watch.Interface { |
StartLogging() 传入的 eventHandler 仅将 events 保存到日志中。
1 | func (eventBroadcaster *eventBroadcasterImpl) StartRecordingToSink(sink EventSink) watch.Interface { |
StartRecordingToSink() 方法先根据当前时间生成一个随机数发生器 randGen，增加随机数是为了在重试时增加随机性，防止 apiserver 重启的时候所有的事件都在同一时间发送事件，接着实例化一个EventCorrelator，EventCorrelator 会对事件做一些预处理的工作，其中包括过滤、聚合、缓存等操作，具体代码不做详细分析，最后将 recordToSink() 函数作为处理函数，recordToSink() 会将处理后的 events 发送到 apiserver，这是 StartEventWatcher() 的整个工作流程。
6、Events 简单实现
了解完 events 的整个处理流程后，可以参考其实现方式写一个 demo，要实现一个完整的 events 需要包含以下几个功能：
- 1、事件的产生
- 2、事件的发送
- 3、事件广播
- 4、事件缓存
- 5、事件过滤和聚合
1 | package main |
此处仅简单实现，将 EventRecorder 处理 events 的功能直接放在了 EventBroadcaster 中实现，对 events 的处理方法仅实现了 StartLogging()，Broadcaster 中的部分功能是直接复制 k8s 中的代码，有一定的精简，其实现值得学习，此处对 EventCorrelator 并没有进行实现。
代码请参考：https://github.com/gosoon/k8s-learning-notes/tree/master/k8s-package/events
7、总结
本文讲述了 k8s 中 events 从产生到展示的一个完整过程，最后也实现了一个简单的 demo，在此将 kubelet 对 events 的整个处理过程再梳理下，其中主要有三个对象 EventBroadcaster、EventRecorder、Broadcaster：
- 1、kubelet 首先会初始化 EventBroadcaster 对象，同时会初始化一个 Broadcaster 对象。
- 2、kubelet 通过 EventBroadcaster 对象的 NewRecorder() 方法初始化 EventRecorder 对象，EventRecorder 对象提供的几个方法会生成 events 并通过 Action() 方法发送 events 到 Broadcaster 的 channel 队列中。
- 3、Broadcaster 的作用就是接收所有的 events 并进行广播，Broadcaster 初始化后会在后台启动一个 goroutine，然后接收所有从 EventRecorder 发来的 events。
- 4、EventBroadcaster 对 events 有三个处理方法：StartEventWatcher()、StartRecordingToSink()、StartLogging()，StartEventWatcher() 是其中的核心方法，会初始化一个 watcher 注册到 Broadcaster，其余两个处理函数对 StartEventWatcher() 进行了封装，并实现了自己的处理函数。
- 5、 Broadcaster 中有一个 map 会保存每一个注册的 watcher，其会将所有的 events 广播给每一个 watcher，每个 watcher 通过它的 ResultChan() 方法从 channel 接收 events。
- 6、kubelet 会使用 StartRecordingToSink() 和 StartLogging() 对 events 进行处理，StartRecordingToSink() 处理函数收到 events 后会进行缓存、过滤、聚合而后发送到 apiserver，apiserver 会将 events 保存到 etcd 中，使用 kubectl 或其他客户端可以查看。StartLogging() 仅将 events 保存到 kubelet 的日志中。