diff --git a/test/framework/functional/metrics.go b/test/framework/functional/metrics.go new file mode 100644 index 0000000000..1b1af99eef --- /dev/null +++ b/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 +} diff --git a/test/functional/outputs/aws/cloudwatch/forward_to_cloudwatch_test.go b/test/functional/outputs/aws/cloudwatch/forward_to_cloudwatch_test.go index 6373391de8..a8054c7418 100644 --- a/test/functional/outputs/aws/cloudwatch/forward_to_cloudwatch_test.go +++ b/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) + } + } + }) + }) }) diff --git a/test/functional/outputs/aws/s3/forward_to_s3_test.go b/test/functional/outputs/aws/s3/forward_to_s3_test.go index 468aa6a045..caff5301c5 100644 --- a/test/functional/outputs/aws/s3/forward_to_s3_test.go +++ b/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) + } + } + }) + }) }) diff --git a/test/functional/outputs/http/forward_to_http_test.go b/test/functional/outputs/http/forward_to_http_test.go index 3cebc4c789..74e1055f49 100644 --- a/test/functional/outputs/http/forward_to_http_test.go +++ b/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) + } + }) + }) })