test(metrics): add functional tests verifying component_sent_bytes_total carries component_id labels

test/framework/functional/metrics.go

@@ -0,0 +1,87 @@
+package functional
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"github.com/openshift/cluster-logging-operator/internal/constants"
+	"github.com/openshift/cluster-logging-operator/internal/runtime"
+)
+
+// SetupMetricsRBAC creates the cluster-scoped RBAC resources needed to scrape
+// the collector's /metrics endpoint from within a test pod. Names are prefixed
+// with the framework's namespace so parallel test packages cannot collide.
+// The returned function deletes all created resources and should be registered
+// with DeferCleanup or called in AfterEach.
+func (f *CollectorFunctionalFramework) SetupMetricsRBAC() (metricsReaderRole *rbacv1.ClusterRole, metricsReaderBinding *rbacv1.ClusterRoleBinding, tokenReviewBinding *rbacv1.ClusterRoleBinding, err error) {
+	roleName := fmt.Sprintf("%s-%s-metrics-reader", f.Test.NS.Name, f.Name)
+
+	metricsReaderRole = runtime.NewClusterRole(
+		roleName,
+		runtime.NewNonResourceURLPolicyRule([]string{"/metrics"}, []string{"get"}),
+	)
+	if err = f.Test.Create(metricsReaderRole); err != nil {
+		return nil, nil, nil, err
+	}
+
+	metricsReaderBinding = runtime.NewClusterRoleBinding(
+		roleName,
+		runtime.NewClusterRoleRef(roleName),
+		runtime.NewServiceAccountSubject("default", f.Namespace),
+	)
+	if err = f.Test.Create(metricsReaderBinding); err != nil {
+		return nil, nil, nil, err
+	}
+
+	tokenReviewName := fmt.Sprintf("%s-%s-token-reviewer", f.Test.NS.Name, f.Name)
+	tokenReviewBinding = runtime.NewClusterRoleBinding(
+		tokenReviewName,
+		runtime.NewClusterRoleRef("system:auth-delegator"),
+		runtime.NewServiceAccountSubject("default", f.Namespace),
+	)
+	if err = f.Test.Create(tokenReviewBinding); err != nil {
+		return nil, nil, nil, err
+	}
+
+	return metricsReaderRole, metricsReaderBinding, tokenReviewBinding, nil
+}
+
+// CollectMetricLines polls the collector's Prometheus endpoint until a line
+// matching both metricName and waitFor is found, then returns all lines
+// matching metricName.
+func (f *CollectorFunctionalFramework) CollectMetricLines(metricName, waitFor string, timeout time.Duration) ([]string, error) {
+	var matched []string
+	var lastErr error
+	err := wait.PollUntilContextTimeout(context.TODO(), 3*time.Second, timeout, true, func(ctx context.Context) (bool, error) {
+		raw, err := f.RunCommand(constants.CollectorName, "bash", "-c",
+			fmt.Sprintf("curl -ks --max-time 10 -H \"Authorization: Bearer $(cat /var/run/secrets/kubernetes.io/serviceaccount/token)\" https://%s.%s:24231/metrics", f.Name, f.Namespace))
+		if err != nil {
+			lastErr = err
+			return false, nil
+		}
+		matched = nil
+		for _, line := range strings.Split(raw, "\n") {
+			if strings.HasPrefix(line, "#") {
+				continue
+			}
+			if strings.Contains(line, metricName) {
+				matched = append(matched, line)
+			}
+		}
+		for _, line := range matched {
+			if strings.Contains(line, waitFor) {
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+	if err != nil && lastErr != nil {
+		return matched, fmt.Errorf("%w (last scrape error: %v)", err, lastErr)
+	}
+	return matched, err
+}

test/functional/outputs/aws/cloudwatch/forward_to_cloudwatch_test.go

@@ -241,4 +241,49 @@ var _ = Describe("[Functional][Outputs][CloudWatch] Forward Output to CloudWatch
 			Entry("should pass with zstd", "zstd"),
 		)
 	})
+
+	// Regression test for https://redhat.atlassian.net/browse/LOG-7893
+	//
+	// vector_component_sent_bytes_total must carry component_id, component_kind,
+	// and component_type labels so dashboards can filter by component_kind="sink".
+	// A bug in Vector's AwsBytesSent causes these labels to be lost when the
+	// metric is emitted from tower::buffer worker tasks outside the component
+	// tracing span.
+	Context("When checking collector metrics for CloudWatch output", func() {
+		BeforeEach(func() {
+			role, binding, tokenBinding, err := framework.SetupMetricsRBAC()
+			Expect(err).To(Succeed())
+			DeferCleanup(func() {
+				_ = framework.Test.Delete(tokenBinding)
+				_ = framework.Test.Delete(binding)
+				_ = framework.Test.Delete(role)
+			})
+		})
+
+		It("should emit component_sent_bytes_total with component_id and region labels, and no unlabeled duplicates (LOG-7893)", func() {
+			obstestruntime.NewClusterLogForwarderBuilder(framework.Forwarder).
+				FromInput(obs.InputTypeApplication).
+				ToCloudwatchOutput(*obsCwAuth)
+			framework.Secrets = append(framework.Secrets, secret)
+
+			Expect(framework.Deploy()).To(BeNil())
+
+			Expect(framework.WritesNApplicationLogsOfSize(numOfLogs, logSize, 0)).To(BeNil())
+			// Allow logs to populate in CloudWatch
+			time.Sleep(10 * time.Second)
+
+			lines, err := framework.CollectMetricLines("component_sent_bytes_total", `component_id="output_cloudwatch"`, 30*time.Second)
+			Expect(err).To(BeNil(), "Timed out waiting for component_sent_bytes_total metric")
+
+			for _, line := range lines {
+				log.V(2).Info("component_sent_bytes_total line", "line", line)
+				Expect(line).To(ContainSubstring(`component_id=`),
+					"component_sent_bytes_total without component_id label (transport-layer duplicate): %s", line)
+				if strings.Contains(line, `component_id="output_cloudwatch"`) {
+					Expect(line).To(ContainSubstring(`region=`),
+						"component_sent_bytes_total for CloudWatch output is missing region label: %s", line)
+				}
+			}
+		})
+	})
 })

