原標題：K8s Events之捉妖記（下）

經過前兩回的“踏血尋妖”，一個完整的Events原形逐漸浮出水面。我們已經摸清了它的由來和身世，本回將一起探索Events的去向，這是一個終點卻也是另一個起點。

蜜汁去向

前面已經瞭解到，Event是由一個叫EventRecorder的東西幻化而生。通過研究原始碼經典發現，在Kubelet啟動的時候獲取一個EventBroadcaster的例項，以及根據KubeletConfig獲取一個EventRecorder例項。EventRecorder自不必多說。EventBroadcaster用來接收Event並且把它們轉交給EventSink、Watcher和Log。

EventBroadcaster定義了包括四個方法的一組介面，分別是：

// 將收到的Events交於相應的處理函式
 StartEventWatcher(eventHandler func(*api.Event)) watch.Interface

// 將收到的Events交於EventSink
 StartRecordingToSink(sink EventSink) watch.Interface

// 將收到的Events交於相應的Log供日誌輸出
 StartLogging(logf func(format string, args ...interface{})) watch.Interface

// 初始化一個EventRecorder，並向EventBroadcaster傳送Events
 NewRecorder(source api.EventSource) EventRecorder

EventBroadcaster由定義在kubernetes/pkg/client/record/event.go裡的NewBroadcaster()方法進行初始化，實際上靠呼叫kubernetes/pkg/watch/mux.go裡的NewBroadcaster()方法實現。在定義裡，每一個EventBroadcaster都包含一列watcher，而對於每個watcher，都監視同一個長度為1000的Events Queue，由此保證分發時佇列按Events發生的時間排序。但是同一個Events傳送至Watcher的順序得不到保證。為了防止短時間內湧入的Events導致來不及處理，每個EventBroadcaster都擁有一個長度為25的接收緩衝佇列。定義的最後指定了佇列滿時的相應操作。

當完成初始化並加入waitGroup之後，EventBroadcaster便進入無限迴圈。在這個迴圈中，Broadcaster會不停地從緩衝佇列裡取走Event。如果獲取失敗就將退出迴圈，並清空所有的watcher。如果獲取成功就將該Event分發至各個watcher。在分發的時候需要加鎖，如果佇列已滿則不會阻塞，直接跳過到下一個watcher。如果佇列未滿，則會阻塞，直到寫入後再分發下一個watcher。

在Kubelet執行過程初始化EventBroadcaster之後，如果KubeletConfig裡的EventClient不為空，即指定對應的EventSink(EventSink是一組介面，包含儲存Events的Create、Update、Patch方法，實際由對應的Client實現):

eventBroadcaster.StartRecordingToSink(&unversionedcore.EventSinkImpl{Interface: kcfg.EventClient.Events("")})

StartRecordingToSink()方法先根據當前時間生成一個隨機數發生器randGen，接著例項化一個EventCorrelator，最後將recordToSink()函式作為處理函式，實現了StartEventWatcher。StartLogging()類似地將用於輸出日誌的匿名函式作為處理函式，實現了StartEventWatcher。

總鑽風StartEventWatcher

StartEventWatcher()首先例項化watcher，每個watcher都被塞入該Broadcaster的watcher列表中，並且新例項化的watcher只能獲得後續的Events，不能獲取整個Events歷史。入佇列的時候加鎖以保證安全。接著啟動一個goroutine用來監視Broadcaster發來的Events。EventBroadcaster會在分發Event的時候將所有的Events都送入一個ResultChan。watcher不斷從ResultChan取走每個Event，如果獲取過程傳送錯誤，將Crash並記錄日誌。否則在獲得該Events後，交於對應的處理函式進行處理。

StartEventWatcher()方法使用recordToSink()函式作為處理。因為同一個Event可能被多個watcher監聽，所以在對Events進行處理前，先要拷貝一份備用。接著同樣使用EventCorrelator對Events進行整理，然後在有限的重試次數裡通過recordEvent()方法對該Event進行記錄。

