6.3 k8s的事件event和kube-scheduler中的事件广播器

什么是k8s的events

• k8s的events是向您展示集群内部发生的事情的对象
  • 例如调度程序做出了哪些决定
  • 或者为什么某些 Pod 从节点中被逐出

哪些组件可以产生events

• 所有核心组件和扩展（操作符）都可以通过 API Server 创建事件
• k8s 多个组件均会产生 event

如何获取event 数据

get events直接获取

[root@k8s-master01 k8s-leaderelection]# kubectl get events -A 
NAMESPACE         LAST SEEN   TYPE      REASON        OBJECT                      MESSAGE
injection         3m21s       Warning   FailedMount   pod/alpine01                MountVolume.SetUp failed for volume "nginx-conf" : configmap "nginx-configmap" not found
nginx-injection   32m         Normal    Pulled        pod/test-alpine-inject-01   Container image "alpine" already present on machine
nginx-injection   32m         Normal    Created       pod/test-alpine-inject-01   Created container alpine
nginx-injection   32m         Normal    Started       pod/test-alpine-inject-01   Started container alpine
nginx-injection   27m         Normal    Pulled        pod/test-alpine-inject-02   Container image "alpine" already present on machine
nginx-injection   27m         Normal    Created       pod/test-alpine-inject-02   Created container alpine
nginx-injection   27m         Normal    Started       pod/test-alpine-inject-02   Started container alpine

describe 资源获取

• 比如创建pod时故意将容器的image 仓库名字写错

apiVersion: v1
kind: Pod
metadata:name: nginx-pod-test
spec:containers:- name: nginximage: nginxa:1.8

• 创建之后就可以describe 这个pod获取events，可以看到拉取镜像失败的events

[root@k8s-master01 app]# kubectl describe pod nginx-pod-test 
Name:         nginx-pod-test
Namespace:    default
Priority:     0
Node:         k8s-node01/172.20.70.215
Start Time:   Wed, 15 Sep 2021 19:13:20 +0800
Labels:       <none>
Annotations:  cni.projectcalico.org/podIP: 10.100.85.220/32cni.projectcalico.org/podIPs: 10.100.85.220/32
Status:       Pending
IP:           10.100.85.220
IPs:IP:  10.100.85.220
Containers:nginx:Container ID:   Image:          nginxa:1.8Image ID:   Port:           <none>Host Port:      <none>State:          WaitingReason:       ImagePullBackOffReady:          FalseRestart Count:  0Environment:    <none>Mounts:/var/run/secrets/kubernetes.io/serviceaccount from default-token-k46nh (ro)
Conditions:Type              StatusInitialized       True Ready             False ContainersReady   False PodScheduled      True 
Volumes:default-token-k46nh:Type:        Secret (a volume populated by a Secret)SecretName:  default-token-k46nhOptional:    false
QoS Class:       BestEffort
Node-Selectors:  <none>
Tolerations:     node.kubernetes.io/not-ready:NoExecute op=Exists for 300snode.kubernetes.io/unreachable:NoExecute op=Exists for 300s
Events:Type     Reason     Age                  From               Message----     ------     ----                 ----               -------Normal   Scheduled  2m57s                default-scheduler  Successfully assigned default/nginx-pod-test to k8s-node01Normal   Pulling    67s (x4 over 2m57s)  kubelet            Pulling image "nginxa:1.8"Warning  Failed     63s (x4 over 2m49s)  kubelet            Failed to pull image "nginxa:1.8": rpc error: code = Unknown desc = failed to pull and unpack image "docker.io/library/nginxa:1.8": failed to resolve reference "docker.io/library/nginxa:1.8": pull access denied, repository does not exist or may require authorization: server message: insufficient_scope: authorization failedWarning  Failed     63s (x4 over 2m49s)  kubelet            Error: ErrImagePullNormal   BackOff    50s (x6 over 2m49s)  kubelet            Back-off pulling image "nginxa:1.8"Warning  Failed     39s (x7 over 2m49s)  kubelet            Error: ImagePullBackOff

Event事件管理机制要有三部分组成

• EventRecorder：是事件生成者，k8s组件通过调用它的方法来生成事件；
• EventBroadcaster：事件广播器，负责消费EventRecorder产生的事件，然后分发给broadcasterWatcher；
• broadcasterWatcher：用于定义事件的处理方式，如上报apiserver；

events保存问题

• Events 量非常大，只能存在一个很短的时间
• Event 一般会通过apiserver暂存在etcd集群中 (最好是单独的etcd集群存储events，和集群业务数据的etcd分开)
• 为避免磁盘空间被填满，故强制执行保留策略：在最后一次的事件发生后，删除1小时之前发生的事件。

event有什么作用

• 可以基于event对k8s集群监控，下面图片来自网络
• 

kube-scheduler中的event解读

EventBroadcaster的初始化

• 在初始化配置的Config中 ，位置 D:\go_path\src\github.com\kubernetes\kubernetes\cmd\kube-scheduler\app\options\options.go

	c.EventBroadcaster = events.NewEventBroadcasterAdapter(eventClient)

• new

// NewEventBroadcasterAdapter creates a wrapper around new and legacy broadcasters to simplify
// migration of individual components to the new Event API.
func NewEventBroadcasterAdapter(client clientset.Interface) EventBroadcasterAdapter {eventClient := &eventBroadcasterAdapterImpl{}if _, err := client.Discovery().ServerResourcesForGroupVersion(eventsv1.SchemeGroupVersion.String()); err == nil {eventClient.eventsv1Client = client.EventsV1()eventClient.eventsv1Broadcaster = NewBroadcaster(&EventSinkImpl{Interface: eventClient.eventsv1Client})}// Even though there can soon exist cases when coreBroadcaster won't really be needed,// we create it unconditionally because its overhead is minor and will simplify using usage// patterns of this library in all components.eventClient.coreClient = client.CoreV1()eventClient.coreBroadcaster = record.NewBroadcaster()return eventClient
}

• 底层使用client-go中的eventBroadcasterImpl，位置 D:\go_path\src\github.com\kubernetes\kubernetes\staging\src\k8s.io\client-go\tools\record\event.go

type eventBroadcasterImpl struct {*watch.BroadcastersleepDuration time.Durationoptions       CorrelatorOptions
}

eventRecorder初始化

• 初始化scheduler的Setup函数中， 位置 D:\go_path\src\github.com\kubernetes\kubernetes\cmd\kube-scheduler\app\server.go

recorderFactory := getRecorderFactory(&cc)

• recorderFactory代表生成eventRecorder的工厂函数
• 最终使用的是client-go中的recorderImpl ,位置 D:\go_path\src\github.com\kubernetes\kubernetes\staging\src\k8s.io\client-go\tools\events\event_recorder.go

type recorderImpl struct {scheme              *runtime.SchemereportingController stringreportingInstance   string*watch.Broadcasterclock clock.Clock
}

开启event 事件广播器

• 在scheduler的run中，位置在D:\go_path\src\github.com\kubernetes\kubernetes\cmd\kube-scheduler\app\server.go

	// Prepare the event broadcaster.cc.EventBroadcaster.StartRecordingToSink(ctx.Done())

client-go中的StartRecordingToSink解读

// StartRecordingToSink starts sending events received from the specified eventBroadcaster to the given sink.
func (e *eventBroadcasterImpl) StartRecordingToSink(stopCh <-chan struct{}) {go wait.Until(e.refreshExistingEventSeries, refreshTime, stopCh)go wait.Until(e.finishSeries, finishTime, stopCh)e.startRecordingEvents(stopCh)
}

startRecordingEvents解读

• 位置D:\go_path\src\github.com\kubernetes\kubernetes\staging\src\k8s.io\client-go\tools\events\event_broadcaster.go

func (e *eventBroadcasterImpl) startRecordingEvents(stopCh <-chan struct{}) {eventHandler := func(obj runtime.Object) {event, ok := obj.(*eventsv1.Event)if !ok {klog.Errorf("unexpected type, expected eventsv1.Event")return}e.recordToSink(event, clock.RealClock{})}stopWatcher := e.StartEventWatcher(eventHandler)go func() {<-stopChstopWatcher()}()
}

