fix(outputs.influxdb_v2): Use dynamic token secret (#16628)

plugins/outputs/influxdb_v2/http.go

@@ -72,17 +72,9 @@ type httpClient struct {
 
 func (c *httpClient) Init() error {
 	if c.headers == nil {
-		c.headers = make(map[string]string, 2)
+		c.headers = make(map[string]string, 1)
 	}
 
-	if _, ok := c.headers["Authorization"]; !ok {
-		token, err := c.token.Get()
-		if err != nil {
-			return fmt.Errorf("getting token failed: %w", err)
-		}
-		c.headers["Authorization"] = "Token " + token.String()
-		token.Destroy()
-	}
 	if _, ok := c.headers["User-Agent"]; !ok {
 		c.headers["User-Agent"] = c.userAgent
 	}
@@ -278,6 +270,15 @@ func (c *httpClient) writeBatch(ctx context.Context, bucket string, metrics []te
 		req.Header.Set("Content-Encoding", c.contentEncoding)
 	}
 	req.Header.Set("Content-Type", "text/plain; charset=utf-8")
+
+	// Set authorization
+	token, err := c.token.Get()
+	if err != nil {
+		return fmt.Errorf("getting token failed: %w", err)
+	}
+	req.Header.Set("Authorization", "Token "+token.String())
+	token.Destroy()
+
 	c.addHeaders(req)
 
 	// Execute the request

plugins/outputs/influxdb_v2/influxdb_v2_test.go

@@ -859,3 +859,50 @@ func TestStatusCodeUnexpected(t *testing.T) {
 		})
 	}
 }
+
+func TestUseDynamicSecret(t *testing.T) {
+	token := "welcome"
+	// Setup a test server
+	ts := httptest.NewServer(
+		http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Header.Get("Authorization") != "Token "+token {
+				w.WriteHeader(http.StatusForbidden)
+			} else {
+				w.WriteHeader(http.StatusNoContent)
+			}
+		}),
+	)
+	defer ts.Close()
+
+	secretToken := config.NewSecret([]byte("wrongtk"))
+	// Setup plugin and connect
+	plugin := &influxdb.InfluxDB{
+		URLs:   []string{"http://" + ts.Listener.Addr().String()},
+		Log:    &testutil.Logger{},
+		Bucket: "my_bucket",
+		Token:  secretToken,
+	}
+	require.NoError(t, plugin.Init())
+	require.NoError(t, plugin.Connect())
+	defer plugin.Close()
+
+	metrics := []telegraf.Metric{
+		metric.New(
+			"cpu",
+			map[string]string{
+				"bucket": "foo",
+			},
+			map[string]interface{}{
+				"value": 0.0,
+			},
+			time.Unix(0, 3),
+		),
+	}
+	// Write the metrics the first time and check for the expected errors
+	err := plugin.Write(metrics)
+	require.ErrorContains(t, err, "failed to write metric to my_bucket")
+	require.ErrorContains(t, err, strconv.Itoa(http.StatusForbidden))
+
+	require.NoError(t, secretToken.Set([]byte(token)))
+	require.NoError(t, plugin.Write(metrics))
+}