cpu_ctrl/consumer/influx_stat.go

package consumer

import (
	"amuz.es/src/infra/cpu_ctrl/util"
	"amuz.es/src/infra/cpu_ctrl/processor"
	"amuz.es/src/infra/cpu_ctrl/logger"
	"github.com/influxdata/influxdb/client/v2"
	"time"
	"strconv"
)

var (
	influxLogger = logger.NewLogger("influx")
)

type influxMetric struct {
	host              string
	processorCount    int
	handler           util.Handler
	fanSpeedConsumer  chan processor.FanspeedInfo
	tempetureConsumer chan processor.TempetureInfo
}

type InfluxMetric interface {
	FanSpeedConsumer() chan<- processor.FanspeedInfo
	TempetureConsumer() chan<- processor.TempetureInfo
	StartLogging()
}

func NewInfluxMetric(host string, processorCount int, handler util.Handler) InfluxMetric {
	return &influxMetric{
		host:              host,
		processorCount:    processorCount,
		handler:           handler,
		fanSpeedConsumer:  make(chan processor.FanspeedInfo, 1),
		tempetureConsumer: make(chan processor.TempetureInfo, 1),
	}
}

func (m *influxMetric) FanSpeedConsumer() chan<- processor.FanspeedInfo   { return m.fanSpeedConsumer }
func (m *influxMetric) TempetureConsumer() chan<- processor.TempetureInfo { return m.tempetureConsumer }

func (m *influxMetric) StartLogging() {
	defer m.handler.DecreaseWait()

	defer func() {
		if err := recover(); err != nil {
			m.handler.NotifyError(err.(error))
		}
	}()
	defer close(m.fanSpeedConsumer)
	defer close(m.tempetureConsumer)

	defer influxLogger.Info("Metric logging stopped")
	influxLogger.Info("Metric logging started")

	var influxDbConn client.Client
	for {
		conn, err := client.NewUDPClient(client.UDPConfig{Addr: m.host,})
		if err != nil {
			influxLogger.Error(err)
		} else {
			influxDbConn = conn
			break
		}
		cont := time.After(1 * time.Second)
		select {
		case <-cont:
			continue
		case <-m.handler.Done():
			return
		}
	}

	// Create a new point batch
	batchPoint, err := client.NewBatchPoints(client.BatchPointsConfig{
		Database:  "core",
		Precision: "s",
	})
	if err != nil {
		panic(err)
	}
	ticker := time.Tick(time.Second)
	fanspeedList, tempetureList := make([]processor.FanspeedInfo, 0, 0), make([]processor.TempetureInfo, 0, 0)

	for {
		select {
		case <-ticker:
			fanspeedListTemp, tempetureListTemp := fanspeedList, tempetureList
			fanspeedList, tempetureList = make([]processor.FanspeedInfo, 0, 0), make([]processor.TempetureInfo, 0, 0)
			go m.sendPoint(influxDbConn, batchPoint, fanspeedListTemp, tempetureListTemp)
		case changedSpeed := <-m.fanSpeedConsumer:
			fanspeedList = append(fanspeedList, changedSpeed)

		case changedTempeture := <-m.tempetureConsumer:
			tempetureList = append(tempetureList, changedTempeture)
		case <-m.handler.Done():
			return
		}
	}
}
func (m *influxMetric) sendPoint(
	influxDbConn client.Client,
	batchPoint client.BatchPoints,
	fanspeeds []processor.FanspeedInfo, tempetures []processor.TempetureInfo) {
	pointList := make([]*client.Point, 0, 0)

	for _, fanspeed := range fanspeeds {
		if point, err := m.getFanspeedPoint(fanspeed); err == nil {
			pointList = append(pointList, point)
		} else {
			influxLogger.Debugf("id %d speed err %s", fanspeed.Id, err)
		}
	}

	for _, tempeture := range tempetures {
		if point, err := m.getTempeturePoint(tempeture); err == nil {
			pointList = append(pointList, point)
		} else {
			influxLogger.Debugf("id %d tempeture err %s", tempeture.Id, err)
		}
	}

	if len(pointList) == 0 {
		return
	}

	batchPoint.AddPoints(pointList)
	if err := influxDbConn.Write(batchPoint); err != nil {
		influxLogger.Warn(err)
	}

}

func (m *influxMetric) getTempeturePoint(info processor.TempetureInfo) (*client.Point, error) {

	// Create a point and add to batch
	tags := map[string]string{"processor": strconv.Itoa(info.Id)}
	fields := map[string]interface{}{
		"tempeture": info.Tempeture,
	}

	return client.NewPoint("processor_tempeture", tags, fields, info.At)
}

func (m *influxMetric) getFanspeedPoint(info processor.FanspeedInfo) (*client.Point, error) {

	// Create a point and add to batch
	tags := map[string]string{"processor": strconv.Itoa(info.Id)}
	fields := map[string]interface{}{
		"noob": info.FanSpeed,
	}

	return client.NewPoint("processor_cooling_fanspeed", tags, fields, info.At)
}