Day29 go-elasticsearch(三)

上一篇我们已经完成了使用telegram来发送讯息，那麽今日我们决定不使用Kibana Alerting，改用metrics资料来看要如何实作，才能达到类似警报器的功能。

目标

我们将使用索引metricbeat-*的的原始数据资料，搭配使用Query DSL的统计工具 Aggregation 语法，来完成警报讯息发送功能。

更多详细的Aggregation请参考

实作

接下范例是使用的索引是metricbeat-*+Aggregation，语法结构如下:

{
  "aggs": {
    "avg_data": {
      "avg": {
        "field": "docker.cpu.total.pct"
      }
    }
  },
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "@timestamp": {
              "from": "2021-10-06T02:55:50Z",
              "include_lower": true,
              "include_upper": true,
              "to": "2021-10-06T02:56:04Z"
            }
          }
        },
        {
          "match": {
            "event.dataset": {
              "query": "docker.cpu"
            }
          }
        },
        {
          "match": {
            "container.name": {
              "query": "es01-test"
            }
          }
        }
      ]
    }
  },
  "from": 0,
  "size": 100
}

完整范例程序

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/elastic/go-elasticsearch/v8"
	"github.com/elastic/go-elasticsearch/v8/esapi"
	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api"
)

var (
	es  *elasticsearch.Client
	bot *tgbotapi.BotAPI
)

func setElkClient() {
	var err error
	cfg := elasticsearch.Config{
		Addresses: []string{"http://127.0.0.1:9200"},
	}
	es, err = elasticsearch.NewClient(cfg)
	if err != nil {
		panic(err)
	}
}

func setNewBotAPI() {
	var err error
	bot, err = tgbotapi.NewBotAPI("youToken")
	if err != nil {
		panic(err)
	}
	bot.Debug = false
}

func main() {
    //初始设定
	setElkClient()
	setNewBotAPI()
    //搜寻并发送讯息
	searchRequest()
}

func sendTelegramMsg(msg string) error {
	NewMsg := tgbotapi.NewMessage(chatID, msg) //传送讯息给使用者
	NewMsg.ParseMode = tgbotapi.ModeHTML
	_, err := bot.Send(NewMsg)
	if err != nil {
		return err
	}
	return nil
}

func searchRequest() {
	var r map[string]interface{}

	t := time.Now().UTC().Add(-1 * time.Minute).Format(time.RFC3339)
	nt := time.Now().UTC().Format(time.RFC3339)

	query := map[string]interface{}{
		"query": map[string]interface{}{
			"bool": map[string]interface{}{
				"must": []map[string]interface{}{
					{"range": map[string]interface{}{
						"@timestamp": map[string]interface{}{
							"from":          t,
							"to":            nt,
							"include_lower": true,
							"include_upper": true,
						},
					}},
					{"match": map[string]interface{}{
						"event.dataset": map[string]interface{}{
							"query": "docker.cpu",
						},
					}},
					{"match": map[string]interface{}{
						"container.name": map[string]interface{}{
							"query": "es01-test",
						},
					}},
				},
			},
		},
		"aggs": map[string]interface{}{
			"avg_data": map[string]interface{}{
				"avg": map[string]interface{}{
					"field": "docker.cpu.total.pct",
				},
			},
		},
		"size": 100,
		"from": 0,
	}

	jsonBody, _ := json.Marshal(query)
	req := esapi.SearchRequest{
		Index: []string{"metricbeat-*"}, // 索引名称
		Body:  bytes.NewReader(jsonBody),
	}
	res, err := req.Do(context.Background(), es)
	if err != nil {
		panic(err)
	}
	defer res.Body.Close()
	if err = json.NewDecoder(res.Body).Decode(&r); err != nil {
		panic(err)
	}
	aggregations := r["aggregations"].(map[string]interface{})["avg_data"]
	avg := aggregations.(map[string]interface{})["value"].(float64)
	if avg <= 40 {
		msg := fmt.Sprintf("es01-test CPU %v%s异常 for go-elasticsearch", avg, "%")
		if err = sendTelegramMsg(msg); err != nil {
			fmt.Println("send msg error")
		}
	}
}

发送的讯息内容如下:

总结

今日我们的go-elasticsearch实作终於完成，经过Aggregation语法使用，我们可以发现警报器产生索引资料和metrics原始数据使用上的差别，在警报器上已经把一些较复杂性的语法都包装好，方便快速使用，而且透过警报器可以快速的修改警报发送内容，不必去修改程序并且重新部署。通过今天的简单的范例程序，我们了解在使用metrics资料，会需要组合许多的条件语法，不是使用单一的DSL语法就能达到想要的结果，而大幅增加进入的门槛。当然如果你是用付费版本的elk，在讯息通知上就不用这麽麻烦啦。