Merge pull request #1607 from rancher/logging/otel-traces

logging: create Trace pipeline from OTel to OpenSearch

apis/core/v1beta1/collector_meta.go

@@ -1,5 +1,5 @@
 package v1beta1
 
 func (c *CollectorSpec) IsEmpty() bool {
-	return c.LoggingConfig == nil && c.MetricsConfig == nil
+	return c.LoggingConfig == nil && c.MetricsConfig == nil && c.TracesConfig == nil
 }

apis/core/v1beta1/collector_types.go

@@ -21,6 +21,7 @@ type CollectorSpec struct {
 	AgentEndpoint            string                       `json:"agentEndpoint,omitempty"`
 	SystemNamespace          string                       `json:"systemNamespace,omitempty"`
 	LoggingConfig            *corev1.LocalObjectReference `json:"loggingConfig,omitempty"`
+	TracesConfig             *corev1.LocalObjectReference `json:"tracesConfig,omitempty"`
 	MetricsConfig            *corev1.LocalObjectReference `json:"metricsConfig,omitempty"`
 	ConfigReloader           *ConfigReloaderSpec          `json:"configReloader,omitempty"`
 	LogLevel                 string                       `json:"logLevel,omitempty"`

controllers/core_collector_controller.go

@@ -81,7 +81,7 @@ func (r *CoreCollectorReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		For(&corev1beta1.Collector{}).
 		Watches(&opnimonitoringv1beta1.CollectorConfig{}, requestMapper).
 		Watches(&opniloggingv1beta1.CollectorConfig{}, requestMapper).
-		// for metrics, the we want to watch changes to the spec of objects that drive discovery
+		// for metrics, we want to watch changes to the spec of objects that drive discovery
 		Watches(&promoperatorv1.ServiceMonitor{}, watchAllRequestMapper).
 		Watches(&promoperatorv1.PodMonitor{}, watchAllRequestMapper).
 		Watches(&corev1.Service{}, watchAllRequestMapper).
@@ -111,6 +111,15 @@ func reconcileRequestsForCollector(collectors []corev1beta1.Collector, name stri
 				},
 			})
 		}
+		if c.Spec.TracesConfig != nil && c.Spec.TracesConfig.Name == name {
+			reqs = append(reqs, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Namespace: c.Namespace,
+					Name:      c.Name,
+				},
+			})
+		}
+
 	}
 	return
 }

controllers/core_collector_controller_test.go

@@ -358,6 +358,97 @@ var _ = Describe("Core Collector Controller", Ordered, Label("controller", "slow
 			))
 		})
 	})
