1、概述
Kubernetes的事件(Event)是一种资源对象(Resource Object),用于展示集群内发生的情况,Kubernetes系统中的各个组件会将运行时发生的各种事件上报给Kubernetes API Server。例如,调度器做了什么决定,某些Pod为什么被从节点中驱逐。由于Kubernetes的事件是一种资源对象,因此它们存储在Kubernetes API Server的Etcd集群中,为避免磁盘空间被填满,故强制执行保留策略:在最后一次的事件发生后,删除1小时之前发生的事件。Kubernetes Event资源对象的概念、使用及持久化方案请参考《Kubernetes Event详述及持久化方案》。
2、EventBroadcaster事件管理机制原理
2.1 谁会发送事件?
Kubernetes以Pod资源为核心,Deployment、StatefulSet、ReplicaSet、DaemonSet、CronJob等,最终都会创建出Pod。因此事件机制也是围绕 pod 进行的,在Pod生命周期的关键步骤都会产生事件消息。比如 Controller Manager 会记录节点注册和销毁的事件、Deployment 扩容和升级的事件;kubelet 会记录镜像回收事件、volume 无法挂载事件等;Scheduler 会记录调度事件等,这些Kubernetes核心组件都是基于EventBroadcaster事件管理机制进行Kubernetes Event处理。本文主要目的是讲解EventBroadcaster事件管理机制,通过讲解EventBroadcaster事件管理机制,我们在自定义组件的时候可以通过EventBroadcaster事件管理机制来管理我们自定义资源类型产生的事件,通过查看自定义资源类型关联的事件可以便于调试、排查、定位问题。
2.2 EventBroadcaster事件管理机制组成及运行原理
Event事件管理机制主要有三部分组成:
- EventRecorder:事件生产者,也称为事件记录器 ,k8s组件通过调用EventRecorder的方法来生成事件;
- EventBroadcaster:事件消费者, 也称为事件广播器。事件广播器,负责消费EventRecorder产生的事件,然后分发给broadcasterWatcher;分发过程有两种机制,分别是非阻塞(Non-Blocking )分发机制和阻塞( Blocking )分发机制;
- broadcasterWatcher:观察者管理,用于定义事件的处理方式,如上报事件至apiserver;
EventBroadcaster事件管理机制运行原理如下图所示:
如上图所示, Actor 可以是 Kubernetes 系统中的任意组件(当然也可以是自定义组件),当组件中发生了些关键性事件时,可通过 EventRecorder 记录该事件。
注意:此图摘自《Kuberneter源码剖析》,其Kuberneter版本为1.14.0,请注意您使用的Kubernetes版本,不同版本EventBroadcaster三个组件调用的方法可能和上图有所出入,本文以Kuberneter1.21.7进行剖析EventBroadcaster事件管理机制。
3、Event资源数据结构
以下数据结构都来自k8s.io/api/core/v1/types.go文件:
- Event结构体:
//事件是集群中某处事件的报告。 type Event struct { metav1.TypeMeta `json:",inline"` //标准对象的元数据。 metav1.ObjectMeta `json:"metadata" protobuf:"bytes,1,opt,name=metadata"` //与此 event 有直接关联的资源对象(触发event的资源对象) InvolvedObject ObjectReference `json:"involvedObject" protobuf:"bytes,2,opt,name=involvedObject"` //这应该是一个简短的,机器可理解的字符串,该字符串给出了转换为对象当前状态的原因。 // +optional Reason string `json:"reason,omitempty" protobuf:"bytes,3,opt,name=reason"` //此操作状态的可读描述。(给一个更易让人读懂的详细说明) // +optional Message string `json:"message,omitempty" protobuf:"bytes,4,opt,name=message"` //报告此事件的组件。 应该是机器可以理解的短字符串。 // +optional Source EventSource `json:"source,omitempty" protobuf:"bytes,5,opt,name=source"` //首次记录事件的时间。 (服务器收到时间以TypeMeta表示。) // +optional FirstTimestamp metav1.Time `json:"firstTimestamp,omitempty" protobuf:"bytes,6,opt,name=firstTimestamp"` //最近一次记录此事件的时间。 // +optional LastTimestamp metav1.Time `json:"lastTimestamp,omitempty" protobuf:"bytes,7,opt,name=lastTimestamp"` // 此事件发生的次数。 // +optional Count int32 `json:"count,omitempty" protobuf:"varint,8,opt,name=count"` // 此事件的类型(正常,警告),将来可能会添加新的类型 // +optional Type string `json:"type,omitempty" protobuf:"bytes,9,opt,name=type"` //首次观察到此事件的时间。 // +optional EventTime metav1.MicroTime `json:"eventTime,omitempty" protobuf:"bytes,10,opt,name=eventTime"` // 有关此事件表示的事件系列的数据,如果是单例事件,则为nil。 // +optional Series *EventSeries `json:"series,omitempty" protobuf:"bytes,11,opt,name=series"` // 针对对象已采取/未采取什么措施。 // +optional Action string `json:"action,omitempty" protobuf:"bytes,12,opt,name=action"` // 可选的辅助对象,用于更复杂的操作。 // +optional Related *ObjectReference `json:"related,omitempty" protobuf:"bytes,13,opt,name=related"` // 发出此事件的控制器的名称,例如 `kubernetes.io / kubelet`。 // +optional ReportingController string `json:"reportingComponent" protobuf:"bytes,14,opt,name=reportingComponent"` // 控制器实例的ID,例如 `kubelet-xyzf`。 // +optional ReportingInstance string `json:"reportingInstance" protobuf:"bytes,15,opt,name=reportingInstance"` }
-
involvedObject结构体: 定义了与此 Event 有直接关联的资源对象:
type ObjectReference struct { Kind string `json:"kind,omitempty" protobuf:"bytes,1,opt,name=kind"` Namespace string `json:"namespace,omitempty" protobuf:"bytes,2,opt,name=namespace"` Name string `json:"name,omitempty" protobuf:"bytes,3,opt,name=name"` UID types.UID `json:"uid,omitempty" protobuf:"bytes,4,opt,name=uid,casttype=k8s.io/apimachinery/pkg/types.UID"` APIVersion string `json:"apiVersion,omitempty" protobuf:"bytes,5,opt,name=apiVersion"` ResourceVersion string `json:"resourceVersion,omitempty" protobuf:"bytes,6,opt,name=resourceVersion"` FieldPath string `json:"fieldPath,omitempty" protobuf:"bytes,7,opt,name=fieldPath"` }
- EventSource结构体,定义了与此Event直接关联的组件(上报事件的组件信息):
type EventSource struct { // Component from which the event is generated. // +optional Component string `json:"component,omitempty" protobuf:"bytes,1,opt,name=component"` // Node name on which the event is generated. // +optional Host string `json:"host,omitempty" protobuf:"bytes,2,opt,name=host"` }
-
types.go常量:定义了两种Event类型:
const ( // 正常事件 // Information only and will not cause any problems EventTypeNormal string = "Normal" // 警告事件 // These events are to warn that something might go wrong EventTypeWarning string = "Warning" )
4、EventBroadcaster源码分析
4.1 EventRecorder记录事件(事件生产者/事件记录器)
- EventRecorder
在client-go中的tools/record/event.go中定义的EventRecorder接口:
// EventRecorder knows how to record events on behalf of an EventSource. type EventRecorder interface { // 对刚发生的事件进行记录 Event(object runtime.Object, eventtype, reason, message string) // 通过使用fmt.Sprintf格式化输出事件的格式。 Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) // 功能与Eventf一样,但附加了注释(Annotations )字段。 AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) }
EventRecorder定义了记录Event的三种方法,用以帮助kubernetes组件记录Event。其中Event是可以用来记录刚发生的事件;Eventf通过使用fmt.Sprintf格式化输出事件的格式;AnnotatedEventf功能和Eventf一致,但是附加了注释字段。
- recorderImpl
结构体recorderImpl是EventRecorder接口的实现:
// client-go/tools/record/event.go type recorderImpl struct { //k8s资源注册表 scheme *runtime.Scheme //上报事件的组件,例如kubelet,kube-controller-manager source v1.EventSource //事件消费 匿名字段 *watch.Broadcaster clock clock.Clock } // recorderImpl实例化方法,基于事件消费者实现类 eventBroadcasterImpl,建立了生产者和消费者之间的联系 func (e *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder { return &recorderImpl{scheme, source, e.Broadcaster, clock.RealClock{}} } func (recorder *recorderImpl) Event(object runtime.Object, eventtype, reason, message string) { recorder.generateEvent(object, nil, 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) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) { recorder.generateEvent(object, annotations, eventtype, reason, fmt.Sprintf(messageFmt, args...)) }
recorderImpl结构体中包含apimachinery/pkg/watch/mux.go中的Broadcaster结构体对象地址,因此可以调用Broadcaster实现的方法。
recorderImpl实现了EventRecorder接口定义的三个方法,以Event方法为例,调用链为:
recorderImpl.Event方法→ recorderImpl.generateEvent方法→Broadcaster.ActionOrDrop方法:
// client-go/tools/record/event.go func (recorder *recorderImpl) generateEvent(object runtime.Object, annotations map[string]string, eventtype, reason, message string) { //实例化事件直接关联的资源对象 ref, err := ref.GetReference(recorder.scheme, object) if err != nil { klog.Errorf("Could not construct reference to: '%#v' due to: '%v'. Will not report event: '%v' '%v' '%v'", object, err, eventtype, reason, message) return } // 验证事件类型,目前只支持Normal和Warning两种事件类型 if !util.ValidateEventType(eventtype) { klog.Errorf("Unsupported event type: '%v'", eventtype) return } // 实例化Event event := recorder.makeEvent(ref, annotations, eventtype, reason, message) // 设置上报事件的组件信息 event.Source = recorder.source //将Event写入m.incoming Chan中,完成事件生产过程 if sent := recorder.ActionOrDrop(watch.Added, event); !sent { klog.Errorf("unable to record event: too many queued events, dropped event %#v", event) } }
makeEvent方法会创建Event资源实例
// client-go/tools/record/event.go func (recorder *recorderImpl) makeEvent(ref *v1.ObjectReference, annotations map[string]string, eventtype, reason, message string) *v1.Event { t := metav1.Time{Time: recorder.clock.Now()} namespace := ref.Namespace // 如果此event有直接关联的资源对象是集群资源,那么此event将创建在default命名空间下 if namespace == "" { namespace = metav1.NamespaceDefault } return &v1.Event{ ObjectMeta: metav1.ObjectMeta{ Name: fmt.Sprintf("%v.%x", ref.Name, t.UnixNano()), Namespace: namespace, Annotations: annotations, }, // 此event有直接关联的资源对象 InvolvedObject: *ref, Reason: reason, Message: message, FirstTimestamp: t, LastTimestamp: t, Count: 1, Type: eventtype, } }
generateEvent方法会异步的调用ActionOrDrop方法,将事件写入到incoming中:
// apimachinery/pkg/watch/mux.go func (m *Broadcaster) ActionOrDrop(action EventType, obj runtime.Object) bool { select { case m.incoming <- Event{action, obj}: return true default: return false } }
4.2 EventBroadcaster事件广播(事件消费者/事件广播器)
- EventBroadcaster
在client-go中的tools/record/event.go中定义了EventBroadcaster接口:
// EventBroadcaster knows how to receive events and send them to any EventSink, watcher, or log. type EventBroadcaster interface { // StartEventWatcher starts sending events received from this EventBroadcaster to the given // event handler function. The return value can be ignored or used to stop recording, if // desired. StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface // StartRecordingToSink starts sending events received from this EventBroadcaster to the given // sink. The return value can be ignored or used to stop recording, if desired. StartRecordingToSink(sink EventSink) watch.Interface // StartLogging starts sending events received from this EventBroadcaster to the given logging // function. The return value can be ignored or used to stop recording, if desired. StartLogging(logf func(format string, args ...interface{})) watch.Interface // StartStructuredLogging starts sending events received from this EventBroadcaster to the structured // logging function. The return value can be ignored or used to stop recording, if desired. StartStructuredLogging(verbosity klog.Level) watch.Interface // 事件生产者实例化方法,用于发送事件到此事件消费者 // NewRecorder returns an EventRecorder that can be used to send events to this EventBroadcaster // with the event source set to the given event source. NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder // Shutdown shuts down the broadcaster Shutdown() }
EventBroadcaster作为Event消费者和事件广播器,消费EventRecorder记录的事件并将其分发给目前所有已连接的broadcasterWatcher。
结构体eventBroadcasterImpl是其实现:
type eventBroadcasterImpl struct { *watch.Broadcaster sleepDuration time.Duration options CorrelatorOptions }
eventBroadcasterImpl结构体中,同样包含Broadcaster结构体对象地址,因此可以调用Broadcaster实现的方法。
在apimachinery中的pkg/watch/mux.go中定义了Broadcaster结构体:
type Broadcaster struct { watchers map[int64]*broadcasterWatcher nextWatcher int64 distributing sync.WaitGroup incoming chan Event stopped chan struct{} // How large to make watcher's channel. watchQueueLength int // If one of the watch channels is full, don't wait for it to become empty. // Instead just deliver it to the watchers that do have space in their // channels and move on to the next event. // It's more fair to do this on a per-watcher basis than to do it on the // "incoming" channel, which would allow one slow watcher to prevent all // other watchers from getting new events. fullChannelBehavior FullChannelBehavior }
client-go的tools/record/event.go中,提供的实例化eventBroadcasterImpl的函数:
// Creates a new event broadcaster. func NewBroadcaster() EventBroadcaster { return &eventBroadcasterImpl{ Broadcaster: watch.NewLongQueueBroadcaster(maxQueuedEvents, watch.DropIfChannelFull), sleepDuration: defaultSleepDuration, } }
Broadcaster实际由apimachinery/pkg/watch/mux.go中的NewBroadcaster函数创建:
func NewLongQueueBroadcaster(queueLength int, fullChannelBehavior FullChannelBehavior) *Broadcaster { m := &Broadcaster{ watchers: map[int64]*broadcasterWatcher{}, incoming: make(chan Event, queueLength), stopped: make(chan struct{}), watchQueueLength: queueLength, fullChannelBehavior: fullChannelBehavior, } m.distributing.Add(1) go m.loop() return m }
创建时,会在内部启动goroutine,通过m.loop方法监控m.incoming;
// k8s.io/apimachinery/pkg/watch/mux.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.distribute函数分发给所有已连接的BroadcasterWatcher:
func (m *Broadcaster) distribute(event Event) { if m.fullChannelBehavior == DropIfChannelFull { for _, w := range m.watchers { select { case w.result <- event: case <-w.stopped: default: // 队列满时,不阻塞 } } } else { for _, w := range m.watchers { select { case w.result <- event: case <-w.stopped: } } } }
分发过程有两种机制,分别是非阻塞(Non-Blocking)分发机制和阻塞(Blocking)分发机制。
在非阻塞分发机制(默认)下使用DropIfChannelFull标识。DropIfChannelFull标识位于select多路复用中,使用default关键字做非阻塞分发,当w.result缓冲区满的时候,事件会丢失。
在阻塞分发机制下使用WaitIfChannelFull标识。WaitIfChannelFull标识也位于select多路复用中,没有default关键字,当w.result缓冲区满的时候,分发过程会阻塞并等待。
这里之所以需要丢失事件,是因为随着k8s集群越来越大,上报事件也随之增多,那么每次上报都要对etcd进行读写,这样会给etcd集群带来压力。但是事件丢失并不会影响集群的正常工作,所以非阻塞分发机制下事件会丢失。
4.3 broadcasterWatcher事件的处理
eventBroadcasterImpl实现的三种Event的处理方法:
(1)StartLogging:将事件写入日志中。
func (e *eventBroadcasterImpl) StartLogging(logf func(format string, args ...interface{})) watch.Interface { return e.StartEventWatcher( func(e *v1.Event) { logf("Event(%#v): type: '%v' reason: '%v' %v", e.InvolvedObject, e.Type, e.Reason, e.Message) }) }
(2)StartStructuredLogging:将事件写入结构化日志中。
func (e *eventBroadcasterImpl) StartStructuredLogging(verbosity klog.Level) watch.Interface { return e.StartEventWatcher( func(e *v1.Event) { klog.V(verbosity).InfoS("Event occurred", "object", klog.KRef(e.InvolvedObject.Namespace, e.InvolvedObject.Name), "kind", e.InvolvedObject.Kind, "apiVersion", e.InvolvedObject.APIVersion, "type", e.Type, "reason", e.Reason, "message", e.Message) }) }
(3)StartRecordingToSink:将事件存储到相应的sink。
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) }) }
对于StartLogging、StartStructuredLogging方式,都是把事件信息当做日志打印了一下。这里主要看一下StartRecordingToSink方法,调用StartRecordingToSink方法会将数据上报到apiserver。
StartRecordingToSink方法依赖StartEventWatcher方法:
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中。
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 > 1, eventCorrelator) { break } tries++ if tries >= maxTriesPerEvent { klog.Errorf("Unable to write event '%#v' (retry limit exceeded!)", event) break } if tries == 1 { time.Sleep(time.Duration(float64(sleepDuration) * rand.Float64())) } else { time.Sleep(sleepDuration) } } }
recordToSink方法首先会调用EventCorrelate方法对event做预处理,聚合相同的事件,避免产生的事件过多,增加 etcd 和 apiserver 的压力,如果传入的Event太多了,那么result.Skip 就会返回false;
接下来会调用recordEvent方法把事件发送到 apiserver,它会重试很多次(默认是 12 次),并且每次重试都有一定时间间隔(默认是 10 秒钟)。
下面我们分别来看看EventCorrelate方法和recordEvent方法。
- EventCorrelate
文件位置:client-go/tools/record/events_cache.go // client-go/tools/record/events_cache.go func (c *EventCorrelator) EventCorrelate(newEvent *v1.Event) (*EventCorrelateResult, error) { if newEvent == nil { return nil, fmt.Errorf("event is nil") } aggregateEvent, ckey := c.aggregator.EventAggregate(newEvent) observedEvent, patch, err := c.logger.eventObserve(aggregateEvent, ckey) if c.filterFunc(observedEvent) { return &EventCorrelateResult{Skip: true}, nil } return &EventCorrelateResult{Event: observedEvent, Patch: patch}, err }
EventCorrelate方法会调用EventAggregate、eventObserve进行聚合,调用filterFunc会调用到spamFilter.Filter方法进行过滤。
func (e *EventAggregator) EventAggregate(newEvent *v1.Event) (*v1.Event, string) { now := metav1.NewTime(e.clock.Now()) var record aggregateRecord eventKey := getEventKey(newEvent) aggregateKey, localKey := e.keyFunc(newEvent) e.Lock() defer e.Unlock() // 查找缓存里面是否也存在这样的记录 value, found := e.cache.Get(aggregateKey) if found { record = value.(aggregateRecord) } // maxIntervalInSeconds默认时间是600s,这里校验缓存里面的记录是否太老了 // 如果是那么就创建一个新的 // 如果record在缓存里面找不到,那么lastTimestamp是零,那么也创建一个新的 maxInterval := time.Duration(e.maxIntervalInSeconds) * time.Second interval := now.Time.Sub(record.lastTimestamp.Time) if interval > maxInterval { record = aggregateRecord{localKeys: sets.NewString()} } record.localKeys.Insert(localKey) record.lastTimestamp = now // 重新加入到LRU缓存中 e.cache.Add(aggregateKey, record) // 如果没有达到阈值,那么不进行聚合 if uint(record.localKeys.Len()) < e.maxEvents { return newEvent, eventKey } record.localKeys.PopAny() eventCopy := &v1.Event{ ObjectMeta: metav1.ObjectMeta{ Name: fmt.Sprintf("%v.%x", newEvent.InvolvedObject.Name, now.UnixNano()), Namespace: newEvent.Namespace, }, Count: 1, FirstTimestamp: now, InvolvedObject: newEvent.InvolvedObject, LastTimestamp: now, // 将Message进行聚合 Message: e.messageFunc(newEvent), Type: newEvent.Type, Reason: newEvent.Reason, Source: newEvent.Source, } return eventCopy, aggregateKey }
EventAggregate方法也考虑了很多,首先是去缓存里面查找有没有相同的聚合记录aggregateRecord,如果没有的话,那么会在校验时间间隔的时候顺便创建聚合记录aggregateRecord;
由于缓存时lru缓存,所以再将聚合记录重新Add到缓存的头部;
接下来会判断缓存是否已经超过了阈值,如果没有达到阈值,那么直接返回不进行聚合;
如果达到阈值了,那么会重新copy传入的Event,并调用messageFunc方法聚合Message;
- eventObserve
func (e *eventLogger) eventObserve(newEvent *v1.Event, key string) (*v1.Event, []byte, error) { var ( patch []byte err error ) eventCopy := *newEvent event := &eventCopy e.Lock() defer e.Unlock() // 检查是否在缓存中 lastObservation := e.lastEventObservationFromCache(key) // 如果大于0说明存在,并且对Count进行自增 if lastObservation.count > 0 { event.Name = lastObservation.name event.ResourceVersion = lastObservation.resourceVersion event.FirstTimestamp = lastObservation.firstTimestamp event.Count = int32(lastObservation.count) + 1 eventCopy2 := *event eventCopy2.Count = 0 eventCopy2.LastTimestamp = metav1.NewTime(time.Unix(0, 0)) eventCopy2.Message = "" newData, _ := json.Marshal(event) oldData, _ := json.Marshal(eventCopy2) patch, err = strategicpatch.CreateTwoWayMergePatch(oldData, newData, event) } // 最后重新更新缓存记录 e.cache.Add( key, eventLog{ count: uint(event.Count), firstTimestamp: event.FirstTimestamp, name: event.Name, resourceVersion: event.ResourceVersion, }, ) return event, patch, err }
eventObserve方法里面会去查找缓存中的记录,然后对count进行自增后更新到缓存中。
- Filter
func (f *EventSourceObjectSpamFilter) Filter(event *v1.Event) bool { var record spamRecord eventKey := getSpamKey(event) f.Lock() defer f.Unlock() value, found := f.cache.Get(eventKey) if found { record = value.(spamRecord) } if record.rateLimiter == nil { record.rateLimiter = flowcontrol.NewTokenBucketRateLimiterWithClock(f.qps, f.burst, f.clock) } // 使用令牌桶进行过滤 filter := !record.rateLimiter.TryAccept() // update the cache f.cache.Add(eventKey, record) return filter }
Filter主要时起到了一个限速的作用,通过令牌桶来进行过滤操作。
- recordEvent
func recordEvent(sink EventSink, event *v1.Event, patch []byte, updateExistingEvent bool, eventCorrelator *EventCorrelator) bool { var newEvent *v1.Event var err error // 更新已经存在的事件 if updateExistingEvent { newEvent, err = sink.Patch(event, patch) } // 创建一个新的事件 if !updateExistingEvent || (updateExistingEvent && util.IsKeyNotFoundError(err)) { event.ResourceVersion = "" newEvent, err = sink.Create(event) } if err == nil { eventCorrelator.UpdateState(newEvent) return true } // 如果是已知错误,就不要再重试了;否则,返回 false,让上层进行重试 switch err.(type) { case *restclient.RequestConstructionError: klog.Errorf("Unable to construct event '%#v': '%v' (will not retry!)", event, err) return true case *errors.StatusError: if errors.IsAlreadyExists(err) { klog.V(5).Infof("Server rejected event '%#v': '%v' (will not retry!)", event, err) } else { klog.Errorf("Server rejected event '%#v': '%v' (will not retry!)", event, err) } return true case *errors.UnexpectedObjectError: default: } klog.Errorf("Unable to write event: '%v' (may retry after sleeping)", err) return false }
recordEvent方法会根据eventCorrelator返回的结果来决定是新建一个事件还是更新已经存在的事件,并根据请求的结果决定是否需要重试。
5、使用示例
const ( // SuccessSynced is used as part of the Event 'reason' when a Foo is synced successSynced = "Synced" // is synced successfully messageResourceSynced = "User synced successfully" ) eventBroadcaster := record.NewBroadcaster() eventBroadcaster.StartLogging(klog.Infof) eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: k8sClient.CoreV1().Events("")}) recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName}) c.recorder.Event(user, corev1.EventTypeNormal, successSynced, messageResourceSynced)
6、总结
了解完 events 的整个处理流程后,再梳理一下整个流程:
-
首先是初始化 EventBroadcaster 对象,同时会初始化一个 Broadcaster 对象,并开启一个loop循环接收所有的 events 并进行广播;
-
然后通过 EventBroadcaster 对象的 NewRecorder() 方法初始化 EventRecorder 对象,EventRecorder 对象会生成 events 并通过ActionOrDrop() 方法发送 events 到 Broadcaster 的 channel 队列中;
-
EventBroadcaster 会调用StartStructuredLogging、StartRecordingToSink方法调用封装好的StartEventWatcher方法,并执行自己的逻辑;
-
StartRecordingToSink封装的StartEventWatcher方法里面会将所有的 events 广播给每一个 watcher,并调用recordToSink方法对收到 events 后会进行缓存、过滤、聚合而后发送到 apiserver,apiserver 会将 events 保存到 etcd 中。
参考:https://blog.csdn.net/weixin_45413603/article/details/108204904
参考:https://www.cnblogs.com/yangyuliufeng/p/13942789.html
参考:https://www.cnblogs.com/luozhiyun/p/13799901.html