[receiver/k8s_cluster] Do not store unused data in the k8s API cache (o…

…pen-telemetry#23417)

This change removes unused k8s informer data from the cache to reduce RAM utilization.

Tried it on a cluster with 40 nodes and 1000 pods, and it gave up to 30% reduction in RAM usage.

.chloggen/k8scluster-dont-store-unused-data-cache.yaml

@@ -0,0 +1,15 @@
+# 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: receiver/k8s_cluster
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Do not store unused data in the k8s API cache to reduce RAM usage
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [23417]

receiver/k8sclusterreceiver/informer_transform.go

@@ -0,0 +1,31 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package k8sclusterreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver"
+
+import (
+	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/jobs"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/node"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/pod"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/replicaset"
+)
+
+// transformObject transforms the k8s object by removing the data that is not utilized by the receiver.
+// Only highly utilized objects are transformed here while others are kept as is.
+func transformObject(object interface{}) (interface{}, error) {
+	switch o := object.(type) {
+	case *corev1.Pod:
+		return pod.Transform(o), nil
+	case *corev1.Node:
+		return node.Transform(o), nil
+	case *appsv1.ReplicaSet:
+		return replicaset.Transform(o), nil
+	case *batchv1.Job:
+		return jobs.Transform(o), nil
+	}
+	return object, nil
+}

receiver/k8sclusterreceiver/informer_transform_test.go

@@ -0,0 +1,87 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package k8sclusterreceiver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/testutils"
+)
+
+func TestTransformObject(t *testing.T) {
+	i := 1
+	intPtr := &i
+	tests := []struct {
+		name   string
+		object interface{}
+		want   interface{}
+		same   bool
+	}{
+		{
+			name: "pod",
+			object: testutils.NewPodWithContainer(
+				"1",
+				testutils.NewPodSpecWithContainer("container-name"),
+				testutils.NewPodStatusWithContainer("container-name", "container-id"),
+			),
+			want: func() *corev1.Pod {
+				pod := testutils.NewPodWithContainer(
+					"1",
+					testutils.NewPodSpecWithContainer("container-name"),
+					testutils.NewPodStatusWithContainer("container-name", "container-id"),
+				)
+				pod.Spec.Containers[0].Image = ""
+				pod.Status.ContainerStatuses[0].State = corev1.ContainerState{}
+				return pod
+			}(),
+			same: false,
+		},
+		{
+			name:   "node",
+			object: testutils.NewNode("1"),
+			want:   testutils.NewNode("1"),
+			same:   false,
+		},
+		{
+			name:   "replicaset",
+			object: testutils.NewReplicaSet("1"),
+			want:   testutils.NewReplicaSet("1"),
+			same:   false,
+		},
+		{
+			name:   "job",
+			object: testutils.NewJob("1"),
+			want:   testutils.NewJob("1"),
+			same:   false,
+		},
+		{
+			// This is a case where we don't transform the object.
+			name:   "hpa",
+			object: testutils.NewHPA("1"),
+			want:   testutils.NewHPA("1"),
+			same:   true,
+		},
+		{
+			name:   "invalid_type",
+			object: intPtr,
+			want:   intPtr,
+			same:   true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := transformObject(tt.object)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+			if tt.same {
+				assert.Same(t, tt.object, got)
+			} else {
+				assert.NotSame(t, tt.object, got)
+			}
+		})
+	}
+}

receiver/k8sclusterreceiver/internal/collection/collector_test.go

@@ -62,22 +62,6 @@ func TestDataCollectorSyncMetadata(t *testing.T) {
 				},
 			},
 		},