recordEvent()方法試著將Event寫到對應的EventSink裡，如果寫成功或可無視的錯誤將返回true，其他錯誤返回false。如果要寫入的Event已經存在，就將它更新，否則建立一個新的Event。在這個過程中如果出錯，不管是構造新的Event失敗，還是伺服器拒絕了這個event，都屬於可無視的錯誤，將返回true。而HTTP傳輸錯誤，或其他不可預料的物件錯誤，都會返回false，並在上一層函式裡進行重試。在kubernetes/pkg/client/record/event.go裡指定了單個Event的最大重試次數為12次。另外，為了避免在master掛掉之後所有的Event同時重試導致不能同步，所以每次重試的間隔時間將隨機產生(第一次間隔由前面的隨機數發生器randGen生成)。

小鑽風EventCorrelator

EventCorrelator定義包含了三個成員，分別是過濾Events的filterFunc，進行Event聚合的aggregator以及記錄Events的logger。它負責處理收到的所有Events，並執行聚合等操作以防止大量的Events沖垮整個系統。它會過濾頻繁發生的相似Events來防止系統向用戶傳送難以區分的資訊和執行去重操作，以使相同的Events被壓縮為被多次計數單個Event。

EventCorrelator通過NewEventCorrelator()函式進行例項化:

func NewEventCorrelator(clock clock.Clock) *EventCorrelator {    
        cacheSize := maxLruCacheEntries    
        return &EventCorrelator{        
            // 預設對於所有的Events均返回false，表示不可忽略
            filterFunc: DefaultEventFilterFunc,
            aggregator: NewEventAggregator(            
                // 大小為4096
                cacheSize,            
                // 通過相同的Event域來進行分組
                EventAggregatorByReasonFunc,            
                // 生成"根據同樣的原因進行分組"訊息
                EventAggregatorByReasonMessageFunc,            
                // 每個時間間隔裡最多統計10個Events
                defaultAggregateMaxEvents,            
                // 最大時間間隔為10mins
                defaultAggregateIntervalInSeconds,
            clock),
        logger: newEventLogger(cacheSize, clock),
    }
}

Kubernetes的Events可以按照兩種方式分類：相同和相似。相同指的是兩個Events除了時間戳以外的其他資訊均相同。相似指的是兩個Events除了時間戳和訊息(message)以外的其他資訊均相同。按照這個分類方法，為了減少Event流對etcd的衝擊，將相同的Events合併計數和將相似的Events聚合，提出“最大努力”的Event壓縮演算法。最大努力指的是在最壞的情況下，N個Event仍然會產生N條Event記錄。

每個Event物件包含不只一個時間戳域：FirstTimestamp、LastTimestamp，同時還有統計在FirstTimestamp和LastTimestamp之間出現頻次的域Count。同時對於每個可以產生Events的元件，都需要維持一個生成過的Event的歷史記錄：通過Least Recently Used Cache實現。

EventCorrelator的主要方法是EventCorrelate()，每次收到一個Event首先判斷它是否可以被跳過(前面提過預設均不可忽略)。然後對該Event進行Aggregate處理。

EventCorrelator包含兩個子元件：EventAggregator和EventLogger。EventCorrelator檢查每個接收到的Event，並讓每個子元件可以訪問和修改這個Event。其中EventAggregator對每個Event進行聚合操作，它基於aggregateKey將Events進行分組，組內區分的唯一標識是localKey。預設的聚合函式將event.Message作為localKey，使用event.Source、event.InvolvedObject、event.Type和event.Reason一同構成aggregateKey。

aggregator是型別EventAggregator的一個例項，定義如下：

type EventAggregator struct {
 // 讀寫鎖
 sync.RWMutex

// 存放整合狀態的Cache
 cache *lru.Cache

// 用來對Events進行分組的函式、
 keyFunc EventAggregatorKeyFunc

// 為整合的Events生成訊息的函式
 messageFunc EventAggregatorMessageFunc

// 每個時間間隔裡可統計的最大Events數
 maxEvents int

// 相同的Events間最大時間間隔以及一個時鐘
 maxIntervalInSeconds int

clock clock.Clock}

• 通過EventAggregatroKeyFunc，EventAggregator會將10mins內出現過10次的相似Event進行整合：丟棄作為輸入的Event，並且建立一個僅有Message區別的新Event。這條Message標識這是一組相似的Events，並且會被後續的Event操作序列處理。
• EventLogger觀察相同的Event，並通過在Cache裡與它關聯的計數來統計它出現的次數。

