向esm-gateway输入数据流程说明文档

1、背景

由于一些业务需要输出数据进入Influxdb直接使用ESM告警，因此需要一种手段直接绕过ESM-agent直接输入到Influxdb数据库。基于此，esm仅仅提供writers配置，用户获取到writers的配置后，自行向esm-gateway写入数据。

2、设计

2.1 流程

2.2 说明

1. 使用Unix Socket技术，用户通过unix socket请求向esm-agent请求writers的配置
2. esm-agent返回writers配置给用户
3. 用户通过writers配置向esm-gateway写数据并注册schema

3、流程

3.1 Step1: 从esm获取writers配置

通过 UnixSock 获取 Writers 配置

3.1.1 curl

curl --unix-socket /var/run/esm-agent/esm-agent.sock http:/writer

3.1.2 go

package main
import (
    "context"
    "fmt"
    "io"
    "net"
    "net/http"
    "os"
)
func main() {
    client := http.Client{
        Transport: &http.Transport{
            DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
                return net.Dial("unix", "/var/run/esm-agent/esm-agent.sock")
            },
        },
    }
    resp, err := client.Get("http://localhost/writer")
    if err != nil {
        fmt.Println("error: " + err.Error())
        os.Exit(1)
    }
    io.Copy(os.Stdout, resp.Body)
}

3.1.3 python

依赖类库requests_unixsocket

推荐安装方式：

pip install requests_unixsocket

以下为执行代码：

import requests_unixsocket
if __name__=="__main__":
    url = "http+unix://%2Fvar%2Frun%2Fesm-agent%2Fesm-agent.sock"
    res = requests_unixsocket.get(url+"/writer")
    print res.text

3.1.4响应格式：

[
  {
    "database": "esm",
    "addr": "esm.gateway.wg.elenet.me:9100",
    "payload_size": 1024,
    "precision": "s"
  }
]

3.2 Step2: 输出到esm-gateway

3.2.1 go 输出到influxdb

官方demo：

import (
    "time"
    "log"
    client "github.com/influxdata/influxdb/client/v2"
)
func writePoints(clnt client.Client) {
    sampleSize := 1000
    bp, err := client.NewBatchPoints(client.BatchPointsConfig{
        Database:  "systemstats",
        Precision: "us",
    })
    if err != nil {
        log.Fatal(err)
    }
    rand.Seed(time.Now().UnixNano())
    for i := 0; i < sampleSize; i++ {
        regions := []string{"us-west1", "us-west2", "us-west3", "us-east1"}
        tags := map[string]string{
            "cpu":    "cpu-total",
            "host":   fmt.Sprintf("host%d", rand.Intn(1000)),
            "region": regions[rand.Intn(len(regions))],
        }
        idle := rand.Float64() * 100.0
        fields := map[string]interface{}{
            "idle": idle,
            "busy": 100.0 - idle,
        }
        pt, err := client.NewPoint(
            "cpu_usage",
            tags,
            fields,
            time.Now(),
        )
        if err != nil {
            log.Fatal(err)
        }
        bp.AddPoint(pt)
    }
    if err := clnt.Write(bp); err != nil {
        log.Fatal(err)
    }
}

3.2.2 python 输出到influxdb

依赖类库influxdb

推荐安装方式：

pip install influxdb

以下为插入的执行代码：

from influxdb import InfluxDBClient
json_body = [
    {
        "measurement": "cpu_load_short",
        "tags": {
            "host": "server01",
            "region": "us-west"
        },
        "time": "2009-11-10T23:00:00Z",
        "fields": {
            "value": 0.64
        }
    }
]
client = InfluxDBClient('localhost', 8086, 'root', 'root', 'example')
client.write_points(json_body)

3.3 Step3: 注册schema

输入的Matrix格式注意点

3.3.1 命名方式

全小写，用_分割单词

3.3.2 matrix name

name必须要明确！

3.3.3 tag

tag 必须包含 host 和 hostgroup 两个，其余可以有自定义tag，但是必须明确类型数量，且不能过多！

host为主机名

hostgroup为一个或多个appid，用,隔开！

ps： 注册schema时若有问题请联系 ping.liush !