mirror of
https://github.com/Luzifer/continuous-spark.git
synced 2024-10-18 04:44:24 +00:00
108 lines
2 KiB
Go
108 lines
2 KiB
Go
|
package main
|
||
|
|
||
|
import (
|
||
|
"sync"
|
||
|
"time"
|
||
|
|
||
|
influx "github.com/influxdata/influxdb/client/v2"
|
||
|
"github.com/pkg/errors"
|
||
|
)
|
||
|
|
||
|
const (
|
||
|
influxWriteInterval = 10 * time.Second
|
||
|
)
|
||
|
|
||
|
type metricsSender struct {
|
||
|
batch influx.BatchPoints
|
||
|
batchLock sync.Mutex
|
||
|
client influx.Client
|
||
|
errs chan error
|
||
|
|
||
|
influxDB string
|
||
|
}
|
||
|
|
||
|
func NewMetricsSender(influxHost, influxUser, influxPass, influxDatabase string) (*metricsSender, error) {
|
||
|
out := &metricsSender{
|
||
|
errs: make(chan error, 10),
|
||
|
influxDB: influxDatabase,
|
||
|
}
|
||
|
return out, out.initialize(influxHost, influxUser, influxPass)
|
||
|
}
|
||
|
|
||
|
func (m *metricsSender) Errors() <-chan error {
|
||
|
return m.errs
|
||
|
}
|
||
|
|
||
|
func (m *metricsSender) ForceTransmit() error {
|
||
|
return errors.Wrap(m.transmit(), "Unable to transmit recorded points")
|
||
|
}
|
||
|
|
||
|
func (m *metricsSender) RecordPoint(name string, tags map[string]string, fields map[string]interface{}) error {
|
||
|
pt, err := influx.NewPoint(name, tags, fields, time.Now())
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
m.batchLock.Lock()
|
||
|
defer m.batchLock.Unlock()
|
||
|
m.batch.AddPoint(pt)
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (m *metricsSender) resetBatch() error {
|
||
|
b, err := influx.NewBatchPoints(influx.BatchPointsConfig{
|
||
|
Database: m.influxDB,
|
||
|
})
|
||
|
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
m.batch = b
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (m *metricsSender) sendLoop() {
|
||
|
for range time.Tick(influxWriteInterval) {
|
||
|
|
||
|
if err := m.transmit(); err != nil {
|
||
|
m.errs <- err
|
||
|
}
|
||
|
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (m *metricsSender) transmit() error {
|
||
|
m.batchLock.Lock()
|
||
|
defer m.batchLock.Unlock()
|
||
|
|
||
|
if err := m.client.Write(m.batch); err != nil {
|
||
|
return errors.Wrap(err, "Unable to write recorded points")
|
||
|
}
|
||
|
m.resetBatch()
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (m *metricsSender) initialize(influxHost, influxUser, influxPass string) error {
|
||
|
influxClient, err := influx.NewHTTPClient(influx.HTTPConfig{
|
||
|
Addr: influxHost,
|
||
|
Username: influxUser,
|
||
|
Password: influxPass,
|
||
|
Timeout: 2 * time.Second,
|
||
|
})
|
||
|
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
m.client = influxClient
|
||
|
if err := m.resetBatch(); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
go m.sendLoop()
|
||
|
|
||
|
return nil
|
||
|
}
|