+
+	When("creating a collector resource for traces", func() {
+		var (
+			ns                string
+			traceConfig       *opniloggingv1beta1.CollectorConfig
+			traceCollectorObj *opnicorev1beta1.Collector
+		)
+		It("should succeed in creating the objects", func() {
+			ns = makeTestNamespace()
+			traceConfig = &opniloggingv1beta1.CollectorConfig{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-trace-config",
+				},
+				Spec: opniloggingv1beta1.CollectorConfigSpec{
+					Provider: opniloggingv1beta1.LogProviderGeneric,
+				},
+			}
+			traceCollectorObj = &opnicorev1beta1.Collector{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-trace",
+				},
+				Spec: opnicorev1beta1.CollectorSpec{
+					SystemNamespace: ns,
+					AgentEndpoint:   "http://test-endpoint",
+					TracesConfig: &corev1.LocalObjectReference{
+						Name: traceConfig.Name,
+					},
+				},
+			}
+			Expect(k8sClient.Create(context.Background(), traceConfig)).To(Succeed())
+			Expect(k8sClient.Create(context.Background(), traceCollectorObj)).To(Succeed())
+		})
+
+		It("should create aggregator with trace capabilities", func() {
+			By("checking agent configmap")
+			Eventually(Object(&corev1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("%s-agent-config", traceCollectorObj.Name),
+					Namespace: ns,
+				},
+			})).Should(ExistAnd(
+				HaveData("receivers.yaml", nil, "config.yaml", nil),
+				HaveOwner(traceCollectorObj),
+			))
+			By("checking aggregator configmap")
+			Eventually(Object(&corev1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("%s-aggregator-config", traceCollectorObj.Name),
+					Namespace: ns,
+				},
+			})).Should(ExistAnd(
+				HaveData("aggregator.yaml", nil),
+				HaveOwner(traceCollectorObj),
+			))
+			By("checking deployment")
+			Eventually(Object(&appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("%s-collector-aggregator", traceCollectorObj.Name),
+					Namespace: ns,
+				},
+			})).Should(ExistAnd(
+				HaveMatchingContainer(And(
+					HaveName("otel-collector"),
+					HaveVolumeMounts("collector-config"),
+				)),
+				HaveMatchingVolume(And(
+					HaveVolumeSource("ConfigMap"),
+					HaveName("collector-config"),
+				)),
+				HaveOwner(traceCollectorObj),
+			))
+		})
+
+		It("should not create collector agents", func() {
+			By("checking daemonset")
+			Eventually(Object(&appsv1.DaemonSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      fmt.Sprintf("%s-collector-agent", traceCollectorObj.Name),
+					Namespace: ns,
+				},
+			})).Should(Not(ExistAnd(
+				HaveMatchingContainer(And(
+					HaveName("otel-collector"),
+				)),
+				HaveMatchingVolume(And(
+					HaveVolumeSource("ConfigMap"),
+					HaveName("collector-config"),
+				)),
+			)))
+		})
+	})
 })
 
 func promDiscoveryObejcts(ns string) []client.Object {

go.mod

@@ -169,7 +169,7 @@ require (
 
 require (
 	github.com/go-logr/zapr v1.2.4 // indirect
-	go.uber.org/zap v1.25.0 // indirect
+	go.uber.org/zap v1.25.0
 )
 
 require (
@@ -538,7 +538,7 @@ require (
 	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0014 // indirect
 	go.opentelemetry.io/collector/processor v0.84.0 // indirect
 	go.opentelemetry.io/collector/receiver v0.84.0 // indirect
-	go.opentelemetry.io/collector/semconv v0.84.0 // indirect
+	go.opentelemetry.io/collector/semconv v0.84.0
 	go.opentelemetry.io/contrib/propagators/aws v1.17.0 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.17.0 // indirect
 	go.opentelemetry.io/contrib/propagators/jaeger v1.17.0 // indirect

pkg/otel/types.go

@@ -28,6 +28,7 @@ func AgentEndpoint(serviceName string) string {
 type NodeConfig struct {
 	Instance      string
 	Logs          LoggingConfig
+	Traces        TraceConfig
 	Metrics       MetricsConfig
 	Containerized bool
 	LogLevel      string
@@ -37,6 +38,7 @@ type NodeConfig struct {
 type AggregatorConfig struct {
 	AgentEndpoint string
 	LogsEnabled   bool
+	TracesEnabled bool
 	Metrics       MetricsConfig
 	Containerized bool
 	LogLevel      string
@@ -75,6 +77,10 @@ type LoggingConfig struct {
 	Receivers []string
 }
 
+type TraceConfig struct {
+	Enabled bool
+}
+
 type MetricsConfig struct {
 	Enabled             bool
 	LogLevel            string

pkg/resources/collector/templates.go

@@ -308,6 +308,11 @@ service:
       processors: ["transform", "memory_limiter", "batch"]
       exporters: ["otlphttp"]
   {{- end }}
+  {{- if .TracesEnabled }}
+    traces:
+      receivers: ["otlp"]
+      exporters: ["otlphttp"]
+  {{- end }}
   {{ template "metrics-remotewrite-pipeline" .}}
 `
 )

pkg/resources/collector/workloads.go

@@ -66,6 +66,9 @@ func (r *Reconciler) getDaemonConfig(loggingReceivers []string) otel.NodeConfig
 			Enabled:   r.collector.Spec.LoggingConfig != nil,
 			Receivers: loggingReceivers,
 		},
+		Traces: otel.TraceConfig{
+			Enabled: r.collector.Spec.TracesConfig != nil,
+		},
 		Metrics:       lo.FromPtr(r.getMetricsConfig()),
 		Containerized: true,
 		LogLevel:      r.collector.Spec.LogLevel,
@@ -107,6 +110,7 @@ func (r *Reconciler) getAggregatorConfig(
 ) otel.AggregatorConfig {
 	return otel.AggregatorConfig{
 		LogsEnabled:   r.collector.Spec.LoggingConfig != nil,
+		TracesEnabled: r.collector.Spec.TracesConfig != nil,
 		Metrics:       metricsCfg,
 		AgentEndpoint: r.collector.Spec.AgentEndpoint,
 		Containerized: true,

pkg/resources/loggingcluster/opensearch.go

@@ -28,6 +28,7 @@ var (
 			{
 				IndexPatterns: []string{
 					"logs*",
+					"ss4o_traces-kubernetes-opni*",
 				},
 				AllowedActions: []string{
 					"read",

pkg/resources/multiclusterrolebinding/opensearch.go

@@ -88,7 +88,7 @@ func (r *Reconciler) ReconcileOpensearchObjects(opensearchCluster *opensearchv1.
 
 	templates := []opensearchtypes.IndexTemplateSpec{
 		OpniLogTemplate,
-		opniSpanTemplate,
+		OpniSpanTemplate,
 	}
 
 	for _, template := range templates {
@@ -118,12 +118,12 @@ func (r *Reconciler) ReconcileOpensearchObjects(opensearchCluster *opensearchv1.
 		return
 	}
 
-	retErr = reconciler.MaybeBootstrapIndex(LogIndexPrefix, LogIndexAlias, OldIndexPrefixes)
+	retErr = reconciler.MaybeBootstrapIndex(LogIndexPrefix, LogIndexAlias, OldLogIndexPrefixes)
 	if retErr != nil {
 		return
 	}
 
-	retErr = reconciler.MaybeBootstrapIndex(spanIndexPrefix, spanIndexAlias, oldTracingIndexPrefixes)
+	retErr = reconciler.MaybeBootstrapIndex(SpanIndexPrefix, SpanIndexAlias, OldSpanIndexPrefixes)
 	if retErr != nil {
 		return
 	}