• 启动一个eventHandler，执行recordToSink写入后端存储
• 在StartEventWatcher消费ResultChan队列中的event，传递给eventHandler处理

func (e *eventBroadcasterImpl) StartEventWatcher(eventHandler func(event runtime.Object)) func() {watcher := e.Watch()go func() {defer utilruntime.HandleCrash()for {watchEvent, ok := <-watcher.ResultChan()if !ok {return}eventHandler(watchEvent.Object)}}()return watcher.Stop
}

recordToSink发送逻辑

• 通过getKey生成event 的类型key，作为cache中的标识

func getKey(event *eventsv1.Event) eventKey {key := eventKey{action:              event.Action,reason:              event.Reason,reportingController: event.ReportingController,regarding:           event.Regarding,}if event.Related != nil {key.related = *event.Related}return key
}

• Event.series中记录的是这个event的次数和最近一次的时间
• 用上面的key去eventCache中寻找，如果找到了同时series存在就更新下相关的次数和时间

			isomorphicEvent, isIsomorphic := e.eventCache[eventKey]if isIsomorphic {if isomorphicEvent.Series != nil {isomorphicEvent.Series.Count++isomorphicEvent.Series.LastObservedTime = metav1.MicroTime{Time: clock.Now()}return nil}

• 不然的话创建新的并返回

				isomorphicEvent.Series = &eventsv1.EventSeries{Count:            1,LastObservedTime: metav1.MicroTime{Time: clock.Now()},}return isomorphicEvent

• 然后用获取到的evToRecord 发送，并更新缓存

		if evToRecord != nil {recordedEvent := e.attemptRecording(evToRecord)if recordedEvent != nil {recordedEventKey := getKey(recordedEvent)e.mu.Lock()defer e.mu.Unlock()e.eventCache[recordedEventKey] = recordedEvent}}

• attemptRecording代表带重试发送，底层调用的就是sink的方法，位置D:\go_path\src\github.com\kubernetes\kubernetes\staging\src\k8s.io\client-go\tools\events\interfaces.go

type EventSink interface {Create(event *eventsv1.Event) (*eventsv1.Event, error)Update(event *eventsv1.Event) (*eventsv1.Event, error)Patch(oldEvent *eventsv1.Event, data []byte) (*eventsv1.Event, error)
}

Event.series的作用

• 就像重复的连接不上mysql的错误日志会打很多条一样
• k8s的同一个event也会产生多条，那么去重降噪是有必要的
• 通过 event.Action类型 event.Reason原因 event.ReportingController产生的源等信息组成唯一的key
• 如果cache中有key的记录，那么更新这种event的发生的次数和最近的时间就可以了
• 没必要将全量重复的event发送到存储

kube-scheduler中的调用 eventRecorder产生的事件

eventRecorder对象

• 在scheduler的frameworkImpl中， 位置 D:\go_path\src\github.com\kubernetes\kubernetes\pkg\scheduler\framework\runtime\framework.go

// EventRecorder returns an event recorder.
func (f *frameworkImpl) EventRecorder() events.EventRecorder {return f.eventRecorder
}

scheduler记录调度pod失败的地方

• 位置 D:\go_path\src\github.com\kubernetes\kubernetes\pkg\scheduler\scheduler.go

// recordSchedulingFailure records an event for the pod that indicates the
// pod has failed to schedule. Also, update the pod condition and nominated node name if set.
func (sched *Scheduler) recordSchedulingFailure(fwk framework.Framework, podInfo *framework.QueuedPodInfo, err error, reason string, nominatedNode string) {sched.Error(podInfo, err)// Update the scheduling queue with the nominated pod information. Without// this, there would be a race condition between the next scheduling cycle// and the time the scheduler receives a Pod Update for the nominated pod.// Here we check for nil only for tests.if sched.SchedulingQueue != nil {sched.SchedulingQueue.AddNominatedPod(podInfo.PodInfo, nominatedNode)}pod := podInfo.Podmsg := truncateMessage(err.Error())fwk.EventRecorder().Eventf(pod, nil, v1.EventTypeWarning, "FailedScheduling", "Scheduling", msg)if err := updatePod(sched.client, pod, &v1.PodCondition{Type:    v1.PodScheduled,Status:  v1.ConditionFalse,Reason:  reason,Message: err.Error(),}, nominatedNode); err != nil {klog.ErrorS(err, "Error updating pod", "pod", klog.KObj(pod))}
}

写一个pod的，故意让scheduler调度失败，查看相关的event

• 这里让pod要调度到 disktype=ssd的node上

cat <<EOF> nginx-pod-failed-to-scheduler.yaml
apiVersion: v1
kind: Pod
metadata:name: nginx-pod-failed-to-scheduler
spec:containers:- name: nginximage: nginx:1.8nodeSelector:disktype: ssd
EOF

• 创建后获取event ，可以看到调度失败的event

kubectl get event
LAST SEEN   TYPE      REASON             OBJECT                              MESSAGE
11s         Warning   FailedScheduling   pod/nginx-pod-failed-to-scheduler   0/2 nodes are available: 1 node(s) didn't match Pod's node affinity, 1 node(s) had taint {node-role.kubernetes.io/master: }, that the pod didn't tolerate.
11s         Warning   FailedScheduling   pod/nginx-pod-failed-to-scheduler   0/2 nodes are available: 1 node(s) didn't match Pod's node affinity, 1 node(s) had taint {node-role.kubernetes.io/master: }, that the pod didn't tolerate.

• 分析下kube-scheduler 记录event的代码

fwk.EventRecorder().Eventf(pod, nil, v1.EventTypeWarning, "FailedScheduling", "Scheduling", msg)
Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{})

