kubernetesのイベントを収集、格納、アラームする

背景:
企業内でますます多くのサービスがkubernetesに移行するにつれて、ユーザーはkubernetesに関する情報の取得をより切実にし、彼らは自分のサービスインスタンスが何を「経験した」のかを知りたいと思っています.例えば、業務異常があるとき、自分のインスタンスが再起動されたかどうかを知りたいと思っています.オンラインに失敗したときも、最初に原因を知りたいと思っています.さらに、これらがリアルタイムで通知され、後で表示されることを望んでいます.しかし、通常、kubectlなど、クラスタサーバに直接接触することはできません.そのため、良い解決策が必要です.
需要分解:
上記の要件は、次の4つの基本機能に分解できます.

kubernetes上のイベント情報

をリアルタイムで取得できる.

異常事象に対してリアルタイム警報を行うことができる

.

は、問題が

に遡るようにイベントをより永続的に保持することができる.

保存されたイベントを問い合わせることができる

.
これらの機能はすべて合理的に聞こえ、kubernetes.ioにはEvents in Stackdriverもイベントのストレージについて話しています.
これは、「イベントはkubernetesのオブジェクトリソースとして、並べ替えに役立ちますが、多くのイベントがetcdに格納されると、パフォーマンスと容量の圧力が大きくなるため、etcdではデフォルトで最近1時間しか保存されません.そのため、より永続的に保存するには、サードパーティ製のツールが必要です」と述べています.
具体的には、1.7+バージョンでは、Google Kubernetes Engineでcloud loggingが有効になっている場合、デフォルトでは、最小優先度で可能な限り少ないリソース(通常0.1コアのCPUと0.1 Gのメモリ)を使用するevent-exporterがクラスタ内に配備されます.
よさそうですが、機能1、3、4だけで対応するアラームはありません.またGKEを使用していないものも直接使用できないことは明らかです.
kubewatchやその背後にあるカスタムコントローラのようなオープンソースプロジェクトをもう一度調べました.ここでは、コントローラを自分で実装する方法を簡単に振り返ってみましょう.

はcacheを通過する.NewListWatchFromClient新規リスト

はcacheを通過する.NewSharedIndexInformer+ListWatcher新しいinformer

workqueueを介してRateLimitingQueueまたは他のタイプのキュー

を新規作成

informerにhandlerを追加し、OnAddなどのイベント情報をキューに入れる

.

informer

を起動

データのローカルキャッシュへの同期完了を待つ

worker持続消費queueのイベントを起動し、フィルタリング、アラーム、記憶などの処理を行う.

しかしkubewatchは直接要求を満たすことができず、pod、service、deploymentなどの様々なリソースタイプを傍受しているが、Eventタイプだけが欠けており、異なるタイプに対してinformerを起動し、異なるタイプのリソースのイベントフォーマットが異なり、互換性のために共通情報のみを伝達するため、フィールドは少ない.したがって,処理を再簡略化し,Eventタイプのみを傍受し,すべてのフィールドを解析することができる.次のようになります.

	lw := cache.NewListWatchFromClient(
		kubeClient.CoreV1().RESTClient(), //    
		"events",                         //        
		"",                               //        ,      
		fields.Everything())              //    ，    

	informer := cache.NewSharedIndexInformer(lw, &api_v1.Event{}, 0, cache.Indexers{})
	c := controller.NewResourceController(kubeClient, informer, config)

func NewResourceController(client kubernetes.Interface, informer cache.SharedIndexInformer, config config.Config) *Controller {
	queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
	var newEvent Event
	var err error
	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			evt := obj.(*api_v1.Event)
			newEvent.Key, err = cache.MetaNamespaceKeyFunc(obj)
			if err == nil {
				logrus.Debugf("      ，   :CreateEvent,   :%+v", evt)
				queue.Add(newEvent.Key)
			}
		},
		UpdateFunc: func(old, new interface{}) {
			oldEvt := old.(*api_v1.Event)
			newEvt := new.(*api_v1.Event)
			if !reflect.DeepEqual(newEvt.Source, oldEvt.Source) && oldEvt.Reason != newEvt.Reason {
				newEvent.Obj = newEvt
				newEvent.Key, err = cache.MetaNamespaceKeyFunc(old)
				if err == nil {
					queue.Add(newEvent.Key)
				}
			}
		},
		DeleteFunc: func(obj interface{}) {
			evt := obj.(*api_v1.Event)
			flog.Log().Infof("%+v", evt)
			newEvent.Obj = evt
			newEvent.Type = DeleteEvent
			newEvent.Key, err = cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
			if err == nil {
				logrus.Debugf("     ，   :DeleteEvent,   :%+v", evt)
			}
		},
	})
	return &Controller{
		clientset: client,
		informer:  informer,
		queue:     queue,
		config:    config,
	}
}

// Run starts the kubewatch controller
func (c *Controller) Run(stopCh 

ここで説明するのはeventオブジェクトの削除イベントであり、通常は期限切れのイベントのクリーンアップであり、直接破棄すればよい.
コード:https://github.com/gok8s/kube-eventalert