[processor/k8sattributes] Store only necessary ReplicaSet data (#23338)

.chloggen/feat_k8sattributes_set-transform-replicaset.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# 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: k8sattributesprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Store only necessary ReplicaSet data
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [23226]
+
+# (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:

processor/k8sattributesprocessor/internal/kube/client.go

@@ -119,6 +119,9 @@ func New(logger *zap.Logger, apiCfg k8sconfig.APIConfig, rules ExtractionRules,
 			return removeUnnecessaryPodData(originalPod, c.Rules), nil
 		},
 	)
+	if err != nil {
+		return nil, err
+	}
 
 	if c.extractNamespaceLabelsAnnotations() {
 		c.namespaceInformer = newNamespaceInformer(c.kc)
@@ -131,6 +134,19 @@ func New(logger *zap.Logger, apiCfg k8sconfig.APIConfig, rules ExtractionRules,
 			newReplicaSetInformer = newReplicaSetSharedInformer
 		}
 		c.replicasetInformer = newReplicaSetInformer(c.kc, c.Filters.Namespace)
+		err = c.replicasetInformer.SetTransform(
+			func(object interface{}) (interface{}, error) {
+				originalReplicaset, success := object.(*apps_v1.ReplicaSet)
+				if !success { // means this is a cache.DeletedFinalStateUnknown, in which case we do nothing
+					return object, nil
+				}
+
+				return removeUnnecessaryReplicaSetData(originalReplicaset), nil
+			},
+		)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	return c, err
@@ -846,6 +862,19 @@ func (c *WatchClient) addOrUpdateReplicaSet(replicaset *apps_v1.ReplicaSet) {
 	c.m.Unlock()
 }
 
+// This function removes all data from the ReplicaSet except what is required by extraction rules
+func removeUnnecessaryReplicaSetData(replicaset *apps_v1.ReplicaSet) *apps_v1.ReplicaSet {
+	transformedReplicaset := apps_v1.ReplicaSet{
+		ObjectMeta: meta_v1.ObjectMeta{
+			Name:      replicaset.GetName(),
+			Namespace: replicaset.GetNamespace(),
+			UID:       replicaset.GetUID(),
+		},
+	}
+	transformedReplicaset.SetOwnerReferences(replicaset.GetOwnerReferences())
+	return &transformedReplicaset
+}
+
 func (c *WatchClient) getReplicaSet(uid string) (*ReplicaSet, bool) {
 	c.m.RLock()
 	replicaset, ok := c.ReplicaSets[uid]

processor/k8sattributesprocessor/internal/kube/client_test.go

@@ -847,10 +847,11 @@ func TestExtractionRules(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			c.Rules = tc.rules
 
-			// manually call the data removal function here
+			// manually call the data removal functions here
 			// normally the informer does this, but fully emulating the informer in this test is annoying
 			transformedPod := removeUnnecessaryPodData(pod, c.Rules)
-			c.handleReplicaSetAdd(replicaset)
+			transformedReplicaset := removeUnnecessaryReplicaSetData(replicaset)
+			c.handleReplicaSetAdd(transformedReplicaset)
 			c.handlePodAdd(transformedPod)
 			p, ok := c.GetPod(newPodIdentifier("connection", "", pod.Status.PodIP))
 			require.True(t, ok)
@@ -1001,10 +1002,11 @@ func TestReplicaSetExtractionRules(t *testing.T) {
 			c.Rules = tc.rules
 			replicaset.OwnerReferences = tc.ownerReferences
 
-			// manually call the data removal function here
+			// manually call the data removal functions here
 			// normally the informer does this, but fully emulating the informer in this test is annoying
 			transformedPod := removeUnnecessaryPodData(pod, c.Rules)
-			c.handleReplicaSetAdd(replicaset)
+			transformedReplicaset := removeUnnecessaryReplicaSetData(replicaset)
+			c.handleReplicaSetAdd(transformedReplicaset)
 			c.handlePodAdd(transformedPod)
 			p, ok := c.GetPod(newPodIdentifier("connection", "", pod.Status.PodIP))
 			require.True(t, ok)