今日我們將要使用go-elasticsearch來搭配telegram完成訊息發送。

目標

前面章節我們有介紹過elk的警報器功能，但是由於是使用免費版本，所以就只是單純的寫入索引，來模擬實際發送警報。現在我們打算透過go-elasticsearch+telegram來實現將訊息發送到使用者。

事前準備

一、 需申請 telegram bot，詳細申請步驟請參考連結
二、 使用 Kibana Alerting 建立範例要使用的索引it13-cpu。

實作

接下範例是使用警報器產生的索引it13-cpu，所以資料就不用過濾，直接以時間為區間來搜尋，Query DSL語法結構如下:

{
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "@timestamp": {
              "from": "2021-10-05T03:53:08Z",
              "include_lower": true,
              "include_upper": true,
              "to": "2021-10-05T03:54:08Z"
            }
          }
        }
      ]
    }
  },
  "from": 0,
  "size": 20
}

完整範例程式

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,
						},
					}},
				},
			},
		},
		"size": 20,
		"from": 0,
	}
	jsonBody, _ := json.Marshal(query)
	req := esapi.SearchRequest{
		Index: []string{"it13-cpu"}, // 索引名稱
		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)
	}

	for _, hit := range r["hits"].(map[string]interface{})["hits"].([]interface{}) {
		source := hit.(map[string]interface{})["_source"]
		message := source.(map[string]interface{})["message"].(string)
		if err = sendTelegramMsg(message); err != nil {
			fmt.Println("send msg error")
		}
	}
}

發送的訊息內容如下: