Resource Attributes - refactor and add missing attributes

autoscaler/controllers/datacollection/configmap.go

@@ -144,14 +144,64 @@ func calculateConfigMapData(nodeCG *odigosv1.CollectorsGroup, sources *odigosv1.
 
 	processorsCfg["batch"] = empty
 	processorsCfg["memory_limiter"] = memoryLimiterConfiguration
-	processorsCfg["resource"] = config.GenericMap{
+	processorsCfg["resource/node-name"] = config.GenericMap{
 		"attributes": []config.GenericMap{{
 			"key":    "k8s.node.name",
 			"value":  "${NODE_NAME}",
 			"action": "upsert",
 		}},
 	}
+
+	// This processor is needed because some vanilla OTel instrumentations (Java, Python) emits wrong container.id
+	// TODO(edenfed): configure instrumentations to not emit container.id (disable container resource provider)
+	processorsCfg["resource/remove-container-id"] = config.GenericMap{
+		"attributes": []config.GenericMap{{
+			"key":    "container.id",
+			"action": "delete",
+		}},
+	}
+
+	// Adds k8s / container attributes to the resource - this is important for correlation of traces with metrics
+	processorsCfg["k8sattributes"] = config.GenericMap{
+		"auth_type":   "serviceAccount",
+		"passthrough": false,
+		"filter": config.GenericMap{
+			"node_from_env_var": "NODE_NAME",
+		},
+		"extract": config.GenericMap{
+			"metadata": []string{
+				"k8s.deployment.name",
+				"k8s.statefulset.name",
+				"k8s.daemonset.name",
+				"k8s.cronjob.name",
+				"k8s.job.name",
+				"k8s.cluster.uid",
+				"k8s.node.name",
+				"container.id",
+				"container.image.name",
+				"container.image.tag",
+			},
+		},
+		"pod_association": []config.GenericMap{
+			{
+				"sources": []config.GenericMap{
+					{
+						"from": "resource_attribute",
+						"name": "k8s.pod.name",
+					},
+					{
+						"from": "resource_attribute",
+						"name": "k8s.namespace.name",
+					},
+				},
+			},
+		},
+	}
+
+	// Add cloud provider specific resource attributes
 	processorsCfg["resourcedetection"] = config.GenericMap{"detectors": []string{"ec2", "gcp", "azure"}}
+
+	// Calculate traffic metrics for visibility in the UI
 	processorsCfg["odigostrafficmetrics"] = config.GenericMap{
 		// adding the following resource attributes to the metrics allows to aggregate the metrics by source.
 		"res_attributes_keys": []string{
@@ -329,9 +379,23 @@ func calculateConfigMapData(nodeCG *odigosv1.CollectorsGroup, sources *odigosv1.
 	}
 
 	if collectTraces {
+		procs := append(commonProcessors, tracesProcessors...)
+		// Special handling for traces: remove wrong container.id and add more k8s attributes
+		procs = append(procs, "resource/remove-container-id", "k8sattributes")
+
+		// Move odigostrafficmetrics processor to the end of the pipeline
+		// to ensure that the metrics are calculated after the k8s attributes are added
+		for i, p := range procs {
+			if p == "odigostrafficmetrics" {
+				procs = append(procs[:i], procs[i+1:]...)
+				procs = append(procs, "odigostrafficmetrics")
+				break
+			}
+		}
+
 		cfg.Service.Pipelines["traces"] = config.Pipeline{
 			Receivers:  []string{"otlp"},
-			Processors: append(commonProcessors, tracesProcessors...),
+			Processors: procs,
 			Exporters:  tracesPipelineExporter,
 		}
 	}
@@ -455,6 +519,6 @@ func getCommonProcessors(disableNameProcessor bool) []string {
 	if !disableNameProcessor {
 		processors = append(processors, "odigosresourcename")
 	}
-	processors = append(processors, "resource", "resourcedetection", "odigostrafficmetrics")
+	processors = append(processors, "resource/node-name", "resourcedetection", "odigostrafficmetrics")
 	return processors
 }

cli/cmd/resources/datacollection.go

@@ -34,22 +34,20 @@ func NewDataCollectionClusterRole(psp bool) *rbacv1.ClusterRole {
 			Name: "odigos-data-collection",
 		},
 		Rules: []rbacv1.PolicyRule{
-			{ // TODO: remove this after we remove honeycomb custom exporter config
-				// located at: autoscaler/controllers/datacollection/custom/honeycomb.go
+			{ // Needed for: host metrics processor
 				APIGroups: []string{""},
-				Resources: []string{"nodes/stats", "nodes/proxy"},
-				Verbs:     []string{"get", "list"},
+				Resources: []string{"nodes/stats", "nodes/proxy", "nodes"},
+				Verbs:     []string{"get", "list", "watch"},
 			},
-			{ // Needed to get "resource name" in processor (TODO: remove this after we kill the resource name processor)
+			{ // Needed for: k8s attributes processor
 				APIGroups: []string{""},
-				Resources: []string{"pods"},
-				Verbs:     []string{"get", "list"},
+				Resources: []string{"pods", "namespaces"},
+				Verbs:     []string{"get", "list", "watch"},
 			},
-			{ // Need "replicasets" to get "resource name" in processor (TODO: remove this after we kill the resource name processor),
-				// Others needed to get applications from cluster
+			{ // Needed for: k8s attributes processor
 				APIGroups: []string{"apps"},
 				Resources: []string{"replicasets", "deployments", "daemonsets", "statefulsets"},
-				Verbs:     []string{"get", "list"},
+				Verbs:     []string{"get", "list", "watch"},
 			},
 			{ // Needed for load balancer
 				APIGroups: []string{""},

common/resourceattributes/attributes.go

@@ -0,0 +1,26 @@
+package resourceattributes
+
+import (
+	"fmt"
+	"strings"
+
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
+)
+
+type Attributes []attribute.KeyValue
+
+// ToEnvVarString converts the attributes to a string that can be used as OTEL_RESOURCE_ATTRIBUTES env var
+func (a Attributes) ToEnvVarString() string {
+	var attrs []string
+	for _, attr := range a {
+		attrs = append(attrs, fmt.Sprintf("%s=%s", attr.Key, attr.Value.AsString()))
+	}
+	return strings.Join(attrs, ",")
+}
+
+// IncludeServiceName adds the service name to the attributes
+// OpAMP clients expect the service name to be set in the resource attributes
+func (a Attributes) IncludeServiceName(serviceName string) Attributes {
+	return append(a, semconv.ServiceName(serviceName))
+}

common/resourceattributes/root.go

@@ -0,0 +1,58 @@
+// Package resourceattributes provides a set of functions to create resource attributes for a container.
+// Resource attributes in Odigos are injected in two phases:
+// 1. Before the container starts, the attributes are set to identify the container.
+// 2. After the container starts, additional attributes may be added to the container.
+// Notice that the second phase is not always called, see AfterPodStart for more details.
+//
+// Use the following guidelines when adding new resource attribute:
+// - If the attribute is fast to calculate and is needed by all clients, add it to BeforePodStart.
+// - If the attribute is slow to calculate or is only needed for OpAMP/eBPF clients, add it to AfterPodStart.
+// An example of slow to calculate attribute is the cloud provider metadata that requires a network call.
+package resourceattributes
+
+import (
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
+)
+
+type ContainerIdentifier struct {
+	PodName       string
+	Namespace     string
+	ContainerName string
+}
+
+// BeforePodStart returns the resource attributes that should be set before the container starts.
+// This function is called by the PodWebhook before the container starts.
+// Currently, the attributes returned are the minimal set of attributes that should be set to identify the container.
+// Other attributes (k8s related, cloud provider, etc) are set later by a dedicated processor in odigos-data-collector.
+// Notice that clients that calls AfterPodStart will overwrite the attributes set by this function.
+func BeforePodStart(identifier *ContainerIdentifier) Attributes {
+	if identifier == nil {
+		return nil
+	}
+
+	return identifyingResourceAttributes(identifier)
+}
+
+// AfterPodStart returns the resource attributes that should be set after the container starts.
+// This function is called by the eBPF director or by the OpAMP server after the container starts.
+// Currently, the attributes returned are the minimal set of attributes that should be set to identify the container.
+// Other attributes (k8s related, cloud provider, etc) are set later by a dedicated processor in odigos-data-collector.
+// Notice that this function is not always called, only OpAMP clients or eBPF-based clients will call this function.
+// Vanilla OpenTelemetry clients will only call BeforePodStart.
+// OpAMP/eBPF clients WILL OVERWRITE the attributes set by BeforePodStart with the attributes returned by this function.
+func AfterPodStart(identifier *ContainerIdentifier) Attributes {
+	if identifier == nil {
+		return nil
+	}
+
+	return identifyingResourceAttributes(identifier)
+}
+
+func identifyingResourceAttributes(identifier *ContainerIdentifier) Attributes {
+	return []attribute.KeyValue{
+		semconv.K8SContainerName(identifier.ContainerName),
+		semconv.K8SPodName(identifier.PodName),
+		semconv.K8SNamespaceName(identifier.Namespace),
+	}
+}

helm/odigos/templates/datacollection/clusterrole.yaml

@@ -8,16 +8,20 @@ rules:
     resources:
       - nodes/stats
       - nodes/proxy
+      - nodes
     verbs:
       - get
       - list
+      - watch
   - apiGroups:
       - ''
     resources:
       - pods
+      - namespaces
     verbs:
       - get
       - list
+      - watch
   - apiGroups:
       - apps
     resources:
@@ -28,6 +32,7 @@ rules:
     verbs:
       - get
       - list
+      - watch
   - apiGroups:
       - ''
     resources:

instrumentation/factory.go

@@ -3,7 +3,7 @@ package instrumentation
 import (
 	"context"
 
-	"go.opentelemetry.io/otel/attribute"
+	"github.com/odigos-io/odigos/common/resourceattributes"
 )
 
 type Config any
@@ -15,7 +15,7 @@ type Settings struct {
 	ServiceName string
 	// ResourceAttributes can be used to pass additional resource attributes to the instrumentation
 	// These attributes will be added to the resource attributes of the telemetry data.
-	ResourceAttributes []attribute.KeyValue
+	ResourceAttributes resourceattributes.Attributes
 	// InitialConfig is the initial configuration that should be applied to the instrumentation,
 	// it can be used to enable/disable specific instrumentation libraries, configure sampling, etc.
 	InitialConfig Config

instrumentor/controllers/instrumentationdevice/manager.go

@@ -4,6 +4,7 @@ import (
 	odigosv1 "github.com/odigos-io/odigos/api/odigos/v1alpha1"
 	"github.com/odigos-io/odigos/common"
 	instrumentorpredicate "github.com/odigos-io/odigos/instrumentor/controllers/utils/predicates"
+	"github.com/odigos-io/odigos/instrumentor/controllers/webhook"
 	odigospredicate "github.com/odigos-io/odigos/k8sutils/pkg/predicate"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -166,7 +167,7 @@ func SetupWithManager(mgr ctrl.Manager) error {
 	err = builder.
 		WebhookManagedBy(mgr).
 		For(&corev1.Pod{}).
-		WithDefaulter(&PodsWebhook{
+		WithDefaulter(&webhook.PodsWebhook{
 			Client: mgr.GetClient(),
 		}).
 		Complete()