Merge branch 'influxdata:master' into master

Author: kyoungho

config/config.go

@@ -1012,6 +1012,11 @@ func (c *Config) addSecretStore(name, source string, table *ast.Table) error {
 		return fmt.Errorf("invalid secret-store ID %q, must only contain letters, numbers or underscore", storeID)
 	}
 
+	tags := map[string]string{
+		"_id":         storeID,
+		"secretstore": name,
+	}
+
 	creator, ok := secretstores.SecretStores[name]
 	if !ok {
 		// Handle removed, deprecated plugins
@@ -1033,6 +1038,7 @@ func (c *Config) addSecretStore(name, source string, table *ast.Table) error {
 
 	logger := logging.New("secretstores", name, "")
 	models.SetLoggerOnPlugin(store, logger)
+	models.SetStatisticsOnPlugin(store, logger, tags)
 
 	if err := store.Init(); err != nil {
 		return fmt.Errorf("error initializing secret-store %q: %w", storeID, err)

models/running_aggregator.go

@@ -42,6 +42,7 @@ func NewRunningAggregator(aggregator telegraf.Aggregator, config *AggregatorConf
 		logger.Error(err)
 	}
 	SetLoggerOnPlugin(aggregator, logger)
+	SetStatisticsOnPlugin(aggregator, logger, tags)
 
 	return &RunningAggregator{
 		Aggregator: aggregator,

models/running_input.go

@@ -55,6 +55,7 @@ func NewRunningInput(input telegraf.Input, config *InputConfig) *RunningInput {
 		logger.Error(err)
 	}
 	SetLoggerOnPlugin(input, logger)
+	SetStatisticsOnPlugin(input, logger, tags)
 
 	return &RunningInput{
 		Input:  input,

models/running_parsers.go

@@ -32,6 +32,7 @@ func NewRunningParser(parser telegraf.Parser, config *ParserConfig) *RunningPars
 		logger.Error(err)
 	}
 	SetLoggerOnPlugin(parser, logger)
+	SetStatisticsOnPlugin(parser, logger, tags)
 
 	return &RunningParser{
 		Parser: parser,

models/running_processor.go

@@ -50,6 +50,7 @@ func NewRunningProcessor(processor telegraf.StreamingProcessor, config *Processo
 		logger.Error(err)
 	}
 	SetLoggerOnPlugin(processor, logger)
+	SetStatisticsOnPlugin(processor, logger, tags)
 
 	return &RunningProcessor{
 		Processor: processor,

models/running_serializer.go

@@ -42,6 +42,7 @@ func NewRunningSerializer(serializer telegraf.Serializer, config *SerializerConf
 		logger.Error(err)
 	}
 	SetLoggerOnPlugin(serializer, logger)
+	SetStatisticsOnPlugin(serializer, logger, tags)
 
 	return &RunningSerializer{
 		Serializer: serializer,

plugins/inputs/influxdb_v2_listener/README.md

@@ -87,6 +87,9 @@ to use them.
   ## 'internal' is the default. 'upstream' is a newer parser that is faster
   ## and more memory efficient.
   # parser_type = "internal"
+
+  ## Use new internal metrics. When true, it adds tag for alias if set.
+  # use_internal_statistics = false
 ```
 
 ## Metrics

plugins/inputs/influxdb_v2_listener/influxdb_v2_listener.go

@@ -45,15 +45,16 @@ type InfluxDBV2Listener struct {
 	port           int
 	common_tls.ServerConfig
 
-	MaxUndeliveredMetrics int             `toml:"max_undelivered_metrics"`
-	ReadTimeout           config.Duration `toml:"read_timeout"`
-	WriteTimeout          config.Duration `toml:"write_timeout"`
-	MaxBodySize           config.Size     `toml:"max_body_size"`
-	Token                 config.Secret   `toml:"token"`
-	BucketTag             string          `toml:"bucket_tag"`
-	ParserType            string          `toml:"parser_type"`
-
-	Log telegraf.Logger `toml:"-"`
+	MaxUndeliveredMetrics int                 `toml:"max_undelivered_metrics"`
+	ReadTimeout           config.Duration     `toml:"read_timeout"`
+	WriteTimeout          config.Duration     `toml:"write_timeout"`
+	MaxBodySize           config.Size         `toml:"max_body_size"`
+	Token                 config.Secret       `toml:"token"`
+	BucketTag             string              `toml:"bucket_tag"`
+	ParserType            string              `toml:"parser_type"`
+	UseInternalStatistics bool                `toml:"use_internal_statistics"`
+	Statistics            *selfstat.Collector `toml:"-"`
+	Log                   telegraf.Logger     `toml:"-"`
 
 	ctx                 context.Context
 	cancel              context.CancelFunc
@@ -97,15 +98,33 @@ func (h *InfluxDBV2Listener) Init() error {
 	tags := map[string]string{
 		"address": h.ServiceAddress,
 	}
-	h.bytesRecv = selfstat.Register("influxdb_v2_listener", "bytes_received", tags)
-	h.requestsServed = selfstat.Register("influxdb_v2_listener", "requests_served", tags)
-	h.writesServed = selfstat.Register("influxdb_v2_listener", "writes_served", tags)
-	h.healthsServed = selfstat.Register("influxdb_v2_listener", "healths_served", tags)
-	h.readysServed = selfstat.Register("influxdb_v2_listener", "readys_served", tags)
-	h.requestsRecv = selfstat.Register("influxdb_v2_listener", "requests_received", tags)
-	h.notFoundsServed = selfstat.Register("influxdb_v2_listener", "not_founds_served", tags)
-	h.pingsServed = selfstat.Register("influxdb_v2_listener", "pings_served", tags)
-	h.authFailures = selfstat.Register("influxdb_v2_listener", "auth_failures", tags)
+	if !h.UseInternalStatistics {
+		h.bytesRecv = selfstat.Register("influxdb_v2_listener", "bytes_received", tags)
+		h.requestsServed = selfstat.Register("influxdb_v2_listener", "requests_served", tags)
+		h.writesServed = selfstat.Register("influxdb_v2_listener", "writes_served", tags)
+		h.healthsServed = selfstat.Register("influxdb_v2_listener", "healths_served", tags)
+		h.readysServed = selfstat.Register("influxdb_v2_listener", "readys_served", tags)
+		h.requestsRecv = selfstat.Register("influxdb_v2_listener", "requests_received", tags)
+		h.notFoundsServed = selfstat.Register("influxdb_v2_listener", "not_founds_served", tags)
+		h.pingsServed = selfstat.Register("influxdb_v2_listener", "pings_served", tags)
+		h.authFailures = selfstat.Register("influxdb_v2_listener", "auth_failures", tags)
+		config.PrintOptionValueDeprecationNotice("inputs.influxdb_v2_listener", "use_internal_statistics", false, telegraf.DeprecationInfo{
+			Since:     "1.37.0",
+			RemovalIn: "1.45.0",
+			Notice:    "please update to 'use_internal_statistics = true'",
+		})
+	} else {
+		h.bytesRecv = h.Statistics.Register("influxdb_v2_listener", "bytes_received", tags)
+		h.requestsServed = h.Statistics.Register("influxdb_v2_listener", "requests_served", tags)
+		h.writesServed = h.Statistics.Register("influxdb_v2_listener", "writes_served", tags)
+		h.healthsServed = h.Statistics.Register("influxdb_v2_listener", "healths_served", tags)
+		h.readysServed = h.Statistics.Register("influxdb_v2_listener", "readys_served", tags)
+		h.requestsRecv = h.Statistics.Register("influxdb_v2_listener", "requests_received", tags)
+		h.notFoundsServed = h.Statistics.Register("influxdb_v2_listener", "not_founds_served", tags)
+		h.pingsServed = h.Statistics.Register("influxdb_v2_listener", "pings_served", tags)
+		h.authFailures = h.Statistics.Register("influxdb_v2_listener", "auth_failures", tags)
+	}
+
 	if err := h.routes(); err != nil {
 		return err
 	}

plugins/inputs/influxdb_v2_listener/influxdb_v2_listener_benchmark_test.go

@@ -8,20 +8,37 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gofrs/uuid/v5"
+	"github.com/stretchr/testify/require"
+
 	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/selfstat"
 	"github.com/influxdata/telegraf/testutil"
 )
 
 // newListener is the minimal InfluxDBV2Listener construction to serve writes.
-func newListener() *InfluxDBV2Listener {
+func newListener(b *testing.B) *InfluxDBV2Listener {
+	// Make sure we do have a unique stats for this instance to avoid side effects
+	id, err := uuid.NewV4()
+	require.NoError(b, err)
+
 	listener := &InfluxDBV2Listener{
-		timeFunc:     time.Now,
-		acc:          &testutil.NopAccumulator{},
-		bytesRecv:    selfstat.Register("influxdb_v2_listener", "bytes_received", map[string]string{}),
-		writesServed: selfstat.Register("influxdb_v2_listener", "writes_served", map[string]string{}),
-		MaxBodySize:  config.Size(defaultMaxBodySize),
+		timeFunc: time.Now,
+		acc:      &testutil.NopAccumulator{},
+		bytesRecv: selfstat.Register("influxdb_v2_listener", "bytes_received", map[string]string{
+			"id": id.String(),
+		}),
+		writesServed: selfstat.Register("influxdb_v2_listener", "writes_served", map[string]string{
+			"id": id.String(),
+		}),
+		MaxBodySize: config.Size(defaultMaxBodySize),
 	}
+
+	b.Cleanup(func() {
+		listener.bytesRecv.Unregister()
+		listener.writesServed.Unregister()
+	})
+
 	return listener
 }
 
@@ -65,18 +82,14 @@ func BenchmarkInfluxDBV2Listener_serveWrite(b *testing.B) {
 
 	for _, bm := range benchmarks {
 		b.Run(bm.name, func(b *testing.B) {
-			listener := newListener()
+			listener := newListener(b)
 
 			b.ResetTimer()
 			for n := 0; n < b.N; n++ {
 				req, err := http.NewRequest("POST", addr, strings.NewReader(bm.lines))
-				if err != nil {
-					b.Error(err)
-				}
+				require.NoError(b, err)
 				listener.handleWrite()(res, req)
-				if res.Code != http.StatusNoContent {
-					b.Errorf("unexpected status %d", res.Code)
-				}
+				require.Equal(b, http.StatusNoContent, res.Code, "unexpected status")
 			}
 		})
 	}