node: metrics: add metrics about cpu pool sizes

Add metrics about the sizing of the cpu pools.
Currently the cpumanager maintains 2 cpu pools:
- shared pool: this is where all pods with non-exclusive
  cpu allocation run
- exclusive pool: this is the union of the set of exclusive
  cpus allocated to containers, if any (requires static policy in use).

By reporting the size of the pools, the users (humans or machines)
can get better insights and more feedback about how the resources
actually allocated to the workload and how the node resources are used.

Author: ffromani

pkg/kubelet/cm/cpumanager/policy_static.go

@@ -194,6 +194,7 @@ func (p *staticPolicy) Start(s state.State) error {
 		klog.ErrorS(err, "Static policy invalid state, please drain node and remove policy state file")
 		return err
 	}
+	p.initializeMetrics(s)
 	return nil
 }
 
@@ -370,8 +371,10 @@ func (p *staticPolicy) Allocate(s state.State, pod *v1.Pod, container *v1.Contai
 		klog.ErrorS(err, "Unable to allocate CPUs", "pod", klog.KObj(pod), "containerName", container.Name, "numCPUs", numCPUs)
 		return err
 	}
+
 	s.SetCPUSet(string(pod.UID), container.Name, cpuset)
 	p.updateCPUsToReuse(pod, container, cpuset)
+	p.updateMetricsOnAllocate(cpuset)
 
 	return nil
 }
@@ -397,6 +400,7 @@ func (p *staticPolicy) RemoveContainer(s state.State, podUID string, containerNa
 		// Mutate the shared pool, adding released cpus.
 		toRelease = toRelease.Difference(cpusInUse)
 		s.SetDefaultCPUSet(s.GetDefaultCPUSet().Union(toRelease))
+		p.updateMetricsOnRelease(toRelease)
 	}
 	return nil
 }
@@ -720,3 +724,30 @@ func (p *staticPolicy) getAlignedCPUs(numaAffinity bitmask.BitMask, allocatableC
 
 	return alignedCPUs
 }
+
+func (p *staticPolicy) initializeMetrics(s state.State) {
+	metrics.CPUManagerSharedPoolSizeMilliCores.Set(float64(p.GetAvailableCPUs(s).Size() * 1000))
+	metrics.CPUManagerExclusiveCPUsAllocationCount.Set(float64(countExclusiveCPUs(s)))
+}
+
+func (p *staticPolicy) updateMetricsOnAllocate(cset cpuset.CPUSet) {
+	ncpus := cset.Size()
+	metrics.CPUManagerExclusiveCPUsAllocationCount.Add(float64(ncpus))
+	metrics.CPUManagerSharedPoolSizeMilliCores.Add(float64(-ncpus * 1000))
+}
+
+func (p *staticPolicy) updateMetricsOnRelease(cset cpuset.CPUSet) {
+	ncpus := cset.Size()
+	metrics.CPUManagerExclusiveCPUsAllocationCount.Add(float64(-ncpus))
+	metrics.CPUManagerSharedPoolSizeMilliCores.Add(float64(ncpus * 1000))
+}
+
+func countExclusiveCPUs(s state.State) int {
+	exclusiveCPUs := 0
+	for _, cpuAssign := range s.GetCPUAssignments() {
+		for _, cset := range cpuAssign {
+			exclusiveCPUs += cset.Size()
+		}
+	}
+	return exclusiveCPUs
+}

pkg/kubelet/metrics/metrics.go

@@ -108,8 +108,10 @@ const (
 	ManagedEphemeralContainersKey = "managed_ephemeral_containers"
 
 	// Metrics to track the CPU manager behavior
-	CPUManagerPinningRequestsTotalKey = "cpu_manager_pinning_requests_total"
-	CPUManagerPinningErrorsTotalKey   = "cpu_manager_pinning_errors_total"
+	CPUManagerPinningRequestsTotalKey         = "cpu_manager_pinning_requests_total"
+	CPUManagerPinningErrorsTotalKey           = "cpu_manager_pinning_errors_total"
+	CPUManagerSharedPoolSizeMilliCoresKey     = "cpu_manager_shared_pool_size_millicores"
+	CPUManagerExclusiveCPUsAllocationCountKey = "cpu_manager_exclusive_cpu_allocation_count"
 
 	// Metrics to track the Memory manager behavior
 	MemoryManagerPinningRequestsTotalKey = "memory_manager_pinning_requests_total"
@@ -773,6 +775,27 @@ var (
 		},
 	)
 
