From 14ec0edd10eaeca0139a00d48548513b5efccff0 Mon Sep 17 00:00:00 2001 From: Francesco Romani Date: Tue, 24 Sep 2024 09:53:56 +0200 Subject: [PATCH] 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. --- pkg/kubelet/cm/cpumanager/policy_static.go | 31 ++++++++ pkg/kubelet/metrics/metrics.go | 30 ++++++- test/e2e_node/cpu_manager_metrics_test.go | 92 ++++++++++++++++++++++ test/e2e_node/podresources_test.go | 30 ++++--- 4 files changed, 168 insertions(+), 15 deletions(-) diff --git a/pkg/kubelet/cm/cpumanager/policy_static.go b/pkg/kubelet/cm/cpumanager/policy_static.go index db8770f96f6..d0278c03f87 100644 --- a/pkg/kubelet/cm/cpumanager/policy_static.go +++ b/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 +} diff --git a/pkg/kubelet/metrics/metrics.go b/pkg/kubelet/metrics/metrics.go index 30a194a3a66..da8492ab78f 100644 --- a/pkg/kubelet/metrics/metrics.go +++ b/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) diff --git a/test/e2e_node/cpu_manager_metrics_test.go b/test/e2e_node/cpu_manager_metrics_test.go index aacde715cab..7ec1db48451 100644 --- a/test/e2e_node/cpu_manager_metrics_test.go +++ b/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) + }) }) }) diff --git a/test/e2e_node/podresources_test.go b/test/e2e_node/podresources_test.go index 494730414a3..b29eb5fba65 100644 --- a/test/e2e_node/podresources_test.go +++ b/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,23 +737,32 @@ 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) { //var err error ginkgo.By("checking the output when no pods are present")