Merge branch 'main' of https://github.com/kyma-project/telemetry-manager

Author: NHingerl

.github/instructions/githubactions.instructions.md

@@ -4,9 +4,13 @@
 /docs/ @kyma-project/technical-writers
 /docs/user/integration/sample-app @kyma-project/observability
 
+
 # All .md files
 *.md @kyma-project/technical-writers
 
+# provide a more specific rule for the /.github directory, to remove @kyma-project/technical-writers ownership
+/.github/**/*.md @kyma-project/observability
+
 # JSON, NDJSON files
 *.ndjson @kyma-project/observability
 *.json @kyma-project/observability

.github/instructions/golang.instructions.md

@@ -219,6 +219,18 @@ docker-build: ## Build docker image with the manager
 docker-push: ## Push docker image with the manager
 	docker push ${MANAGER_IMAGE}
 
+.PHONY: docker-build-selfmonitor
+docker-build-selfmonitor: ## Build docker image for telemetry self-monitor
+	@set -a && . dependencies/telemetry-self-monitor/envs && set +a && \
+	docker build -t ${SELF_MONITOR_IMAGE} \
+		--build-arg ALPINE_VERSION=$${ALPINE_VERSION} \
+		--build-arg PROMETHEUS_VERSION=$${PROMETHEUS_VERSION} \
+		dependencies/telemetry-self-monitor
+
+.PHONY: docker-push-selfmonitor
+docker-push-selfmonitor: ## Push docker image for telemetry self-monitor
+	docker push ${SELF_MONITOR_IMAGE}
+
 ##@ Development
 
 .PHONY: run

CODEOWNERS