+	// CPUManagerSharedPoolSizeMilliCores reports the current size of the shared CPU pool for non-guaranteed pods
+	CPUManagerSharedPoolSizeMilliCores = metrics.NewGauge(
+		&metrics.GaugeOpts{
+			Subsystem:      KubeletSubsystem,
+			Name:           CPUManagerSharedPoolSizeMilliCoresKey,
+			Help:           "The size of the shared CPU pool for non-guaranteed QoS pods, in millicores.",
+			StabilityLevel: metrics.ALPHA,
+		},
+	)
+
+	// CPUManagerExclusiveCPUsAllocationCount reports the total number of CPUs exclusively allocated to containers running on this node
+	CPUManagerExclusiveCPUsAllocationCount = metrics.NewGauge(
+		&metrics.GaugeOpts{
+			Subsystem:      KubeletSubsystem,
+			Name:           CPUManagerExclusiveCPUsAllocationCountKey,
+			Help:           "The total number of CPUs exclusively allocated to containers running on this node",
+			StabilityLevel: metrics.ALPHA,
+		},
+	)
+
+	// ContainerAlignedComputeResources reports the count of resources allocation which granted aligned resources, per alignment boundary
 	ContainerAlignedComputeResources = metrics.NewCounterVec(
 		&metrics.CounterOpts{
 			Subsystem:      KubeletSubsystem,
@@ -782,7 +805,6 @@ var (
 		},
 		[]string{ContainerAlignedComputeResourcesScopeLabelKey, ContainerAlignedComputeResourcesBoundaryLabelKey},
 	)
-
 	// MemoryManagerPinningRequestTotal tracks the number of times the pod spec required the memory manager to pin memory pages
 	MemoryManagerPinningRequestTotal = metrics.NewCounter(
 		&metrics.CounterOpts{
@@ -1006,6 +1028,8 @@ func Register(collectors ...metrics.StableCollector) {
 		legacyregistry.MustRegister(RunPodSandboxErrors)
 		legacyregistry.MustRegister(CPUManagerPinningRequestsTotal)
 		legacyregistry.MustRegister(CPUManagerPinningErrorsTotal)
+		legacyregistry.MustRegister(CPUManagerSharedPoolSizeMilliCores)
+		legacyregistry.MustRegister(CPUManagerExclusiveCPUsAllocationCount)
 		legacyregistry.MustRegister(ContainerAlignedComputeResources)
 		if utilfeature.DefaultFeatureGate.Enabled(features.MemoryManager) {
 			legacyregistry.MustRegister(MemoryManagerPinningRequestTotal)

test/e2e_node/cpu_manager_metrics_test.go

@@ -29,8 +29,11 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubeletpodresourcesv1 "k8s.io/kubelet/pkg/apis/podresources/v1"
 	kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
+	"k8s.io/kubernetes/pkg/kubelet/apis/podresources"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
+	"k8s.io/kubernetes/pkg/kubelet/util"
 	"k8s.io/kubernetes/test/e2e/feature"
 	"k8s.io/kubernetes/test/e2e/framework"
 	e2emetrics "k8s.io/kubernetes/test/e2e/framework/metrics"
@@ -182,6 +185,95 @@ var _ = SIGDescribe("CPU Manager Metrics", framework.WithSerial(), feature.CPUMa
 			ginkgo.By("Ensuring the metrics match the expectations about alignment metrics a few more times")
 			gomega.Consistently(ctx, getKubeletMetrics, 1*time.Minute, 15*time.Second).Should(matchAlignmentMetrics)
 		})
+
+		ginkgo.It("should report the default idle cpu pool size", func(ctx context.Context) {
+			ginkgo.By("Querying the podresources endpoint to get the baseline")
+			endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
+			framework.ExpectNoError(err, "LocalEndpoint() failed err: %v", err)
+
+			cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
+			framework.ExpectNoError(err, "GetV1Client() failed err: %v", err)
+			defer func() {
+				framework.ExpectNoError(conn.Close())
+			}()
+
+			ginkgo.By("Checking the pool allocatable resources from the kubelet")
+			resp, err := cli.GetAllocatableResources(ctx, &kubeletpodresourcesv1.AllocatableResourcesRequest{})
+			framework.ExpectNoError(err, "failed to get the kubelet allocatable resources")
+			allocatableCPUs, _ := demuxCPUsAndDevicesFromGetAllocatableResources(resp)
+
+			matchResourceMetrics := gstruct.MatchKeys(gstruct.IgnoreExtras, gstruct.Keys{
+				"kubelet_cpu_manager_shared_pool_size_millicores": gstruct.MatchAllElements(nodeID, gstruct.Elements{
+					"": timelessSample(int(allocatableCPUs.Size() * 1000)),
+				}),
+				"kubelet_cpu_manager_exclusive_cpu_allocation_count": gstruct.MatchAllElements(nodeID, gstruct.Elements{
+					"": timelessSample(0),
+				}),
+			})
+
+			ginkgo.By("Giving the Kubelet time to start up and produce metrics about idle pool size")
+			gomega.Eventually(ctx, getKubeletMetrics, 1*time.Minute, 10*time.Second).Should(matchResourceMetrics)
+			ginkgo.By("Ensuring the metrics match the expectations about idle pool size a few more times")
+			gomega.Consistently(ctx, getKubeletMetrics, 30*time.Second, 10*time.Second).Should(matchResourceMetrics)
+		})
+
+		ginkgo.It("should report mutating cpu pool size when handling guaranteed pods", func(ctx context.Context) {
+			ginkgo.By("Querying the podresources endpoint to get the baseline")
+			endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
+			framework.ExpectNoError(err, "LocalEndpoint() failed err: %v", err)
+
+			cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize)
+			framework.ExpectNoError(err, "GetV1Client() failed err: %v", err)
+			defer func() {
+				framework.ExpectNoError(conn.Close())
+			}()
+
+			ginkgo.By("Checking the pool allocatable resources from the kubelet")
+			resp, err := cli.GetAllocatableResources(ctx, &kubeletpodresourcesv1.AllocatableResourcesRequest{})
+			framework.ExpectNoError(err, "failed to get the kubelet allocatable resources")
+			allocatableCPUs, _ := demuxCPUsAndDevicesFromGetAllocatableResources(resp)
+
+			allocatableCPUsIdleMillis := int(allocatableCPUs.Size() * 1000)
+
+			matchResourceMetricsIdle := gstruct.MatchKeys(gstruct.IgnoreExtras, gstruct.Keys{
+				"kubelet_cpu_manager_shared_pool_size_millicores": gstruct.MatchAllElements(nodeID, gstruct.Elements{
+					"": timelessSample(allocatableCPUsIdleMillis),
+				}),
+				"kubelet_cpu_manager_exclusive_cpu_allocation_count": gstruct.MatchAllElements(nodeID, gstruct.Elements{
+					"": timelessSample(0),
+				}),
+			})
+			ginkgo.By(fmt.Sprintf("Pool allocatable resources from the kubelet: shared pool %d cpus %d millis", allocatableCPUs.Size(), allocatableCPUsIdleMillis))
+
+			ginkgo.By("Giving the Kubelet time to start up and produce metrics about idle pool size")
+			gomega.Eventually(ctx, getKubeletMetrics, 1*time.Minute, 10*time.Second).Should(matchResourceMetricsIdle)
+			ginkgo.By("Ensuring the metrics match the expectations about idle pool size a few more times")
+			gomega.Consistently(ctx, getKubeletMetrics, 30*time.Second, 10*time.Second).Should(matchResourceMetricsIdle)
+
+			ginkgo.By("Creating the test pod to consume exclusive cpus from the pool")
+			testPod = e2epod.NewPodClient(f).CreateSync(ctx, makeGuaranteedCPUExclusiveSleeperPod("smt-cpupool", smtLevel))
+
+			matchResourceMetricsBusy := gstruct.MatchKeys(gstruct.IgnoreExtras, gstruct.Keys{
+				"kubelet_cpu_manager_shared_pool_size_millicores": gstruct.MatchAllElements(nodeID, gstruct.Elements{
+					"": timelessSample(allocatableCPUsIdleMillis - (smtLevel * 1000)),
+				}),
+				"kubelet_cpu_manager_exclusive_cpu_allocation_count": gstruct.MatchAllElements(nodeID, gstruct.Elements{
+					"": timelessSample(smtLevel),
+				}),
+			})
+
+			ginkgo.By("Giving the Kubelet time to start up and produce metrics")
+			gomega.Eventually(ctx, getKubeletMetrics, 1*time.Minute, 10*time.Second).Should(matchResourceMetricsBusy)
+			ginkgo.By("Ensuring the metrics match the expectations a few more times")
+			gomega.Consistently(ctx, getKubeletMetrics, 30*time.Second, 10*time.Second).Should(matchResourceMetricsBusy)
+
+			deletePodSyncByName(ctx, f, testPod.Name)
+
+			ginkgo.By("Giving the Kubelet time to start up and produce metrics")
+			gomega.Eventually(ctx, getKubeletMetrics, 1*time.Minute, 10*time.Second).Should(matchResourceMetricsIdle)
+			ginkgo.By("Ensuring the metrics match the expectations a few more times")
+			gomega.Consistently(ctx, getKubeletMetrics, 30*time.Second, 10*time.Second).Should(matchResourceMetricsIdle)
+		})
 	})
 })
 

