feat(processors/k8sattributesprocessor): support metadata enrichment based on multiple attributes

processor/k8sattributesprocessor/config.go

@@ -15,9 +15,12 @@
 package k8sattributesprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor"
 
 import (
+	"fmt"
+
 	"go.opentelemetry.io/collector/config"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor/internal/kube"
 )
 
 // Config defines configuration for k8s attributes processor.
@@ -50,7 +53,17 @@ type Config struct {
 }
 
 func (cfg *Config) Validate() error {
-	return cfg.APIConfig.Validate()
+	if err := cfg.APIConfig.Validate(); err != nil {
+		return err
+	}
+
+	for _, assoc := range cfg.Association {
+		if len(assoc.Sources) > kube.PodIdentifierMaxLength {
+			return fmt.Errorf("too many association sources. limit is %v", kube.PodIdentifierMaxLength)
+		}
+	}
+
+	return nil
 }
 
 // ExtractConfig section allows specifying extraction rules to extract
@@ -214,13 +227,21 @@ type FieldFilterConfig struct {
 // PodAssociationConfig contain single rule how to associate Pod metadata
 // with logs, spans and metrics
 type PodAssociationConfig struct {
+	// Deprecated: Sources should be used to provide From and Name.
+	// If this is set, From and Name are going to be used as Sources' ones
 	// From represents the source of the association.
 	// Allowed values are "connection" and "resource_attribute".
 	From string `mapstructure:"from"`
 
+	// Deprecated: Sources should be used to provide From and Name.
+	// If this is set, From and Name are going to be used as Sources' ones
 	// Name represents extracted key name.
 	// e.g. ip, pod_uid, k8s.pod.ip
 	Name string `mapstructure:"name"`
+
+	// List of pod association sources which should be taken
+	// to identify pod
+	Sources []PodAssociationSourceConfig `mapstructure:"sources"`
 }
 
 // ExcludeConfig represent a list of Pods to exclude
@@ -232,3 +253,13 @@ type ExcludeConfig struct {
 type ExcludePodConfig struct {
 	Name string `mapstructure:"name"`
 }
+
+type PodAssociationSourceConfig struct {
+	// From represents the source of the association.
+	// Allowed values are "connection" and "resource_attribute".
+	From string `mapstructure:"from"`
+
+	// Name represents extracted key name.
+	// e.g. ip, pod_uid, k8s.pod.ip
+	Name string `mapstructure:"name"`
+}

processor/k8sattributesprocessor/config_test.go

@@ -84,21 +84,38 @@ func TestLoadConfig(t *testing.T) {
 			},
 			Association: []PodAssociationConfig{
 				{
-					From: "resource_attribute",
-					Name: "ip",
+					Sources: []PodAssociationSourceConfig{
+						{
+							From: "resource_attribute",
+							Name: "ip",
+						},
+					},
 				},
 				{
-					From: "resource_attribute",
-					Name: "k8s.pod.ip",
+					Sources: []PodAssociationSourceConfig{
+						{
+							From: "resource_attribute",
+							Name: "k8s.pod.ip",
+						},
+					},
 				},
 				{
-					From: "resource_attribute",
-					Name: "host.name",
+					Sources: []PodAssociationSourceConfig{
+						{
+							From: "resource_attribute",
+							Name: "host.name",
+						},
+					},
 				},
 				{
-					From: "connection",
-					Name: "ip",
+					Sources: []PodAssociationSourceConfig{
+						{
+							From: "connection",
+							Name: "ip",
+						},
+					},
 				},
+				// Deprecated way
 				{
 					From: "resource_attribute",
 					Name: "k8s.pod.uid",

processor/k8sattributesprocessor/doc.go

@@ -18,25 +18,29 @@
 // extracted metadata to the relevant spans, metrics and logs. The processor uses the kubernetes API to discover all pods
 // running in a cluster, keeps a record of their IP addresses, pod UIDs and interesting metadata.
 // The rules for associating the data passing through the processor (spans, metrics and logs) with specific Pod Metadata are configured via "pod_association" key.
-// It represents a list of rules that are executed in the specified order until the first one is able to do the match.
-// Each rule is specified as a pair of from (representing the rule type) and name (representing the extracted key name).
+// It represents a list of associations that are executed in the specified order until the first one is able to do the match.
+//
+// Each association is specified as a list of sources of association.
+// Sources represents list of rules. All rules are going to be executed and combination of result is going to be a pod metadata cache key.
+// In order to get an association applied, all the data defined in each source has to be successfully fetched from a log, trace or metric.
+//
+// Each sources rule is specified as a pair of `from` (representing the rule type) and `name` (representing the attribute name if `From` is set to `resource_attribute`).
 // Following rule types are available:
-//   from: "resource_attribute" - allows to specify the attribute name to lookup up in the list of attributes of the received Resource. The specified attribute, if it is present, identifies the Pod that is represented by the Resource.
-//     (the value can contain either IP address or Pod UID)
-//   from: "connection" - takes the IP attribute from connection context (if available) and automatically
-//     associates it with "k8s.pod.ip" attribute
+//   from: "connection" - takes the IP attribute from connection context (if available)
+//   from: "resource_attribute" - allows to specify the attribute name to lookup up in the list of attributes of the received Resource.
+//                                Semantic convention should be used for naming.
+//
 // Pod association configuration.
 // pod_association:
-//  - from: resource_attribute
-//    name: ip
-//  - from: resource_attribute
-//    name: k8s.pod.ip
-//  - from: resource_attribute
-//    name: host.name
-//  - from: connection
-//    name: ip
-//  - from: resource_attribute
-//    name: k8s.pod.uid
+//  - sources:
+//     - from: resource_attribute
+//       name: k8s.pod.ip
+//  # below association matches for pair `k8s.pod.name` and `k8s.namespace.name`
+//  - sources:
+//    - from: resource_attribute
+//      name: k8s.pod.name
+//    - from: resource_attribute
+//      name: k8s.namespace.name
 //
 // If Pod association rules are not configured resources are associated with metadata only by connection's IP Address.
 //
@@ -52,6 +56,9 @@
 //   - k8s.node.name
 // Not all the attributes are guaranteed to be added.
 //
+// Only attribute names from `metadata` should be used for pod_association's `resource_attribute`,
+// because empty or non-existing values will be ignored.
+//
 // The following container level attributes require additional attributes to identify a particular container in a pod:
 //   1. Container spec attributes - will be set only if container identifying attribute `k8s.container.name` is set
 //      as a resource attribute (similar to all other attributes, pod has to be identified as well):

processor/k8sattributesprocessor/factory.go

@@ -155,6 +155,7 @@ func createKubernetesProcessor(
 	kp := &kubernetesprocessor{logger: params.Logger}
 
 	warnDeprecatedMetadataConfig(kp.logger, cfg)
+	warnDeprecatedPodAssociationConfig(kp.logger, cfg)
 
 	err := errWrongKeyConfig(cfg)
 	if err != nil {
@@ -252,3 +253,42 @@ func errWrongKeyConfig(cfg config.Processor) error {
 
 	return nil
 }
+
+func warnDeprecatedPodAssociationConfig(logger *zap.Logger, cfg config.Processor) {
+	oCfg := cfg.(*Config)
+	deprecated := ""
+	actual := ""
+	for _, assoc := range oCfg.Association {
+		if assoc.From == "" && assoc.Name == "" {
+			continue
+		}
+
+		deprecated += fmt.Sprintf(`
+- from: %s`, assoc.From)
+		actual += fmt.Sprintf(`
+- sources:
+  - from: %s`, assoc.From)
+
+		if assoc.Name != "" {
+			deprecated += fmt.Sprintf(`
+  name: %s`, assoc.Name)
+		}
+
+		if assoc.From != kube.ConnectionSource {
+			actual += fmt.Sprintf(`
+    name: %s`, assoc.Name)
+		}
+	}
+
+	if deprecated != "" {
+		logger.Warn(fmt.Sprintf(`Deprecated pod_association configuration detected. Please replace:
+
+pod_association:%s
+
+with
+
+pod_association:%s
+
+`, deprecated, actual))
+	}
+}

processor/k8sattributesprocessor/internal/kube/client.go

@@ -369,7 +369,7 @@ func (c *WatchClient) extractNamespaceAttributes(namespace *api_v1.Namespace) ma
 	return tags
 }
 
-func (c *WatchClient) addOrUpdatePod(pod *api_v1.Pod) {
+func (c *WatchClient) podFromAPI(pod *api_v1.Pod) *Pod {
 	newPod := &Pod{
 		Name:      pod.Name,
 		Namespace: pod.GetNamespace(),
@@ -387,41 +387,98 @@ func (c *WatchClient) addOrUpdatePod(pod *api_v1.Pod) {
 		}
 	}
 
+	return newPod
+}
+
+// getIdentifiersFromAssoc returns list of PodIdentifiers for given pod
+func (c *WatchClient) getIdentifiersFromAssoc(pod *Pod) []PodIdentifier {
+	ids := []PodIdentifier{}
+	for _, assoc := range c.Associations {
+		ret := PodIdentifier{}
+		skip := false
+		for i, source := range assoc.Sources {
+			// If association configured to take IP address from connection
+			switch {
+			case source.From == ConnectionSource:
+				if pod.Address == "" {
+					skip = true
+					break
+				}
+				ret[i] = PodIdentifierAttributeFromSource(source, pod.Address)
+			case source.From == ResourceSource:
+				attr := ""
+				switch source.Name {
+				case conventions.AttributeK8SNamespaceName:
+					attr = pod.Namespace
+				case conventions.AttributeK8SPodName:
+					attr = pod.Name
+				case conventions.AttributeK8SPodUID:
+					attr = pod.PodUID
+				case conventions.AttributeHostName:
+					attr = pod.Address
+				default:
+					if v, ok := pod.Attributes[source.Name]; ok {
+						attr = v
+					}
+				}
+
+				if attr == "" {
+					skip = true
+					break
+				}
+				ret[i] = PodIdentifierAttributeFromSource(source, attr)
+			}
+		}
+
+		if !skip {
+			ids = append(ids, ret)
+		}
+	}
+
+	// Ensure backward compatibility
+	if pod.PodUID != "" {
+		ids = append(ids, PodIdentifier{
+			PodIdentifierAttributeFromResourceAttribute(conventions.AttributeK8SPodUID, pod.PodUID),
+		})
+	}
+
+	if pod.Address != "" {
+		ids = append(ids, PodIdentifier{
+			PodIdentifierAttributeFromConnection(pod.Address),
+		})
+	}
+
+	return ids
+}
+
+func (c *WatchClient) addOrUpdatePod(pod *api_v1.Pod) {
+	newPod := c.podFromAPI(pod)
+
 	c.m.Lock()
 	defer c.m.Unlock()
 
-	if pod.UID != "" {
-		c.Pods[PodIdentifier(pod.UID)] = newPod
-	}
-	if pod.Status.PodIP != "" {
-		// compare initial scheduled timestamp for existing pod and new pod with same IP
-		// and only replace old pod if scheduled time of new pod is newer? This should fix
-		// the case where scheduler has assigned the same IP to a new pod but update event for
-		// the old pod came in later.
-		if p, ok := c.Pods[PodIdentifier(pod.Status.PodIP)]; ok {
-			if p.StartTime != nil && pod.Status.StartTime.Before(p.StartTime) {
+	for _, id := range c.getIdentifiersFromAssoc(newPod) {
+		// compare initial scheduled timestamp for existing pod and new pod with same identifier
+		// and only replace old pod if scheduled time of new pod is newer or equal.
+		// This should fix the case where scheduler has assigned the same attribtues (like IP address)
+		// to a new pod but update event for the old pod came in later.
+		if p, ok := c.Pods[id]; ok {
+			if p.StartTime != nil && !p.StartTime.Before(pod.Status.StartTime) {
 				return
 			}
 		}
-		c.Pods[PodIdentifier(pod.Status.PodIP)] = newPod
+		c.Pods[id] = newPod
 	}
 }
 
 func (c *WatchClient) forgetPod(pod *api_v1.Pod) {
-	c.m.RLock()
-	p, ok := c.GetPod(PodIdentifier(pod.Status.PodIP))
-	c.m.RUnlock()
-
-	if ok && p.Name == pod.Name {
-		c.appendDeleteQueue(PodIdentifier(pod.Status.PodIP), pod.Name)
-	}
+	podToRemove := c.podFromAPI(pod)
+	for _, id := range c.getIdentifiersFromAssoc(podToRemove) {
+		p, ok := c.GetPod(id)
 
-	c.m.RLock()
-	p, ok = c.GetPod(PodIdentifier(pod.UID))
-	c.m.RUnlock()
-
-	if ok && p.Name == pod.Name {
-		c.appendDeleteQueue(PodIdentifier(pod.UID), pod.Name)
+		if ok && p.Name == pod.Name {
+			c.appendDeleteQueue(id, pod.Name)
+		}
 	}
 }