@@ -126,10 +126,9 @@ func NewLogPipelineController(config LogPipelineControllerConfig, client client.
 
 	reconciler := logpipeline.New(
 		client,
-		overrides.New(config.Global, client),
-		pipelineSyncer,
-		fluentBitReconciler,
-		otelReconciler,
+		logpipeline.WithOverridesHandler(overrides.New(config.Global, client)),
+		logpipeline.WithPipelineSyncer(pipelineSyncer),
+		logpipeline.WithReconcilers(fluentBitReconciler, otelReconciler),
 	)
 
 	return &LogPipelineController{
@@ -195,12 +194,12 @@ func (r *LogPipelineController) mapTelemetryChanges(ctx context.Context, object
 }
 
 func configureFluentBitReconciler(config LogPipelineControllerConfig, client client.Client, flowHealthProber *prober.FluentBitProber, pipelineLock logpipelinefluentbit.PipelineLock) (*logpipelinefluentbit.Reconciler, error) {
-	pipelineValidator := &logpipelinefluentbit.Validator{
-		EndpointValidator:  &endpoint.Validator{Client: client},
-		TLSCertValidator:   tlscert.New(client),
-		SecretRefValidator: &secretref.Validator{Client: client},
-		PipelineLock:       pipelineLock,
-	}
+	pipelineValidator := logpipelinefluentbit.NewValidator(
+		logpipelinefluentbit.WithEndpointValidator(&endpoint.Validator{Client: client}),
+		logpipelinefluentbit.WithTLSCertValidator(tlscert.New(client)),
+		logpipelinefluentbit.WithSecretRefValidator(&secretref.Validator{Client: client}),
+		logpipelinefluentbit.WithValidatorPipelineLock(pipelineLock),
+	)
 
 	fluentBitApplierDeleter := fluentbit.NewFluentBitApplierDeleter(
 		config.TargetNamespace(),
@@ -218,16 +217,19 @@ func configureFluentBitReconciler(config LogPipelineControllerConfig, client cli
 	}
 
 	fbReconciler := logpipelinefluentbit.New(
-		config.Global,
-		client,
-		fluentBitConfigBuilder,
-		fluentBitApplierDeleter,
-		&workloadstatus.DaemonSetProber{Client: client},
-		flowHealthProber,
-		istiostatus.NewChecker(discoveryClient),
-		pipelineLock,
-		pipelineValidator,
-		&conditions.ErrorToMessageConverter{})
+		logpipelinefluentbit.WithClient(client),
+		logpipelinefluentbit.WithGlobals(config.Global),
+
+		logpipelinefluentbit.WithAgentApplierDeleter(fluentBitApplierDeleter),
+		logpipelinefluentbit.WithAgentConfigBuilder(fluentBitConfigBuilder),
+		logpipelinefluentbit.WithAgentProber(&workloadstatus.DaemonSetProber{Client: client}),
+
+		logpipelinefluentbit.WithErrorToMessageConverter(&conditions.ErrorToMessageConverter{}),
+		logpipelinefluentbit.WithFlowHealthProber(flowHealthProber),
+		logpipelinefluentbit.WithIstioStatusChecker(istiostatus.NewChecker(discoveryClient)),
+		logpipelinefluentbit.WithPipelineLock(pipelineLock),
+		logpipelinefluentbit.WithPipelineValidator(pipelineValidator),
+	)
 
 	return fbReconciler, nil
 }
@@ -244,14 +246,14 @@ func configureOTelReconciler(config LogPipelineControllerConfig, client client.C
 		return nil, err
 	}
 
-	pipelineValidator := &logpipelineotel.Validator{
-		PipelineLock:           pipelineLock,
-		EndpointValidator:      &endpoint.Validator{Client: client},
-		TLSCertValidator:       tlscert.New(client),
-		SecretRefValidator:     &secretref.Validator{Client: client},
-		TransformSpecValidator: transformSpecValidator,
-		FilterSpecValidator:    filterSpecValidator,
-	}
+	pipelineValidator := logpipelineotel.NewValidator(
+		logpipelineotel.WithValidatorPipelineLock(pipelineLock),
+		logpipelineotel.WithEndpointValidator(&endpoint.Validator{Client: client}),
+		logpipelineotel.WithTLSCertValidator(tlscert.New(client)),
+		logpipelineotel.WithSecretRefValidator(&secretref.Validator{Client: client}),
+		logpipelineotel.WithTransformSpecValidator(transformSpecValidator),
+		logpipelineotel.WithFilterSpecValidator(filterSpecValidator),
+	)
 
 	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config.RestConfig)
 	if err != nil {
@@ -263,20 +265,25 @@ func configureOTelReconciler(config LogPipelineControllerConfig, client client.C
 	}
 
 	otelReconciler := logpipelineotel.New(
-		config.Global,
-		client,
-		gatewayFlowHealthProber,
-		agentFlowHealthProber,
-		agentConfigBuilder,
-		otelcollector.NewLogAgentApplierDeleter(config.Global, config.OTelCollectorImage, config.LogAgentPriorityClassName),
-		&workloadstatus.DaemonSetProber{Client: client},
-		otelcollector.NewLogGatewayApplierDeleter(config.Global, config.OTelCollectorImage, config.LogGatewayPriorityClassName),
-		&loggateway.Builder{Reader: client},
-		&workloadstatus.DeploymentProber{Client: client},
-		istiostatus.NewChecker(discoveryClient),
-		pipelineLock,
-		pipelineValidator,
-		&conditions.ErrorToMessageConverter{})
+		logpipelineotel.WithClient(client),
+		logpipelineotel.WithGlobals(config.Global),
+
+		logpipelineotel.WithAgentApplierDeleter(otelcollector.NewLogAgentApplierDeleter(config.Global, config.OTelCollectorImage, config.LogAgentPriorityClassName)),
+		logpipelineotel.WithAgentConfigBuilder(agentConfigBuilder),
+		logpipelineotel.WithAgentFlowHealthProber(agentFlowHealthProber),
+		logpipelineotel.WithAgentProber(&workloadstatus.DaemonSetProber{Client: client}),
+
+		logpipelineotel.WithErrorToMessageConverter(&conditions.ErrorToMessageConverter{}),
+
+		logpipelineotel.WithGatewayApplierDeleter(otelcollector.NewLogGatewayApplierDeleter(config.Global, config.OTelCollectorImage, config.LogGatewayPriorityClassName)),
+		logpipelineotel.WithGatewayConfigBuilder(&loggateway.Builder{Reader: client}),
+		logpipelineotel.WithGatewayFlowHealthProber(gatewayFlowHealthProber),
+		logpipelineotel.WithGatewayProber(&workloadstatus.DeploymentProber{Client: client}),
+
+		logpipelineotel.WithIstioStatusChecker(istiostatus.NewChecker(discoveryClient)),
+		logpipelineotel.WithPipelineLock(pipelineLock),
+		logpipelineotel.WithPipelineValidator(pipelineValidator),
+	)
 
 	return otelReconciler, nil
 }

Makefile

@@ -112,14 +112,14 @@ func NewMetricPipelineController(config MetricPipelineControllerConfig, client c
 		return nil, err
 	}
 