在Cache裡的Key是Event物件除去Timestamp/Counts等剩餘部分構成的。下面的任意組合都可以唯一構造Cache裡Event唯一的Key：

event.Source.Component
event.Source.Host
event.InvolvedObject.Kind
event.InvolvedObject.Namespace
event.InvolvedObject.Name
event.InvolvedObject.UID
event.InvolvedObject.APIVersion
event.Reason
event.Message

不管對於EventAggregator或EventLogger，LRU Cache大小僅為4096。這也意味著當一個元件（比如Kubelet）執行很長時間，並且產生了大量的不重複Event，先前產生的未被檢查的Events並不會讓Cache大小繼續增長，而將最老的Event從Cache中排除。當一個Event被產生，先前產生的Event Cache會被檢查:

• 如果新產生的Event的Key跟先前產生的Event的Key相匹配（意味著前面所有的域都相匹配），那麼它被認為是重複的，並且在etcd裡已存在的這條記錄將被更新。
  使用PUT方法來更新etcd裡存放的這條記錄，僅更新它的LastTimestamp和Count域。
  同時還會更新先前生成的Event Cache裡對應記錄的Count、LastTimestamp、Name以及新的ResourceVersion。
• 如果新產生的Event的Key並不能跟先前產生的Event相匹配（意味著前面所有的域都不匹配），這個Event將被認為是新的且是唯一的記錄，並寫入etcd裡。
  使用POST方法來在etcd裡建立該記錄
  對該Event的記錄同樣被加入到先前生成的Event Cache裡

當然這樣還存在一些問題。對於每個元件來說，Event歷史都存放在記憶體裡，如果該程式重啟，那麼歷史將被清空。另外，如果產生了大量的唯一Event，舊的Event將從Cache裡去除。只有從Cache裡去除的Event才會被壓縮，同時任何一個此Event的新例項都會在etcd裡建立新記錄。

舉個例子，下面的kubectl結果表示有20條相互獨立的Event記錄(請看錶示排程錯誤的記錄:Scheduling Failure)被壓縮至5條。

FIRSTSEEN LASTSEEN COUNT NAME KIND SUBOBJECT REASON SOURCE MESSAGE
 Thu, 12 Feb 2015 01:13:02 +0000 Thu, 12 Feb 2015 01:13:02 +0000 1 kubernetes-node-4.c.saad-dev-vms.internal Node starting {kubelet kubernetes-node-4.c.saad-dev-vms.internal} Starting kubelet.
 Thu, 12 Feb 2015 01:13:09 +0000 Thu, 12 Feb 2015 01:13:09 +0000 1 kubernetes-node-1.c.saad-dev-vms.internal Node starting {kubelet kubernetes-node-1.c.saad-dev-vms.internal} Starting kubelet.
 Thu, 12 Feb 2015 01:13:09 +0000 Thu, 12 Feb 2015 01:13:09 +0000 1 kubernetes-node-3.c.saad-dev-vms.internal Node starting {kubelet kubernetes-node-3.c.saad-dev-vms.internal} Starting kubelet.
 Thu, 12 Feb 2015 01:13:09 +0000 Thu, 12 Feb 2015 01:13:09 +0000 1 kubernetes-node-2.c.saad-dev-vms.internal Node starting {kubelet kubernetes-node-2.c.saad-dev-vms.internal} Starting kubelet.
 Thu, 12 Feb 2015 01:13:05 +0000 Thu, 12 Feb 2015 01:13:12 +0000 4 monitoring-influx-grafana-controller-0133o Pod failedScheduling {scheduler } Error scheduling: no nodes available to schedule pods
 Thu, 12 Feb 2015 01:13:05 +0000 Thu, 12 Feb 2015 01:13:12 +0000 4 elasticsearch-logging-controller-fplln Pod failedScheduling {scheduler } Error scheduling: no nodes available to schedule pods
 Thu, 12 Feb 2015 01:13:05 +0000 Thu, 12 Feb 2015 01:13:12 +0000 4 kibana-logging-controller-gziey Pod failedScheduling {scheduler } Error scheduling: no nodes available to schedule pods
 Thu, 12 Feb 2015 01:13:05 +0000 Thu, 12 Feb 2015 01:13:12 +0000 4 skydns-ls6k1 Pod failedScheduling {scheduler } Error scheduling: no nodes available to schedule pods
 Thu, 12 Feb 2015 01:13:05 +0000 Thu, 12 Feb 2015 01:13:12 +0000 4 monitoring-heapster-controller-oh43e Pod failedScheduling {scheduler } Error scheduling: no nodes available to schedule pods
 Thu, 12 Feb 2015 01:13:20 +0000 Thu, 12 Feb 2015 01:13:20 +0000 1 kibana-logging-controller-gziey BoundPod implicitly required container POD pulled {kubelet kubernetes-node-4.c.saad-dev-vms.internal} Successfully pulled image "kubernetes/pause:latest"Thu, 12 Feb 2015 01:13:20 +0000 Thu, 12 Feb 2015 01:13:20 +0000 1 kibana-logging-controller-gziey Pod scheduled {scheduler } Successfully assigned kibana-logging-controller-gziey to kubernetes-node-4.c.saad-dev-vms.internal

