[processor/k8sattribute] Support adding labels and annotations from n…

…ode (open-telemetry#28570)

**Description:**
support adding labels and annotations from the node as additional
resource attributes on telemetry processed through the `k8sattributes`
processor.

**Link to tracking Issue:** Resolve  open-telemetry#22620

---------

Co-authored-by: Tyler Helmuth <12352919+TylerHelmuth@users.noreply.github.com>

.chloggen/feat-22620.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: processor/k8sattribute
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: support adding labels and annotations from node
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [22620]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

.github/workflows/configs/e2e-kind-config.yaml

@@ -4,3 +4,8 @@ kubeadmConfigPatches:
   - |
     kind: KubeletConfiguration
     serverTLSBootstrap: true
+nodes:
+  - role: control-plane
+    labels:
+      # used in k8sattributesprocessor e2e test
+      foo: too

processor/k8sattributesprocessor/README.md

@@ -88,12 +88,12 @@ Additional container level attributes can be extracted provided that certain res
    instance. If it's not set, the latest container instance will be used:
    - container.id (not added by default, has to be specified in `metadata`)
 
-The k8sattributesprocessor can also set resource attributes from k8s labels and annotations of pods and namespaces.
-The config for associating the data passing through the processor (spans, metrics and logs) with specific Pod/Namespace annotations/labels is configured via "annotations"  and "labels" keys.
-This config represents a list of annotations/labels that are extracted from pods/namespaces and added to spans, metrics and logs.
+The k8sattributesprocessor can also set resource attributes from k8s labels and annotations of pods, namespaces and nodes.
+The config for associating the data passing through the processor (spans, metrics and logs) with specific Pod/Namespace/Node annotations/labels is configured via "annotations"  and "labels" keys.
+This config represents a list of annotations/labels that are extracted from pods/namespaces/nodes and added to spans, metrics and logs.
 Each item is specified as a config of tag_name (representing the tag name to tag the spans with),
 key (representing the key used to extract value) and from (representing the kubernetes object used to extract the value).
-The "from" field has only two possible values "pod" and "namespace" and defaults to "pod" if none is specified.
+The "from" field has only three possible values "pod", "namespace" and "node" and defaults to "pod" if none is specified.
 
 A few examples to use this config are as follows:
 
@@ -106,6 +106,10 @@ annotations:
     key: annotation-two
     regex: field=(?P<value>.+)
     from: namespace
+  - tag_name: a3 # extracts value of annotation from nodes with key `annotation-three` with regexp and inserts it as a tag with key `a3`
+    key: annotation-three
+    regex: field=(?P<value>.+)
+    from: node
 
 labels:
   - tag_name: l1 # extracts value of label from namespaces with key `label1` and inserts it as a tag with key `l1`
@@ -115,6 +119,9 @@ labels:
     key: label2
     regex: field=(?P<value>.+)
     from: pod
+  - tag_name: l3 # extracts value of label from nodes with key `label3` and inserts it as a tag with key `l3`
+    key: label3
+    from: node
 ```
 
 ### Config example
@@ -147,7 +154,7 @@ k8sattributes/2:
 
 ## Role-based access control
 
-The k8sattributesprocessor needs `get`, `watch` and `list` permissions on both `pods` and `namespaces` resources, for all namespaces and pods included in the configured filters. Additionally, when using `k8s.deployment.uid` or `k8s.deployment.name` the processor also needs `get`, `watch` and `list` permissions for `replicaset` resources.
+The k8sattributesprocessor needs `get`, `watch` and `list` permissions on both `pods` and `namespaces` resources, for all namespaces and pods included in the configured filters. Additionally, when using `k8s.deployment.uid` or `k8s.deployment.name` the processor also needs `get`, `watch` and `list` permissions for `replicaset` resources. When extracting metadatas from `node`, the processor needs `get`, `watch` and `list` permissions for `node` resources.
 
 Here is an example of a `ClusterRole` to give a `ServiceAccount` the necessary permissions for all pods and namespaces in the cluster (replace `<OTEL_COL_NAMESPACE>` with a namespace where collector is deployed):
 

processor/k8sattributesprocessor/client_test.go

@@ -23,7 +23,9 @@ type fakeClient struct {
 	Informer           cache.SharedInformer
 	NamespaceInformer  cache.SharedInformer
 	ReplicaSetInformer cache.SharedInformer
+	NodeInformer       cache.SharedInformer
 	Namespaces         map[string]*kube.Namespace
+	Nodes              map[string]*kube.Node
 	StopCh             chan struct{}
 }
 
@@ -44,6 +46,7 @@ func newFakeClient(_ *zap.Logger, _ k8sconfig.APIConfig, rules kube.ExtractionRu
 		Associations:       associations,
 		Informer:           kube.NewFakeInformer(cs, "", ls, fs),
 		NamespaceInformer:  kube.NewFakeInformer(cs, "", ls, fs),
+		NodeInformer:       kube.NewFakeInformer(cs, "", ls, fs),
 		ReplicaSetInformer: kube.NewFakeInformer(cs, "", ls, fs),
 		StopCh:             make(chan struct{}),
 	}, nil
@@ -61,6 +64,11 @@ func (f *fakeClient) GetNamespace(namespace string) (*kube.Namespace, bool) {
 	return ns, ok
 }
 
+func (f *fakeClient) GetNode(nodeName string) (*kube.Node, bool) {
+	node, ok := f.Nodes[nodeName]
+	return node, ok
+}
+
 // Start is a noop for FakeClient.
 func (f *fakeClient) Start() {
 	if f.Informer != nil {

processor/k8sattributesprocessor/config.go

@@ -57,9 +57,9 @@ func (cfg *Config) Validate() error {
 		}
 
 		switch f.From {
-		case "", kube.MetadataFromPod, kube.MetadataFromNamespace:
+		case "", kube.MetadataFromPod, kube.MetadataFromNamespace, kube.MetadataFromNode:
 		default:
-			return fmt.Errorf("%s is not a valid choice for From. Must be one of: pod, namespace", f.From)
+			return fmt.Errorf("%s is not a valid choice for From. Must be one of: pod, namespace, node", f.From)
 		}
 
 		if f.Regex != "" {
@@ -117,7 +117,7 @@ func (cfg *Config) Validate() error {
 // ExtractConfig section allows specifying extraction rules to extract
 // data from k8s pod specs.
 type ExtractConfig struct {
-	// Metadata allows to extract pod/namespace metadata from a list of metadata fields.
+	// Metadata allows to extract pod/namespace/node metadata from a list of metadata fields.
 	// The field accepts a list of strings.
 	//
 	// Metadata fields supported right now are,

processor/k8sattributesprocessor/e2e_test.go

@@ -107,6 +107,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -129,6 +130,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -175,6 +177,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -197,6 +200,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -219,6 +223,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -265,6 +270,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -287,6 +293,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -333,6 +340,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 		{
@@ -355,6 +363,7 @@ func TestE2E(t *testing.T) {
 				"container.image.name":     newExpectedValue(equal, "ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen"),
 				"container.image.tag":      newExpectedValue(equal, "latest"),
 				"container.id":             newExpectedValue(exist, ""),
+				"k8s.node.labels.foo":      newExpectedValue(equal, "too"),
 			},
 		},
 	}

processor/k8sattributesprocessor/internal/kube/client.go

@@ -42,6 +42,7 @@ type WatchClient struct {
 	kc                 kubernetes.Interface
 	informer           cache.SharedInformer
 	namespaceInformer  cache.SharedInformer
+	nodeInformer       cache.SharedInformer
 	replicasetInformer cache.SharedInformer
 	replicasetRegex    *regexp.Regexp
 	cronJobRegex       *regexp.Regexp
@@ -60,6 +61,10 @@ type WatchClient struct {
 	// Key is namespace name
 	Namespaces map[string]*Namespace
 
+	// A map containing Node related data, used to associate them with resources.
+	// Key is node name
+	Nodes map[string]*Node
+
 	// A map containing ReplicaSets related data, used to associate them with resources.
 	// Key is replicaset uid
 	ReplicaSets map[string]*ReplicaSet
@@ -89,6 +94,7 @@ func New(logger *zap.Logger, apiCfg k8sconfig.APIConfig, rules ExtractionRules,
 
 	c.Pods = map[PodIdentifier]*Pod{}
 	c.Namespaces = map[string]*Namespace{}
+	c.Nodes = map[string]*Node{}
 	c.ReplicaSets = map[string]*ReplicaSet{}
 	if newClientSet == nil {
 		newClientSet = k8sconfig.MakeClient
@@ -162,6 +168,10 @@ func New(logger *zap.Logger, apiCfg k8sconfig.APIConfig, rules ExtractionRules,
 		}
 	}
 
+	if c.extractNodeLabelsAnnotations() {
+		c.nodeInformer = newNodeSharedInformer(c.kc, c.Filters.Node)
+	}
+
 	return c, err
 }
 
@@ -198,6 +208,18 @@ func (c *WatchClient) Start() {
 		}
 		go c.replicasetInformer.Run(c.stopCh)
 	}
+
+	if c.nodeInformer != nil {
+		_, err = c.nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc:    c.handleNodeAdd,
+			UpdateFunc: c.handleNodeUpdate,
+			DeleteFunc: c.handleNodeDelete,
+		})
+		if err != nil {
+			c.logger.Error("error adding event handler to node informer", zap.Error(err))
+		}
+		go c.nodeInformer.Run(c.stopCh)
+	}
 }
 
 // Stop signals the the k8s watcher/informer to stop watching for new events.
@@ -273,6 +295,37 @@ func (c *WatchClient) handleNamespaceDelete(obj interface{}) {
 	}
 }
 
+func (c *WatchClient) handleNodeAdd(obj interface{}) {
+	observability.RecordNodeAdded()
+	if node, ok := obj.(*api_v1.Node); ok {
+		c.addOrUpdateNode(node)
+	} else {
+		c.logger.Error("object received was not of type api_v1.Node", zap.Any("received", obj))
+	}
+}
+
+func (c *WatchClient) handleNodeUpdate(_, newNode interface{}) {
+	observability.RecordNodeUpdated()
+	if node, ok := newNode.(*api_v1.Node); ok {
+		c.addOrUpdateNode(node)
+	} else {
+		c.logger.Error("object received was not of type api_v1.Node", zap.Any("received", newNode))
+	}
+}
+
+func (c *WatchClient) handleNodeDelete(obj interface{}) {
+	observability.RecordNodeDeleted()
+	if node, ok := ignoreDeletedFinalStateUnknown(obj).(*api_v1.Node); ok {
+		c.m.Lock()
+		if n, ok := c.Nodes[node.Name]; ok {
+			delete(c.Nodes, n.Name)
+		}
+		c.m.Unlock()
+	} else {
+		c.logger.Error("object received was not of type api_v1.Node", zap.Any("received", obj))
+	}
+}
+
 func (c *WatchClient) deleteLoop(interval time.Duration, gracePeriod time.Duration) {
 	// This loop runs after N seconds and deletes pods from cache.
 	// It iterates over the delete queue and deletes all that aren't
@@ -339,6 +392,17 @@ func (c *WatchClient) GetNamespace(namespace string) (*Namespace, bool) {
 	return nil, false
 }
 
+// GetNode takes a node name and returns the node object the node name is associated with.
+func (c *WatchClient) GetNode(nodeName string) (*Node, bool) {
+	c.m.RLock()
+	node, ok := c.Nodes[nodeName]
+	c.m.RUnlock()
+	if ok {
+		return node, ok
+	}
+	return nil, false
+}
+
 func (c *WatchClient) extractPodAttributes(pod *api_v1.Pod) map[string]string {
 	tags := map[string]string{}
 	if c.Rules.PodName {
@@ -614,10 +678,25 @@ func (c *WatchClient) extractNamespaceAttributes(namespace *api_v1.Namespace) ma
 	return tags
 }
 
+func (c *WatchClient) extractNodeAttributes(node *api_v1.Node) map[string]string {
+	tags := map[string]string{}
+
+	for _, r := range c.Rules.Labels {
+		r.extractFromNodeMetadata(node.Labels, tags, "k8s.node.labels.%s")
+	}
+
+	for _, r := range c.Rules.Annotations {
+		r.extractFromNodeMetadata(node.Annotations, tags, "k8s.node.annotations.%s")
+	}
+
+	return tags
+}
+
 func (c *WatchClient) podFromAPI(pod *api_v1.Pod) *Pod {
 	newPod := &Pod{
 		Name:        pod.Name,
 		Namespace:   pod.GetNamespace(),
+		NodeName:    pod.Spec.NodeName,
 		Address:     pod.Status.PodIP,
 		HostNetwork: pod.Spec.HostNetwork,
 		PodUID:      string(pod.UID),
@@ -832,6 +911,36 @@ func (c *WatchClient) extractNamespaceLabelsAnnotations() bool {
 	return false
 }
 
+func (c *WatchClient) extractNodeLabelsAnnotations() bool {
+	for _, r := range c.Rules.Labels {
+		if r.From == MetadataFromNode {
+			return true
+		}
+	}
+
+	for _, r := range c.Rules.Annotations {
+		if r.From == MetadataFromNode {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (c *WatchClient) addOrUpdateNode(node *api_v1.Node) {
+	newNode := &Node{
+		Name:    node.Name,
+		NodeUID: string(node.UID),
+	}
+	newNode.Attributes = c.extractNodeAttributes(node)
+
+	c.m.Lock()
+	if node.Name != "" {
+		c.Nodes[node.Name] = newNode
+	}
+	c.m.Unlock()
+}
+
 func needContainerAttributes(rules ExtractionRules) bool {
 	return rules.ContainerImageName ||
 		rules.ContainerName ||