test(processors): Fix unit-test for tracking metrics (#14844)

plugins/processors/converter/converter_test.go

@@ -836,6 +836,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/date/date_test.go

@@ -289,6 +289,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/defaults/defaults_test.go

@@ -195,6 +195,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/execd/execd_test.go

@@ -391,6 +391,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/filepath/filepath_test.go

@@ -129,6 +129,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/ifname/ifname_test.go

@@ -228,6 +228,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/lookup/lookup_test.go

@@ -162,7 +162,7 @@ func TestCasesTracking(t *testing.T) {
 			testutil.RequireMetricsEqual(t, expected, actual)
 
 			// Simulate output acknowledging delivery
-			for _, m := range input {
+			for _, m := range actual {
 				m.Accept()
 			}
 
@@ -170,7 +170,7 @@ func TestCasesTracking(t *testing.T) {
 			require.Eventuallyf(t, func() bool {
 				mu.Lock()
 				defer mu.Unlock()
-				return len(expected) == len(delivered)
+				return len(input) == len(delivered)
 			}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 		})
 	}

plugins/processors/regex/regex_test.go

@@ -1,6 +1,7 @@
 package regex
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -992,6 +993,56 @@ func TestAnyFieldConversion(t *testing.T) {
 }
 
 func TestTrackedMetricNotLost(t *testing.T) {
+	now := time.Now()
+
+	// Setup raw input and expected output
+	inputRaw := testutil.MustMetric(
+		"access_log",
+		map[string]string{
+			"verb":      "GET",
+			"resp_code": "200",
+		},
+		map[string]interface{}{
+			"request":       "/api/search/?category=plugins&q=regex&sort=asc",
+			"ignore_number": int64(200),
+			"ignore_bool":   true,
+		},
+		now,
+	)
+
+	expected := []telegraf.Metric{
+		metric.New(
+			"access_log",
+			map[string]string{
+				"verb":            "GET",
+				"resp_code":       "200",
+				"resp_code_group": "2xx",
+				"resp_code_text":  "OK",
+			},
+			map[string]interface{}{
+				"request":         "/api/search/?category=plugins&q=regex&sort=asc",
+				"method":          "/search/",
+				"search_category": "plugins",
+				"ignore_number":   int64(200),
+				"ignore_bool":     true,
+			},
+			now,
+		),
+	}
+
+	// Create fake notification for testing
+	var mu sync.Mutex
+	delivered := make([]telegraf.DeliveryInfo, 0, 1)
+	notify := func(di telegraf.DeliveryInfo) {
+		mu.Lock()
+		defer mu.Unlock()
+		delivered = append(delivered, di)
+	}
+
+	// Convert raw input to tracking metric
+	input, _ := metric.WithTracking(inputRaw, notify)
+
+	// Prepare and start the plugin
 	regex := Regex{
 		Tags: []converter{
 			{
@@ -1025,32 +1076,19 @@ func TestTrackedMetricNotLost(t *testing.T) {
 	}
 	require.NoError(t, regex.Init())
 
-	m := newM2().Copy()
+	// Process expected metrics and compare with resulting metrics
-	var delivered bool
+	actual := regex.Apply(input)
-	notify := func(telegraf.DeliveryInfo) {
+	testutil.RequireMetricsEqual(t, expected, actual)
-		delivered = true
-	}
-	m, _ = metric.WithTracking(m, notify)
-	processed := regex.Apply(m)
-	processed[0].Accept()
 
-	expectedFields := map[string]interface{}{
+	// Simulate output acknowledging delivery
-		"request":         "/api/search/?category=plugins&q=regex&sort=asc",
+	for _, m := range actual {
-		"method":          "/search/",
+		m.Accept()
-		"search_category": "plugins",
-		"ignore_number":   int64(200),
-		"ignore_bool":     true,
-	}
-	expectedTags := map[string]string{
-		"verb":            "GET",
-		"resp_code":       "200",
-		"resp_code_group": "2xx",
-		"resp_code_text":  "OK",
 	}
 
-	require.Equal(t, expectedFields, processed[0].Fields())
+	// Check delivery
-	require.Equal(t, expectedTags, processed[0].Tags())
+	require.Eventuallyf(t, func() bool {
-	require.Eventually(t, func() bool {
+		mu.Lock()
-		return delivered
+		defer mu.Unlock()
-	}, time.Second, 100*time.Millisecond, "metric not delivered")
+		return len(delivered) == 1
+	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/rename/rename_test.go

@@ -124,6 +124,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/reverse_dns/reverse_dns_test.go

@@ -128,6 +128,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/s2geo/s2geo_test.go

@@ -118,6 +118,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/scale/scale_test.go

@@ -546,6 +546,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/tag_limit/tag_limit_test.go

@@ -148,6 +148,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }

plugins/processors/template/template_test.go

@@ -292,12 +292,16 @@ func TestTracking(t *testing.T) {
 	testutil.RequireMetricsEqual(t, expected, actual)
 
 	// Simulate output acknowledging delivery
-	input.Accept()
+	for _, m := range actual {
+		m.Accept()
+	}
+
+	// Check delivery
 
 	// Check delivery
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(delivered) > 0
+		return len(delivered) == 1
 	}, time.Second, 100*time.Millisecond, "%d delivered but 1 expected", len(delivered))
 }

plugins/processors/topk/topk_test.go

@@ -576,6 +576,6 @@ func TestTracking(t *testing.T) {
 	require.Eventuallyf(t, func() bool {
 		mu.Lock()
 		defer mu.Unlock()
-		return len(expected) == len(delivered)
+		return len(input) == len(delivered)
 	}, time.Second, 100*time.Millisecond, "%d delivered but %d expected", len(delivered), len(expected))
 }