diff --git a/Makefile b/Makefile index 47bf484d8..02b81f92d 100644 --- a/Makefile +++ b/Makefile @@ -26,7 +26,7 @@ OPENSHIFT_VERSIONS?="v4.20-v4.23" export LOGGING_VERSION?=6.6 export VERSION=$(LOGGING_VERSION).0 export NAMESPACE?=openshift-logging -export LOKI_OPERATOR_CHANNEL?=stable-6.4 +export LOKI_OPERATOR_CHANNEL?=stable-6.6 IMAGE_LOGGING_VECTOR?=quay.io/openshift-logging/vector:v0.54.0 IMAGE_VECTOR_RECEIVER?=quay.io/openshift-logging/vector:v0.54.0-devel diff --git a/test/e2e/logforwarding/lokistack/forward_to_lokistack_test.go b/test/e2e/logforwarding/lokistack/forward_to_lokistack_test.go index 798fac099..d6533a952 100644 --- a/test/e2e/logforwarding/lokistack/forward_to_lokistack_test.go +++ b/test/e2e/logforwarding/lokistack/forward_to_lokistack_test.go @@ -15,11 +15,30 @@ import ( "k8s.io/apimachinery/pkg/api/resource" ) +// suiteFramework owns the loki operator, which is a cluster singleton shared by +// every spec. It is deployed once before any spec runs and torn down once after +// all specs complete, rather than being redeployed in each spec's BeforeEach. +var suiteFramework *framework.E2ETestFramework + +var _ = BeforeSuite(func() { + suiteFramework = framework.NewE2ETestFramework() + if err := suiteFramework.DeployLokiOperator(); err != nil { + Fail(err.Error()) + } +}) + +var _ = AfterSuite(func() { + if suiteFramework != nil { + suiteFramework.Cleanup() + } +}) + var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { const ( forwarderName = "my-forwarder" logGenName = "log-generator" outputName = "lokistack-output" + otlpKey = "k8s_container_name" ) var ( err error @@ -38,9 +57,7 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { if err = e2e.DeployMinio(); err != nil { Fail(err.Error()) } - if err = e2e.DeployLokiOperator(); err != nil { - Fail(err.Error()) - } + // The loki operator is deployed once for the suite (see BeforeSuite). if lokistackReceiver, err = e2e.DeployLokistackInNamespace(deployNS); err != nil { Fail(err.Error()) } @@ -99,7 +116,12 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { // Deploy log generator logGenNS = e2e.CreateTestNamespaceWithPrefix("clo-test-loader") generatorOpt := framework.NewDefaultLogGeneratorOptions() + // Generate continuously but at a low rate. An unbounded, zero-delay stream + // overruns the lokistack per-stream ingestion rate limit (especially in the + // Otel data model, whose entries carry more labels), so the generator's logs + // get rejected and are never ingested within the query timeout. generatorOpt.Count = -1 + generatorOpt.Delay = 100 * time.Millisecond if err = e2e.DeployLogGeneratorWithNamespaceName(logGenNS, logGenName, generatorOpt); err != nil { Fail(fmt.Sprintf("unable to deploy log generator %v.", err)) } @@ -116,11 +138,10 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { Fail(err.Error()) } - found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout) + found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout, otlpKey, logGenName) Expect(err).To(BeNil()) Expect(found).To(BeTrue()) - otlpKey := "k8s_container_name" res, err := lokistackReceiver.GetApplicationLogsByKeyValue(serviceAccount.Name, otlpKey, logGenName, framework.DefaultWaitForLogsTimeout) Expect(err).To(BeNil()) Expect(res).ToNot(BeEmpty()) @@ -145,7 +166,7 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { Fail(err.Error()) } - found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout) + found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout, otlpKey, logGenName) Expect(err).To(BeNil()) Expect(found).To(BeTrue()) @@ -298,7 +319,7 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { Fail(err.Error()) } - found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout) + found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout, otlpKey, logGenName) Expect(err).To(BeNil()) Expect(found).To(BeTrue()) @@ -312,7 +333,7 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { stream := res[0].Stream Expect(len(stream)).To(Equal(10)) wantStreamLabels := []string{ - "k8s_container_name", + otlpKey, "k8s_namespace_name", "k8s_pod_name", "k8s_node_name", @@ -342,7 +363,7 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { Fail(err.Error()) } - found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout) + found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout, otlpKey, logGenName) Expect(err).To(BeNil()) Expect(found).To(BeTrue()) @@ -391,7 +412,7 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { Fail(fmt.Sprintf("unable to deploy trace log generator %v.", err)) } - found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout) + found, err := lokistackReceiver.HasApplicationLogs(serviceAccount.Name, framework.DefaultWaitForLogsTimeout, otlpKey, logGenName) Expect(err).To(BeNil()) Expect(found).To(BeTrue()) @@ -407,8 +428,8 @@ var _ = Describe("[ClusterLogForwarder] Forward to Lokistack", func() { streamLabels := stream.Stream _, hasK8sNamespace := streamLabels["k8s_namespace_name"] Expect(hasK8sNamespace).To(BeTrue(), "Expected OTEL-style label k8s_namespace_name") - _, hasK8sContainer := streamLabels["k8s_container_name"] - Expect(hasK8sContainer).To(BeTrue(), "Expected OTEL-style label k8s_container_name") + _, hasK8sContainer := streamLabels[otlpKey] + Expect(hasK8sContainer).To(BeTrue(), "Expected OTEL-style label otlpKey") if verifyTraceContextLabels { actTraceId, found := streamLabels["trace_id"] diff --git a/test/framework/e2e/lokistack.go b/test/framework/e2e/lokistack.go index 047fda7b9..f4bd66706 100644 --- a/test/framework/e2e/lokistack.go +++ b/test/framework/e2e/lokistack.go @@ -23,6 +23,7 @@ import ( lokitesthelper "github.com/openshift/cluster-logging-operator/test/helpers/loki" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -58,7 +59,7 @@ const ( minioImage = "quay.io/openshift-logging/minio-mirror:RELEASE.2025-09-07T16-13-09Z" ) -var lokiOperatorChannel = "stable-6.4" +var lokiOperatorChannel = "stable-6.6" func init() { if value := os.Getenv("LOKI_OPERATOR_CHANNEL"); value != "" { @@ -196,6 +197,28 @@ minio server /data --console-address ":9001" } func (tc *E2ETestFramework) DeployLokiOperator() error { + // Only install when the operator deployment is confirmed absent. If it already + // exists (e.g. pre-installed on the cluster or left over from a prior run), + // reinstalling would register cleanups that delete an operator we did not + // create, so handle the existing deployment without entering the install path. + dep, err := tc.KubeClient.AppsV1().Deployments(test.OpenshiftOperatorsRedhatNS).Get(context.TODO(), lokiOperatorDeploymentName, metav1.GetOptions{}) + switch { + case err == nil: + if dep.Spec.Replicas != nil && dep.Status.AvailableReplicas == *dep.Spec.Replicas { + clolog.V(1).Info("loki operator already available; skipping install", + "namespace", test.OpenshiftOperatorsRedhatNS, "deployment", lokiOperatorDeploymentName) + return nil + } + // Deployment exists but is not yet fully available; wait for it rather + // than attempting a second install. + clolog.V(1).Info("loki operator deployment exists but not yet available; waiting", + "namespace", test.OpenshiftOperatorsRedhatNS, "deployment", lokiOperatorDeploymentName) + return tc.WaitForDeployment(test.OpenshiftOperatorsRedhatNS, lokiOperatorDeploymentName, defaultRetryInterval, defaultTimeout) + case !apierrors.IsNotFound(err): + // Unexpected error looking up the deployment; do not guess that it is absent. + return err + } + clolog.V(1).Info("deploying loki operator", "namespace", test.OpenshiftOperatorsRedhatNS, "channel", lokiOperatorChannel) operatorGroupYaml := ` apiVersion: operators.coreos.com/v1 @@ -228,7 +251,7 @@ spec: // Create operator group ogUri := fmt.Sprintf(operatorGroupURI, test.OpenshiftOperatorsRedhatNS, "loki-operator-group") - err := tc.Client().RESTClient().Post(). + err = tc.Client().RESTClient().Post(). RequestURI(ogUri). SetHeader("Content-Type", "application/yaml"). Body([]byte(operatorGroupYaml)). @@ -297,6 +320,14 @@ spec: storageClassName: gp3-csi tenants: mode: openshift-logging + openshift: + otlp: + # Promote k8s.container.name, k8s.pod.name, etc. to Loki stream labels so + # that OTLP-mode queries can use {k8s_container_name="..."} as a stream + # selector. Without this the loki operator only promotes the required base + # attributes (log_type, log_source, …) and container-scoped queries never + # match. + enableConsoleLabels: true rules: enabled: true selector: @@ -308,8 +339,8 @@ spec: limits: global: ingestion: - ingestionBurstSize: 10 - ingestionRate: 10 + ingestionBurstSize: 50 + ingestionRate: 50 `, LokistackName, namespace, minioName+"-secret") uri := fmt.Sprintf(lokistackURI, namespace, LokistackName) @@ -457,8 +488,17 @@ func (ls LokistackLogStore) GetApplicationLogsWithPipeline(saName, expression st return result, errors.Wrap(err, "error getting application logs with log pipeline expression") } -func (ls LokistackLogStore) HasApplicationLogs(saName string, timeToWait time.Duration) (bool, error) { - query := fmt.Sprintf(`{log_type=%q}`, obs.InputTypeApplication) +// HasApplicationLogs reports whether any application logs have been ingested. Optional +// labelKeyValues (alternating key, value arguments) are appended to the stream selector +// as additional {key="value"} matchers to narrow the query, e.g. to a specific +// container so the check is not satisfied by collateral application logs from another +// workload. An incomplete trailing key without a value is ignored. +func (ls LokistackLogStore) HasApplicationLogs(saName string, timeToWait time.Duration, labelKeyValues ...string) (bool, error) { + selector := fmt.Sprintf("log_type=%q", obs.InputTypeApplication) + for i := 0; i+1 < len(labelKeyValues); i += 2 { + selector += fmt.Sprintf(", %s=%q", labelKeyValues[i], labelKeyValues[i+1]) + } + query := fmt.Sprintf("{%s}", selector) result, err := ls.QueryUntil(query, "", string(obs.InputTypeApplication), saName, 1, timeToWait) return len(result) > 0, errors.Wrap(err, "error determining if logstore has application logs") } @@ -481,9 +521,50 @@ func (ls LokistackLogStore) HasAuditLogs(saName string, timeToWait time.Duration return len(result) > 0, errors.Wrap(err, "error determining if logstore has audit logs") } +// waitForLokistackReadyCondition waits until the LokiStack custom resource reports its +// "Ready" condition as true. The loki operator sets this only once all components — +// including the ingesters — are available and the stack can actually ingest. Gating on +// the workloads alone can return before the stack is serving, because a Deployment/ +// StatefulSet can report ready replicas a moment before Loki's ring is active. +func (tc *E2ETestFramework) waitForLokistackReadyCondition(namespace, name string) error { + uri := fmt.Sprintf(lokistackURI, namespace, name) + return wait.PollUntilContextTimeout(context.TODO(), defaultRetryInterval, defaultTimeout, true, func(ctx context.Context) (bool, error) { + raw, err := tc.Client().RESTClient().Get().RequestURI(uri).DoRaw(ctx) + if err != nil { + // The LokiStack may not exist yet while it is being created, so keep + // polling on NotFound. Any other error (e.g. Forbidden) will not be + // resolved by waiting, so return it rather than run out the timeout. + if !apierrors.IsNotFound(err) { + return false, err + } + return false, nil + } + lokistack := struct { + Status struct { + Conditions []metav1.Condition `json:"conditions"` + } `json:"status"` + }{} + if err := json.Unmarshal(raw, &lokistack); err != nil { + return false, nil + } + for _, c := range lokistack.Status.Conditions { + if c.Type == "Ready" && c.Status == metav1.ConditionTrue { + return true, nil + } + } + return false, nil + }) +} + func (tc *E2ETestFramework) WaitForLokistackReady(namespace string) error { var err error + // Gate on the operator's own Ready condition first: it reports ready only when all + // components (including the ingesters) are up and the stack can ingest. + if err = tc.waitForLokistackReadyCondition(namespace, LokistackName); err != nil { + return errors.Wrap(err, "error waiting for lokistack Ready condition") + } + // Wait stateful sets if err = tc.waitForStatefulSet(namespace, LokistackName+lokistackCompactor, defaultRetryInterval, defaultTimeout); err != nil { return errors.Wrap(err, "error waiting for lokistack-compactor")