go監(jiān)控方案的實(shí)現(xiàn)是怎樣的

這期內(nèi)容當(dāng)中小編將會(huì)給大家?guī)?lái)有關(guān)go監(jiān)控方案的實(shí)現(xiàn)是怎樣的，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

metrics 客戶端

數(shù)據(jù)采集使用go-metrics

傳輸使用UDP, 仿StatsD上傳采集數(shù)據(jù), InfluxDB進(jìn)行數(shù)據(jù)存儲(chǔ), Grafana進(jìn)行展示。

實(shí)現(xiàn)github 地址

https://github.com/solate/metrics

該地址有已經(jīng)改好的配置文件可以直接使用。

使用的all-in-one :

git docker-statsd-influxdb-grafana

docker hub 地址

數(shù)據(jù)封裝

//掛載配置文件，已修改statsd模版
docker run --ulimit nofile=66000:66000  -v /root/telegraf.conf:/etc/telegraf/telegraf.conf   -d   --name docker-statsd-influxdb-grafana   -p 3003:3003   -p 3004:8888   -p 8086:8086   -p 8125:8125/udp   samuelebistoletti/docker-statsd-influxdb-grafana:latest

register

register 使用的name 必須是不同的

telegraf 配置修改

將 [[inputs.statsd]] 部分配置打開(kāi), 修改templates為:

   templates = [
      "* measurement.measurement.field"
   ]

表示傳值prefix.name.field 最好表示為 prefix_name field

代碼實(shí)現(xiàn)

package client

import (
	"bufio"
	"bytes"
	"github.com/rcrowley/go-metrics"
	"log"
	"net"
	"strconv"
	"strings"
	"time"
)

const (
	// UDP packet limit, see
	// https://en.wikipedia.org/wiki/User_Datagram_Protocol#Packet_structure
	UDP_MAX_PACKET_SIZE int = 64 * 1024
)

// Config provides a container with configuration parameters for
// the StatsD exporter
type Config struct {
	Network       string           // Network: tcp, udp.
	Addr          string           // Network address to connect to | 地址
	Registry      metrics.Registry // Registry to be exported | metrics注冊(cè)
	FlushInterval time.Duration    // Flush interval | 刷新間隔時(shí)間
	Prefix        string           // Prefix to be prepended to metric names | 前綴名字
	Rate          float32          // Rate
	Tags          string           // tag //TODO

	conn net.Conn
}

func StatsD(r metrics.Registry, d time.Duration, prefix string, network string, addr string, rate float32) {

	conn, err := net.Dial(network, addr)
	if err != nil {
		panic("conn remote err!")
	}

	StatsDWithConfig(Config{
		Network:       network,
		Addr:          addr,
		Registry:      r,
		FlushInterval: d,
		Prefix:        prefix,
		Rate:          rate,
		conn:          conn,
	})

}

// WithConfig is a blocking exporter function
func StatsDWithConfig(c Config) {
	for _ = range time.Tick(c.FlushInterval) {
		if err := statsd(&c); err != nil {
			log.Println(err)
			c.conn.Close()
		}
	}
}

