kubelet 事件機制
對于 kubernetes? 來說陕赃，apiserver 是整個集群的交互中心，客戶端主要和它打交道颁股，kubelet 是各個節(jié)點上的 worker么库，負(fù)責(zé)執(zhí)行具體的任務(wù)。kubelet 需要把關(guān)鍵步驟中的執(zhí)行事件發(fā)送到 apiserver甘有，這樣客戶端就能通過查詢知道整個流程發(fā)生了哪些事情诉儒，不需要登錄到 kubelet 所在的節(jié)點查看日志的內(nèi)容或者容器的運行狀態(tài)。
kubernetes 是以 pod 為核心概念的亏掀，不管是 deployment忱反、statefulSet、replicaSet滤愕，最終都會創(chuàng)建出來 pod温算。因此事件機制也是圍繞 pod 進行的，在 pod 生命周期的關(guān)鍵步驟都會產(chǎn)生事件消息该互。比如 Controller Manager 會記錄節(jié)點注冊和銷毀的事件米者、Deployment 擴容和升級的事件；kubelet 會記錄鏡像回收事件宇智、volume 無法掛載事件等蔓搞；Scheduler 會記錄調(diào)度事件等。
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?的信息打印機制，內(nèi)部也會調(diào)用?Event机蔗，而?PastEventf?允許用戶傳進來自定義的時間戳蒲祈，因此可以設(shè)置事件產(chǎn)生的時間甘萧。
在?cmd/kubelet/app/server.go RunKubelet 方法中
eventBroadcaster?是個事件廣播器，StartLogging?和?StartRecordingToSink?創(chuàng)建了兩個不同的事件處理函數(shù)梆掸，分別把事件記錄到日志和發(fā)送給 apiserver扬卷。而?NewRecorder?新建了一個?Recoder?對象，通過它的?Event酸钦、Eventf?和?PastEventf?方法条霜，用戶可以往里面發(fā)送事件经瓷，eventBroadcaster?會把接收到的事件發(fā)送個多個處理函數(shù)，比如這里提到的寫日志和發(fā)送到 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?對象匿垄，它就像一個事件記錄儀嚣州，用戶可以通過它記錄事件，它在內(nèi)部會把事件發(fā)送給?EventBroadcaster硝拧。
此外径筏，EventBroadcaster?定義了三個?Start?開頭的方法，它們用來添加事件處理 handler 河爹。其中核心方法是?StartEventWatcher匠璧，它會在后臺啟動一個 goroutine，不斷從 EventBroadcaster 提供的管道中接收事件咸这，然后調(diào)用?eventHandler?處理函數(shù)對事件進行處理夷恍。StartRecordingToSink?和?StartLogging?是對?StartEventWatcher?的封裝，分別實現(xiàn)了不同的處理函數(shù)（發(fā)送給 apiserver 和寫日志）媳维。
watcher := eventBroadcaster.Watch() 意思是注冊了一個 watcher 到 broadcaster 里面酿雪。
EventBroadcaster? 通過?EventRecorder?提供接口供用戶寫事件，內(nèi)部把接收到的事件發(fā)送給處理函數(shù)侄刽。處理函數(shù)是可以擴展的指黎，用戶可以通過?StartEventWatcher?來編寫自己的事件處理邏輯，kubelet?默認(rèn)會使用?StartRecordingToSink和?StartLogging州丹，也就是說任何一個事件會同時發(fā)送給 apiserver醋安，并打印到日志中。
發(fā)送事件的過程
事件是通過?EventRecorder?對象發(fā)送出來的墓毒，具體實現(xiàn)在?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?是具體的實現(xiàn)吓揪，eventBroadcaster.NewRecorder?會創(chuàng)建一個指定?EventSource?的?EventRecorder，EventSource?指明了哪個節(jié)點的哪個組件所计。
recorder 對外暴露了三個方法：Event柠辞、Eventf?和?PastEventf，它們的內(nèi)部最終都是調(diào)用?generateEvent?方法：
generateEvent?就是根據(jù)傳入的參數(shù)主胧，生成一個?api.Event?對象叭首，并發(fā)送出去习勤。它各個參數(shù)的意思是：
????????object：哪個組件/對象發(fā)出的事件，比如 kubelet 產(chǎn)生的事件會使用 node 對象
????????timestamp：事件產(chǎn)生的時間
????????eventtype：事件類型焙格，目前有兩種：Normal?和?Warning图毕，分別代表正常的事件和可能有問題的事件，定義在?pkg/api/types.go?文件中眷唉，未來可能有其他類型的事件擴展
????????reason：事件產(chǎn)生的原因吴旋，可以在?pkg/kubelet/events/event.go?看到 kubelet 定義的所有事件類型
????????message：事件的具體內(nèi)容，用戶可以理解的語句
makeEvent?就是根據(jù)參數(shù)構(gòu)建?api.Event?對象厢破，自動填充時間戳和 namespace：
注意 Event 事件的名字的構(gòu)成，它有兩部分：事件關(guān)聯(lián)對象的名字和當(dāng)前的時間治拿，中間用點隔開摩泪。
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（資源的類型和版本，對應(yīng)了 yaml 文件的?Kind?和?apiVersion?字段） 和?ObjectMera?字段（資源的元數(shù)據(jù)劫谅，比如 name见坑、nemspace、labels捏检、uuid荞驴、創(chuàng)建時間等）之外，還有和事件本身息息相關(guān)的字段贯城，比如事件消息熊楼、來源、類型能犯，以及數(shù)量（kubernetes 會把多個相同的事件匯聚到一起）和第一個事件的發(fā)生的時間等鲫骗。
中間有個?InvolvedObject?字段，它其實指向了和事件關(guān)聯(lián)的對象踩晶，如果是啟動容器的事件执泰，這個對象就是 Pod。
事件的發(fā)送是通過?recorder.Action()?實現(xiàn)的渡蜻。代碼在apimachinery/pkg/watch/mux.go
僅僅把對象封裝并放入 channel 里面术吝。Broadcaster?是?Recoder?內(nèi)部的對象，調(diào)用?NewRecoder?的時候?EventBroadcaster?傳給它的茸苇。
EventBroadcaster?也在?pkg/event/record/event.go?文件中
它的核心組件是?watch.Broadcaster排苍，Broadcaster?就是廣播的意思，主要功能就是把發(fā)給它的消息税弃，廣播給所有的監(jiān)聽者（watcher）纪岁。
簡單來說，watch.Broadcaster?是一個分發(fā)器则果，內(nèi)部保存了一個消息隊列幔翰，可以通過?Watch創(chuàng)建監(jiān)聽它內(nèi)部的 watcher漩氨。當(dāng)有消息發(fā)送到隊列中，watch.Broadcaster?后臺運行的 goroutine 會接收消息并通過 distribute() 發(fā)送給所有的 watcher遗增。而每個?watcher?都有一個接收消息的 channel叫惊，用戶可以通過它的?ResultChan()?獲取這個 channel 從中讀取數(shù)據(jù)進行處理。
StartRecordingToSink 是一個事件處理函數(shù)
StartLogging 也是一個事件處理函數(shù)
其實這兩個函數(shù)都只是對?StartEventWatcher 函數(shù)的封裝
它啟動一個 goroutine做修，不斷從?watcher.ResultChan()?中讀取消息霍狰，然后調(diào)用?eventHandler(event)?對事件進行處理。
事件的處理過程
recordToSink?負(fù)責(zé)把事件發(fā)送到 apiserver饰及，這里的 sink 其實就是和 apiserver 交互的 restclient蔗坯， event 是要發(fā)送的事件，eventCorrelator?在發(fā)送事件之前先對事件進行預(yù)處理燎含。
eventCorrelator.EventCorrelate?會對事件做預(yù)處理宾濒，主要是聚合相同的事件（避免產(chǎn)生的事件過多，增加 etcd 和 apiserver 的壓力屏箍，也會導(dǎo)致查看 pod 事件很不清晰）
result, err := eventCorrelator.EventCorrelate(event)
recordEvent?負(fù)責(zé)最終把事件發(fā)送到 apiserver绘梦，它會重試很多次（默認(rèn)是 12 次），并且每次重試都有一定時間間隔（默認(rèn)是 10 秒鐘）
recordEvent(sink, result.Event, result.Patch, result.Event.Count > 1, eventCorrelator)
它根據(jù)?eventCorrelator?的結(jié)果來決定是新建一個事件還是更新已經(jīng)存在的事件赴魁，并根據(jù)請求的結(jié)果決定是否需要重試（返回值為 false 說明需要重試卸奉，返回值為 true 表明已經(jīng)操作成功或者忽略請求錯誤）。sink.Create?和?sink.Patch?是自動生成的 apiserver 的 client颖御，對應(yīng)的代碼在：?pkg/client/clientset_generated/internalclientset/typed/core/internalversion/event_expansion.go?榄棵。
EventCorrelator?內(nèi)部有三個對象：filterFunc、aggregator?和?logger郎嫁，它們分別對事件進行過濾秉继、把相似的事件匯聚在一起、把相同的事件記錄到一起泽铛。使用?NewEventCorrelator?初始化的時候內(nèi)部會自動創(chuàng)建各個對象的默認(rèn)值尚辑，EventCorrelate?會以此調(diào)用三個對象的方法，并返回最終的結(jié)果】唬現(xiàn)在它們的邏輯是這樣的：
filterFunc：目前不做過濾杠茬，也就是說所有的事件都要經(jīng)過后續(xù)處理，后面可能會做擴展
aggregator：如果在最近 10 分鐘出現(xiàn)過 10 個相似的事件（除了 message 和時間戳之外其他關(guān)鍵字段都相同的事件）弛随，aggregator 會把它們的 message 設(shè)置為?events with common reason combined瓢喉，這樣它們就完全一樣了
logger：這個變量的名字有點奇怪，其實它會把相同的事件（除了時間戳之外其他字段都相同）變成同一個事件舀透，通過增加事件的?Count?字段來記錄該事件發(fā)生了多少次栓票。經(jīng)過?aggregator?的事件會在這里變成同一個事件
aggregator?和?logger?都會在內(nèi)部維護一個緩存（默認(rèn)長度是 4096），事件的相似性和相同性比較是和緩存中的事件進行的，也就是說它并在乎 kubelet 啟動之前的事件走贪，而且如果事件超過 4096 的長度佛猛，最近沒有被訪問的事件也會被從緩存中移除。這也是這個文件中帶有?cache?的原因坠狡。
總結(jié)
1. kubelet 通過?recorder?對象提供的?Event继找、Eventf?和?PastEventf?方法產(chǎn)生特性的事件
2. recorder?根據(jù)傳遞過來的參數(shù)新建一個?Event?對象，并把它發(fā)送給?EventBroadcaster?的管道
3. EventBroadcaster?后臺運行的 goroutine 從管道中讀取事件消息逃沿，把它廣播給之前注冊的 handler 進行處理
4. kubelet 有兩個 handler婴渡，它們分別把事件記錄到日志和發(fā)送給 apiserver。記錄到日志很簡單凯亮，直接打印就行
5. 發(fā)送給 apiserver 的 handler 叫做?EventSink边臼，它在發(fā)送事件給 apiserver 之前會先做預(yù)處理
6. 預(yù)處理操作是?EventCorrelator?完成的，它會對事件做過濾假消、匯聚和去重操作硼瓣，返回處理后的事件（可能是原來的事件，也可能是新創(chuàng)建的事件）
7. 最后通過 restclient （eventClient） 調(diào)用對應(yīng)的方法置谦，給 apiserver 發(fā)送請求，這個過程如果出錯會進行重試
8. apiserver 接收到事件的請求把數(shù)據(jù)更新到 etcd
事件一般用于調(diào)試亿傅，用戶可以通過?kubectl?命令獲取整個集群或者某個 pod 的事件信息媒峡。kubectl get events?可以看到所有的事件，kubectl describe pod PODNAME?能看到關(guān)于某個 pod 的事件葵擎。對于前者很好理解谅阿，kubectl 會直接訪問 apiserver 的 event 資源，而對于后者 kubectl 還根據(jù) pod 的名字進行搜索酬滤，匹配 InvolvedObject 名稱和 pod 名稱匹配的事件签餐。