test/functional/outputs/aws/s3/forward_to_s3_test.go

@@ -2,6 +2,7 @@ package s3
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	log "github.com/ViaQ/logerr/v2/log/static"
@@ -187,4 +188,49 @@ var _ = Describe("[Functional][Outputs][S3] Forward Output to s3 (minIO)", func(
 			Entry("should pass with zstd", "zstd"),
 		)
 	})
+
+	// Regression test for https://redhat.atlassian.net/browse/LOG-7893
+	//
+	// vector_component_sent_bytes_total must carry component_id, component_kind,
+	// and component_type labels so dashboards can filter by component_kind="sink".
+	// A bug in Vector's AwsBytesSent causes these labels to be lost when the
+	// metric is emitted from tower::buffer worker tasks outside the component
+	// tracing span.
+	Context("When checking collector metrics for S3 output", func() {
+		BeforeEach(func() {
+			role, binding, tokenBinding, err := framework.SetupMetricsRBAC()
+			Expect(err).To(Succeed())
+			DeferCleanup(func() {
+				_ = framework.Test.Delete(tokenBinding)
+				_ = framework.Test.Delete(binding)
+				_ = framework.Test.Delete(role)
+			})
+		})
+
+		It("should emit component_sent_bytes_total with component_id and region labels, and no unlabeled duplicates (LOG-7893)", func() {
+			const keyPrefix = "metrics-test/"
+
+			setupS3Output(obs.InputTypeApplication, "none", keyPrefix)
+			Expect(framework.Deploy()).To(BeNil())
+
+			Expect(framework.SetupS3Bucket(TestBucketName)).
+				To(Succeed(), "should set up the minIO test bucket")
+
+			Expect(framework.WritesNApplicationLogsOfSize(numOfLogs, logSize, 0)).To(BeNil())
+			time.Sleep(15 * time.Second)
+
+			lines, err := framework.CollectMetricLines("component_sent_bytes_total", `component_id="output_s3"`, 30*time.Second)
+			Expect(err).To(BeNil(), "Timed out waiting for component_sent_bytes_total metric")
+
+			for _, line := range lines {
+				log.V(2).Info("component_sent_bytes_total line", "line", line)
+				Expect(line).To(ContainSubstring(`component_id=`),
+					"component_sent_bytes_total without component_id label (transport-layer duplicate): %s", line)
+				if strings.Contains(line, `component_id="output_s3"`) {
+					Expect(line).To(ContainSubstring(`region=`),
+						"component_sent_bytes_total for S3 output is missing region label: %s", line)
+				}
+			}
+		})
+	})
 })

