【ES三周年】Informer实战之持久化K8s事件至ElasticSearch

 docker run --name es01 -d -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" -e  "ES_JAVA_OPTS=-Xms512m -Xmx512m" elasticsearch:latest
 ​
 # 使用head客户端链接
 docker pull mobz/elasticsearch-head:5
 # 启动header 容器
 docker run -d --name my-es_admin -p 9100:9100 mobz/elasticsearch-head:5
 ​
 ​

 # 进入容器
 $ docker exec -it es01 /bin/bash
 ​
 # 进入容器后设置参数
 # http.cors.enabled: true
 # http.cors.allow-origin: "*"
 echo 'http.cors.enabled: true' >> config/elasticsearch.yml
 echo 'http.cors.allow-origin: "*"' >> config/elasticsearch.yml
 # 设置完成，退出后重启容器
 docker restart es01

 var client *elastic.Client
 var host = "http://127.0.0.1:9200/"
 ​
 //初始化
 func init() {
   errorlog := log.New(os.Stdout, "APP", log.LstdFlags)
   var err error
   // 这个地方有个小坑 不加上elastic.SetSniff(false) 会连接不上
   client, err = elastic.NewClient(elastic.SetSniff(false), elastic.SetErrorLog(errorlog), elastic.SetURL(host))
   if err != nil {
     panic(err)
   }
   info, code, err := client.Ping(host).Do(context.Background())
   if err != nil {
     panic(err)
   }
   fmt.Printf("Elasticsearch returned with code %d and version %s\n", code, info.Version.Number)
 ​
   esversion, err := client.ElasticsearchVersion(host)
   if err != nil {
     panic(err)
   }
   fmt.Printf("Elasticsearch version %s\n", esversion)
 ​
 }
 ​
 func Must(err error) {
   if err != nil {
     panic(err)
   }
 }
 ​
 func main() {
   rand.Seed(time.Now().UnixNano())
   config, err := clientcmd.BuildConfigFromFlags("", "/Users/xuel/.kube/config")
   Must(err)
 ​
   clientset, err := kubernetes.NewForConfig(config)
   Must(err)
   sharedInformers := informers.NewSharedInformerFactory(clientset, 0)
   stopChan := make(chan struct{})
   defer close(stopChan)
 ​
   // 在此使用event informer，
   eventInformer := sharedInformers.Events().V1beta1().Events().Informer()
   addChan := make(chan v1beta1.Event)
   deleteChan := make(chan v1beta1.Event)
   eventInformer.AddEventHandlerWithResyncPeriod(cache.ResourceEventHandlerFuncs{
     AddFunc: func(obj interface{}) {
       unstructObj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
       Must(err)
       event := &v1beta1.Event{}
       err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructObj, event)
       Must(err)
       addChan <- *event
     },
     UpdateFunc: func(oldObj, newObj interface{}) {
     },
     DeleteFunc: func(obj interface{}) {
     },
   }, 0)
 ​
   go func() {
     for {
       select {
       case event := <-addChan:
         str, err := json.Marshal(&event)
         Must(err)
         fmt.Printf("插入k8s事件内容：%s", string(str))
         esinsert(str)
         break
       case <-deleteChan:
         break
       }
     }
   }()
   eventInformer.Run(stopChan)
 }
 ​
 func esinsert(str []byte) {
   index := "k8s_informer"
   dbtype := "doc"
 ​
   put1, err := client.Index().
     Index(index).
     Type(dbtype).
     Id("1").BodyString(string(str)).
     Do(context.Background())
   if err != nil {
     fmt.Println("Insert es error: %s", err)
   }
   fmt.Println("insert success", put1)
 }
 ​

 curl  -H "Content-Type: application/json" -XGET 'http://127.0.0.1:9200/k8s_informer/doc/_search?pretty' -d '{"query":{"match_all":{}}}'