[k8sprocessor] Add ability to associate metadata tags using pod UID rather than just IP

processor/k8sprocessor/client_test.go

@@ -27,11 +27,12 @@ import (
 
 // fakeClient is used as a replacement for WatchClient in test cases.
 type fakeClient struct {
-	Pods     map[string]*kube.Pod
-	Rules    kube.ExtractionRules
-	Filters  kube.Filters
-	Informer cache.SharedInformer
-	StopCh   chan struct{}
+	Pods         map[string]*kube.Pod
+	Rules        kube.ExtractionRules
+	Filters      kube.Filters
+	Associations []kube.Association
+	Informer     cache.SharedInformer
+	StopCh       chan struct{}
 }
 
 func selectors() (labels.Selector, fields.Selector) {
@@ -40,22 +41,24 @@ func selectors() (labels.Selector, fields.Selector) {
 }
 
 // newFakeClient instantiates a new FakeClient object and satisfies the ClientProvider type
-func newFakeClient(_ *zap.Logger, apiCfg k8sconfig.APIConfig, rules kube.ExtractionRules, filters kube.Filters, _ kube.APIClientsetProvider, _ kube.InformerProvider) (kube.Client, error) {
+func newFakeClient(_ *zap.Logger, apiCfg k8sconfig.APIConfig, rules kube.ExtractionRules, filters kube.Filters, associations []kube.Association, _ kube.APIClientsetProvider, _ kube.InformerProvider) (kube.Client, error) {
 	cs := fake.NewSimpleClientset()
 
 	ls, fs := selectors()
 	return &fakeClient{
-		Pods:     map[string]*kube.Pod{},
-		Rules:    rules,
-		Filters:  filters,
-		Informer: kube.NewFakeInformer(cs, "", ls, fs),
-		StopCh:   make(chan struct{}),
+		Pods:         map[string]*kube.Pod{},
+		Rules:        rules,
+		Filters:      filters,
+		Associations: associations,
+		Informer:     kube.NewFakeInformer(cs, "", ls, fs),
+		StopCh:       make(chan struct{}),
 	}, nil
 }
 
-// GetPodByIP looks up FakeClient.Pods map by the provided string.
-func (f *fakeClient) GetPodByIP(ip string) (*kube.Pod, bool) {
-	p, ok := f.Pods[ip]
+// GetPod looks up FakeClient.Pods map by the provided string,
+// which might represent either IP address or Pod UID.
+func (f *fakeClient) GetPod(identifier string) (*kube.Pod, bool) {
+	p, ok := f.Pods[identifier]
 	return p, ok
 }
 

processor/k8sprocessor/config.go

@@ -39,6 +39,8 @@ type Config struct {
 	// Filter section allows specifying filters to filter
 	// pods by labels, fields, namespaces, nodes, etc.
 	Filter FilterConfig `mapstructure:"filter"`
+
+	Association []PodAssociationConfig `mapstructure:"pod_association"`
 }
 
 // ExtractConfig section allows specifying extraction rules to extract
@@ -173,3 +175,15 @@ type FieldFilterConfig struct {
 	//   equals, not-equals, exists, does-not-exist.
 	Op string `mapstructure:"op"`
 }
+
+// PodAssociationConfig allows specifying rules for associating resources
+// with pod metadata
+type PodAssociationConfig struct {
+	// From represents the source of the association.
+	// Allowed values are connection and labels
+	From string `mapstructure:"from"`
+
+	// Name represents the name of the association.
+	// e.g. ip, pod_uid
+	Name string `mapstructure:"name"`
+}

processor/k8sprocessor/config_test.go

@@ -89,5 +89,27 @@ func TestLoadConfig(t *testing.T) {
 					{Key: "key2", Value: "value2", Op: "not-equals"},
 				},
 			},
+			Association: []PodAssociationConfig{
+				{
+					From: "labels",
+					Name: "ip",
+				},
+				{
+					From: "labels",
+					Name: "k8s.pod.ip",
+				},
+				{
+					From: "labels",
+					Name: "host.hostname",
+				},
+				{
+					From: "connection",
+					Name: "ip",
+				},
+				{
+					From: "labels",
+					Name: "pod_uid",
+				},
+			},
 		})
 }

processor/k8sprocessor/doc.go

