OpenTelemetry filter processor

components/telemetry-operator/controller/tracepipeline/config_builder.go

@@ -92,6 +92,15 @@ type ProcessorsConfig struct {
 	MemoryLimiter MemoryLimiterConfig          `yaml:"memory_limiter,omitempty"`
 	K8sAttributes K8sAttributesProcessorConfig `yaml:"k8sattributes,omitempty"`
 	Resource      ResourceProcessorConfig      `yaml:"resource,omitempty"`
+	Filter        FilterProcessorConfig        `yaml:"filter,omitempty"`
+}
+
+type FilterProcessorConfig struct {
+	Traces TraceConfig `yaml:"traces,omitempty"`
+}
+
+type TraceConfig struct {
+	Span []string `yaml:"span"`
 }
 
 type PipelineConfig struct {
@@ -256,6 +265,25 @@ func makeProcessorsConfig() ProcessorsConfig {
 				},
 			},
 		},
+		Filter: FilterProcessorConfig{
+			Traces: TraceConfig{
+				Span: makeSpanFilterConfig(),
+			},
+		},
+	}
+}
+
+func makeSpanFilterConfig() []string {
+	return []string{
+		"(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (resource.attributes[\"service.name\"] == \"jaeger.kyma-system\")",
+		"(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (resource.attributes[\"service.name\"] == \"grafana.kyma-system\")",
+		"(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (IsMatch(attributes[\"http.url\"], \".+/metrics\") == true) and (resource.attributes[\"k8s.namespace.name\"] == \"kyma-system\")",
+		"(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (IsMatch(attributes[\"http.url\"], \".+/healthz(/.*)?\") == true) and (resource.attributes[\"k8s.namespace.name\"] == \"kyma-system\")",
+		"(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (attributes[\"user_agent\"] == \"vm_promscrape\")",
+		"(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (IsMatch(attributes[\"http.url\"], \"http(s)?:\\\\/\\\\/telemetry-otlp-traces\\\\.kyma-system(\\\\..*)?:(4318|4317).*\") == true)",
+		"(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (IsMatch(attributes[\"http.url\"], \"http(s)?:\\\\/\\\\/telemetry-trace-collector-internal\\\\.kyma-system(\\\\..*)?:(55678).*\") == true)",
+		"(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (resource.attributes[\"service.name\"] == \"loki.kyma-system\")",
+		"(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (resource.attributes[\"service.name\"] == \"telemetry-fluent-bit.kyma-system\")",
 	}
 }
 