-	pipelineValidator := &metricpipeline.Validator{
-		EndpointValidator:      &endpoint.Validator{Client: client},
-		TLSCertValidator:       tlscert.New(client),
-		SecretRefValidator:     &secretref.Validator{Client: client},
-		PipelineLock:           pipelineLock,
-		TransformSpecValidator: transformSpecValidator,
-		FilterSpecValidator:    filterSpecValidator,
-	}
+	pipelineValidator := metricpipeline.NewValidator(
+		metricpipeline.WithEndpointValidator(&endpoint.Validator{Client: client}),
+		metricpipeline.WithTLSCertValidator(tlscert.New(client)),
+		metricpipeline.WithSecretRefValidator(&secretref.Validator{Client: client}),
+		metricpipeline.WithValidatorPipelineLock(pipelineLock),
+		metricpipeline.WithTransformSpecValidator(transformSpecValidator),
+		metricpipeline.WithFilterSpecValidator(filterSpecValidator),
+	)
 
 	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config.RestConfig)
 	if err != nil {
@@ -130,22 +130,26 @@ func NewMetricPipelineController(config MetricPipelineControllerConfig, client c
 	gatewayConfigBuilder := &metricgateway.Builder{Reader: client}
 
 	reconciler := metricpipeline.New(
-		config.Global,
-		client,
-		otelcollector.NewMetricAgentApplierDeleter(config.Global, config.OTelCollectorImage, config.MetricAgentPriorityClassName),
-		agentConfigBuilder,
-		&workloadstatus.DaemonSetProber{Client: client},
-		gatewayFlowHealthProber,
-		agentFlowHealthProber,
-		otelcollector.NewMetricGatewayApplierDeleter(config.Global, config.OTelCollectorImage, config.MetricGatewayPriorityClassName),
-		gatewayConfigBuilder,
-		&workloadstatus.DeploymentProber{Client: client},
-		istiostatus.NewChecker(discoveryClient),
-		overrides.New(config.Global, client),
-		pipelineLock,
-		pipelineSync,
-		pipelineValidator,
-		&conditions.ErrorToMessageConverter{},
+		metricpipeline.WithClient(client),
+		metricpipeline.WithGlobals(config.Global),
+
+		metricpipeline.WithAgentApplierDeleter(otelcollector.NewMetricAgentApplierDeleter(config.Global, config.OTelCollectorImage, config.MetricAgentPriorityClassName)),
+		metricpipeline.WithAgentConfigBuilder(agentConfigBuilder),
+		metricpipeline.WithAgentFlowHealthProber(agentFlowHealthProber),
+		metricpipeline.WithAgentProber(&workloadstatus.DaemonSetProber{Client: client}),
+
+		metricpipeline.WithGatewayApplierDeleter(otelcollector.NewMetricGatewayApplierDeleter(config.Global, config.OTelCollectorImage, config.MetricGatewayPriorityClassName)),
+		metricpipeline.WithGatewayConfigBuilder(gatewayConfigBuilder),
+		metricpipeline.WithGatewayFlowHealthProber(gatewayFlowHealthProber),
+		metricpipeline.WithGatewayProber(&workloadstatus.DeploymentProber{Client: client}),
+
+		metricpipeline.WithErrorToMessageConverter(&conditions.ErrorToMessageConverter{}),
+		metricpipeline.WithIstioStatusChecker(istiostatus.NewChecker(discoveryClient)),
+		metricpipeline.WithOverridesHandler(overrides.New(config.Global, client)),
+		metricpipeline.WithPipelineValidator(pipelineValidator),
+
+		metricpipeline.WithPipelineLock(pipelineLock),
+		metricpipeline.WithPipelineSyncer(pipelineSync),
 	)
 
 	return &MetricPipelineController{

controllers/telemetry/logpipeline_controller.go

@@ -105,33 +105,37 @@ func NewTracePipelineController(config TracePipelineControllerConfig, client cli
 		return nil, err
 	}
 
-	pipelineValidator := &tracepipeline.Validator{
-		EndpointValidator:      &endpoint.Validator{Client: client},
-		TLSCertValidator:       tlscert.New(client),
-		SecretRefValidator:     &secretref.Validator{Client: client},
-		PipelineLock:           pipelineLock,
-		TransformSpecValidator: transformSpecValidator,
-		FilterSpecValidator:    filterSpecValidator,
-	}
+	pipelineValidator := tracepipeline.NewValidator(
+		tracepipeline.WithEndpointValidator(&endpoint.Validator{Client: client}),
+		tracepipeline.WithTLSCertValidator(tlscert.New(client)),
+		tracepipeline.WithSecretRefValidator(&secretref.Validator{Client: client}),
+		tracepipeline.WithValidatorPipelineLock(pipelineLock),
+		tracepipeline.WithTransformSpecValidator(transformSpecValidator),
+		tracepipeline.WithFilterSpecValidator(filterSpecValidator),
+	)
 
 	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config.RestConfig)
 	if err != nil {
 		return nil, err
 	}
 
 	reconciler := tracepipeline.New(
-		client,
-		config.Global,
-		flowHealthProber,
-		otelcollector.NewTraceGatewayApplierDeleter(config.Global, config.OTelCollectorImage, config.TraceGatewayPriorityClassName),
-		&tracegateway.Builder{Reader: client},
-		&workloadstatus.DeploymentProber{Client: client},
-		istiostatus.NewChecker(discoveryClient),
-		overrides.New(config.Global, client),
-		pipelineLock,
-		pipelineSync,
-		pipelineValidator,
-		&conditions.ErrorToMessageConverter{})
+		tracepipeline.WithClient(client),
+		tracepipeline.WithGlobal(config.Global),
+
+		tracepipeline.WithGatewayApplierDeleter(otelcollector.NewTraceGatewayApplierDeleter(config.Global, config.OTelCollectorImage, config.TraceGatewayPriorityClassName)),
+		tracepipeline.WithGatewayConfigBuilder(&tracegateway.Builder{Reader: client}),
+		tracepipeline.WithGatewayProber(&workloadstatus.DeploymentProber{Client: client}),
+
+		tracepipeline.WithFlowHealthProber(flowHealthProber),
+		tracepipeline.WithIstioStatusChecker(istiostatus.NewChecker(discoveryClient)),
+		tracepipeline.WithOverridesHandler(overrides.New(config.Global, client)),
+		tracepipeline.WithErrorToMessageConverter(&conditions.ErrorToMessageConverter{}),
+
+		tracepipeline.WithPipelineLock(pipelineLock),
+		tracepipeline.WithPipelineSyncer(pipelineSync),
+		tracepipeline.WithPipelineValidator(pipelineValidator),
+	)
 
 	return &TracePipelineController{
 		Client:               client,

controllers/telemetry/metricpipeline_controller.go

@@ -113,3 +113,34 @@ We conducted the following PoCs:
 ## Conclusion
 
 The original motivation for the gateway concept is no longer relevant. Transitioning to the agent approach resolves many issues while introducing only minor drawbacks. However, inadequate retry handling by OTel SDKs and Istio proxies remains a challenge. Before proceeding, we must ensure this issue is addressed, considering its alignment with the OTLP specification.
+
+### Agent Rollout and Zero-Downtime Updates
+
+To perform a zero-downtime rollout of a `DaemonSet`, use the `RollingUpdate` update strategy with `maxUnavailable: 0` and `maxSurge: 1`. See the following example:
+
+```yaml
+apiVersion: apps/v1
+kind: DaemonSet
+metadata:
+  name: otel
+  namespace: otel
+spec:
+  selector:
+    matchLabels:
+      app: otel-col
+  updateStrategy:
+    type: RollingUpdate
+    rollingUpdate:
+      maxUnavailable: 0
+      maxSurge: 1
+...
+```
+
+With this configuration, the `DaemonSet` creates a new Pod on each node before terminating the old one, ensuring that a replacement Pod is fully running before any disruption occurs.
+
+This setup was tested using the OpenTelemetry Demo application to verify that no telemetry data is lost during the rollout. Tests were executed for all three signal types: metrics, traces, and logs. The demo application, which includes services written in various programming languages and uses the OTel SDK, continued sending telemetry to the test collector throughout the rollout without any data loss.
+
+The setup was also tested using `TelemetryGen.` TelemetryGen successfully detected endpoint changes after the rollout and continued sending telemetry to the new endpoints.
+
+The same behavior was confirmed with `Istio access logs`. After the rollout, log data continued flowing to the new endpoints without data loss or duplication.
+

controllers/telemetry/tracepipeline_controller.go

@@ -170,6 +170,30 @@ We recommend direct integration with the Dynatrace server. This approach reduces
     metadata:
         name: dynatrace
     spec:
+      transform:
+        - statements:
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.statefulset.name"]) where IsString(resource.attributes["k8s.statefulset.name"])
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.replicaset.name"]) where IsString(resource.attributes["k8s.replicaset.name"])
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.job.name"]) where IsString(resource.attributes["k8s.job.name"])
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.deployment.name"]) where IsString(resource.attributes["k8s.deployment.name"])
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.daemonset.name"]) where IsString(resource.attributes["k8s.daemonset.name"])
+          - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.cronjob.name"]) where IsString(resource.attributes["k8s.cronjob.name"])
+          - set(resource.attributes["k8s.workload.kind"], "statefulset") where IsString(resource.attributes["k8s.statefulset.name"])
+          - set(resource.attributes["k8s.workload.kind"], "replicaset") where IsString(resource.attributes["k8s.replicaset.name"])
+          - set(resource.attributes["k8s.workload.kind"], "job") where IsString(resource.attributes["k8s.job.name"])
+          - set(resource.attributes["k8s.workload.kind"], "deployment") where IsString(resource.attributes["k8s.deployment.name"])
+          - set(resource.attributes["k8s.workload.kind"], "daemonset") where IsString(resource.attributes["k8s.daemonset.name"])
+          - set(resource.attributes["k8s.workload.kind"], "cronjob") where IsString(resource.attributes["k8s.cronjob.name"])
+          - set(resource.attributes["dt.kubernetes.workload.name"], resource.attributes["k8s.workload.name"])
+          - set(resource.attributes["dt.kubernetes.workload.kind"], resource.attributes["k8s.workload.kind"])
+          - delete_key(resource.attributes, "k8s.statefulset.name")
+          - delete_key(resource.attributes, "k8s.replicaset.name")
+          - delete_key(resource.attributes, "k8s.job.name")
+          - delete_key(resource.attributes, "k8s.deployment.name")
+          - delete_key(resource.attributes, "k8s.daemonset.name")
+          - delete_key(resource.attributes, "k8s.cronjob.name")
+        - statements:
+          - set(resource.attributes["k8s.pod.ip"], resource.attributes["ip"]) where resource.attributes["k8s.pod.ip"] == nil
         output:
             otlp:
                 endpoint:
@@ -261,6 +285,30 @@ Depending on your metrics source and temporality, choose one of the following me
         metadata:
             name: dynatrace
         spec:
+          transform:
+            - statements:
+              - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.statefulset.name"]) where IsString(resource.attributes["k8s.statefulset.name"])
+              - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.replicaset.name"]) where IsString(resource.attributes["k8s.replicaset.name"])
+              - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.job.name"]) where IsString(resource.attributes["k8s.job.name"])
+              - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.deployment.name"]) where IsString(resource.attributes["k8s.deployment.name"])
+              - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.daemonset.name"]) where IsString(resource.attributes["k8s.daemonset.name"])
+              - set(resource.attributes["k8s.workload.name"], resource.attributes["k8s.cronjob.name"]) where IsString(resource.attributes["k8s.cronjob.name"])
+              - set(resource.attributes["k8s.workload.kind"], "statefulset") where IsString(resource.attributes["k8s.statefulset.name"])
+              - set(resource.attributes["k8s.workload.kind"], "replicaset") where IsString(resource.attributes["k8s.replicaset.name"])
+              - set(resource.attributes["k8s.workload.kind"], "job") where IsString(resource.attributes["k8s.job.name"])
+              - set(resource.attributes["k8s.workload.kind"], "deployment") where IsString(resource.attributes["k8s.deployment.name"])
+              - set(resource.attributes["k8s.workload.kind"], "daemonset") where IsString(resource.attributes["k8s.daemonset.name"])
+              - set(resource.attributes["k8s.workload.kind"], "cronjob") where IsString(resource.attributes["k8s.cronjob.name"])
+              - set(resource.attributes["dt.kubernetes.workload.name"], resource.attributes["k8s.workload.name"])
+              - set(resource.attributes["dt.kubernetes.workload.kind"], resource.attributes["k8s.workload.kind"])
+              - delete_key(resource.attributes, "k8s.statefulset.name")
+              - delete_key(resource.attributes, "k8s.replicaset.name")
+              - delete_key(resource.attributes, "k8s.job.name")
+              - delete_key(resource.attributes, "k8s.deployment.name")
+              - delete_key(resource.attributes, "k8s.daemonset.name")
+              - delete_key(resource.attributes, "k8s.cronjob.name")
+            - statements:
+              - set(resource.attributes["k8s.pod.ip"], resource.attributes["ip"]) where resource.attributes["k8s.pod.ip"] == nil
             output:
                 otlp:
                     endpoint: