[receiver/kubeletstats] Add k8s.container.cpu.node.utilization metr…

…ic (#32295)

**Description:** <Describe what has changed.>
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->

At the moment. We calculate the `k8s.container.cpu_limit_utilization` as
[a ratio of the container's
limits](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/receiver/kubeletstatsreceiver/documentation.md#k8scontainercpu_limit_utilization)
at
https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/867d6700c31446172e6998e602c55fbf7351831f/receiver/kubeletstatsreceiver/internal/kubelet/cpu.go#L30.

Similarly we can calculate the cpu utilization as ratio of the whole
node's allocatable cpu, if we divide by the total number of node's
cores.

We can retrieve this information from the Node's `Status.Capacity`, for
example:

```console
$ k get nodes kind-control-plane -ojsonpath='{.status.capacity}'
{"cpu":"8","ephemeral-storage":"485961008Ki","hugepages-1Gi":"0","hugepages-2Mi":"0","memory":"32564732Ki","pods":"110"}
```




## Performance concerns

In order to get the Node's capacity we need an API call to the k8s API
in order to get the Node object.
Something to consider here is the performance impact that this extra API
call would bring. We can always choose to have this metric as disabled
by default and clearly specify in the docs that this metric comes with
an extra API call to get the Node of the Pods.

The good thing is that `kubeletstats` receiver target's only one node so
I believe it's a safe assumption to only fetch the current node because
all the observed Pods will belong to the one single local node. Correct
me if I miss anything here.

In addition, instead of performing the API call explicitly on every
single `scrape` we can use an informer instead and leverage its cache. I
can change this patch to this direction if we agree on this.

Would love to hear other's opinions on this. 

## Todos

✅ 1) Apply this change behind a feature gate as it was indicated at
#27885 (comment)
✅  2) Use an Informer instead of direct API calls.

**Link to tracking Issue:** <Issue number if applicable>
ref:
#27885

**Testing:** <Describe what testing was performed and which tests were
added.>

I experimented with this approach and the results look correct. In order
to verify this I deployed a stress Pod on my machine to consume a target
cpu of 4 cores:

```yaml
apiVersion: v1
kind: Pod
metadata:
  name: cpu-stress
spec:
  containers:
  - name: cpu-stress
    image: polinux/stress
    command: ["stress"]
    args: ["-c", "4"]
```

And then the collected `container.cpu.utilization` for that Pod's
container was at `0,5` as exepcted, based that my machine-node comes
with 8 cores in total:


![cpu-stress](https://github.com/open-telemetry/opentelemetry-collector-contrib/assets/11754898/3abe4a0d-6c99-4b4e-a704-da5789dde01b)

Unit test is also included.

**Documentation:** <Describe the documentation added.>

Added:
https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/32295/files#diff-8ad3b506fb1132c961e8da99b677abd31f0108e3f9ed6999dd96ad3297b51e08

---------

Signed-off-by: ChrsMark <[email protected]>

.chloggen/k8s_container_metrics.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: kubeletstatsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add k8s.container.cpu.node.utilization metric
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [27885]
+
+# (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: []

receiver/kubeletstatsreceiver/README.md

@@ -218,6 +218,36 @@ receivers:
       - pod
 ```
 
+### Collect k8s.container.cpu.node.utilization as ratio of total node's capacity
+
+In order to calculate the `k8s.container.cpu.node.utilization` metric, the information of the node's capacity
+must be retrieved from the k8s API. In this, the `k8s_api_config` needs to be set.
+In addition, the node name must be identified properly. The `K8S_NODE_NAME` env var can be set using the
+downward API inside the collector pod spec as follows:
+
+```yaml
+env:
+  - name: K8S_NODE_NAME
+    valueFrom:
+      fieldRef:
+        fieldPath: spec.nodeName
+```
+Then set `node` value to `${env:K8S_NODE_NAME}` in the receiver's configuration:
+
+```yaml
+receivers:
+    kubeletstats:
+      collection_interval: 10s
+      auth_type: 'serviceAccount'
+      endpoint: '${env:K8S_NODE_NAME}:10250'
+      node: '${env:K8S_NODE_NAME}'
+      k8s_api_config:
+        auth_type: serviceAccount
+      metrics:
+        k8s.container.cpu.node.utilization:
+          enabled: true
+```
+
 ### Optional parameters
 
 The following parameters can also be specified:

receiver/kubeletstatsreceiver/config.go

@@ -40,6 +40,19 @@ type Config struct {
 	// Configuration of the Kubernetes API client.
 	K8sAPIConfig *k8sconfig.APIConfig `mapstructure:"k8s_api_config"`
 
+	// NodeName is the node name to limit the discovery of nodes.
+	// For example, node name can be set using the downward API inside the collector
+	// pod spec as follows:
+	//
+	// env:
+	//   - name: K8S_NODE_NAME
+	//     valueFrom:
+	//       fieldRef:
+	//         fieldPath: spec.nodeName
+	//
+	// Then set this value to ${env:K8S_NODE_NAME} in the configuration.
+	NodeName string `mapstructure:"node"`
+
 	// MetricsBuilderConfig allows customizing scraped metrics/attributes representation.
 	metadata.MetricsBuilderConfig `mapstructure:",squash"`
 }
@@ -105,3 +118,10 @@ func (cfg *Config) Unmarshal(componentParser *confmap.Conf) error {
 
 	return nil
 }
+
+func (cfg *Config) Validate() error {
+	if cfg.Metrics.K8sContainerCPUNodeUtilization.Enabled && cfg.NodeName == "" {
+		return errors.New("for k8s.container.cpu.node.utilization node setting is required. Check the readme on how to set the required setting")
+	}
+	return nil
+}

receiver/kubeletstatsreceiver/config_test.go

@@ -31,9 +31,9 @@ func TestLoadConfig(t *testing.T) {
 	duration := 10 * time.Second
 
 	tests := []struct {
-		id          component.ID
-		expected    component.Config
-		expectedErr error
+		id                    component.ID
+		expected              component.Config
+		expectedValidationErr string
 	}{
 		{
 			id: component.NewIDWithName(metadata.Type, "default"),
@@ -173,6 +173,34 @@ func TestLoadConfig(t *testing.T) {
 				MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
 			},
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "container_cpu_node_utilization"),
+			expected: &Config{
+				ControllerConfig: scraperhelper.ControllerConfig{
+					CollectionInterval: duration,
+					InitialDelay:       time.Second,
+				},
+				ClientConfig: kube.ClientConfig{
+					APIConfig: k8sconfig.APIConfig{
+						AuthType: "tls",
+					},
+				},
+				MetricGroupsToCollect: []kubelet.MetricGroup{
+					kubelet.ContainerMetricGroup,
+					kubelet.PodMetricGroup,
+					kubelet.NodeMetricGroup,
+				},
+				MetricsBuilderConfig: metadata.MetricsBuilderConfig{
+					Metrics: metadata.MetricsConfig{
+						K8sContainerCPUNodeUtilization: metadata.MetricConfig{
+							Enabled: true,
+						},
+					},
+					ResourceAttributes: metadata.DefaultResourceAttributesConfig(),
+				},
+			},
+			expectedValidationErr: "for k8s.container.cpu.node.utilization node setting is required. Check the readme on how to set the required setting",
+		},
 	}
 
 	for _, tt := range tests {
@@ -184,8 +212,14 @@ func TestLoadConfig(t *testing.T) {
 			require.NoError(t, err)
 			require.NoError(t, component.UnmarshalConfig(sub, cfg))
 
-			assert.NoError(t, component.ValidateConfig(cfg))
-			assert.Equal(t, tt.expected, cfg)
+			err = component.ValidateConfig(cfg)
+			if tt.expectedValidationErr != "" {
+				assert.EqualError(t, err, tt.expectedValidationErr)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tt.expected, cfg)
+			}
+
 		})
 	}
 }

receiver/kubeletstatsreceiver/documentation.md

@@ -402,6 +402,14 @@ The time since the container started
 | ---- | ----------- | ---------- | ----------------------- | --------- |
 | s | Sum | Int | Cumulative | true |
 
+### k8s.container.cpu.node.utilization
+
+Container cpu utilization as a ratio of the node's capacity
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| 1 | Gauge | Double |
+
 ### k8s.container.cpu_limit_utilization
 
 Container cpu utilization as a ratio of the container's limits

receiver/kubeletstatsreceiver/factory.go

@@ -68,7 +68,7 @@ func createMetricsReceiver(
 		return nil, err
 	}
 
-	scrp, err := newKubletScraper(rest, set, rOptions, cfg.MetricsBuilderConfig)
+	scrp, err := newKubletScraper(rest, set, rOptions, cfg.MetricsBuilderConfig, cfg.NodeName)
 	if err != nil {
 		return nil, err
 	}

receiver/kubeletstatsreceiver/internal/kubelet/accumulator.go

@@ -56,7 +56,7 @@ func (a *metricDataAccumulator) nodeStats(s stats.NodeStats) {
 
 	currentTime := pcommon.NewTimestampFromTime(a.time)
 	addUptimeMetric(a.mbs.NodeMetricsBuilder, metadata.NodeUptimeMetrics.Uptime, s.StartTime, currentTime)
-	addCPUMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeCPUMetrics, s.CPU, currentTime, resources{})
+	addCPUMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeCPUMetrics, s.CPU, currentTime, resources{}, 0)
 	addMemoryMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeMemoryMetrics, s.Memory, currentTime, resources{})
 	addFilesystemMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeFilesystemMetrics, s.Fs, currentTime)
 	addNetworkMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeNetworkMetrics, s.Network, currentTime)
@@ -76,7 +76,7 @@ func (a *metricDataAccumulator) podStats(s stats.PodStats) {
 
 	currentTime := pcommon.NewTimestampFromTime(a.time)
 	addUptimeMetric(a.mbs.PodMetricsBuilder, metadata.PodUptimeMetrics.Uptime, s.StartTime, currentTime)
-	addCPUMetrics(a.mbs.PodMetricsBuilder, metadata.PodCPUMetrics, s.CPU, currentTime, a.metadata.podResources[s.PodRef.UID])
+	addCPUMetrics(a.mbs.PodMetricsBuilder, metadata.PodCPUMetrics, s.CPU, currentTime, a.metadata.podResources[s.PodRef.UID], 0)
 	addMemoryMetrics(a.mbs.PodMetricsBuilder, metadata.PodMemoryMetrics, s.Memory, currentTime, a.metadata.podResources[s.PodRef.UID])
 	addFilesystemMetrics(a.mbs.PodMetricsBuilder, metadata.PodFilesystemMetrics, s.EphemeralStorage, currentTime)
 	addNetworkMetrics(a.mbs.PodMetricsBuilder, metadata.PodNetworkMetrics, s.Network, currentTime)
@@ -110,7 +110,7 @@ func (a *metricDataAccumulator) containerStats(sPod stats.PodStats, s stats.Cont
 	currentTime := pcommon.NewTimestampFromTime(a.time)
 	resourceKey := sPod.PodRef.UID + s.Name
 	addUptimeMetric(a.mbs.ContainerMetricsBuilder, metadata.ContainerUptimeMetrics.Uptime, s.StartTime, currentTime)
-	addCPUMetrics(a.mbs.ContainerMetricsBuilder, metadata.ContainerCPUMetrics, s.CPU, currentTime, a.metadata.containerResources[resourceKey])
+	addCPUMetrics(a.mbs.ContainerMetricsBuilder, metadata.ContainerCPUMetrics, s.CPU, currentTime, a.metadata.containerResources[resourceKey], a.metadata.cpuNodeLimit)
 	addMemoryMetrics(a.mbs.ContainerMetricsBuilder, metadata.ContainerMemoryMetrics, s.Memory, currentTime, a.metadata.containerResources[resourceKey])
 	addFilesystemMetrics(a.mbs.ContainerMetricsBuilder, metadata.ContainerFilesystemMetrics, s.Rootfs, currentTime)
 

receiver/kubeletstatsreceiver/internal/kubelet/accumulator_test.go

@@ -53,7 +53,7 @@ func TestMetadataErrorCases(t *testing.T) {
 						},
 					},
 				},
-			}, nil),
+			}, NodeLimits{}, nil),
 			testScenario: func(acc metricDataAccumulator) {
 				now := metav1.Now()
 				podStats := stats.PodStats{
@@ -79,7 +79,7 @@ func TestMetadataErrorCases(t *testing.T) {
 			metricGroupsToCollect: map[MetricGroup]bool{
 				VolumeMetricGroup: true,
 			},
-			metadata: NewMetadata([]MetadataLabel{MetadataLabelVolumeType}, nil, nil),
+			metadata: NewMetadata([]MetadataLabel{MetadataLabelVolumeType}, nil, NodeLimits{}, nil),
 			testScenario: func(acc metricDataAccumulator) {
 				podStats := stats.PodStats{
 					PodRef: stats.PodReference{
@@ -121,7 +121,7 @@ func TestMetadataErrorCases(t *testing.T) {
 						},
 					},
 				},
-			}, nil),
+			}, NodeLimits{}, nil),
 			testScenario: func(acc metricDataAccumulator) {
 				podStats := stats.PodStats{
 					PodRef: stats.PodReference{
@@ -165,7 +165,7 @@ func TestMetadataErrorCases(t *testing.T) {
 						},
 					},
 				},
-			}, nil),
+			}, NodeLimits{}, nil),
 			detailedPVCLabelsSetterOverride: func(*metadata.ResourceBuilder, string, string, string) error {
 				// Mock failure cases.
 				return errors.New("")

receiver/kubeletstatsreceiver/internal/kubelet/cpu.go

@@ -10,22 +10,37 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver/internal/metadata"
 )
 
-func addCPUMetrics(mb *metadata.MetricsBuilder, cpuMetrics metadata.CPUMetrics, s *stats.CPUStats, currentTime pcommon.Timestamp, r resources) {
+func addCPUMetrics(
+	mb *metadata.MetricsBuilder,
+	cpuMetrics metadata.CPUMetrics,
+	s *stats.CPUStats,
+	currentTime pcommon.Timestamp,
+	r resources,
+	nodeCPULimit float64) {
 	if s == nil {
 		return
 	}
-	addCPUUsageMetric(mb, cpuMetrics, s, currentTime, r)
+	addCPUUsageMetric(mb, cpuMetrics, s, currentTime, r, nodeCPULimit)
 	addCPUTimeMetric(mb, cpuMetrics.Time, s, currentTime)
 }
 
-func addCPUUsageMetric(mb *metadata.MetricsBuilder, cpuMetrics metadata.CPUMetrics, s *stats.CPUStats, currentTime pcommon.Timestamp, r resources) {
+func addCPUUsageMetric(
+	mb *metadata.MetricsBuilder,
+	cpuMetrics metadata.CPUMetrics,
+	s *stats.CPUStats,
+	currentTime pcommon.Timestamp,
+	r resources,
+	nodeCPULimit float64) {
 	if s.UsageNanoCores == nil {
 		return
 	}
 	value := float64(*s.UsageNanoCores) / 1_000_000_000
 	cpuMetrics.Utilization(mb, currentTime, value)
 	cpuMetrics.Usage(mb, currentTime, value)
 
+	if nodeCPULimit > 0 {
+		cpuMetrics.NodeUtilization(mb, currentTime, value/nodeCPULimit)
+	}
 	if r.cpuLimit > 0 {
 		cpuMetrics.LimitUtilization(mb, currentTime, value/r.cpuLimit)
 	}

receiver/kubeletstatsreceiver/internal/kubelet/metadata.go

@@ -52,6 +52,7 @@ type Metadata struct {
 	DetailedPVCResourceSetter func(rb *metadata.ResourceBuilder, volCacheID, volumeClaim, namespace string) error
 	podResources              map[string]resources
 	containerResources        map[string]resources
+	cpuNodeLimit              float64
 }
 
 type resources struct {
@@ -61,6 +62,11 @@ type resources struct {
 	memoryLimit   int64
 }
 
+type NodeLimits struct {
+	Name              string
+	CPUNanoCoresLimit float64
+}
+
 func getContainerResources(r *v1.ResourceRequirements) resources {
 	if r == nil {
 		return resources{}
@@ -74,14 +80,15 @@ func getContainerResources(r *v1.ResourceRequirements) resources {
 	}
 }
 
-func NewMetadata(labels []MetadataLabel, podsMetadata *v1.PodList,
+func NewMetadata(labels []MetadataLabel, podsMetadata *v1.PodList, nodeResourceLimits NodeLimits,
 	detailedPVCResourceSetter func(rb *metadata.ResourceBuilder, volCacheID, volumeClaim, namespace string) error) Metadata {
 	m := Metadata{
 		Labels:                    getLabelsMap(labels),
 		PodsMetadata:              podsMetadata,
 		DetailedPVCResourceSetter: detailedPVCResourceSetter,
 		podResources:              make(map[string]resources),
 		containerResources:        make(map[string]resources),
+		cpuNodeLimit:              nodeResourceLimits.CPUNanoCoresLimit,
 	}
 
 	if podsMetadata != nil {