背景：在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的日志