[kubeletstats] Calculate cpu usage on the fly behind feature gate

Signed-off-by: ChrsMark <chrismarkou92@gmail.com>

Author: ChrsMark

.chloggen/kubeletstats-cpu-usage-scrape-based.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. receiver/filelog)
+component: receiver/kubelet_stats
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add the `receiver.kubeletstats.cpuUsageScrapeBased` feature gate. When enabled, `container.cpu.usage`, `k8s.pod.cpu.usage` and `k8s.node.cpu.usage` (and the cpu utilization metrics derived from them) are calculated as the rate of the corresponding `*.cpu.time` counter between consecutive scrapes, instead of being read directly from the kubelet's `UsageNanoCores` value.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [49477]
+
+# (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/documentation.md

@@ -580,3 +580,13 @@ The number of used bytes in the pod volume.
 | k8s.volume.name | The name of the Volume | Any Str | true | - | - |
 | k8s.volume.type | The type of the Volume | Any Str | true | - | - |
 | partition | [DEPRECATED] The partition in the Volume | Any Str | false | - | - |
+
+## Feature Gates
+
+This component has the following feature gates:
+
+| Feature Gate | Stage | Description | From Version | To Version | Reference |
+| ------------ | ----- | ----------- | ------------ | ---------- | --------- |
+| `receiver.kubeletstats.cpuUsageScrapeBased` | alpha | When enabled, container.cpu.usage, k8s.pod.cpu.usage and k8s.node.cpu.usage (and the cpu utilization metrics derived from them) are calculated by the receiver as the rate of the corresponding *.cpu.time counter between consecutive scrapes, instead of being read directly from the kubelet's UsageNanoCores value. | v0.156.0 | N/A | [Link](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/49477) |
+
+For more information about feature gates, see the [Feature Gates](https://github.com/open-telemetry/opentelemetry-collector/blob/main/featuregate/README.md) documentation.

receiver/kubeletstatsreceiver/go.mod

@@ -5,6 +5,7 @@ go 1.25.0
 require (
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.156.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig v0.156.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/kubelet v0.156.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.156.0
@@ -20,6 +21,7 @@ require (
 	go.opentelemetry.io/collector/confmap/xconfmap v0.156.0
 	go.opentelemetry.io/collector/consumer v1.62.0
 	go.opentelemetry.io/collector/consumer/consumertest v0.156.0
+	go.opentelemetry.io/collector/featuregate v1.62.0
 	go.opentelemetry.io/collector/filter v0.156.0
 	go.opentelemetry.io/collector/pdata v1.62.0
 	go.opentelemetry.io/collector/pipeline v1.62.0
@@ -88,7 +90,6 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.156.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.156.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.1 // indirect
@@ -113,7 +114,6 @@ require (
 	go.opentelemetry.io/collector/consumer/xconsumer v0.156.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionauth v1.62.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionmiddleware v0.156.0 // indirect
-	go.opentelemetry.io/collector/featuregate v1.62.0 // indirect
 	go.opentelemetry.io/collector/internal/componentalias v0.156.0 // indirect
 	go.opentelemetry.io/collector/internal/sharedcomponent v0.156.0 // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.156.0 // indirect

receiver/kubeletstatsreceiver/internal/kubelet/accumulator.go

@@ -41,6 +41,7 @@ type metricDataAccumulator struct {
 	allNetworkInterfaces  map[MetricGroup]bool
 	time                  time.Time
 	mbs                   *metadata.MetricsBuilders
+	cpuUsageCalculator    *CPUUsageCalculator
 }
 
 func addUptimeMetric(mb *metadata.MetricsBuilder, uptimeMetric metadata.RecordIntDataPointFunc, startTime v1.Time, currentTime pcommon.Timestamp) {
@@ -57,7 +58,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{}, 0)
+	addCPUMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeCPUMetrics, s.CPU, currentTime, resources{}, 0, a.cpuUsageCalculator, "node/"+s.NodeName)
 	addMemoryMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeMemoryMetrics, s.Memory, currentTime, resources{}, 0)
 	addFilesystemMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeFilesystemMetrics, s.Fs, currentTime)
 	addNetworkMetrics(a.mbs.NodeMetricsBuilder, metadata.NodeNetworkMetrics, s.Network, currentTime, a.allNetworkInterfaces[NodeMetricGroup])
@@ -75,7 +76,7 @@ func (a *metricDataAccumulator) nodeStats(s stats.NodeStats) {
 		rb.SetK8sNodeName(s.NodeName)
 		rb.SetK8sNodeSystemContainerName(cs.Name)
 
-		addCPUMetrics(a.mbs.NodeMetricsBuilder, metadata.SystemContainerCPUMetrics, cs.CPU, currentTime, resources{}, 0)
+		addCPUMetrics(a.mbs.NodeMetricsBuilder, metadata.SystemContainerCPUMetrics, cs.CPU, currentTime, resources{}, 0, a.cpuUsageCalculator, "node-system-container/"+s.NodeName+"/"+cs.Name)
 		addMemoryMetrics(a.mbs.NodeMetricsBuilder, metadata.SystemContainerMemoryMetrics, cs.Memory, currentTime, resources{}, 0)
 
 		a.m = append(a.m, a.mbs.NodeMetricsBuilder.Emit(
@@ -92,7 +93,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], a.metadata.nodeInfo.CPUCapacity)
+	addCPUMetrics(a.mbs.PodMetricsBuilder, metadata.PodCPUMetrics, s.CPU, currentTime, a.metadata.podResources[s.PodRef.UID], a.metadata.nodeInfo.CPUCapacity, a.cpuUsageCalculator, "pod/"+s.PodRef.UID)
 	addMemoryMetrics(a.mbs.PodMetricsBuilder, metadata.PodMemoryMetrics, s.Memory, currentTime, a.metadata.podResources[s.PodRef.UID], a.metadata.nodeInfo.MemoryCapacity)
 	addFilesystemMetrics(a.mbs.PodMetricsBuilder, metadata.PodFilesystemMetrics, s.EphemeralStorage, currentTime)
 	addNetworkMetrics(a.mbs.PodMetricsBuilder, metadata.PodNetworkMetrics, s.Network, currentTime, a.allNetworkInterfaces[PodMetricGroup])
@@ -126,7 +127,7 @@ func (a *metricDataAccumulator) containerStats(sPod *stats.PodStats, s *stats.Co
 	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], a.metadata.nodeInfo.CPUCapacity)
+	addCPUMetrics(a.mbs.ContainerMetricsBuilder, metadata.ContainerCPUMetrics, s.CPU, currentTime, a.metadata.containerResources[resourceKey], a.metadata.nodeInfo.CPUCapacity, a.cpuUsageCalculator, "container/"+resourceKey)
 	addMemoryMetrics(a.mbs.ContainerMetricsBuilder, metadata.ContainerMemoryMetrics, s.Memory, currentTime, a.metadata.containerResources[resourceKey], a.metadata.nodeInfo.MemoryCapacity)
 	addFilesystemMetrics(a.mbs.ContainerMetricsBuilder, metadata.ContainerFilesystemMetrics, s.Rootfs, currentTime)
 

receiver/kubeletstatsreceiver/internal/kubelet/cpu.go

@@ -17,18 +17,42 @@ func addCPUMetrics(
 	currentTime pcommon.Timestamp,
 	r resources,
 	nodeCPULimit float64,
+	calc *CPUUsageCalculator,
+	cpuUsageKey string,
 ) {
 	if s == nil {
 		return
 	}
-	if s.UsageNanoCores != nil {
-		usageCores := float64(*s.UsageNanoCores) / 1_000_000_000
+
+	var usageCores float64
+	var ok bool
+	if metadata.ReceiverKubeletstatsCPUUsageScrapeBasedFeatureGate.IsEnabled() {
+		usageCores, ok = calculateScrapeBasedCPUUsage(s, calc, cpuUsageKey)
+	} else if s.UsageNanoCores != nil {
+		usageCores, ok = float64(*s.UsageNanoCores)/1_000_000_000, true
+	}
+	if ok {
 		cpuMetrics.Usage(mb, currentTime, usageCores)
 		addCPUUtilizationMetrics(mb, cpuMetrics, usageCores, currentTime, r, nodeCPULimit)
 	}
+
 	addCPUTimeMetric(mb, cpuMetrics.Time, s, currentTime)
 }
 
+// calculateScrapeBasedCPUUsage calculates CPU usage in cores as the rate of change of
+// the cumulative CPU time (UsageCoreNanoSeconds) between this scrape and the previous
+// one, instead of using the value reported directly by the kubelet (UsageNanoCores).
+// This is gated behind the kubeletstats.cpuUsageScrapeBased feature gate. It returns
+// ok=false when usage cannot be calculated this scrape (see CPUUsageCalculator).
+func calculateScrapeBasedCPUUsage(s *stats.CPUStats, calc *CPUUsageCalculator, cpuUsageKey string) (usageCores float64, ok bool) {
+	if s.UsageCoreNanoSeconds == nil || s.Time.IsZero() {
+		return 0, false
+	}
+	cpuTimeSeconds := float64(*s.UsageCoreNanoSeconds) / 1_000_000_000
+	sampleTime := pcommon.NewTimestampFromTime(s.Time.Time)
+	return calc.calculateUsage(cpuUsageKey, cpuTimeSeconds, sampleTime)
+}
+
 func addCPUUtilizationMetrics(
 	mb *metadata.MetricsBuilder,
 	cpuMetrics metadata.CPUMetrics,

receiver/kubeletstatsreceiver/internal/kubelet/cpu_test.go

@@ -0,0 +1,93 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kubelet
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/receiver/receivertest"
+	"go.uber.org/zap"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	stats "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver/internal/metadata"
+)
+
+// nodeSummary builds a minimal stats.Summary for a single node with the given cumulative
+// CPU time (in nanocore-seconds) sampled at the given time.
+func nodeSummary(usageCoreNanoSeconds uint64, sampleTime time.Time) *stats.Summary {
+	return &stats.Summary{
+		Node: stats.NodeStats{
+			NodeName:  "worker-42",
+			StartTime: v1.Time{Time: sampleTime.Add(-time.Hour)},
+			CPU: &stats.CPUStats{
+				Time:                 v1.Time{Time: sampleTime},
+				UsageCoreNanoSeconds: &usageCoreNanoSeconds,
+				// UsageNanoCores is deliberately left unset: with the feature gate
+				// enabled it must not be consulted at all.
+			},
+		},
+	}
+}
+
+func TestScrapeBasedCPUUsage(t *testing.T) {
+	defer testutil.SetFeatureGateForTest(t, metadata.ReceiverKubeletstatsCPUUsageScrapeBasedFeatureGate, true)()
+
+	mbs := &metadata.MetricsBuilders{
+		NodeMetricsBuilder: metadata.NewMetricsBuilder(metadata.NewDefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
+	}
+	nodeGroup := map[MetricGroup]bool{NodeMetricGroup: true}
+	calc := NewCPUUsageCalculator()
+
+	base := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	// First scrape: no previous sample, so no usage should be recorded, but cpu.time
+	// (the cumulative counter) is always recorded regardless of the gate.
+	firstMetrics := indexedFakeMetrics(MetricsData(zap.NewNop(), nodeSummary(10_000_000_000, base), Metadata{}, nodeGroup, nil, mbs, calc))
+	_, found := firstMetrics["k8s.node.cpu.usage"]
+	require.False(t, found)
+	requireContains(t, firstMetrics, "k8s.node.cpu.time")
+	require.InDelta(t, 10.0, firstMetrics["k8s.node.cpu.time"][0].Sum().DataPoints().At(0).DoubleValue(), 1e-9)
+
+	// Second scrape, 2s later: cumulative CPU time advanced by 4 CPU-seconds ->
+	// (14-10)/2 = 2 cores of usage.
+	secondMetrics := indexedFakeMetrics(MetricsData(zap.NewNop(), nodeSummary(14_000_000_000, base.Add(2*time.Second)), Metadata{}, nodeGroup, nil, mbs, calc))
+	usageMetrics, found := secondMetrics["k8s.node.cpu.usage"]
+	require.True(t, found)
+	require.InDelta(t, 2.0, usageMetrics[0].Gauge().DataPoints().At(0).DoubleValue(), 1e-9)
+}
+
+func TestScrapeBasedCPUUsage_DisabledGateUsesUsageNanoCores(t *testing.T) {
+	// Gate defaults to disabled; explicitly assert the default (legacy) behavior is
+	// unaffected by the new code path.
+	require.False(t, metadata.ReceiverKubeletstatsCPUUsageScrapeBasedFeatureGate.IsEnabled())
+
+	mbs := &metadata.MetricsBuilders{
+		NodeMetricsBuilder: metadata.NewMetricsBuilder(metadata.NewDefaultMetricsBuilderConfig(), receivertest.NewNopSettings(metadata.Type)),
+	}
+	nodeGroup := map[MetricGroup]bool{NodeMetricGroup: true}
+	calc := NewCPUUsageCalculator()
+
+	usageNanoCores := uint64(1_500_000_000) // 1.5 cores
+	usageCoreNanoSeconds := uint64(10_000_000_000)
+	summary := &stats.Summary{
+		Node: stats.NodeStats{
+			NodeName:  "worker-42",
+			StartTime: v1.Time{Time: time.Now().Add(-time.Hour)},
+			CPU: &stats.CPUStats{
+				Time:                 v1.Time{Time: time.Now()},
+				UsageNanoCores:       &usageNanoCores,
+				UsageCoreNanoSeconds: &usageCoreNanoSeconds,
+			},
+		},
+	}
+
+	metrics := indexedFakeMetrics(MetricsData(zap.NewNop(), summary, Metadata{}, nodeGroup, nil, mbs, calc))
+	usageMetrics, found := metrics["k8s.node.cpu.usage"]
+	require.True(t, found)
+	require.InDelta(t, 1.5, usageMetrics[0].Gauge().DataPoints().At(0).DoubleValue(), 1e-9)
+}

receiver/kubeletstatsreceiver/internal/kubelet/cpu_usage.go

@@ -0,0 +1,90 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kubelet // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver/internal/kubelet"
+
+import (
+	"time"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+)
+
+// cpuUsageSample stores the cumulative CPU time (in seconds) reported for a resource
+// at a given point in time, as observed on a previous scrape.
+type cpuUsageSample struct {
+	cpuTimeSeconds float64
+	sampleTime     pcommon.Timestamp
+}
+
+// CPUUsageCalculator calculates CPU usage (in cores) as the rate of change of a
+// cumulative CPU time counter between two consecutive scrapes:
+//
+//	usageCores = (cpuTimeEnd - cpuTimeStart) / elapsedSeconds
+//
+// This mirrors the approach used by the hostmetrics receiver to calculate CPU
+// utilization from cumulative CPU time. Samples are keyed by an arbitrary
+// per-resource key (e.g. node name, pod UID, or container UID+name) so that a single
+// calculator instance can be shared across all node/pod/container/system-container
+// resources scraped by the receiver.
+type CPUUsageCalculator struct {
+	previous map[string]cpuUsageSample
+	touched  map[string]bool
+}
+
+// NewCPUUsageCalculator creates a new, empty CPUUsageCalculator.
+func NewCPUUsageCalculator() *CPUUsageCalculator {
+	return &CPUUsageCalculator{
+		previous: make(map[string]cpuUsageSample),
+	}
+}
+
+// startScrape prepares the calculator for a new scrape. It must be called once before
+// any calls to calculateUsage for a given scrape, and paired with a following call to
+// endScrape once the scrape completes.
+func (c *CPUUsageCalculator) startScrape() {
+	c.touched = make(map[string]bool)
+}
+
+// endScrape prunes stored samples for any resource key that was not observed during
+// the scrape (e.g. because the pod/container no longer exists), to avoid unbounded
+// growth of the underlying map as resources come and go over time.
+func (c *CPUUsageCalculator) endScrape() {
+	for key := range c.previous {
+		if !c.touched[key] {
+			delete(c.previous, key)
+		}
+	}
+	c.touched = nil
+}
+
+// calculateUsage returns the CPU usage in cores for the resource identified by key,
+// computed as the rate of change of cpuTimeSeconds (a cumulative CPU-seconds counter)
+// between sampleTime and the previous sample's time for the same key.
+//
+// It returns ok=false when there is no previous sample for key (e.g. on the first
+// scrape after startup, or after the resource disappeared and reappeared), when the
+// elapsed time between samples is not strictly positive (e.g. the kubelet reported a
+// stale or duplicate sample time), or when the counter decreased (e.g. a container
+// restart reset it). In all of these cases no usage value should be recorded for this
+// scrape.
+func (c *CPUUsageCalculator) calculateUsage(key string, cpuTimeSeconds float64, sampleTime pcommon.Timestamp) (usageCores float64, ok bool) {
+	c.touched[key] = true
+
+	prev, hasPrev := c.previous[key]
+	c.previous[key] = cpuUsageSample{cpuTimeSeconds: cpuTimeSeconds, sampleTime: sampleTime}
+	if !hasPrev {
+		return 0, false
+	}
+
+	elapsedSeconds := time.Duration(sampleTime - prev.sampleTime).Seconds()
+	if elapsedSeconds <= 0 {
+		return 0, false
+	}
+
+	delta := cpuTimeSeconds - prev.cpuTimeSeconds
+	if delta < 0 {
+		return 0, false
+	}
+
+	return delta / elapsedSeconds, true
+}