@@ -15,13 +15,27 @@
 // Package k8sprocessor allow automatic tagging of spans, metrics and logs with k8s metadata.
 //
 // The processor automatically discovers k8s resources (pods), extracts metadata from them and adds the
-// extracted metadata to the relevant spans, metrics and logs. The processor use the kubernetes API to discover all pods
-// running in a cluster, keeps a record of their IP addresses and interesting metadata. Upon receiving telemetry data,
-// the processor looks for presence of well-known resource attributes which might contain IP address ("ip",
-// "k8s.pod.ip" for logs, metrics or traces and "host.name" for metrics). If this field is not available, or it
-// does not contain a valid IP address, the processor tries to identify the source IP address of the service
-// that sent the telemetry data.
-// If a match is found, the cached metadata is added to the data as resource attributes.
+// 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 source record with specific Pod Metadata are configured via "pod_association" key.
+// It represents a list of rules that are executed in 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).
+// Following rule types are available:
+//   from: "labels" - takes the attribute from resource attributes (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
+// Pod association configuration.
+// pod_association:
+//  - from: labels
+//    name: ip
+//  - from: labels
+//    name: k8s.pod.ip
+//  - from: labels
+//    name: host.hostname
+//  - from: connection
+//    name: ip
+//  - from: labels
+//    name: pod_uid
 //
 // RBAC
 //

processor/k8sprocessor/factory.go

@@ -190,5 +190,7 @@ func createProcessorOpts(cfg configmodels.Processor) []Option {
 	opts = append(opts, WithFilterFields(oCfg.Filter.Fields...))
 	opts = append(opts, WithAPIConfig(oCfg.APIConfig))
 
+	opts = append(opts, WithExtractPodAssociations(oCfg.Association...))
+
 	return opts
 }

processor/k8sprocessor/kube/client.go

@@ -45,18 +45,26 @@ type WatchClient struct {
 	deleteQueue     []deleteRequest
 	stopCh          chan struct{}
 
-	Pods    map[string]*Pod
-	Rules   ExtractionRules
-	Filters Filters
+	Pods         map[string]*Pod
+	Rules        ExtractionRules
+	Filters      Filters
+	Associations []Association
 }
 
 // Extract deployment name from the pod name. Pod name is created using
 // format: [deployment-name]-[Random-String-For-ReplicaSet]-[Random-String-For-Pod]
 var dRegex = regexp.MustCompile(`^(.*)-[0-9a-zA-Z]*-[0-9a-zA-Z]*$`)
 
 // New initializes a new k8s Client.