test/functional/outputs/http/forward_to_http_test.go

@@ -10,6 +10,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/resource"
 
+	log "github.com/ViaQ/logerr/v2/log/static"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	obs "github.com/openshift/cluster-logging-operator/api/observability/v1"
@@ -130,27 +131,15 @@ var _ = Describe("[Functional][Outputs][Http] Functional tests", func() {
 	})
 
 	Context("with tuning parameters", func() {
-		var (
-			addDestinationContainer func(f *functional.CollectorFunctionalFramework) runtime.PodBuilderVisitor
-		)
 		DescribeTable("with compression", func(compression string) {
-			framework = functional.NewCollectorFunctionalFramework()
-			obstestruntime.NewClusterLogForwarderBuilder(framework.Forwarder).
-				FromInput(obs.InputTypeApplication).
-				ToHttpOutput(func(output *obs.OutputSpec) {
-					output.HTTP.Tuning = &obs.HTTPTuningSpec{
-						Compression: compression,
-					}
-				})
-
-			addDestinationContainer = func(f *functional.CollectorFunctionalFramework) runtime.PodBuilderVisitor {
-				return func(b *runtime.PodBuilder) error {
-					return f.AddVectorHttpOutput(b, f.Forwarder.Spec.Outputs[0])
-				}
+			framework.Forwarder.Spec.Outputs[0].HTTP.Tuning = &obs.HTTPTuningSpec{
+				Compression: compression,
 			}
 
 			Expect(framework.DeployWithVisitors([]runtime.PodBuilderVisitor{
-				addDestinationContainer(framework),
+				func(b *runtime.PodBuilder) error {
+					return framework.AddVectorHttpOutput(b, framework.Forwarder.Spec.Outputs[0])
+				},
 				func(builder *runtime.PodBuilder) error {
 					builder.AddLabels(map[string]string{
 						"app.kubernetes.io/name": "somevalue",
@@ -204,4 +193,41 @@ var _ = Describe("[Functional][Outputs][Http] Functional tests", func() {
 			}, obs.AuditSourceOpenShift),
 		)
 	})
+
+	// Verify that component_sent_bytes_total carries component_id for the HTTP
+	// output. This serves as a positive control alongside the CloudWatch
+	// regression test for LOG-7893 — the HTTP sink emits bytes metrics from
+	// within the Driver's future context (not a spawned buffer worker), so its
+	// labels should always be correct.
+	Context("When checking collector metrics for HTTP output", func() {
+		BeforeEach(func() {
+			role, binding, tokenBinding, err := framework.SetupMetricsRBAC()
+			Expect(err).To(Succeed())
+			DeferCleanup(func() {
+				_ = framework.Test.Delete(tokenBinding)
+				_ = framework.Test.Delete(binding)
+				_ = framework.Test.Delete(role)
+			})
+		})
+
+		It("should emit component_sent_bytes_total with component_id label", func() {
+			Expect(framework.DeployWithVisitors([]runtime.PodBuilderVisitor{
+				func(b *runtime.PodBuilder) error {
+					return framework.AddVectorHttpOutput(b, framework.Forwarder.Spec.Outputs[0])
+				},
+			})).To(BeNil())
+
+			msg := functional.NewCRIOLogMessage(functional.CRIOTime(time.Now()), "metrics test message", false)
+			Expect(framework.WriteMessagesToApplicationLog(msg, 10)).To(BeNil())
+
+			lines, err := framework.CollectMetricLines("component_sent_bytes_total", `component_id="output_http"`, 30*time.Second)
+			Expect(err).To(BeNil(), "Timed out waiting for component_sent_bytes_total with component_id label")
+
+			for _, line := range lines {
+				log.V(2).Info("component_sent_bytes_total line", "line", line)
+				Expect(line).To(ContainSubstring(`component_id=`),
+					"component_sent_bytes_total without component_id label (transport-layer duplicate): %s", line)
+			}
+		})
+	})
 })