@@ -264,7 +292,7 @@ func makeServiceConfig(outputType string) OTLPServiceConfig {
 		Pipelines: PipelinesConfig{
 			Traces: PipelineConfig{
 				Receivers:  []string{"opencensus", "otlp"},
-				Processors: []string{"memory_limiter", "k8sattributes", "resource", "batch"},
+				Processors: []string{"memory_limiter", "k8sattributes", "filter", "resource", "batch"},
 				Exporters:  []string{outputType, "logging"},
 			},
 		},

components/telemetry-operator/controller/tracepipeline/config_builder_test.go

@@ -149,10 +149,11 @@ func TestMakeServiceConfig(t *testing.T) {
 	require.Contains(t, serviceConfig.Pipelines.Traces.Receivers, "otlp")
 	require.Contains(t, serviceConfig.Pipelines.Traces.Receivers, "opencensus")
 
-	require.Contains(t, serviceConfig.Pipelines.Traces.Processors, "memory_limiter")
-	require.Contains(t, serviceConfig.Pipelines.Traces.Processors, "k8sattributes")
-	require.Contains(t, serviceConfig.Pipelines.Traces.Processors, "resource")
-	require.Contains(t, serviceConfig.Pipelines.Traces.Processors, "batch")
+	require.Equal(t, serviceConfig.Pipelines.Traces.Processors[0], "memory_limiter")
+	require.Equal(t, serviceConfig.Pipelines.Traces.Processors[1], "k8sattributes")
+	require.Equal(t, serviceConfig.Pipelines.Traces.Processors[2], "filter")
+	require.Equal(t, serviceConfig.Pipelines.Traces.Processors[3], "resource")
+	require.Equal(t, serviceConfig.Pipelines.Traces.Processors[4], "batch")
 
 	require.Contains(t, serviceConfig.Pipelines.Traces.Exporters, "otlp")
 	require.Contains(t, serviceConfig.Pipelines.Traces.Exporters, "logging")
@@ -214,6 +215,20 @@ func TestK8sAttributesProcessor(t *testing.T) {
 	require.Equal(t, "connection", processors.K8sAttributes.PodAssociation[2].Sources[0].From)
 }
 
+func TestFilterProcessor(t *testing.T) {
+	processors := makeProcessorsConfig()
+	require.Equal(t, len(processors.Filter.Traces.Span), 9, "Span filter list size is wrong")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (resource.attributes[\"service.name\"] == \"jaeger.kyma-system\")", "Jaeger span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (resource.attributes[\"service.name\"] == \"grafana.kyma-system\")", "Grafana span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (IsMatch(attributes[\"http.url\"], \".+/metrics\") == true) and (resource.attributes[\"k8s.namespace.name\"] == \"kyma-system\")", "/metrics endpoint span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (IsMatch(attributes[\"http.url\"], \".+/healthz(/.*)?\") == true) and (resource.attributes[\"k8s.namespace.name\"] == \"kyma-system\")", "/healthz endpoint span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"GET\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (attributes[\"user_agent\"] == \"vm_promscrape\")", "Victoria Metrics agent span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (IsMatch(attributes[\"http.url\"], \"http(s)?:\\\\/\\\\/telemetry-otlp-traces\\\\.kyma-system(\\\\..*)?:(4318|4317).*\") == true)", "Telemetry OTLP service span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (IsMatch(attributes[\"http.url\"], \"http(s)?:\\\\/\\\\/telemetry-trace-collector-internal\\\\.kyma-system(\\\\..*)?:(55678).*\") == true)", "Telemetry Opencensus service span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Ingress\") and (resource.attributes[\"service.name\"] == \"loki.kyma-system\")", "Loki service span filter missing")
+	require.Contains(t, processors.Filter.Traces.Span, "(attributes[\"http.method\"] == \"POST\") and (attributes[\"component\"] == \"proxy\") and (attributes[\"OperationName\"] == \"Egress\") and (resource.attributes[\"service.name\"] == \"telemetry-fluent-bit.kyma-system\")", "Fluent-Bit service span filter missing")
+}
+
 func TestCollectorConfigMarshalling(t *testing.T) {
 	expected := `receivers:
   opencensus: {}
@@ -272,6 +287,39 @@ processors:
     - action: insert
       key: k8s.cluster.name
       value: ${KUBERNETES_SERVICE_HOST}
+  filter:
+    traces:
+      span:
+      - (attributes["http.method"] == "POST") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Ingress") and (resource.attributes["service.name"]
+        == "jaeger.kyma-system")
+      - (attributes["http.method"] == "GET") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Egress") and (resource.attributes["service.name"]
+        == "grafana.kyma-system")
+      - (attributes["http.method"] == "GET") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Ingress") and (IsMatch(attributes["http.url"],
+        ".+/metrics") == true) and (resource.attributes["k8s.namespace.name"] == "kyma-system")
+      - (attributes["http.method"] == "GET") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Ingress") and (IsMatch(attributes["http.url"],
+        ".+/healthz(/.*)?") == true) and (resource.attributes["k8s.namespace.name"]
+        == "kyma-system")
+      - (attributes["http.method"] == "GET") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Ingress") and (attributes["user_agent"]
+        == "vm_promscrape")
+      - (attributes["http.method"] == "POST") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Egress") and (IsMatch(attributes["http.url"],
+        "http(s)?:\\/\\/telemetry-otlp-traces\\.kyma-system(\\..*)?:(4318|4317).*")
+        == true)
+      - (attributes["http.method"] == "POST") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Egress") and (IsMatch(attributes["http.url"],
+        "http(s)?:\\/\\/telemetry-trace-collector-internal\\.kyma-system(\\..*)?:(55678).*")
+        == true)
+      - (attributes["http.method"] == "POST") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Ingress") and (resource.attributes["service.name"]
+        == "loki.kyma-system")
+      - (attributes["http.method"] == "POST") and (attributes["component"] == "proxy")
+        and (attributes["OperationName"] == "Egress") and (resource.attributes["service.name"]
+        == "telemetry-fluent-bit.kyma-system")
 extensions:
   health_check: {}
 service:
@@ -283,6 +331,7 @@ service:
       processors:
       - memory_limiter
       - k8sattributes
+      - filter
       - resource
       - batch
       exporters:

components/telemetry-operator/main.go

@@ -98,7 +98,7 @@ var (
 	maxLogPipelines            int
 )
 
-const otelImage = "eu.gcr.io/kyma-project/tpi/otel-collector:0.66.0-a80d981f"
+const otelImage = "eu.gcr.io/kyma-project/tpi/otel-collector:0.66.0-8bb1c644"
 
 //nolint:gochecknoinits
 func init() {

docs/01-overview/main-areas/telemetry/telemetry-03-traces.md

@@ -278,6 +278,13 @@ The used buffers are volatile, and the data can be lost on the otel-collector in
 
 Only one TracePipeline resource at a time is supported at the moment.
 
+### System span filtering
+System related spans reported by Istio will be filtered out without option to opt-out, especially the following:
+- `/healtz` endpoint of any component deployed on `kyma-system` namespace
+- `/metrics` endpoint of any component deployed on `kyma-system` namespace
+- All outgoing spans reported by Grafana and Jaeger
+- All spans in reference to `fluent-bit` and `loki` communication
+
 ## Frequently Asked Questions
 
 1. Traces are not arriving at the destination at all

resources/telemetry/values.yaml

@@ -7,13 +7,13 @@ global:
       version: "PR-16573"
     telemetry_operator:
       name: "telemetry-operator"
-      version: "PR-16592"
+      version: "PR-16558"
     telemetry_webhook_cert_init:
       name: "webhook-cert-init"
       version: "PR-16573"
     telemetry_otel_collector:
       name: "otel-collector"
-      version: "0.66.0-a80d981f"
+      version: "0.66.0-8bb1c644"
       directory: "tpi"
     fluent_bit:
       name: "fluent-bit"