dom/telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Add support for credentials file to nats_consumer and nats output (#7022)

Browse Source
This commit is contained in:
R.I.Pienaar
2020-02-20 22:30:04 +00:00
committed by GitHub
parent 82a358b910
commit 79ff743064
8 changed files with 78 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
plugins/inputs/nats_consumer/README.md
View File

@@ -23,6 +23,9 @@ instances of telegraf can read from a NATS cluster in parallel.
# username = ""
# password = ""
## Optional NATS 2.0 and NATS NGS compatible user credentials
# credentials = "/etc/telegraf/nats.creds"
## Use Transport Layer Security
# secure = false

56
plugins/inputs/nats_consumer/nats_consumer.go
View File

@@ -3,13 +3,14 @@ package natsconsumer
import (
"context"
"fmt"
"strings"
"sync"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal/tls"
"github.com/influxdata/telegraf/plugins/inputs"
"github.com/influxdata/telegraf/plugins/parsers"
nats "github.com/nats-io/go-nats"
"github.com/nats-io/nats.go"
)
var (
@@ -31,12 +32,14 @@ func (e natsError) Error() string {
}
type natsConsumer struct {
QueueGroup string `toml:"queue_group"`
Subjects []string `toml:"subjects"`
Servers []string `toml:"servers"`
Secure bool `toml:"secure"`
Username string `toml:"username"`
Password string `toml:"password"`
QueueGroup string `toml:"queue_group"`
Subjects []string `toml:"subjects"`
Servers []string `toml:"servers"`
Secure bool `toml:"secure"`
Username string `toml:"username"`
Password string `toml:"password"`
Credentials string `toml:"credentials"`
tls.ClientConfig
Log telegraf.Logger
@@ -77,6 +80,9 @@ var sampleConfig = `
# username = ""
# password = ""
## Optional NATS 2.0 and NATS NGS compatible user credentials
# credentials = "/etc/telegraf/nats.creds"
## Use Transport Layer Security
# secure = false
@@ -135,19 +141,18 @@ func (n *natsConsumer) Start(acc telegraf.Accumulator) error {
var connectErr error
// set default NATS connection options
opts := nats.DefaultOptions
// override max reconnection tries
opts.MaxReconnect = -1
// override servers if any were specified
opts.Servers = n.Servers
options := []nats.Option{
nats.MaxReconnects(-1),
nats.ErrorHandler(n.natsErrHandler),
}
// override authentication, if any was specified
if n.Username != "" {
opts.User = n.Username
opts.Password = n.Password
if n.Username != "" && n.Password != "" {
options = append(options, nats.UserInfo(n.Username, n.Password))
}
if n.Credentials != "" {
options = append(options, nats.UserCredentials(n.Credentials))
}
if n.Secure {
@@ -156,19 +161,17 @@ func (n *natsConsumer) Start(acc telegraf.Accumulator) error {
return err
}
opts.Secure = true
opts.TLSConfig = tlsConfig
options = append(options, nats.Secure(tlsConfig))
}
if n.conn == nil || n.conn.IsClosed() {
n.conn, connectErr = opts.Connect()
n.conn, connectErr = nats.Connect(strings.Join(n.Servers, ","), options...)
if connectErr != nil {
return connectErr
}
// Setup message and error channels
n.errs = make(chan error)
n.conn.SetErrorHandler(n.natsErrHandler)
n.in = make(chan *nats.Msg, 1000)
for _, subj := range n.Subjects {
@@ -178,14 +181,13 @@ func (n *natsConsumer) Start(acc telegraf.Accumulator) error {
if err != nil {
return err
}
// ensure that the subscription has been processed by the server
if err = n.conn.Flush(); err != nil {
return err
}
// set the subscription pending limits
if err = sub.SetPendingLimits(n.PendingMessageLimit, n.PendingBytesLimit); err != nil {
err = sub.SetPendingLimits(n.PendingMessageLimit, n.PendingBytesLimit)
if err != nil {
return err
}
n.subs = append(n.subs, sub)
}
}