為處理函式，實現了StartEventWatcher。

小結

到此基本上捋出了Events的來龍去脈：Event由Kubernetes的核心元件Kubelet和ControllerManager等產生，用來記錄系統一些重要的狀態變更。ControllerManager裡包含了一些小controller，比如deployment_controller，它們擁有EventBroadCaster的物件，負責將採集到的Event進行廣播。Kubelet包含一些小的manager，比如docker_manager，它們會通過EventRecorder輸出各種Event。當然，Kubelet本身也擁有EventBroadCaster物件和EventRecorder物件。

EventRecorder通過generateEvent()實際生成各種Event，並將其新增到監視佇列。我們通過kubectl get events看到的NAME並不是Events的真名，而是與該Event相關的資源的名稱，真正的Event名稱還包含了一個時間戳。Event物件通過InvolvedObject成員與發生該Event的資源建立關聯。Kubernetes的資源分為“可被描述資源”和“不可被描述資源”。當我們kubectl describe可描述資源，比如Pod時，除了獲取Pod的相應資訊，還會通過FieldSelector獲取相應的Event列表。Kubelet在初始化的時候已經指明瞭該Event的Source為Kubelet。

EventBroadcaster會將收到的Event交於各個處理函式進行處理。接收Event的緩衝佇列長為25，不停地取走Event並廣播給各個watcher。watcher由StartEventWatcher()例項產生，並被塞入EventBroadcaster的watcher列表裡，後例項化的watcher只能獲取後面的Event歷史，不能獲取全部歷史。watcher通過recordEvent()方法將Event寫入對應的EventSink裡，最大重試次數為12次，重試間隔隨機生成。

在寫入EventSink前，會對所有的Events進行聚合等操作。將Events分為相同和相似兩類，分別使用EventLogger和EventAggregator進行操作。EventLogger將相同的Event去重為1個，並通過計數表示它出現的次數。EventAggregator將對10分鐘內出現10次的Event進行分組，依據是Event的Source、InvolvedObject、Type和Reason域。這樣可以避免系統長時間執行時產生的大量Event衝擊etcd，或佔用大量記憶體。EventAggregator和EventLogger採用大小為4096的LRU Cache，存放先前已產生的不重複Events。超出Cache範圍的Events會被壓縮。

後記

這篇文章僅150行文字，但花了整整一天，結果並沒有成為“捉妖”系列的完美收官之作，系列三篇文章也仍沒有完完整整地梳理出Event的全貌。一個小小的Event研究起來卻這麼複雜，讓我想起探花兄曾經說過“我們要時刻保持敬畏之心，不管對人還是對技術”。不管是學術上，還是工程上，每種技術的實現和發展無不凝聚了很多人的智慧和汗水，Kubernetes這樣龐大的系統更是。我還只是剛“識字”的初學者，更有必要時刻保持敬畏之心。跟Event的故事仍未完結，後面的文章會繼續圍繞Event展開，敬請關注！

作者立堯微信公眾號