• eventf参数和字段分析
  • regarding 关注哪种资源的event 这里传入的是pod
  • related 还关联哪些资源，这里传入的是nil
  • eventtype 代表是warning的还是normal的，这里是v1.EventTypeWarning
  • reason 原因 ，这里传入的是 FailedScheduling调度失败
  • action 代表执行哪个动作，这里传入的是Scheduling
  • note 代表详细信息，这里传入的是错误信息msg

最后给node节点打上 disktype=ssd的标签让pod正常调度

[root@k8s-master01 app]# kubectl get pod 
NAME                                           READY   STATUS             RESTARTS   AGE
grafana-d5d85bcd6-f74ch                        1/1     Running            0          26d
grafana-d5d85bcd6-l44mx                        1/1     Running            0          26d
ink8s-pod-metrics-deployment-85d9795d6-95lsp   1/1     Running            0          23d
nginx-pod                                      1/1     Running            0          12d
nginx-pod-failed-to-scheduler                  0/1     Pending            0          12m
nginx-pod-test                                 0/1     ImagePullBackOff   0          90m
prometheus-operator-54c4665c6b-7j9jr           1/1     Running            0          13d
sleep-557747455f-w79wf                         1/1     Running            0          6d1h
[root@k8s-master01 app]# kubectl get node 
NAME           STATUS   ROLES                  AGE    VERSION
k8s-master01   Ready    control-plane,master   162d   v1.20.1
k8s-node01     Ready    <none>                 64d    v1.20.1
[root@k8s-master01 app]# kubectl label nodes k8s-node01 disktype=ssd
node/k8s-node01 labeled
[root@k8s-master01 app]# kubectl get node 
NAME           STATUS   ROLES                  AGE    VERSION
k8s-master01   Ready    control-plane,master   162d   v1.20.1
k8s-node01     Ready    <none>                 64d    v1.20.1
[root@k8s-master01 app]# kubectl get pod  
NAME                                           READY   STATUS             RESTARTS   AGE
grafana-d5d85bcd6-f74ch                        1/1     Running            0          26d
grafana-d5d85bcd6-l44mx                        1/1     Running            0          26d
ink8s-pod-metrics-deployment-85d9795d6-95lsp   1/1     Running            0          23d
nginx-pod                                      1/1     Running            0          12d
nginx-pod-failed-to-scheduler                  1/1     Running            0          13m
nginx-pod-test                                 0/1     ImagePullBackOff   0          91m
prometheus-operator-54c4665c6b-7j9jr           1/1     Running            0          13d
sleep-557747455f-w79wf                         1/1     Running            0          6d1h

