152 lines
3.2 KiB
Go
152 lines
3.2 KiB
Go
package main
|
|
|
|
import (
|
|
"log"
|
|
"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
|
|
}
|
|
|
|
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 {
|
|
panic(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 {
|
|
panic(err)
|
|
}
|
|
writeNow = false
|
|
bp = nil
|
|
}
|
|
}
|
|
|
|
timer.Stop()
|
|
c.wg.Done()
|
|
}
|