-		{
-			name:          "Empty container id skips container resource",
-			metadataStore: &metadata.Store{},
-			resource: testutils.NewPodWithContainer(
-				"0",
-				testutils.NewPodSpecWithContainer("container-name"),
-				testutils.NewPodStatusWithContainer("container-name", ""),
-			),
-			want: map[experimentalmetricmetadata.ResourceID]*metadata.KubernetesMetadata{
-				experimentalmetricmetadata.ResourceID("test-pod-0-uid"): {
-					ResourceIDKey: "k8s.pod.uid",
-					ResourceID:    "test-pod-0-uid",
-					Metadata:      commonPodMetadata,
-				},
-			},
-		},
 		{
 			name:          "Pod with Owner Reference",
 			metadataStore: &metadata.Store{},

receiver/k8sclusterreceiver/internal/jobs/jobs.go

@@ -9,6 +9,7 @@ import (
 	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
 	conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
 	batchv1 "k8s.io/api/batch/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/constants"
@@ -51,6 +52,28 @@ var podsSuccessfulMetric = &metricspb.MetricDescriptor{
 	Type:        metricspb.MetricDescriptor_GAUGE_INT64,
 }
 
+// Transform transforms the job to remove the fields that we don't use to reduce RAM utilization.
+// IMPORTANT: Make sure to update this function when using a new job fields.
+func Transform(job *batchv1.Job) *batchv1.Job {
+	return &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      job.ObjectMeta.Name,
+			Namespace: job.ObjectMeta.Namespace,
+			UID:       job.ObjectMeta.UID,
+			Labels:    job.ObjectMeta.Labels,
+		},
+		Spec: batchv1.JobSpec{
+			Completions: job.Spec.Completions,
+			Parallelism: job.Spec.Parallelism,
+		},
+		Status: batchv1.JobStatus{
+			Active:    job.Status.Active,
+			Succeeded: job.Status.Succeeded,
+			Failed:    job.Status.Failed,
+		},
+	}
+}
+
 func GetMetrics(j *batchv1.Job) []*agentmetricspb.ExportMetricsServiceRequest {
 	metrics := make([]*metricspb.Metric, 0, 5)
 	metrics = append(metrics, []*metricspb.Metric{

receiver/k8sclusterreceiver/internal/jobs/jobs_test.go

@@ -7,7 +7,11 @@ import (
 	"testing"
 
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/constants"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/testutils"
@@ -60,3 +64,69 @@ func TestJobMetrics(t *testing.T) {
 	testutils.AssertMetricsInt(t, actualResourceMetrics[0].Metrics[2], "k8s.job.successful_pods",
 		metricspb.MetricDescriptor_GAUGE_INT64, 3)
 }
+
+func TestTransform(t *testing.T) {
+	originalJob := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-job",
+			Namespace: "default",
+			UID:       "my-job-uid",
+			Labels: map[string]string{
+				"app": "my-app",
+			},
+		},
+		Spec: batchv1.JobSpec{
+			Completions: func() *int32 { completions := int32(1); return &completions }(),
+			Parallelism: func() *int32 { parallelism := int32(1); return &parallelism }(),
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"app": "my-app",
+					},
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:            "my-container",
+							Image:           "busybox",
+							Command:         []string{"echo", "Hello, World!"},
+							ImagePullPolicy: corev1.PullAlways,
+						},
+					},
+					RestartPolicy: corev1.RestartPolicyNever,
+				},
+			},
+		},
+		Status: batchv1.JobStatus{
+			Active:    1,
+			Succeeded: 2,
+			Failed:    3,
+			Conditions: []batchv1.JobCondition{
+				{
+					Type:   batchv1.JobComplete,
+					Status: corev1.ConditionTrue,
+				},
+			},
+		},
+	}
+	wantJob := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-job",
+			Namespace: "default",
+			UID:       "my-job-uid",
+			Labels: map[string]string{
+				"app": "my-app",
+			},
+		},
+		Spec: batchv1.JobSpec{
+			Completions: func() *int32 { completions := int32(1); return &completions }(),
+			Parallelism: func() *int32 { parallelism := int32(1); return &parallelism }(),
+		},
+		Status: batchv1.JobStatus{
+			Active:    1,
+			Succeeded: 2,
+			Failed:    3,
+		},
+	}
+	assert.Equal(t, wantJob, Transform(originalJob))
+}

receiver/k8sclusterreceiver/internal/node/nodes.go

@@ -14,6 +14,7 @@ import (
 	conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/maps"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata"
@@ -34,6 +35,28 @@ var allocatableDesciption = map[string]string{
 	"storage":           "How many bytes of storage remaining that the node can allocate to pods",
 }
 
+// Transform transforms the node to remove the fields that we don't use to reduce RAM utilization.
+// IMPORTANT: Make sure to update this function when using a new node fields.
+func Transform(node *corev1.Node) *corev1.Node {
+	newNode := &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   node.ObjectMeta.Name,
+			UID:    node.ObjectMeta.UID,
+			Labels: node.ObjectMeta.Labels,
+		},
+		Status: corev1.NodeStatus{
+			Allocatable: node.Status.Allocatable,
+		},
+	}
+	for _, c := range node.Status.Conditions {
+		newNode.Status.Conditions = append(newNode.Status.Conditions, corev1.NodeCondition{
+			Type:   c.Type,
+			Status: c.Status,
+		})
+	}
+	return newNode
+}
+
 func GetMetrics(node *corev1.Node, nodeConditionTypesToReport, allocatableTypesToReport []string, logger *zap.Logger) []*agentmetricspb.ExportMetricsServiceRequest {
 	metrics := make([]*metricspb.Metric, 0, len(nodeConditionTypesToReport)+len(allocatableTypesToReport))
 	// Adding 'node condition type' metrics

receiver/k8sclusterreceiver/internal/node/nodes_test.go

@@ -7,9 +7,12 @@ import (
 	"testing"
 
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/constants"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/testutils"
@@ -140,3 +143,63 @@ func TestNodeConditionValue(t *testing.T) {
 		})
 	}
 }
+
+func TestTransform(t *testing.T) {
+	originalNode := &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-node",
+			UID:  "my-node-uid",
+			Labels: map[string]string{
+				"node-role": "worker",
+			},
+		},
+		Status: corev1.NodeStatus{
+			Conditions: []corev1.NodeCondition{
+				{
+					Type:   corev1.NodeReady,
+					Status: corev1.ConditionTrue,
+				},
+			},
+			Capacity: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("8"),
+				corev1.ResourceMemory: resource.MustParse("16Gi"),
+			},
+			Allocatable: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("4"),
+				corev1.ResourceMemory: resource.MustParse("8Gi"),
+			},
+			Addresses: []corev1.NodeAddress{
+				{
+					Type:    corev1.NodeHostName,
+					Address: "my-node-hostname",
+				},
+				{
+					Type:    corev1.NodeInternalIP,
+					Address: "192.168.1.100",
+				},
+			},
+		},
+	}
+	wantNode := &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "my-node",
+			UID:  "my-node-uid",
+			Labels: map[string]string{
+				"node-role": "worker",
+			},
+		},
+		Status: corev1.NodeStatus{
+			Conditions: []corev1.NodeCondition{
+				{
+					Type:   corev1.NodeReady,
+					Status: corev1.ConditionTrue,
+				},
+			},
+			Allocatable: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("4"),
+				corev1.ResourceMemory: resource.MustParse("8Gi"),
+			},
+		},
+	}
+	assert.Equal(t, wantNode, Transform(originalNode))
+}