CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(secretstores): convert many output plugins (#12497)

This commit is contained in:
Sven Rebhan 2023-01-25 21:02:29 +01:00 committed by GitHub
parent 607bfdbc97
commit f5c2c4abf0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
41 changed files with 546 additions and 273 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
config/secret.go
View File

@ -91,6 +91,26 @@ func (s *Secret) Empty() bool {
return !s.notempty
}
// EqualTo performs a constant-time comparison of the secret to the given reference
func (s *Secret) EqualTo(ref []byte) (bool, error) {
if s.enclave == nil {
return false, nil
}
if len(s.unlinked) > 0 {
return false, fmt.Errorf("unlinked parts in secret: %v", strings.Join(s.unlinked, ";"))
}
// Get a locked-buffer of the secret to perform the comparison
lockbuf, err := s.enclave.Open()
if err != nil {
return false, fmt.Errorf("opening enclave failed: %v", err)
}
defer lockbuf.Destroy()
return lockbuf.EqualTo(ref), nil
}
// Get return the string representation of the secret
func (s *Secret) Get() ([]byte, error) {
if s.enclave == nil {

14
config/secret_test.go
View File

@ -433,6 +433,20 @@ func TestSecretStoreInvalidKeys(t *testing.T) {
}
}
func TestSecretEqualTo(t *testing.T) {
mysecret := "a wonderful test"
s := NewSecret([]byte(mysecret))
defer s.Destroy()
equal, err := s.EqualTo([]byte(mysecret))
require.NoError(t, err)
require.True(t, equal)
equal, err = s.EqualTo([]byte("some random text"))
require.NoError(t, err)
require.False(t, equal)
}
func TestSecretStoreInvalidReference(t *testing.T) {
// Make sure we clean-up our mess
defer func() { unlinkedSecrets = make([]*Secret, 0) }()

21
plugins/outputs/amqp/amqp.go
View File

@ -4,6 +4,7 @@ package amqp
import (
"bytes"
_ "embed"
"fmt"
"strings"
"time"
@ -47,8 +48,8 @@ type AMQP struct {
ExchangePassive bool `toml:"exchange_passive"`
ExchangeDurability string `toml:"exchange_durability"`
ExchangeArguments map[string]string `toml:"exchange_arguments"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
MaxMessages int `toml:"max_messages"`
AuthMethod string `toml:"auth_method"`
RoutingTag string `toml:"routing_tag"`
@ -293,11 +294,21 @@ func (q *AMQP) makeClientConfig() (*ClientConfig, error) {
var auth []amqp.Authentication
if strings.ToUpper(q.AuthMethod) == "EXTERNAL" {
auth = []amqp.Authentication{&externalAuth{}}
} else if q.Username != "" || q.Password != "" {
} else if !q.Username.Empty() || !q.Password.Empty() {
username, err := q.Username.Get()
if err != nil {
return nil, fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := q.Password.Get()
if err != nil {
return nil, fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
auth = []amqp.Authentication{
&amqp.PlainAuth{
Username: q.Username,
Password: q.Password,
Username: string(username),
Password: string(password),
},
}
}

4
plugins/outputs/amqp/amqp_test.go
View File

@ -114,8 +114,8 @@ func TestConnect(t *testing.T) {
name: "username password",
output: &AMQP{
URL: "amqp://foo:bar@localhost",
Username: "telegraf",
Password: "pa$$word",
Username: config.NewSecret([]byte("telegraf")),
Password: config.NewSecret([]byte("pa$$word")),
connect: func(_ *ClientConfig) (Client, error) {
return NewMockClient(), nil
},

13
plugins/outputs/dynatrace/dynatrace.go
View File

@ -26,7 +26,7 @@ var sampleConfig string
// Dynatrace Configuration for the Dynatrace output plugin
type Dynatrace struct {
URL string `toml:"url"`
APIToken string `toml:"api_token"`
APIToken config.Secret `toml:"api_token"`
Prefix string `toml:"prefix"`
Log telegraf.Logger `toml:"-"`
Timeout config.Duration `toml:"timeout"`
@ -151,8 +151,13 @@ func (d *Dynatrace) send(msg string) error {
}
req.Header.Add("Content-Type", "text/plain; charset=UTF-8")
if len(d.APIToken) != 0 {
req.Header.Add("Authorization", "Api-Token "+d.APIToken)
if !d.APIToken.Empty() {
token, err := d.APIToken.Get()
if err != nil {
return fmt.Errorf("getting token failed: %w", err)
}
req.Header.Add("Authorization", "Api-Token "+string(token))
config.ReleaseSecret(token)
}
// add user-agent header to identify metric source
req.Header.Add("User-Agent", "telegraf")
@ -184,7 +189,7 @@ func (d *Dynatrace) Init() error {
d.Log.Infof("Dynatrace URL is empty, defaulting to OneAgent metrics interface")
d.URL = apiconstants.GetDefaultOneAgentEndpoint()
}
if d.URL != apiconstants.GetDefaultOneAgentEndpoint() && len(d.APIToken) == 0 {
if d.URL != apiconstants.GetDefaultOneAgentEndpoint() && d.APIToken.Empty() {
d.Log.Errorf("Dynatrace api_token is a required field for Dynatrace output")
return fmt.Errorf("api_token is a required field for Dynatrace output")
}

24
plugins/outputs/dynatrace/dynatrace_test.go
View File

@ -35,7 +35,7 @@ func TestNilMetrics(t *testing.T) {
}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -58,7 +58,7 @@ func TestEmptyMetricsSlice(t *testing.T) {
d := &Dynatrace{}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
@ -82,7 +82,7 @@ func TestMockURL(t *testing.T) {
d := &Dynatrace{}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
@ -153,7 +153,7 @@ func TestSendMetrics(t *testing.T) {
d := &Dynatrace{
URL: ts.URL,
APIToken: "123",
APIToken: config.NewSecret([]byte("123")),
Log: testutil.Logger{},
AddCounterMetrics: []string{},
}
@ -230,7 +230,7 @@ func TestSendSingleMetricWithUnorderedTags(t *testing.T) {
d := &Dynatrace{}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -271,7 +271,7 @@ func TestSendMetricWithoutTags(t *testing.T) {
d := &Dynatrace{}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -318,7 +318,7 @@ func TestSendMetricWithUpperCaseTagKeys(t *testing.T) {
d := &Dynatrace{}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -359,7 +359,7 @@ func TestSendBooleanMetricWithoutTags(t *testing.T) {
d := &Dynatrace{}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -402,7 +402,7 @@ func TestSendMetricWithDefaultDimensions(t *testing.T) {
d := &Dynatrace{DefaultDimensions: map[string]string{"dim": "value"}}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -445,7 +445,7 @@ func TestMetricDimensionsOverrideDefault(t *testing.T) {
d := &Dynatrace{DefaultDimensions: map[string]string{"dim": "default"}}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -487,7 +487,7 @@ func TestStaticDimensionsOverrideMetric(t *testing.T) {
d := &Dynatrace{DefaultDimensions: map[string]string{"dim": "default"}}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = testutil.Logger{}
err := d.Init()
require.NoError(t, err)
@ -533,7 +533,7 @@ func TestSendUnsupportedMetric(t *testing.T) {
logStub := loggerStub{}
d.URL = ts.URL
d.APIToken = "123"
d.APIToken = config.NewSecret([]byte("123"))
d.Log = logStub
err := d.Init()
require.NoError(t, err)

53
plugins/outputs/elasticsearch/elasticsearch.go
View File

@ -27,7 +27,7 @@ import (
var sampleConfig string
type Elasticsearch struct {
AuthBearerToken string `toml:"auth_bearer_token"`
AuthBearerToken config.Secret `toml:"auth_bearer_token"`
DefaultPipeline string `toml:"default_pipeline"`
DefaultTagValue string `toml:"default_tag_value"`
EnableGzip bool `toml:"enable_gzip"`
@ -40,12 +40,12 @@ type Elasticsearch struct {
IndexName string `toml:"index_name"`
ManageTemplate bool `toml:"manage_template"`
OverwriteTemplate bool `toml:"overwrite_template"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
TemplateName string `toml:"template_name"`
Timeout config.Duration `toml:"timeout"`
URLs []string `toml:"urls"`
UsePipeline string `toml:"use_pipeline"`
Username string `toml:"username"`
Log telegraf.Logger `toml:"-"`
majorReleaseNumber int
pipelineName string
@ -182,19 +182,11 @@ func (a *Elasticsearch) Connect() error {
elastic.SetGzip(a.EnableGzip),
)
if a.Username != "" && a.Password != "" {
clientOptions = append(clientOptions,
elastic.SetBasicAuth(a.Username, a.Password),
)
}
if a.AuthBearerToken != "" {
clientOptions = append(clientOptions,
elastic.SetHeaders(http.Header{
"Authorization": []string{fmt.Sprintf("Bearer %s", a.AuthBearerToken)},
}),
)
authOptions, err := a.getAuthOptions()
if err != nil {
return err
}
clientOptions = append(clientOptions, authOptions...)
if time.Duration(a.HealthCheckInterval) == 0 {
clientOptions = append(clientOptions,
@ -470,6 +462,37 @@ func (a *Elasticsearch) Close() error {
return nil
}
func (a *Elasticsearch) getAuthOptions() ([]elastic.ClientOptionFunc, error) {
var fns []elastic.ClientOptionFunc
if !a.Username.Empty() && !a.Password.Empty() {
username, err := a.Username.Get()
if err != nil {
return nil, fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := a.Password.Get()
if err != nil {
return nil, fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
fns = append(fns, elastic.SetBasicAuth(string(username), string(password)))
}
if !a.AuthBearerToken.Empty() {
token, err := a.AuthBearerToken.Get()
if err != nil {
return nil, fmt.Errorf("getting token failed: %w", err)
}
defer config.ReleaseSecret(token)
auth := []string{"Bearer " + string(token)}
fns = append(fns, elastic.SetHeaders(http.Header{"Authorization": auth}))
}
return fns, nil
}
func init() {
outputs.Add("elasticsearch", func() telegraf.Output {
return &Elasticsearch{

2
plugins/outputs/elasticsearch/elasticsearch_test.go
View File

@ -746,7 +746,7 @@ func TestAuthorizationHeaderWhenBearerTokenIsPresent(t *testing.T) {
EnableGzip: false,
ManageTemplate: false,
Log: testutil.Logger{},
AuthBearerToken: "0123456789abcdef",
AuthBearerToken: config.NewSecret([]byte("0123456789abcdef")),
}
err := e.Connect()

23
plugins/outputs/groundwork/groundwork.go
View File

@ -17,6 +17,7 @@ import (
"github.com/hashicorp/go-uuid"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/plugins/outputs"
)
@ -31,8 +32,8 @@ type metricMeta struct {
type Groundwork struct {
Server string `toml:"url"`
AgentID string `toml:"agent_id"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
DefaultAppType string `toml:"default_app_type"`
DefaultHost string `toml:"default_host"`
DefaultServiceState string `toml:"default_service_state"`
@ -53,10 +54,10 @@ func (g *Groundwork) Init() error {
if g.AgentID == "" {
return errors.New(`no "agent_id" provided`)
}
if g.Username == "" {
if g.Username.Empty() {
return errors.New(`no "username" provided`)
}
if g.Password == "" {
if g.Password.Empty() {
return errors.New(`no "password" provided`)
}
if g.DefaultAppType == "" {
@ -72,13 +73,23 @@ func (g *Groundwork) Init() error {
return errors.New(`invalid "default_service_state" provided`)
}
username, err := g.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := g.Password.Get()
if err != nil {
return fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
g.client = clients.GWClient{
AppName: "telegraf",
AppType: g.DefaultAppType,
GWConnection: &clients.GWConnection{
HostName: g.Server,
UserName: g.Username,
Password: g.Password,
UserName: string(username),
Password: string(password),
IsDynamicInventory: true,
},
}

20
plugins/outputs/http/http.go
View File

@ -17,6 +17,7 @@ import (
v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal"
internalaws "github.com/influxdata/telegraf/plugins/common/aws"
httpconfig "github.com/influxdata/telegraf/plugins/common/http"
@ -43,8 +44,8 @@ const (
type HTTP struct {
URL string `toml:"url"`
Method string `toml:"method"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
Headers map[string]string `toml:"headers"`
ContentEncoding string `toml:"content_encoding"`
UseBatchFormat bool `toml:"use_batch_format"`
@ -180,8 +181,19 @@ func (h *HTTP) writeMetric(reqBody []byte) error {
}
}
if h.Username != "" || h.Password != "" {
req.SetBasicAuth(h.Username, h.Password)
if !h.Username.Empty() || !h.Password.Empty() {
username, err := h.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := h.Password.Get()
if err != nil {
return fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
req.SetBasicAuth(string(username), string(password))
}
// google api auth

49
plugins/outputs/http/http_test.go
View File

@ -395,55 +395,46 @@ func TestBasicAuth(t *testing.T) {
require.NoError(t, err)
tests := []struct {
name string
plugin *HTTP
name string
username string
password string
}{
{
name: "default",
plugin: &HTTP{
URL: u.String(),
},
},
{
name: "username only",
plugin: &HTTP{
URL: u.String(),
Username: "username",
},
name: "username only",
username: "username",
},
{
name: "password only",
plugin: &HTTP{
URL: u.String(),
Password: "pa$$word",
},
name: "password only",
password: "pa$$word",
},
{
name: "username and password",
plugin: &HTTP{
URL: u.String(),
Username: "username",
Password: "pa$$word",
},
name: "username and password",
username: "username",
password: "pa$$word",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
plugin := &HTTP{
URL: u.String(),
Username: config.NewSecret([]byte(tt.username)),
Password: config.NewSecret([]byte(tt.password)),
}
ts.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
username, password, _ := r.BasicAuth()
require.Equal(t, tt.plugin.Username, username)
require.Equal(t, tt.plugin.Password, password)
require.Equal(t, tt.username, username)
require.Equal(t, tt.password, password)
w.WriteHeader(http.StatusOK)
})
serializer := influx.NewSerializer()
tt.plugin.SetSerializer(serializer)
err = tt.plugin.Connect()
require.NoError(t, err)
err = tt.plugin.Write([]telegraf.Metric{getMetric()})
require.NoError(t, err)
plugin.SetSerializer(serializer)
require.NoError(t, plugin.Connect())
require.NoError(t, plugin.Write([]telegraf.Metric{getMetric()}))
})
}
}

34
plugins/outputs/influxdb/http.go
View File

@ -16,6 +16,7 @@ import (
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/serializers/influx"
)
@ -88,8 +89,8 @@ type HTTPConfig struct {
URL *url.URL
UserAgent string
Timeout time.Duration
Username string
Password string
Username config.Secret
Password config.Secret
TLSConfig *tls.Config
Proxy *url.URL
Headers map[string]string
@ -451,9 +452,11 @@ func (c *httpClient) makeQueryRequest(query string) (*http.Request, error) {
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
c.addHeaders(req)
if err := c.addHeaders(req); err != nil {
return nil, err
}
return req, nil
return req, err
}
func (c *httpClient) makeWriteRequest(address string, body io.Reader) (*http.Request, error) {
@ -465,7 +468,9 @@ func (c *httpClient) makeWriteRequest(address string, body io.Reader) (*http.Req
}
req.Header.Set("Content-Type", "text/plain; charset=utf-8")
c.addHeaders(req)
if err := c.addHeaders(req); err != nil {
return nil, err
}
if c.config.ContentEncoding == "gzip" {
req.Header.Set("Content-Encoding", "gzip")
@ -491,14 +496,27 @@ func (c *httpClient) requestBodyReader(metrics []telegraf.Metric) (io.ReadCloser
return io.NopCloser(reader), nil
}
func (c *httpClient) addHeaders(req *http.Request) {
if c.config.Username != "" || c.config.Password != "" {
req.SetBasicAuth(c.config.Username, c.config.Password)
func (c *httpClient) addHeaders(req *http.Request) error {
if !c.config.Username.Empty() || !c.config.Password.Empty() {
username, err := c.config.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := c.config.Password.Get()
if err != nil {
return fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
req.SetBasicAuth(string(username), string(password))
}
for header, value := range c.config.Headers {
req.Header.Set(header, value)
}
return nil
}
func (c *httpClient) validateResponse(response io.ReadCloser) (io.ReadCloser, error) {

9
plugins/outputs/influxdb/http_test.go
View File

@ -18,6 +18,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/outputs/influxdb"
@ -91,8 +92,8 @@ func TestHTTP_CreateDatabase(t *testing.T) {
name: "send basic auth",
config: influxdb.HTTPConfig{
URL: u,
Username: "guy",
Password: "smiley",
Username: config.NewSecret([]byte("guy")),
Password: config.NewSecret([]byte("smiley")),
Database: "telegraf",
},
database: "telegraf",
@ -302,8 +303,8 @@ func TestHTTP_Write(t *testing.T) {
config: influxdb.HTTPConfig{
URL: u,
Database: "telegraf",
Username: "guy",
Password: "smiley",
Username: config.NewSecret([]byte("guy")),
Password: config.NewSecret([]byte("smiley")),
Log: testutil.Logger{},
},
queryHandlerFunc: func(t *testing.T, w http.ResponseWriter, r *http.Request) {

4
plugins/outputs/influxdb/influxdb.go
View File

@ -38,8 +38,8 @@ type Client interface {
type InfluxDB struct {
URL string `toml:"url" deprecated:"0.1.9;2.0.0;use 'urls' instead"`
URLs []string `toml:"urls"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
Database string `toml:"database"`
DatabaseTag string `toml:"database_tag"`
ExcludeDatabaseTag bool `toml:"exclude_database_tag"`

4
plugins/outputs/influxdb/influxdb_test.go
View File

@ -121,8 +121,8 @@ func TestConnectHTTPConfig(t *testing.T) {
RetentionPolicy: "default",
WriteConsistency: "any",
Timeout: config.Duration(5 * time.Second),
Username: "guy",
Password: "smiley",
Username: config.NewSecret([]byte("guy")),
Password: config.NewSecret([]byte("smiley")),
UserAgent: "telegraf",
HTTPProxy: "http://localhost:8086",
HTTPHeaders: map[string]string{

54
plugins/outputs/influxdb_v2/http.go
View File

@ -16,6 +16,7 @@ import (
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/plugins/serializers/influx"
)
@ -41,7 +42,7 @@ const (
type HTTPConfig struct {
URL *url.URL
Token string
Token config.Secret
Organization string
Bucket string
BucketTag string
@ -74,59 +75,66 @@ type httpClient struct {
log telegraf.Logger
}
func NewHTTPClient(config *HTTPConfig) (*httpClient, error) {
if config.URL == nil {
func NewHTTPClient(cfg *HTTPConfig) (*httpClient, error) {
if cfg.URL == nil {
return nil, ErrMissingURL
}
timeout := config.Timeout
timeout := cfg.Timeout
if timeout == 0 {
timeout = defaultRequestTimeout
}
userAgent := config.UserAgent
userAgent := cfg.UserAgent
if userAgent == "" {
userAgent = internal.ProductToken()
}
var headers = make(map[string]string, len(config.Headers)+2)
var headers = make(map[string]string, len(cfg.Headers)+2)
headers["User-Agent"] = userAgent
headers["Authorization"] = "Token " + config.Token
for k, v := range config.Headers {
token, err := cfg.Token.Get()
if err != nil {
return nil, fmt.Errorf("getting token failed: %w", err)
}
headers["Authorization"] = "Token " + string(token)
config.ReleaseSecret(token)
for k, v := range cfg.Headers {
headers[k] = v
}
var proxy func(*http.Request) (*url.URL, error)
if config.Proxy != nil {
proxy = http.ProxyURL(config.Proxy)
if cfg.Proxy != nil {
proxy = http.ProxyURL(cfg.Proxy)
} else {
proxy = http.ProxyFromEnvironment
}
serializer := config.Serializer
serializer := cfg.Serializer
if serializer == nil {
serializer = influx.NewSerializer()
}
var transport *http.Transport
switch config.URL.Scheme {
switch cfg.URL.Scheme {
case "http", "https":
transport = &http.Transport{
Proxy: proxy,
TLSClientConfig: config.TLSConfig,
TLSClientConfig: cfg.TLSConfig,
}
case "unix":
transport = &http.Transport{
Dial: func(_, _ string) (net.Conn, error) {
return net.DialTimeout(
config.URL.Scheme,
config.URL.Path,
cfg.URL.Scheme,
cfg.URL.Path,
timeout,
)
},
}
default:
return nil, fmt.Errorf("unsupported scheme %q", config.URL.Scheme)
return nil, fmt.Errorf("unsupported scheme %q", cfg.URL.Scheme)
}
client := &httpClient{
@ -135,15 +143,15 @@ func NewHTTPClient(config *HTTPConfig) (*httpClient, error) {
Timeout: timeout,
Transport: transport,
},
url: config.URL,
ContentEncoding: config.ContentEncoding,
url: cfg.URL,
ContentEncoding: cfg.ContentEncoding,
Timeout: timeout,
Headers: headers,
Organization: config.Organization,
Bucket: config.Bucket,
BucketTag: config.BucketTag,
ExcludeBucketTag: config.ExcludeBucketTag,
log: config.Log,
Organization: cfg.Organization,
Bucket: cfg.Bucket,
BucketTag: cfg.BucketTag,
ExcludeBucketTag: cfg.ExcludeBucketTag,
log: cfg.Log,
}
return client, nil
}

2
plugins/outputs/influxdb_v2/influxdb_v2.go
View File

@ -35,7 +35,7 @@ type Client interface {
type InfluxDB struct {
URLs []string `toml:"urls"`
Token string `toml:"token"`
Token config.Secret `toml:"token"`
Organization string `toml:"organization"`
Bucket string `toml:"bucket"`
BucketTag string `toml:"bucket_tag"`

9
plugins/outputs/instrumental/instrumental.go
View File

@ -28,7 +28,7 @@ var (
type Instrumental struct {
Host string `toml:"host"`
APIToken string `toml:"api_token"`
APIToken config.Secret `toml:"api_token"`
Prefix string `toml:"prefix"`
DataFormat string `toml:"data_format"`
Template string `toml:"template"`
@ -164,8 +164,13 @@ func (i *Instrumental) Write(metrics []telegraf.Metric) error {
}
func (i *Instrumental) authenticate(conn net.Conn) error {
_, err := fmt.Fprintf(conn, HandshakeFormat, i.APIToken)
token, err := i.APIToken.Get()
if err != nil {
return fmt.Errorf("getting token failed: %w", err)
}
defer config.ReleaseSecret(token)
if _, err := fmt.Fprintf(conn, HandshakeFormat, string(token)); err != nil {
return err
}

3
plugins/outputs/instrumental/instrumental_test.go
View File

@ -11,6 +11,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/metric"
)
@ -21,7 +22,7 @@ func TestWrite(t *testing.T) {
i := Instrumental{
Host: "127.0.0.1",
APIToken: "abc123token",
APIToken: config.NewSecret([]byte("abc123token")),
Prefix: "my.prefix",
}

30
plugins/outputs/iotdb/iotdb.go
View File

@ -24,8 +24,8 @@ var sampleConfig string
type IoTDB struct {
Host string `toml:"host"`
Port string `toml:"port"`
User string `toml:"user"`
Password string `toml:"password"`
User config.Secret `toml:"user"`
Password config.Secret `toml:"password"`
Timeout config.Duration `toml:"timeout"`
ConvertUint64To string `toml:"uint64_conversion"`
TimeStampUnit string `toml:"timestamp_precision"`
@ -64,16 +64,36 @@ func (s *IoTDB) Init() error {
if !choice.Contains(s.TreatTagsAs, []string{"fields", "device_id"}) {
return fmt.Errorf("unknown 'convert_tags_to' method %q", s.TreatTagsAs)
}
if s.User.Empty() {
s.User.Destroy()
s.User = config.NewSecret([]byte("root"))
}
if s.Password.Empty() {
s.Password.Destroy()
s.Password = config.NewSecret([]byte("root"))
}
s.Log.Info("Initialization completed.")
return nil
}
func (s *IoTDB) Connect() error {
username, err := s.User.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := s.Password.Get()
if err != nil {
return fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
sessionConf := &client.Config{
Host: s.Host,
Port: s.Port,
UserName: s.User,
Password: s.Password,
UserName: string(username),
Password: string(password),
}
var ss = client.NewSession(sessionConf)
s.session = &ss
@ -265,8 +285,6 @@ func newIoTDB() *IoTDB {
return &IoTDB{
Host: "localhost",
Port: "6667",
User: "root",
Password: "root",
Timeout: config.Duration(time.Second * 5),
ConvertUint64To: "int64_clip",
TimeStampUnit: "nanosecond",

4
plugins/outputs/iotdb/iotdb_test.go
View File

@ -511,8 +511,8 @@ func TestIntegrationInserts(t *testing.T) {
testClient := &IoTDB{
Host: container.Address,
Port: container.Ports[iotdbPort],
User: "root",
Password: "root",
User: config.NewSecret([]byte("root")),
Password: config.NewSecret([]byte("root")),
Timeout: config.Duration(time.Second * 5),
ConvertUint64To: "int64_clip",
TimeStampUnit: "nanosecond",

24
plugins/outputs/librato/librato.go
View File

@ -5,6 +5,7 @@ import (
"bytes"
_ "embed"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
@ -22,8 +23,8 @@ var sampleConfig string
// Librato structure for configuration and client
type Librato struct {
APIUser string `toml:"api_user"`
APIToken string `toml:"api_token"`
APIUser config.Secret `toml:"api_user"`
APIToken config.Secret `toml:"api_token"`
Debug bool `toml:"debug"`
SourceTag string `toml:"source_tag" deprecated:"1.0.0;use 'template' instead"`
Timeout config.Duration `toml:"timeout"`
@ -67,9 +68,8 @@ func (*Librato) SampleConfig() string {
// Connect is the default output plugin connection function who make sure it
// can connect to the endpoint
func (l *Librato) Connect() error {
if l.APIUser == "" || l.APIToken == "" {
return fmt.Errorf(
"api_user and api_token are required fields for librato output")
if l.APIUser.Empty() || l.APIToken.Empty() {
return errors.New("api_user and api_token required")
}
l.client = &http.Client{
Transport: &http.Transport{
@ -142,7 +142,19 @@ func (l *Librato) writeBatch(start int, sizeBatch int, metricCounter int, tempGa
return fmt.Errorf("unable to create http.Request, %s", err.Error())
}
req.Header.Add("Content-Type", "application/json")
req.SetBasicAuth(l.APIUser, l.APIToken)
user, err := l.APIUser.Get()
if err != nil {
return fmt.Errorf("getting user failed: %w", err)
}
token, err := l.APIToken.Get()
if err != nil {
config.ReleaseSecret(user)
return fmt.Errorf("getting token failed: %w", err)
}
req.SetBasicAuth(string(user), string(token))
config.ReleaseSecret(user)
config.ReleaseSecret(token)
resp, err := l.client.Do(req)
if err != nil {

28
plugins/outputs/librato/librato_test.go
View File

@ -9,6 +9,7 @@ import (
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
@ -33,12 +34,10 @@ func TestUriOverride(t *testing.T) {
defer ts.Close()
l := newTestLibrato(ts.URL)
l.APIUser = "telegraf@influxdb.com"
l.APIToken = "123456"
err := l.Connect()
require.NoError(t, err)
err = l.Write([]telegraf.Metric{newHostMetric(int32(0), "name", "host")})
require.NoError(t, err)
l.APIUser = config.NewSecret([]byte("telegraf@influxdb.com"))
l.APIToken = config.NewSecret([]byte("123456"))
require.NoError(t, l.Connect())
require.NoError(t, l.Write([]telegraf.Metric{newHostMetric(int32(0), "name", "host")}))
}
func TestBadStatusCode(t *testing.T) {
@ -49,18 +48,11 @@ func TestBadStatusCode(t *testing.T) {
defer ts.Close()
l := newTestLibrato(ts.URL)
l.APIUser = "telegraf@influxdb.com"
l.APIToken = "123456"
err := l.Connect()
require.NoError(t, err)
err = l.Write([]telegraf.Metric{newHostMetric(int32(0), "name", "host")})
if err == nil {
t.Errorf("error expected but none returned")
} else {
require.EqualError(
t,
fmt.Errorf("received bad status code, 503\n "), err.Error())
}
l.APIUser = config.NewSecret([]byte("telegraf@influxdb.com"))
l.APIToken = config.NewSecret([]byte("123456"))
require.NoError(t, l.Connect())
err := l.Write([]telegraf.Metric{newHostMetric(int32(0), "name", "host")})
require.ErrorContains(t, err, "received bad status code, 503")
}
func TestBuildGauge(t *testing.T) {

30
plugins/outputs/logzio/logzio.go
View File

@ -25,10 +25,10 @@ const (
)
type Logzio struct {
Log telegraf.Logger `toml:"-"`
Timeout config.Duration `toml:"timeout"`
Token string `toml:"token"`
URL string `toml:"url"`
Token config.Secret `toml:"token"`
Timeout config.Duration `toml:"timeout"`
Log telegraf.Logger `toml:"-"`
tls.ClientConfig
client *http.Client
@ -53,9 +53,14 @@ func (*Logzio) SampleConfig() string {
func (l *Logzio) Connect() error {
l.Log.Debug("Connecting to logz.io output...")
if l.Token == "" || l.Token == "your logz.io token" {
if l.Token.Empty() {
return fmt.Errorf("token is required")
}
if equal, err := l.Token.EqualTo([]byte("your logz.io token")); err != nil {
return err
} else if equal {
return fmt.Errorf("please replace 'token' with your actual token")
}
tlsCfg, err := l.ClientConfig.TLSConfig()
if err != nil {
@ -110,7 +115,12 @@ func (l *Logzio) Write(metrics []telegraf.Metric) error {
}
func (l *Logzio) send(metrics []byte) error {
req, err := http.NewRequest("POST", l.authURL(), bytes.NewBuffer(metrics))
url, err := l.authURL()
if err != nil {
return err
}
req, err := http.NewRequest("POST", url, bytes.NewBuffer(metrics))
if err != nil {
return fmt.Errorf("unable to create http.Request, %s", err.Error())
}
@ -130,8 +140,14 @@ func (l *Logzio) send(metrics []byte) error {
return nil
}
func (l *Logzio) authURL() string {
return fmt.Sprintf("%s/?token=%s", l.URL, l.Token)
func (l *Logzio) authURL() (string, error) {
token, err := l.Token.Get()
if err != nil {
return "", fmt.Errorf("getting token failed: %w", err)
}
defer config.ReleaseSecret(token)
return fmt.Sprintf("%s/?token=%s", l.URL, string(token)), nil
}
func (l *Logzio) parseMetric(metric telegraf.Metric) *Metric {

42
plugins/outputs/logzio/logzio_test.go
View File

@ -4,13 +4,15 @@ import (
"bytes"
"compress/gzip"
"encoding/json"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
"io"
"net/http"
"net/http/httptest"
"testing"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
)
const (
@ -18,13 +20,23 @@ const (
testURL = "https://logzio.com"
)
func TestConnetWithoutToken(t *testing.T) {
func TestConnectWithoutToken(t *testing.T) {
l := &Logzio{
URL: testURL,
Log: testutil.Logger{},
}
err := l.Connect()
require.Error(t, err)
require.ErrorContains(t, err, "token is required")
}
func TestConnectWithDefaultToken(t *testing.T) {
l := &Logzio{
URL: testURL,
Token: config.NewSecret([]byte("your logz.io token")),
Log: testutil.Logger{},
}
err := l.Connect()
require.ErrorContains(t, err, "please replace 'token'")
}
func TestParseMetric(t *testing.T) {
@ -45,16 +57,12 @@ func TestBadStatusCode(t *testing.T) {
defer ts.Close()
l := &Logzio{
Token: testToken,
Token: config.NewSecret([]byte(testToken)),
URL: ts.URL,
Log: testutil.Logger{},
}
err := l.Connect()
require.NoError(t, err)
err = l.Write(testutil.MockMetrics())
require.Error(t, err)
require.NoError(t, l.Connect())
require.Error(t, l.Write(testutil.MockMetrics()))
}
func TestWrite(t *testing.T) {
@ -81,14 +89,10 @@ func TestWrite(t *testing.T) {
defer ts.Close()
l := &Logzio{
Token: testToken,
Token: config.NewSecret([]byte(testToken)),
URL: ts.URL,
Log: testutil.Logger{},
}
err := l.Connect()
require.NoError(t, err)
err = l.Write([]telegraf.Metric{tm})
require.NoError(t, err)
require.NoError(t, l.Connect())
require.NoError(t, l.Write([]telegraf.Metric{tm}))
}

19
plugins/outputs/loki/loki.go
View File

@ -35,8 +35,8 @@ type Loki struct {
Domain string `toml:"domain"`
Endpoint string `toml:"endpoint"`
Timeout config.Duration `toml:"timeout"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
Headers map[string]string `toml:"http_headers"`
ClientID string `toml:"client_id"`
ClientSecret string `toml:"client_secret"`
@ -156,8 +156,19 @@ func (l *Loki) writeMetrics(s Streams) error {
return err
}
if l.Username != "" {
req.SetBasicAuth(l.Username, l.Password)
if !l.Username.Empty() {
username, err := l.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := l.Password.Get()
if err != nil {
return fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
req.SetBasicAuth(string(username), string(password))
}
for k, v := range l.Headers {

33
plugins/outputs/loki/loki_test.go
View File

@ -14,6 +14,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/testutil"
)
@ -250,22 +251,17 @@ func TestBasicAuth(t *testing.T) {
require.NoError(t, err)
tests := []struct {
name string
plugin *Loki
name string
username string
password string
}{
{
name: "default",
plugin: &Loki{
Domain: u.String(),
},
},
{
name: "username and password",
plugin: &Loki{
Domain: u.String(),
Username: "username",
Password: "pa$$word",
},
name: "username and password",
username: "username",
password: "pa$$word",
},
}
@ -273,16 +269,19 @@ func TestBasicAuth(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
ts.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
username, password, _ := r.BasicAuth()
require.Equal(t, tt.plugin.Username, username)
require.Equal(t, tt.plugin.Password, password)
require.Equal(t, tt.username, username)
require.Equal(t, tt.password, password)
w.WriteHeader(http.StatusOK)
})
err = tt.plugin.Connect()
require.NoError(t, err)
plugin := &Loki{
Domain: u.String(),
Username: config.NewSecret([]byte(tt.username)),
Password: config.NewSecret([]byte(tt.password)),
}
require.NoError(t, plugin.Connect())
err = tt.plugin.Write([]telegraf.Metric{getMetric()})
require.NoError(t, err)
require.NoError(t, plugin.Write([]telegraf.Metric{getMetric()}))
})
}
}

23
plugins/outputs/mongodb/mongodb.go
View File

@ -55,8 +55,8 @@ type MongoDB struct {
AuthenticationType string `toml:"authentication"`
MetricDatabase string `toml:"database"`
MetricGranularity string `toml:"granularity"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
ServerSelectTimeout config.Duration `toml:"timeout"`
TTL config.Duration `toml:"ttl"`
Log telegraf.Logger `toml:"-"`
@ -95,17 +95,28 @@ func (s *MongoDB) Init() error {
switch s.AuthenticationType {
case "SCRAM":
if s.Username == "" {
if s.Username.Empty() {
return fmt.Errorf("SCRAM authentication must specify a username")
}
if s.Password == "" {
if s.Password.Empty() {
return fmt.Errorf("SCRAM authentication must specify a password")
}
username, err := s.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
password, err := s.Password.Get()
if err != nil {
config.ReleaseSecret(username)
return fmt.Errorf("getting password failed: %w", err)
}
credential := options.Credential{
AuthMechanism: "SCRAM-SHA-256",
Username: s.Username,
Password: s.Password,
Username: string(username),
Password: string(password),
}
config.ReleaseSecret(username)
config.ReleaseSecret(password)
s.clientOptions.SetAuth(credential)
case "X509":
//format connection string to include tls/x509 options

12
plugins/outputs/mongodb/mongodb_test.go
View File

@ -86,8 +86,8 @@ func TestConnectAndWriteIntegrationSCRAMAuth(t *testing.T) {
Dsn: fmt.Sprintf("mongodb://%s:%s/admin",
container.Address, container.Ports[servicePort]),
AuthenticationType: "SCRAM",
Username: "root",
Password: "changeme",
Username: config.NewSecret([]byte("root")),
Password: config.NewSecret([]byte("changeme")),
MetricDatabase: "telegraf_test",
MetricGranularity: "seconds",
},
@ -101,8 +101,8 @@ func TestConnectAndWriteIntegrationSCRAMAuth(t *testing.T) {
Dsn: fmt.Sprintf("mongodb://%s:%s/admin",
container.Address, container.Ports[servicePort]),
AuthenticationType: "SCRAM",
Username: "root",
Password: "root",
Username: config.NewSecret([]byte("root")),
Password: config.NewSecret([]byte("root")),
MetricDatabase: "telegraf_test",
MetricGranularity: "seconds",
ServerSelectTimeout: config.Duration(time.Duration(5) * time.Second),
@ -380,7 +380,7 @@ func TestConfiguration(t *testing.T) {
plugin: &MongoDB{
Dsn: "mongodb://localhost:27017",
AuthenticationType: "SCRAM",
Password: "somerandompasswordthatwontwork",
Password: config.NewSecret([]byte("somerandompasswordthatwontwork")),
MetricDatabase: "telegraf_test",
MetricGranularity: "seconds",
},
@ -390,7 +390,7 @@ func TestConfiguration(t *testing.T) {
plugin: &MongoDB{
Dsn: "mongodb://localhost:27017",
AuthenticationType: "SCRAM",
Username: "somerandomusernamethatwontwork",
Username: config.NewSecret([]byte("somerandomusernamethatwontwork")),
MetricDatabase: "telegraf_test",
MetricGranularity: "seconds",
},

8
plugins/outputs/mqtt/mqtt.go
View File

@ -24,10 +24,10 @@ const (
)
type MQTT struct {
Servers []string `toml:"servers"`
Protocol string `toml:"protocol"`
Username string `toml:"username"`
Password string `toml:"password"`
Servers []string `toml:"servers"`
Protocol string `toml:"protocol"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
Database string
Timeout config.Duration `toml:"timeout"`
TopicPrefix string `toml:"topic_prefix" deprecated:"1.25.0;use 'topic' instead"`

20
plugins/outputs/mqtt/mqtt_v3.go
View File

@ -40,13 +40,21 @@ func (m *mqttv311Client) Connect() error {
}
opts.SetTLSConfig(tlsCfg)
user := m.Username
if user != "" {
opts.SetUsername(user)
if !m.Username.Empty() {
user, err := m.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
opts.SetUsername(string(user))
config.ReleaseSecret(user)
}
password := m.Password
if password != "" {
opts.SetPassword(password)
if !m.Password.Empty() {
password, err := m.Password.Get()
if err != nil {
return fmt.Errorf("getting password failed: %w", err)
}
opts.SetPassword(string(password))
config.ReleaseSecret(password)
}
if len(m.Servers) == 0 {

15
plugins/outputs/mqtt/mqtt_v5.go
View File

@ -29,9 +29,18 @@ func (m *mqttv5Client) Connect() error {
opts.ClientID = "Telegraf-Output-" + internal.RandomString(5)
}
user := m.Username
pass := m.Password
opts.SetUsernamePassword(user, []byte(pass))
user, err := m.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
pass, err := m.Password.Get()
if err != nil {
config.ReleaseSecret(user)
return fmt.Errorf("getting password failed: %w", err)
}
opts.SetUsernamePassword(string(user), pass)
config.ReleaseSecret(user)
config.ReleaseSecret(pass)
tlsCfg, err := m.ClientConfig.TLSConfig()
if err != nil {

39
plugins/outputs/nats/nats.go
View File

@ -9,6 +9,7 @@ import (
"github.com/nats-io/nats.go"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/plugins/common/tls"
"github.com/influxdata/telegraf/plugins/outputs"
"github.com/influxdata/telegraf/plugins/serializers"
@ -18,13 +19,13 @@ import (
var sampleConfig string
type NATS struct {
Servers []string `toml:"servers"`
Secure bool `toml:"secure"`
Name string `toml:"name"`
Username string `toml:"username"`
Password string `toml:"password"`
Credentials string `toml:"credentials"`
Subject string `toml:"subject"`
Servers []string `toml:"servers"`
Secure bool `toml:"secure"`
Name string `toml:"name"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
Credentials config.Secret `toml:"credentials"`
Subject string `toml:"subject"`
tls.ClientConfig
@ -50,12 +51,28 @@ func (n *NATS) Connect() error {
}
// override authentication, if any was specified
if n.Username != "" && n.Password != "" {
opts = append(opts, nats.UserInfo(n.Username, n.Password))
if !n.Username.Empty() && !n.Password.Empty() {
username, err := n.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
password, err := n.Password.Get()
if err != nil {
config.ReleaseSecret(username)
return fmt.Errorf("getting password failed: %w", err)
}
opts = append(opts, nats.UserInfo(string(username), string(password)))
config.ReleaseSecret(username)
config.ReleaseSecret(password)
}
if n.Credentials != "" {
opts = append(opts, nats.UserCredentials(n.Credentials))
if !n.Credentials.Empty() {
credentials, err := n.Credentials.Get()
if err != nil {
return fmt.Errorf("getting credentials failed: %w", err)
}
opts = append(opts, nats.UserCredentials(string(credentials)))
config.ReleaseSecret(credentials)
}
if n.Name != "" {

22
plugins/outputs/redistimeseries/redistimeseries.go
View File

@ -8,6 +8,7 @@ import (
"github.com/go-redis/redis/v7"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/plugins/common/tls"
"github.com/influxdata/telegraf/plugins/outputs"
)
@ -17,8 +18,8 @@ var sampleConfig string
type RedisTimeSeries struct {
Address string `toml:"address"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
Database int `toml:"database"`
Log telegraf.Logger `toml:"-"`
tls.ClientConfig
@ -29,10 +30,23 @@ func (r *RedisTimeSeries) Connect() error {
if r.Address == "" {
return errors.New("redis address must be specified")
}
username, err := r.Username.Get()
if err != nil {
return fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := r.Password.Get()
if err != nil {
return fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
r.client = redis.NewClient(&redis.Options{
Addr: r.Address,
Password: r.Password,
Username: r.Username,
Username: string(username),
Password: string(password),
DB: r.Database,
})
return r.client.Ping().Err()

20
plugins/outputs/signalfx/signalfx.go
View File

@ -14,6 +14,7 @@ import (
"github.com/signalfx/golib/v3/sfxclient"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/plugins/outputs"
)
@ -29,10 +30,10 @@ func init() {
// SignalFx plugin context
type SignalFx struct {
AccessToken string `toml:"access_token"`
SignalFxRealm string `toml:"signalfx_realm"`
IngestURL string `toml:"ingest_url"`
IncludedEventNames []string `toml:"included_event_names"`
AccessToken config.Secret `toml:"access_token"`
SignalFxRealm string `toml:"signalfx_realm"`
IngestURL string `toml:"ingest_url"`
IncludedEventNames []string `toml:"included_event_names"`
Log telegraf.Logger `toml:"-"`
@ -66,9 +67,6 @@ func GetMetricType(mtype telegraf.ValueType) (metricType datapoint.MetricType) {
func NewSignalFx() *SignalFx {
ctx, cancel := context.WithCancel(context.Background())
return &SignalFx{
AccessToken: "",
SignalFxRealm: "",
IngestURL: "",
IncludedEventNames: []string{""},
ctx: ctx,
cancel: cancel,
@ -83,7 +81,13 @@ func (*SignalFx) SampleConfig() string {
// Connect establishes a connection to SignalFx
func (s *SignalFx) Connect() error {
client := s.client.(*sfxclient.HTTPSink)
client.AuthToken = s.AccessToken
token, err := s.AccessToken.Get()
if err != nil {
return fmt.Errorf("getting token failed: %w", err)
}
client.AuthToken = string(token)
config.ReleaseSecret(token)
if s.IngestURL != "" {
client.DatapointEndpoint = datapointEndpointForIngestURL(s.IngestURL)

30
plugins/outputs/stomp/stomp.go
View File

@ -22,8 +22,8 @@ var sampleConfig string
type STOMP struct {
Host string `toml:"host"`
Username string `toml:"username"`
Password string `toml:"password"`
Username config.Secret `toml:"username"`
Password config.Secret `toml:"password"`
QueueName string `toml:"queueName"`
Log telegraf.Logger `toml:"-"`
@ -55,11 +55,15 @@ func (q *STOMP) Connect() error {
}
}
q.stomp, err = stomp.Connect(
q.conn,
stomp.ConnOpt.HeartBeat(time.Duration(q.HeartBeatSend), time.Duration(q.HeartBeatRec)),
stomp.ConnOpt.Login(q.Username, q.Password),
authOption, err := q.getAuthOption()
if err != nil {
return err
}
heartbeatOption := stomp.ConnOpt.HeartBeat(
time.Duration(q.HeartBeatSend),
time.Duration(q.HeartBeatRec),
)
q.stomp, err = stomp.Connect(q.conn, heartbeatOption, authOption)
if err != nil {
return err
}
@ -92,6 +96,20 @@ func (q *STOMP) Close() error {
return q.stomp.Disconnect()
}
func (q *STOMP) getAuthOption() (func(*stomp.Conn) error, error) {
username, err := q.Username.Get()
if err != nil {
return nil, fmt.Errorf("getting username failed: %w", err)
}
defer config.ReleaseSecret(username)
password, err := q.Password.Get()
if err != nil {
return nil, fmt.Errorf("getting password failed: %w", err)
}
defer config.ReleaseSecret(password)
return stomp.ConnOpt.Login(string(username), string(password)), nil
}
func init() {
outputs.Add("stomp", func() telegraf.Output {
return &STOMP{

7
plugins/outputs/stomp/stomp_test.go
View File

@ -39,11 +39,10 @@ func TestConnectAndWrite(t *testing.T) {
QueueName: "test_queue",
HeartBeatSend: 0,
HeartBeatRec: 0,
Log: testutil.Logger{},
serialize: s,
}
err = st.Connect()
require.NoError(t, err)
require.NoError(t, st.Connect())
err = st.Write(testutil.MockMetrics())
require.NoError(t, err)
require.NoError(t, st.Write(testutil.MockMetrics()))
}

9
plugins/outputs/warp10/warp10.go
View File

@ -31,7 +31,7 @@ const (
type Warp10 struct {
Prefix string `toml:"prefix"`
WarpURL string `toml:"warp_url"`
Token string `toml:"token"`
Token config.Secret `toml:"token"`
Timeout config.Duration `toml:"timeout"`
PrintErrorBody bool `toml:"print_error_body"`
MaxStringErrorSize int `toml:"max_string_error_size"`
@ -125,8 +125,13 @@ func (w *Warp10) Write(metrics []telegraf.Metric) error {
return fmt.Errorf("unable to create new request '%s': %s", addr, err)
}
req.Header.Set("X-Warp10-Token", w.Token)
req.Header.Set("Content-Type", "text/plain")
token, err := w.Token.Get()
if err != nil {
return fmt.Errorf("getting token failed: %w", err)
}
req.Header.Set("X-Warp10-Token", string(token))
config.ReleaseSecret(token)
resp, err := w.client.Do(req)
if err != nil {

7
plugins/outputs/warp10/warp10_test.go
View File

@ -3,6 +3,7 @@ package warp10
import (
"testing"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
)
@ -16,7 +17,7 @@ func TestWriteWarp10(t *testing.T) {
w := Warp10{
Prefix: "unit.test",
WarpURL: "http://localhost:8090",
Token: "WRITE",
Token: config.NewSecret([]byte("WRITE")),
}
payload := w.GenWarp10Payload(testutil.MockMetrics())
@ -27,7 +28,7 @@ func TestWriteWarp10EncodedTags(t *testing.T) {
w := Warp10{
Prefix: "unit.test",
WarpURL: "http://localhost:8090",
Token: "WRITE",
Token: config.NewSecret([]byte("WRITE")),
}
metrics := testutil.MockMetrics()
@ -43,7 +44,7 @@ func TestHandleWarp10Error(t *testing.T) {
w := Warp10{
Prefix: "unit.test",
WarpURL: "http://localhost:8090",
Token: "WRITE",
Token: config.NewSecret([]byte("WRITE")),
}
tests := [...]*ErrorTest{
{

22
plugins/outputs/wavefront/wavefront.go
View File

@ -11,6 +11,7 @@ import (
wavefront "github.com/wavefronthq/wavefront-sdk-go/senders"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/plugins/outputs"
serializer "github.com/influxdata/telegraf/plugins/serializers/wavefront"
)
@ -22,7 +23,7 @@ const maxTagLength = 254
type Wavefront struct {
URL string `toml:"url"`
Token string `toml:"token"`
Token config.Secret `toml:"token"`
Host string `toml:"host" deprecated:"2.4.0;use url instead"`
Port int `toml:"port" deprecated:"2.4.0;use url instead"`
Prefix string `toml:"prefix"`
@ -53,10 +54,20 @@ func (*Wavefront) SampleConfig() string {
return sampleConfig
}
func senderURLFromURLAndToken(rawURL, token string) (string, error) {
newURL, err := url.Parse(rawURL)
func (w *Wavefront) senderURLFromURLAndToken() (string, error) {
newURL, err := url.Parse(w.URL)
if err != nil {
return "", fmt.Errorf("could not parse the provided Url: %s", rawURL)
return "", fmt.Errorf("could not parse the provided URL: %s", w.URL)
}
token := "DUMMY_TOKEN"
if !w.Token.Empty() {
b, err := w.Token.Get()
if err != nil {
return "", fmt.Errorf("getting token failed: %w", err)
}
token = string(b)
config.ReleaseSecret(b)
}
newURL.User = url.User(token)
@ -75,7 +86,7 @@ func (w *Wavefront) Connect() error {
var connectionURL string
if w.URL != "" {
w.Log.Debug("connecting over http/https using Url: %s", w.URL)
connectionURLWithToken, err := senderURLFromURLAndToken(w.URL, w.Token)
connectionURLWithToken, err := w.senderURLFromURLAndToken()
if err != nil {
return err
}
@ -287,7 +298,6 @@ func (w *Wavefront) Close() error {
func init() {
outputs.Add("wavefront", func() telegraf.Output {
return &Wavefront{
Token: "DUMMY_TOKEN",
MetricSeparator: ".",
ConvertPaths: true,
ConvertBool: true,

13
plugins/outputs/wavefront/wavefront_test.go
View File

@ -9,6 +9,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/plugins/outputs"
serializer "github.com/influxdata/telegraf/plugins/serializers/wavefront"
@ -369,10 +370,14 @@ func TestSenderURLFromHostAndPort(t *testing.T) {
}
func TestSenderURLFromURLAndToken(t *testing.T) {
url, err := senderURLFromURLAndToken("https://surf.wavefront.com", "11111111-2222-3333-4444-555555555555")
require.Nil(t, err)
require.Equal(t, "https://11111111-2222-3333-4444-555555555555@surf.wavefront.com",
url)
w := &Wavefront{
URL: "https://surf.wavefront.com",
Token: config.NewSecret([]byte("11111111-2222-3333-4444-555555555555")),
}
url, err := w.senderURLFromURLAndToken()
require.NoError(t, err)
require.Equal(t, "https://11111111-2222-3333-4444-555555555555@surf.wavefront.com", url)
}
func TestDefaults(t *testing.T) {