test/e2e_node/podresources_test.go

@@ -719,19 +719,16 @@ func podresourcesListTests(ctx context.Context, f *framework.Framework, cli kube
 }
 
 func podresourcesGetAllocatableResourcesTests(ctx context.Context, cli kubeletpodresourcesv1.PodResourcesListerClient, sd *sriovData, onlineCPUs, reservedSystemCPUs cpuset.CPUSet) {
+	ginkgo.GinkgoHelper()
+
 	ginkgo.By("checking the devices known to the kubelet")
 	resp, err := cli.GetAllocatableResources(ctx, &kubeletpodresourcesv1.AllocatableResourcesRequest{})
-	framework.ExpectNoErrorWithOffset(1, err)
-	devs := resp.GetDevices()
-	var cpus []int
-	for _, cpuid := range resp.GetCpuIds() {
-		cpus = append(cpus, int(cpuid))
-	}
-	allocatableCPUs := cpuset.New(cpus...)
+	framework.ExpectNoError(err, "cannot get allocatable CPUs from podresources")
+	allocatableCPUs, devs := demuxCPUsAndDevicesFromGetAllocatableResources(resp)
 
 	if onlineCPUs.Size() == 0 {
 		ginkgo.By("expecting no CPUs reported")
-		gomega.ExpectWithOffset(1, onlineCPUs.Size()).To(gomega.Equal(reservedSystemCPUs.Size()), "with no online CPUs, no CPUs should be reserved")
+		gomega.Expect(onlineCPUs.Size()).To(gomega.Equal(reservedSystemCPUs.Size()), "with no online CPUs, no CPUs should be reserved")
 	} else {
 		ginkgo.By(fmt.Sprintf("expecting online CPUs reported - online=%v (%d) reserved=%v (%d)", onlineCPUs, onlineCPUs.Size(), reservedSystemCPUs, reservedSystemCPUs.Size()))
 		if reservedSystemCPUs.Size() > onlineCPUs.Size() {
@@ -740,21 +737,30 @@ func podresourcesGetAllocatableResourcesTests(ctx context.Context, cli kubeletpo
 		expectedCPUs := onlineCPUs.Difference(reservedSystemCPUs)
 
 		ginkgo.By(fmt.Sprintf("expecting CPUs '%v'='%v'", allocatableCPUs, expectedCPUs))
-		gomega.ExpectWithOffset(1, allocatableCPUs.Equals(expectedCPUs)).To(gomega.BeTrueBecause("mismatch expecting CPUs"))
+		gomega.Expect(allocatableCPUs.Equals(expectedCPUs)).To(gomega.BeTrueBecause("mismatch expecting CPUs"))
 	}
 
 	if sd == nil { // no devices in the environment, so expect no devices
 		ginkgo.By("expecting no devices reported")
-		gomega.ExpectWithOffset(1, devs).To(gomega.BeEmpty(), fmt.Sprintf("got unexpected devices %#v", devs))
+		gomega.Expect(devs).To(gomega.BeEmpty(), fmt.Sprintf("got unexpected devices %#v", devs))
 		return
 	}
 
 	ginkgo.By(fmt.Sprintf("expecting some %q devices reported", sd.resourceName))
-	gomega.ExpectWithOffset(1, devs).ToNot(gomega.BeEmpty())
+	gomega.Expect(devs).ToNot(gomega.BeEmpty())
 	for _, dev := range devs {
 		gomega.Expect(dev.ResourceName).To(gomega.Equal(sd.resourceName))
-		gomega.ExpectWithOffset(1, dev.DeviceIds).ToNot(gomega.BeEmpty())
+		gomega.Expect(dev.DeviceIds).ToNot(gomega.BeEmpty())
+	}
+}
+
+func demuxCPUsAndDevicesFromGetAllocatableResources(resp *kubeletpodresourcesv1.AllocatableResourcesResponse) (cpuset.CPUSet, []*kubeletpodresourcesv1.ContainerDevices) {
+	devs := resp.GetDevices()
+	var cpus []int
+	for _, cpuid := range resp.GetCpuIds() {
+		cpus = append(cpus, int(cpuid))
 	}
+	return cpuset.New(cpus...), devs
 }
 
 func podresourcesGetTests(ctx context.Context, f *framework.Framework, cli kubeletpodresourcesv1.PodResourcesListerClient, sidecarContainersEnabled bool) {