event-export源码解析噜噜噜的博客-

背景：在kubernetes里event记录了集群运行所遇到的各种大事件,有助于排错，但大量的事件如果都存储在etcd中，会带来较大的性能与容量压力，所以etcd中默认只保存最近1小时的，如果我们将该时间改大 会大大增加集群etcd的压力，所以我们需要将该数据存储到其他地方。经过一段时间的寻找 觉的k8s-stackdriver这个项目比较符合目前我的需求，但是它存储到的是promethues，但是目前我司给与promethues的资源并不是很足，并且保存的时间默认只有15天本，地存储也意味着Prometheus无法持久化数据，无法存储大量历史数据，同时也无法灵活扩展。，所以考虑使用了ES作为存储。但是es存储方案目前官方并不支持，只有通过fluentd去收集日志达到效果，所以我这边通过修改源码将存储直接改为es，这样就不需要借助fluentd去收集日志了。

本文主要是先分析event-export源码，之后再讲修改过程写出来。

程序入口 main.go

 sink, err := stackdriver.NewSdSinkFactory().CreateNew(strings.Split(*sinkOpts, " ")) if err != nil {    glog.Fatalf("Failed to initialize sink: %v", err) }

解释:初始化sink对象 并根据sink-opts传递进来的 参数设置对应的参数

参数列表（NewSdSinkFactory）：

flagSet

flushDelay

maxBufferSize

maxConcurrency

resourceModelVersion

endpoint

解析sinkOpts的参数（CreateNew）

 client, err := newKubernetesClient() if err != nil {    glog.Fatalf("Failed to initialize kubernetes client: %v", err) } eventExporter := newEventExporter(client, sink, *resyncPeriod)

newKubernetesClient 初始化go-client 接口

通过集群内部配置创建 k8s 配置信息，通过 KUBERNETES_SERVICE_HOST 和 KUBERNETES_SERVICE_PORT 环境变量方式获取

默认tokenfile rootCAFile 在/var/run/secrets/kubernetes.io/serviceaccount/

newEventExporter函数的作用 我们可以看他具体实现方法：

 func newEventExporter(client kubernetes.Interface, sink sinks.Sink, resyncPeriod time.Duration) *eventExporter {    return &eventExporter{       sink:    sink,       watcher: createWatcher(client, sink, resyncPeriod),    } }  func createWatcher(client kubernetes.Interface, sink sinks.Sink, resyncPeriod time.Duration) watchers.Watcher {    return events.NewEventWatcher(client, &events.EventWatcherConfig{       OnList:       sink.OnList,       ResyncPeriod: resyncPeriod,       Handler:      sink,    }) }

newEventExporter 返回了eventExporter这个结构体 该结构体包含sink 和watcher两个字段。

sink： 包含所有的参数配置等

watcher: 通过createwatcher函数 监听events resource

分解：createWatcher函数

NewEventWatcher配置了需要watche的目标对象为event，以及间隔多久watch一次

events.EventWatcherConfig 结构体包含三个字段

OnList OnListFunc event列表

ResyncPeriod watcher的时间间隔 默认为1分钟watch一次 。 通过resyncPeriod参数设置

Handler 包含三个字段OnAdd OnUpdate OnDelete

OnAdd： 主要作用就是将watch到的数据通过通道存入sdSink下的logEntryChannel字段，并且在promethues插入数据