-func New(logger *zap.Logger, apiCfg k8sconfig.APIConfig, rules ExtractionRules, filters Filters, newClientSet APIClientsetProvider, newInformer InformerProvider) (Client, error) {
-	c := &WatchClient{logger: logger, Rules: rules, Filters: filters, deploymentRegex: dRegex, stopCh: make(chan struct{})}
+func New(logger *zap.Logger, apiCfg k8sconfig.APIConfig, rules ExtractionRules, filters Filters, associations []Association, newClientSet APIClientsetProvider, newInformer InformerProvider) (Client, error) {
+	c := &WatchClient{
+		logger:          logger,
+		Rules:           rules,
+		Filters:         filters,
+		Associations:    associations,
+		deploymentRegex: dRegex,
+		stopCh:          make(chan struct{}),
+	}
 	go c.deleteLoop(time.Second*30, defaultPodDeleteGracePeriod)
 
 	c.Pods = map[string]*Pod{}
@@ -109,6 +117,8 @@ func (c *WatchClient) handlePodAdd(obj interface{}) {
 	} else {
 		c.logger.Error("object received was not of type api_v1.Pod", zap.Any("received", obj))
 	}
+	podTableSize := len(c.Pods)
+	observability.RecordPodTableSize(int64(podTableSize))
 }
 
 func (c *WatchClient) handlePodUpdate(old, new interface{}) {
@@ -119,6 +129,8 @@ func (c *WatchClient) handlePodUpdate(old, new interface{}) {
 	} else {
 		c.logger.Error("object received was not of type api_v1.Pod", zap.Any("received", new))
 	}
+	podTableSize := len(c.Pods)
+	observability.RecordPodTableSize(int64(podTableSize))
 }
 
 func (c *WatchClient) handlePodDelete(obj interface{}) {
@@ -128,6 +140,8 @@ func (c *WatchClient) handlePodDelete(obj interface{}) {
 	} else {
 		c.logger.Error("object received was not of type api_v1.Pod", zap.Any("received", obj))
 	}
+	podTableSize := len(c.Pods)
+	observability.RecordPodTableSize(int64(podTableSize))
 }
 
 func (c *WatchClient) deleteLoop(interval time.Duration, gracePeriod time.Duration) {
@@ -152,14 +166,16 @@ func (c *WatchClient) deleteLoop(interval time.Duration, gracePeriod time.Durati
 
 			c.m.Lock()
 			for _, d := range toDelete {
-				if p, ok := c.Pods[d.ip]; ok {
+				if p, ok := c.Pods[d.id]; ok {
 					// Sanity check: make sure we are deleting the same pod
 					// and the underlying state (ip<>pod mapping) has not changed.
 					if p.Name == d.name {
-						delete(c.Pods, d.ip)
+						delete(c.Pods, d.id)
 					}
 				}
 			}
+			podTableSize := len(c.Pods)
+			observability.RecordPodTableSize(int64(podTableSize))
 			c.m.Unlock()
 
 		case <-c.stopCh:
@@ -168,10 +184,10 @@ func (c *WatchClient) deleteLoop(interval time.Duration, gracePeriod time.Durati
 	}
 }
 
-// GetPodByIP takes an IP address and returns the pod the IP address is associated with.
-func (c *WatchClient) GetPodByIP(ip string) (*Pod, bool) {
+// GetPod takes an IP address or Pod UID and returns the pod the identifier is associated with.
+func (c *WatchClient) GetPod(identifier string) (*Pod, bool) {
 	c.m.RLock()
-	pod, ok := c.Pods[ip]
+	pod, ok := c.Pods[identifier]
 	c.m.RUnlock()
 	if ok {
 		if pod.Ignore {
@@ -253,24 +269,12 @@ func (c *WatchClient) extractField(v string, r FieldExtractionRule) string {
 }
 
 func (c *WatchClient) addOrUpdatePod(pod *api_v1.Pod) {
-	if pod.Status.PodIP == "" {
-		return
-	}
-
 	c.m.Lock()
 	defer c.m.Unlock()
-	// 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[pod.Status.PodIP]; ok {
-		if p.StartTime != nil && pod.Status.StartTime.Before(p.StartTime) {
-			return
-		}
-	}
 	newPod := &Pod{
 		Name:      pod.Name,
 		Address:   pod.Status.PodIP,
+		PodUID:    string(pod.UID),
 		StartTime: pod.Status.StartTime,
 	}
 
@@ -279,21 +283,46 @@ func (c *WatchClient) addOrUpdatePod(pod *api_v1.Pod) {
 	} else {
 		newPod.Attributes = c.extractPodAttributes(pod)
 	}
-	c.Pods[pod.Status.PodIP] = newPod
+	if pod.UID != "" {
+		c.Pods[string(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[pod.Status.PodIP]; ok {
+			if p.StartTime != nil && pod.Status.StartTime.Before(p.StartTime) {
+				return
+			}
+		}
+		c.Pods[pod.Status.PodIP] = newPod
+	}
 }
 
 func (c *WatchClient) forgetPod(pod *api_v1.Pod) {
-	if pod.Status.PodIP == "" {
-		return
+	c.m.RLock()
+	p, ok := c.GetPod(pod.Status.PodIP)
+	c.m.RUnlock()
+
+	if ok && p.Name == pod.Name {
+		c.deleteMut.Lock()
+		c.deleteQueue = append(c.deleteQueue, deleteRequest{
+			id:   pod.Status.PodIP,
+			name: pod.Name,
+			ts:   time.Now(),
+		})
+		c.deleteMut.Unlock()
 	}
+
 	c.m.RLock()
-	p, ok := c.GetPodByIP(pod.Status.PodIP)
+	p, ok = c.GetPod(string(pod.UID))
 	c.m.RUnlock()
 
 	if ok && p.Name == pod.Name {
 		c.deleteMut.Lock()
 		c.deleteQueue = append(c.deleteQueue, deleteRequest{
-			ip:   pod.Status.PodIP,
+			id:   string(pod.UID),
 			name: pod.Name,
 			ts:   time.Now(),
 		})