• 可以看到pod调度正常也会有相关的event

[root@k8s-master01 app]# kubectl get event
LAST SEEN   TYPE      REASON             OBJECT                              MESSAGE
13m         Warning   FailedScheduling   pod/nginx-pod-failed-to-scheduler   0/2 nodes are available: 1 node(s) didn't match Pod's node affinity, 1 node(s) had taint {node-role.kubernetes.io/master: }, that the pod didn't tolerate.
12m         Warning   FailedScheduling   pod/nginx-pod-failed-to-scheduler   0/2 nodes are available: 1 node(s) didn't match Pod's node affinity, 1 node(s) had taint {node-role.kubernetes.io/master: }, that the pod didn't tolerate.
13s         Normal    Scheduled          pod/nginx-pod-failed-to-scheduler   Successfully assigned default/nginx-pod-failed-to-scheduler to k8s-node01
13s         Normal    Pulled             pod/nginx-pod-failed-to-scheduler   Container image "nginx:1.8" already present on machine
13s         Normal    Created            pod/nginx-pod-failed-to-scheduler   Created container nginx
13s         Normal    Started            pod/nginx-pod-failed-to-scheduler   Started container nginx
96s         Normal    BackOff            pod/nginx-pod-test                  Back-off pulling image "nginxa:1.8"
51m         Warning   Failed             pod/nginx-pod-test                  Error: ImagePullBackOff

• 反查正常的eventf源码，位置 D:\go_path\src\github.com\kubernetes\kubernetes\pkg\scheduler\scheduler.go

func (sched *Scheduler) finishBinding(fwk framework.Framework, assumed *v1.Pod, targetNode string, err error) {if finErr := sched.SchedulerCache.FinishBinding(assumed); finErr != nil {klog.ErrorS(finErr, "Scheduler cache FinishBinding failed")}if err != nil {klog.V(1).InfoS("Failed to bind pod", "pod", klog.KObj(assumed))return}fwk.EventRecorder().Eventf(assumed, nil, v1.EventTypeNormal, "Scheduled", "Binding", "Successfully assigned %v/%v to %v", assumed.Namespace, assumed.Name, targetNode)
}

• 这里的 Successfully assigned default/nginx-pod-failed-to-scheduler to k8s-node01 对应的就是"Successfully assigned %v/%v to %v", assumed.Namespace, assumed.Name, targetNode)

本节重点总结 :

• k8s的events是展示集群内部发生的事情的对象
• 很多组件都可以产生event，数据会上传到apiserver，临时存储在etcd中，因为量太大了，有删除策略

Event事件管理机制要有三部分组成

• EventRecorder：是事件生成者，k8s组件通过调用它的方法来生成事件；
• EventBroadcaster：事件广播器，负责消费EventRecorder产生的事件，然后分发给broadcasterWatcher；
• broadcasterWatcher：用于定义事件的处理方式，如上报apiserver；

可以给pod添加nodeSelector故意调度失败

cat <<EOF> nginx-pod-failed-to-scheduler.yaml
apiVersion: v1
kind: Pod
metadata:name: nginx-pod-failed-to-scheduler
spec:containers:- name: nginximage: nginx:1.8nodeSelector:disktype: ssd
EOF

kube-scheduler中会对pod调度成功失败都记录event

• 成功的

13s         Normal    Scheduled          pod/nginx-pod-failed-to-scheduler   Successfully assigned default/nginx-pod-failed-to-scheduler to k8s-node01
fwk.EventRecorder().Eventf(assumed, nil, v1.EventTypeNormal, "Scheduled", "Binding", "Successfully assigned %v/%v to %v", assumed.Namespace, assumed.Name, targetNode)

• 失败的

fwk.EventRecorder().Eventf(pod, nil, v1.EventTypeWarning, "FailedScheduling", "Scheduling", msg)
Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...interface{})