OnUpdate： 更新logEntryChannel字段，并更新promethues里的数据

 go func() {    http.Handle("/metrics", promhttp.Handler())    glog.Fatalf("Prometheus monitoring failed: %v", http.ListenAndServe(*prometheusEndpoint, nil)) }()  stopCh := newSystemStopChannel() eventExporter.Run(stopCh)

 首先 通过gorouteing执行了匿名函数 该匿名函数 主要设置了promethues的/metrics路由并且监听localhost的本地端口 默认为80 通过启动参数prometheusEndpoint设置

 func newSystemStopChannel() chan struct{} {    ch := make(chan struct{})    go func() {       c := make(chan os.Signal)       signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)       sig := <-c       glog.Infof("Received signal %s, terminating", sig.String())        // Close stop channel to make sure every goroutine will receive stop signal.       close(ch)    }()     return ch }

newSystemStopChannel函数，通过gorouteing 执行匿名函数，当接收到SIGINT(2),SIGTERM(9)的时候 打印Received signal sig.String(), terminating的信号

eventExporter.Run(stopCh) 函数：

 func (e *eventExporter) Run(stopCh <-chan struct{}) {    utils.RunConcurrentlyUntil(stopCh, e.sink.Run, e.watcher.Run) }

run函数：

stopCh： 接收newSystemStopChannel传递过来的信号值通道

utils.RunConcurrentlyUntil 方法

 func RunConcurrentlyUntil(stopCh <-chan struct{}, funcs ...StoppableFunc) {    var wg sync.WaitGroup    for i := range funcs {       wg.Add(1)       f := funcs[i]       go func() {          defer wg.Done()          f(stopCh)       }()    }     <-stopCh     wg.Wait() }

并发运行e.sink.Run, e.watcher.Run

通过sync.WaitGroup解决同步阻塞等待的问题。

wg.Add 将计数器设置为1

执行e.sink.Run, e.watcher.Run两个函数

当执行完成的时候 将计数器设置减去1

在计数器不为0的时候Wait() 一直阻塞

只有当计数器为0的时候 或者接收到退出信号的时候才会退出。

e.sink.Run：

在eventExporter的sink字段 之后跳转到

interface.go下的Sink接口下的run方法

因为在newEventExporter函数里sink的值已经被赋值为sink, err := stackdriver.NewSdSinkFactory().CreateNew(strings.Split(*sinkOpts, ” “))

返回的值 createNew

并且CreateNew 函数return newSdSink(writer, clk, config, resourceModelFactory), nil

通过查看newSdSink 该函数返回的是sdSink结构体

所以e.sink.Run 就相当于函数拥有者为sdSink下的Run方法

 func (s *sdSink) Run(stopCh <-chan struct{}) {    glog.Info("Starting Stackdriver sink")    for {       select {       case entry := <-s.logEntryChannel:          s.currentBuffer = append(s.currentBuffer, entry)          if len(s.currentBuffer) >= s.config.MaxBufferSize {             s.flushBuffer()          } else if len(s.currentBuffer) == 1 {             s.setTimer()          }          break       case <-s.getTimerChannel():          s.flushBuffer()          break       case <-stopCh:          glog.Info("Stackdriver sink received stop signal, waiting for all requests to finish")          for i := 0; i < s.config.MaxConcurrency; i++ {             s.concurrencyChannel <- struct{}{}          }          glog.Info("All requests to Stackdriver finished, exiting Stackdriver sink")          return       }    } }

 run方法是一个for死循环 只有当收到newSystemStopChannel传递过来的信号时才会退出，通过s.logEntryChannel传递过来的日志，将该日志存储到s.currentBuffer里 当存储里的个数大于s.config.MaxBufferSize的值时 执行flushBuffer函数

 func (s *sdSink) flushBuffer() {    entries := s.currentBuffer    s.currentBuffer = nil    s.concurrencyChannel <- struct{}{}    go s.sendEntries(entries) }  func (s *sdSink) sendEntries(entries []*sd.LogEntry) {    glog.V(4).Infof("Sending %d entries to Stackdriver", len(entries))     written := s.writer.Write(entries, s.logName, s.sdResourceFactory.defaultMonitoredResource())    successfullySentEntryCount.Add(float64(written))     <-s.concurrencyChannel     glog.V(4).Infof("Successfully sent %d entries to Stackdriver", len(entries)) }

flushBuffer首先会清空s.concurrencyChannel里的值

之后将entries的值传递给sendEntries

sendEntries函数下通过successfullySentEntryCount将日志写入promethues

之后接收s.concurrencyChannel传递过来的值，这里主要解决接收到SIGINT，SIGTERM的时候等待所有请求完成之后再停止

当s.currentBuffer等于1时 充值计时器，默认为5s，如果5s内没有数据代表超时 直接执行flushBuffer函数

e.watcher.Run：

 func (w *watcher) Run(stopCh <-chan struct{}) {    w.reflector.Run()    <-stopCh } 

w.reflector.Run()会跳转到所属者为reflector下的Run函数

 func (r *Reflector) Run() {    glog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedType, r.resyncPeriod, r.name)    go wait.Until(func() {       if err := r.ListAndWatch(wait.NeverStop); err != nil {          utilruntime.HandleError(err)       }    }, r.period, wait.NeverStop) }

 声明一个匿名函数 每隔一段时间执行这个匿名函数

该函数下的主要方法就是ListAndWatch 也就是由它收集event的日志