From 8dc52773d3b65d7afc78af5d3947370dac662a9c Mon Sep 17 00:00:00 2001 From: Calvin Lee Date: Thu, 16 Jul 2026 12:14:40 -0400 Subject: [PATCH] test(metrics): add functional tests verifying component_sent_bytes_total carries component_id labels --- test/framework/functional/metrics.go | 41 +++++++++++++++++ .../cloudwatch/forward_to_cloudwatch_test.go | 35 ++++++++++++++ .../outputs/http/forward_to_http_test.go | 46 ++++++++++++------- 3 files changed, 105 insertions(+), 17 deletions(-) create mode 100644 test/framework/functional/metrics.go diff --git a/test/framework/functional/metrics.go b/test/framework/functional/metrics.go new file mode 100644 index 0000000000..7fa27956e2 --- /dev/null +++ b/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 +} diff --git a/test/functional/outputs/cloudwatch/forward_to_cloudwatch_test.go b/test/functional/outputs/cloudwatch/forward_to_cloudwatch_test.go index 4cbca09887..b95fe43cd9 100644 --- a/test/functional/outputs/cloudwatch/forward_to_cloudwatch_test.go +++ b/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) + } + } + }) + }) }) diff --git a/test/functional/outputs/http/forward_to_http_test.go b/test/functional/outputs/http/forward_to_http_test.go index f00e1ea927..53fdfd8ddd 100644 --- a/test/functional/outputs/http/forward_to_http_test.go +++ b/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) + }) + }) + })