15.深入k8s：Event事件處理及其原始碼分析
阿新 • • 發佈：2020-10-12
> 轉載請宣告出處哦~，本篇文章釋出於luozhiyun的部落格：https://www.luozhiyun.com
>
> 原始碼版本是[1.19](https://github.com/kubernetes/kubernetes/tree/release-1.19)
## 概述
k8s的Event事件是一種資源物件，用於展示叢集內發生的情況，k8s系統中的各個元件會將執行時發生的各種事件上報給apiserver 。可以通過kubectl get event 或 kubectl describe pod podName 命令顯示事件，檢視k8s叢集中發生了哪些事件。
apiserver 會將Event事件存在etcd叢集中，為避免磁碟空間被填滿，故強制執行保留策略：在最後一次的事件發生後，刪除1小時之前發生的事件。
如：
```
Events:
Type Reason Age From Message
---- ------ ---- ---- -------
Normal Scheduled 19s default-scheduler Successfully assigned default/hpatest-bbb44c476-8d45v to 192.168.13.130
Normal Pulled 15s kubelet, 192.168.13.130 Container image "nginx" already present on machine
Normal Created 15s kubelet, 192.168.13.130 Created container hpatest
Normal Started 13s kubelet, 192.168.13.130 Started container hpatest
```
當叢集中的 node 或 pod 異常時，大部分使用者會使用 kubectl 檢視對應的 events，我們通過前面章節的程式碼分析可以看到這樣的程式碼：
```go
recorder.Eventf(cj, v1.EventTypeWarning, "FailedNeedsStart", "Cannot determine if job needs to be started: %v", err)
```
通過查詢也可以確認基本上與node 或 pod相關的模組都會涉及到事件，如：controller-manage、kube-proxy、kube-scheduler、kubelet 等。
Event事件管理機制主要有三部分組成：
* EventRecorder：是事件生成者，k8s元件通過呼叫它的方法來生成事件；
* EventBroadcaster：事件廣播器，負責消費EventRecorder產生的事件，然後分發給broadcasterWatcher；
* broadcasterWatcher：用於定義事件的處理方式，如上報apiserver；
整個事件管理機制的流程大致如圖：
下面我們以kubelet 中的Event事件來作為分析的例子進行講解。
## 原始碼分析
kubelet 在初始化的時候會呼叫makeEventRecorder進行Event初始化。
**makeEventRecorder**
檔案位置：cmd/kubelet/app/server.go
```go
func makeEventRecorder(kubeDeps *kubelet.Dependencies, nodeName types.NodeName) {
if kubeDeps.Recorder != nil {
return
}
// 初始化 EventBroadcaster
eventBroadcaster := record.NewBroadcaster()
// 初始化 EventRecorder
kubeDeps.Recorder = eventBroadcaster.NewRecorder(legacyscheme.Scheme, v1.EventSource{Component: componentKubelet, Host: string(nodeName)})
//記錄Event到log
eventBroadcaster.StartStructuredLogging(3)
if kubeDeps.EventClient != nil {
klog.V(4).Infof("Sending events to api server.")
//上報Event到apiserver並存儲至etcd叢集
eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeDeps.EventClient.Events("")})
} else {
klog.Warning("No api server defined - no events will be sent to API server.")
}
}
```
這個方法建立了一個EventBroadcaster，這是一個事件廣播器，會消費EventRecorder記錄的事件並通過StartStructuredLogging和StartRecordingToSink分別將event傳送給log和apiserver；EventRecorder，用作事件記錄器，k8s系統元件通過它記錄關鍵性事件；
### EventRecorder記錄事件
```go
type EventRecorder interface {
Event(object runtime.Object, eventtype, reason, message string)
Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{})
AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{})
}
```
EventRecorder介面非常的簡單，就3個方法。其中Event是可以用來記錄剛發生的事件；Eventf通過使用fmt.Sprintf格式化輸出事件的格式；AnnotatedEventf功能和Eventf一致，但是附加了註釋欄位。
我們記錄事件的時候上面也提到了，一般如下記錄：
```go
recorder.Eventf(cj, v1.EventTypeWarning, "FailedNeedsStart", "Cannot determine if job needs to be started: %v", err)
```
Eventf會呼叫到EventRecorder的實現類recorderImpl中去，最後呼叫到generateEvent方法中：
**Event**
檔案位置：client-go/tools/record/event.go
```go
func (recorder *recorderImpl) Event(object runtime.Object, eventtype, reason, message string) {
recorder.generateEvent(object, nil, 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...))
}
```
**generateEvent**
```go
func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations map[string]string, timestamp metav1.Time, eventtype, reason, message string) {
...
//例項化Event
event := recorder.makeEvent(ref, annotations, eventtype, reason, message)
event.Source = recorder.source
//非同步呼叫Action方法將事件寫入到incoming中
go func() {
// NOTE: events should be a non-blocking operation
defer utilruntime.HandleCrash()
recorder.Action(watch.Added, event)
}()
}
```
generateEvent方法會非同步的呼叫Action方法，將事件寫入到incoming中：
```go
func (m *Broadcaster) Action(action EventType, obj runtime.Object) {
m.incoming <- Event{action, obj}
}
```
呼叫步驟如下：
### EventBroadcaster事件廣播
EventBroadcaster初始化的時候會呼叫NewBroadcaster方法：
檔案位置：client-go/tools/record/event.go
```go
func NewBroadcaster() EventBroadcaster {
return &eventBroadcasterImpl{
Broadcaster: watch.NewBroadcaster(maxQueuedEvents, watch.DropIfChannelFull),
sleepDuration: defaultSleepDuration,
}
}
```
這裡會建立一個eventBroadcasterImpl例項，並設定兩個欄位Broadcaster和sleepDuration。Broadcaster是這個方法的核心，我們下面接著看：
```go
func NewBroadcaster(queueLength int, fullChannelBehavior FullChannelBehavior) *Broadcaster {
m := &Broadcaster{
watchers: map[int64]*broadcasterWatcher{},
incoming: make(chan Event, incomingQueueLength),
watchQueueLength: queueLength,
fullChannelBehavior: fullChannelBehavior,
}
m.distributing.Add(1)
//開啟事件迴圈
go m.loop()
return m
}
```
在這裡初始化Broadcaster的時候，會初始化一個broadcasterWatcher，用於定義事件處理方式，如上報apiserver等；初始化incoming，用於EventBroadcaster和EventRecorder進行事件傳輸。
**loop**
檔案位置：k8s.io/apimachinery/pkg/watch/mux.go
```go
func (m *Broadcaster) loop() {
//獲取m.incoming管道中的資料
for event := range m.incoming {
if event.Type == internalRunFunctionMarker {
event.Object.(functionFakeRuntimeObject)()
continue
}
//進行事件分發
m.distribute(event)
}
m.closeAll()
m.distributing.Done()
}
```
這個方法會一直後臺等待獲取m.incoming管道中的資料，然後呼叫distribute方法進行事件分發給broadcasterWatcher。incoming管道中的資料是EventRecorder呼叫Event方法傳入的。
**distribute**
```go
func (m *Broadcaster) distribute(event Event) {
m.lock.Lock()
defer m.lock.Unlock()
//如果是非阻塞，那麼使用DropIfChannelFull標識
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:
}
}
}
}
```
如果是非阻塞，那麼使用DropIfChannelFull標識，在w.result管道滿了之後，事件會丟失。如果沒有default關鍵字，那麼，當w.result管道滿了之後，分發過程會阻塞並等待。
這裡之所以需要丟失事件，是因為隨著k8s叢集越來越大，上報事件也隨之增多，那麼每次上報都要對etcd進行讀寫，這樣會給etcd叢集帶來壓力。但是事件丟失並不會影響叢集的正常工作，所以非阻塞分發機制下事件會丟失。
### recordToSink事件的處理
呼叫StartRecordingToSink方法會將資料上報到apiserver。
**StartRecordingToSink**
```go
func (e *eventBroadcasterImpl) StartRecordingToSink(sink EventSink) watch.Interface {
eventCorrelator := NewEventCorrelatorWithOptions(e.options)
return e.StartEventWatcher(
func(event *v1.Event) {
recordToSink(sink, event, eventCorrelator, e.sleepDuration)
})
}
func (e *eventBroadcasterImpl) StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface {
watcher := e.Watch()
go func() {
defer utilruntime.HandleCrash()
for watchEvent := range watcher.ResultChan() {
event, ok := watchEvent.Object.(*v1.Event)
if !ok {
continue
}
//回撥傳入的方法
eventHandler(event)
}
}()
return watcher
}
```
StartRecordingToSink會呼叫StartEventWatcher，StartEventWatcher方法裡面會非同步的呼叫 watcher.ResultChan()方法獲取到broadcasterWatcher的result管道，result管道里面的資料就是Broadcaster的distribute方法進行分發的。
最後會回撥到傳入的方法recordToSink中。
**recordToSink**
```go
func recordToSink(sink EventSink, event *v1.Event, eventCorrelator *EventCorrelator, sleepDuration time.Duration) {
eventCopy := *event
event = &eventCopy
//對事件做預處理，聚合相同的事件
result, err := eventCorrelator.EventCorrelate(event)
if err != nil {
utilruntime.HandleError(err)
}
if result.Skip {
return
}
tries := 0
for {
// 把事件傳送到 apiserver
if recordEvent(sink, result.Event, result.Patch, result.Event.Count >