telegraf/accumulator.go

105 lines
1.8 KiB
Go
Raw Permalink Normal View History

package telegraf
import (
"fmt"
"sort"
"strings"
"sync"
"time"
"github.com/influxdb/influxdb/client"
)
// BatchPoints is used to send a batch of data in a single write from telegraf
// to influx
type BatchPoints struct {
mu sync.Mutex
client.BatchPoints
Debug bool
Prefix string
Config *ConfiguredPlugin
}
// Add adds a measurement
2015-05-29 20:25:48 +00:00
func (bp *BatchPoints) Add(measurement string, val interface{}, tags map[string]string) {
bp.mu.Lock()
defer bp.mu.Unlock()
2015-05-29 20:25:48 +00:00
measurement = bp.Prefix + measurement
if bp.Config != nil {
if !bp.Config.ShouldPass(measurement, tags) {
return
}
}
if bp.Debug {
var tg []string
for k, v := range tags {
tg = append(tg, fmt.Sprintf("%s=\"%s\"", k, v))
}
sort.Strings(tg)
2015-05-29 20:25:48 +00:00
fmt.Printf("> [%s] %s value=%v\n", strings.Join(tg, " "), measurement, val)
}
bp.Points = append(bp.Points, client.Point{
2015-05-29 20:25:48 +00:00
Measurement: measurement,
Tags: tags,
Fields: map[string]interface{}{
"value": val,
},
})
}
// AddValuesWithTime adds a measurement with a provided timestamp
func (bp *BatchPoints) AddValuesWithTime(
2015-05-29 20:25:48 +00:00
measurement string,
values map[string]interface{},
tags map[string]string,
timestamp time.Time,
) {
bp.mu.Lock()
defer bp.mu.Unlock()
2015-05-29 20:25:48 +00:00
measurement = bp.Prefix + measurement
if bp.Config != nil {
if !bp.Config.ShouldPass(measurement, tags) {
return
}
}
if bp.Debug {
var tg []string
for k, v := range tags {
tg = append(tg, fmt.Sprintf("%s=\"%s\"", k, v))
}
var vals []string
for k, v := range values {
vals = append(vals, fmt.Sprintf("%s=%v", k, v))
}
sort.Strings(tg)
sort.Strings(vals)
2015-05-29 20:25:48 +00:00
fmt.Printf("> [%s] %s %s\n", strings.Join(tg, " "), measurement, strings.Join(vals, " "))
}
bp.Points = append(bp.Points, client.Point{
2015-05-29 20:25:48 +00:00
Measurement: measurement,
Tags: tags,
Fields: values,
Time: timestamp,
})
}