func statsd(c *Config) (err error) {

	w := bufio.NewWriter(c.conn)

	c.Registry.Each(func(name string, i interface{}) {
		switch metric := i.(type) {
		case metrics.Counter:
			ms := metric.Snapshot()
			w.Write(statsdLine(c.Prefix, name, "", ms.Count(), "|c", c.Rate))
		case metrics.Gauge:
			ms := metric.Snapshot()
			w.Write(statsdLine(c.Prefix, name, "", ms.Value(), "|g", c.Rate))
		case metrics.GaugeFloat64:
			ms := metric.Snapshot()
			w.Write(statsdLine(c.Prefix, name, "", ms.Value(), "|g", c.Rate))
		case metrics.Histogram:
			ms := metric.Snapshot()
			ps := ms.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999, 0.9999})

			fields := make([][]byte, 12)
			fields = append(fields, statsdLine(c.Prefix, name, "count", ms.Count(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "max", ms.Max(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "mean", ms.Mean(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "min", ms.Min(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "stddev", ms.StdDev(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "variance", ms.Variance(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p50", ps[0], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p75", ps[1], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p95", ps[2], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p99", ps[3], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p999", ps[4], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p9999", ps[5], "|g", c.Rate))

			buf := bytes.Join(fields, []byte{})
			w.Write(buf)
		case metrics.Meter:
			ms := metric.Snapshot()
			fields := make([][]byte, 5)
			fields = append(fields, statsdLine(c.Prefix, name, "count", ms.Count(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "m1", ms.Rate1(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "m5", ms.Rate5(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "m15", ms.Rate15(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "mean", ms.RateMean(), "|g", c.Rate))

			buf := bytes.Join(fields, []byte{})
			w.Write(buf)

		case metrics.Timer:
			ms := metric.Snapshot()
			ps := ms.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999, 0.9999})

			fields := make([][]byte, 12)
			fields = append(fields, statsdLine(c.Prefix, name, "count", ms.Count(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "max", ms.Max(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "mean", ms.Mean(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "min", ms.Min(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "stddev", ms.StdDev(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "variance", ms.Variance(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p50", ps[0], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p75", ps[1], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p95", ps[2], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p99", ps[3], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p999", ps[4], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "p9999", ps[5], "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "m1", ms.Rate1(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "m5", ms.Rate5(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "m15", ms.Rate15(), "|g", c.Rate))
			fields = append(fields, statsdLine(c.Prefix, name, "mean", ms.RateMean(), "|g", c.Rate))

			buf := bytes.Join(fields, []byte{})

			w.Write(buf)

			//case metrics.Healthcheck:
			//	metric.Check()
			//	log.Printf("healthcheck %s\n", name)
			//	log.Printf("  error:       %v\n", metric.Error())
			//
			//case metrics.EWMA:
			//case metrics.Sample:

		}

		w.Flush()
	})

	return
}

//構(gòu)造發(fā)送line
func statsdLine(prefix, name, field string, value interface{}, suffix string, rate float32) []byte {

	//<metricname>:<value>|<type>|@<rate>
	var buffer bytes.Buffer

	//buf := make([]byte, UDP_MAX_PACKET_SIZE)

	//添加前綴
	if prefix != "" {
		//buf = append(buf, prefix...)
		//buf = append(buf, '.')
		buffer.WriteString(prefix)
		buffer.WriteString(".")
	} else {
		buffer.WriteString("statsd")
		buffer.WriteString(".")
	}

	////添加名稱
	//buf = append(buf, name...)
	//buf = append(buf, ':')

	//將name注冊(cè)中的'.'替換成'_', 配合telegraf修改模版,防止將數(shù)據(jù)庫(kù)名字改為屬性
	if strings.Contains(name, ".") {
		name = strings.ReplaceAll(name, ".", "_")
	}
	//添加名稱
	buffer.WriteString(name)

	if field != "" {
		buffer.WriteString(".")
		buffer.WriteString(field)
	}
	buffer.WriteString(":")

	buf := buffer.Bytes()

	switch v := value.(type) {
	case string:
		buf = append(buf, v...)
	case int64:
		buf = strconv.AppendInt(buf, v, 10)
	case float64:
		buf = strconv.AppendFloat(buf, v, 'f', -1, 64)
	default:
		return nil
	}

	if suffix != "" {
		buf = append(buf, suffix...)
	}

	if rate != 0 && rate < 1 {
		buf = append(buf, "|@"...)
		buf = strconv.AppendFloat(buf, float64(rate), 'f', 6, 32)
	}

	buf = append(buf, "\n"...) //每一行打一個(gè)回車，telegraf 是使用回車進(jìn)行讀取的
	return buf

}

上述就是小編為大家分享的go監(jiān)控方案的實(shí)現(xiàn)是怎樣的了，如果剛好有類似的疑惑，不妨參照上述分析進(jìn)行理解。如果想知道更多相關(guān)知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道。