yanic/stats_db.go

package main

import (
	"log"
	"strconv"
	"sync"
	"time"

	"github.com/FreifunkBremen/respond-collector/data"
	"github.com/influxdata/influxdb/client/v2"
)

const (
	batchDuration = time.Second * 5
)

type StatsDb struct {
	points chan *client.Point
	wg     sync.WaitGroup
	client client.Client
}

func NewStatsDb() *StatsDb {
	// Make client
	c, err := client.NewHTTPClient(client.HTTPConfig{
		Addr:     config.Influxdb.Addr,
		Username: config.Influxdb.Username,
		Password: config.Influxdb.Password,
	})

	if err != nil {
		panic(err)
	}

	db := &StatsDb{
		client: c,
		points: make(chan *client.Point, 500),
	}

	// start worker
	db.wg.Add(1)
	go db.worker()

	return db
}

func (c *StatsDb) Add(stats *data.Statistics) {
	tags := map[string]string{
		"nodeid": stats.NodeId,
	}
	fields := map[string]interface{}{
		"load":              stats.LoadAverage,
		"idletime":          int64(stats.Idletime),
		"uptime":            int64(stats.Uptime),
		"processes.running": stats.Processes.Running,
		"clients.wifi":      stats.Clients.Wifi,
		"clients.wifi24":    stats.Clients.Wifi24,
		"clients.wifi5":     stats.Clients.Wifi5,
		"clients.total":     stats.Clients.Total,
		"memory.buffers":    stats.Memory.Buffers,
		"memory.cached":     stats.Memory.Cached,
		"memory.free":       stats.Memory.Free,
		"memory.total":      stats.Memory.Total,
	}

	if t := stats.Traffic.Rx; t != nil {
		fields["traffic.rx.bytes"] = int64(t.Bytes)
		fields["traffic.rx.packets"] = t.Packets
	}
	if t := stats.Traffic.Tx; t != nil {
		fields["traffic.tx.bytes"] = int64(t.Bytes)
		fields["traffic.tx.packets"] = t.Packets
		fields["traffic.tx.dropped"] = t.Dropped
	}
	if t := stats.Traffic.Forward; t != nil {
		fields["traffic.forward.bytes"] = int64(t.Bytes)
		fields["traffic.forward.packets"] = t.Packets
	}
	if t := stats.Traffic.MgmtRx; t != nil {
		fields["traffic.mgmt_rx.bytes"] = int64(t.Bytes)
		fields["traffic.mgmt_rx.packets"] = t.Packets
	}
	if t := stats.Traffic.MgmtTx; t != nil {
		fields["traffic.mgmt_tx.bytes"] = int64(t.Bytes)
		fields["traffic.mgmt_tx.packets"] = t.Packets
	}
	if w := stats.Wireless; w != nil {
		if time := w.Airtime24; time != nil {
			fields["airtime24.active"] = time.Active
			fields["airtime24.busy"] = time.Busy
			fields["airtime24.tx"] = time.Tx
			fields["airtime24.rx"] = time.Rx
			fields["airtime24.noise"] = time.Noise
			fields["airtime24.frequency"] = time.Frequency
			tags["frequency24"] = strconv.Itoa(int(time.Frequency))
		}
		if time := w.Airtime5; time != nil {
			fields["airtime5.active"] = time.Active
			fields["airtime5.busy"] = time.Busy
			fields["airtime5.tx"] = time.Tx
			fields["airtime5.rx"] = time.Rx
			fields["airtime5.noise"] = time.Noise
			fields["airtime5.frequency"] = time.Frequency
			tags["frequency5"] = strconv.Itoa(int(time.Frequency))
		}
	}

	point, err := client.NewPoint("node", tags, fields, time.Now())
	if err != nil {
		panic(err)
	}
	c.points <- point
}

func (c *StatsDb) Close() {
	close(c.points)
	c.wg.Wait()
	c.client.Close()
}

// stores data points in batches into the influxdb
func (c *StatsDb) worker() {
	bpConfig := client.BatchPointsConfig{
		Database:  config.Influxdb.Database,
		Precision: "m",
	}

	var bp client.BatchPoints
	var err error
	var writeNow, closed bool
	timer := time.NewTimer(batchDuration)

	for !closed {

		// wait for new points
		select {
		case point, ok := <-c.points:
			if ok {
				if bp == nil {
					// create new batch
					timer.Reset(batchDuration)
					if bp, err = client.NewBatchPoints(bpConfig); err != nil {
						log.Fatal(err)
					}
				}
				bp.AddPoint(point)
			} else {
				closed = true
			}
		case <-timer.C:
			if bp == nil {
				timer.Reset(batchDuration)
			} else {
				writeNow = true
			}
		}

		// write batch now?
		if bp != nil && (writeNow || closed) {
			log.Println("saving", len(bp.Points()), "points")

			if err = c.client.Write(bp); err != nil {
				log.Fatal(err)
			}
			writeNow = false
			bp = nil
		}
	}

	timer.Stop()
	c.wg.Done()
}