[release-6.2] test(metrics): add functional tests verifying component_sent_bytes_total carries component_id labels

test/framework/functional/metrics.go

@@ -0,0 +1,41 @@
+package functional
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/openshift/cluster-logging-operator/internal/constants"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+// 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
+	err := wait.PollUntilContextTimeout(context.TODO(), 3*time.Second, timeout, true, func(ctx context.Context) (bool, error) {
+		raw, err := f.RunCommand(constants.CollectorName, "curl", "-ks",
+			fmt.Sprintf("https://%s.%s:24231/metrics", f.Name, f.Namespace))
+		if err != nil {
+			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
+	})
+	return matched, err
+}

test/functional/outputs/cloudwatch/forward_to_cloudwatch_test.go

@@ -242,4 +242,39 @@ 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() {
+		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/http/forward_to_http_test.go

@@ -9,6 +9,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/resource"
 
+	log "github.com/ViaQ/logerr/v2/log/static"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/ginkgo/extensions/table"
 	. "github.com/onsi/gomega"
@@ -107,27 +108,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",
@@ -150,4 +139,27 @@ var _ = Describe("[Functional][Outputs][Http] Functional tests", func() {
 			Entry("should pass with no compression", "none"))
 	})
 
+	// 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() {
+		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")
+
+			log.V(2).Info("matched metric lines", "lines", lines)
+		})
+	})
+
 })