Add NatsSink to send data via the NATS messaging system

This commit is contained in:
Thomas Roehl 2021-03-29 14:14:38 +02:00
parent 162f0f14d6
commit 0a70ca1f13
2 changed files with 61 additions and 0 deletions

View File

@ -27,6 +27,7 @@ var Collectors = map[string]collectors.MetricGetter{
var Sinks = map[string]sinks.SinkFuncs{ var Sinks = map[string]sinks.SinkFuncs{
"influxdb": &sinks.InfluxSink{}, "influxdb": &sinks.InfluxSink{},
"stdout": &sinks.StdoutSink{}, "stdout": &sinks.StdoutSink{},
"nats": &sinks.NatsSink{},
} }
// Structure of the configuration file // Structure of the configuration file

60
sinks/natsSink.go Normal file
View File

@ -0,0 +1,60 @@
package sinks
import (
"bytes"
"fmt"
protocol "github.com/influxdata/line-protocol"
nats "github.com/nats-io/nats.go"
"log"
"time"
)
type NatsSink struct {
Sink
client *nats.Conn
encoder *protocol.Encoder
buffer *bytes.Buffer
}
func (s *NatsSink) Init(host string, port string, user string, password string, database string) error {
s.host = host
s.port = port
s.user = user
s.password = password
s.database = database
// Setup Influx line protocol
s.buffer = &bytes.Buffer{}
s.buffer.Grow(1025)
s.encoder = protocol.NewEncoder(s.buffer)
s.encoder.SetPrecision(time.Second)
s.encoder.SetMaxLineBytes(1024)
// Setup infos for connection
uinfo := nats.UserInfo(s.user, s.password)
uri := fmt.Sprintf("nats://%s:%s", s.host, s.port)
nc, err := nats.Connect(uri, uinfo)
if err != nil {
log.Fatal(err)
return err
}
s.client = nc
return nil
}
func (s *NatsSink) Write(measurement string, tags map[string]string, fields map[string]interface{}, t time.Time) error {
m, err := protocol.New(measurement, tags, fields, t)
if err != nil {
log.Print(err)
return err
}
_, err = s.encoder.Encode(m)
if err != nil {
log.Print(err)
return err
}
s.client.Publish(s.database, s.buffer.Bytes())
return nil
}
func (s *NatsSink) Close() {
s.client.Close()
}