From bc56d0e45a24b24ca9727911a891275b81bae69b Mon Sep 17 00:00:00 2001 From: Francesco Romani Date: Wed, 4 Jun 2025 16:35:03 +0200 Subject: [PATCH 1/3] podresources: list: use active pods in list The podresources API List implementation uses the internal data of the resource managers as source of truth. Looking at the implementation here: https://github.com/kubernetes/kubernetes/blob/v1.34.0-alpha.0/pkg/kubelet/apis/podresources/server_v1.go#L60 we take care of syncing the device allocation data before querying the device manager to return its pod->devices assignment. This is needed because otherwise the device manager (and all the other resource managers) would do the cleanup asynchronously, so the `List` call will return incorrect data. But we don't do this syncing neither for CPUs or for memory, so when we report these we will get stale data as the issue #132020 demonstrates. For CPU manager, we however have the reconcile loop which cleans the stale data periodically. Turns out this timing interplay was actually the reason the existing issue #119423 seemed fixed (see: https://github.com/kubernetes/kubernetes/issues/119423#issuecomment-2786891218). But it's actually timing. If in the reproducer we set the `cpuManagerReconcilePeriod` to a time very high (>= 5 minutes), then the issue still reproduces against current master branch (https://github.com/kubernetes/kubernetes/blob/v1.34.0-alpha.0/test/e2e_node/podresources_test.go#L983). Taking a step back, we can see multiple problems: 1. not syncing the resource managers internal data before to query for pod assignment (no removeStaleState calls) but most importantly 2. the List call iterate overs all the pod known to the kubelet. But the resource managers do NOT hold resources for non-running pod, so it is better, actually it's correct to iterate only over the active pods. This will also avoid issue 1 above. Furthermore, the resource managers all iterate over the active pods anyway: `List` is using all the pods known about: 1. https://github.com/kubernetes/kubernetes/blob/v1.34.0-alpha.0/pkg/kubelet/kubelet.go#L3135 goes in 2. https://github.com/kubernetes/kubernetes/blob/v1.34.0-alpha.0/pkg/kubelet/pod/pod_manager.go#L215 But all the resource managers are using the list of active pods: 1. https://github.com/kubernetes/kubernetes/blob/v1.34.0-alpha.0/pkg/kubelet/kubelet.go#L1666 goes in 2. https://github.com/kubernetes/kubernetes/blob/v1.34.0-alpha.0/pkg/kubelet/kubelet_pods.go#L198 So this change will also make the `List` view consistent with the resource managers view, which is also a promise of the API currently broken. We also need to acknowledge the the warning in the docstring of GetActivePods. Arguably, having the endpoint using a different podset wrt the resource managers with the related desync causes way more harm than good. And arguably, it's better to fix this issue in just one place instead of having the `List` use a different pod set for unclear reason. For these reasons, while important, I don't think the warning per se invalidated this change. We need to further acknowledge the `List` endpoint used the full pod list since its inception. So, we will add a Feature Gate to disable this fix and restore the old behavior. We plan to keep this Feature Gate for quite a long time (at least 4 more releases) considering how stable this change was. Should a consumer of the API being broken by this change, we have the option to restore the old behavior and to craft a more elaborate fix. The old `v1alpha1` endpoint will be not modified intentionally. Signed-off-by: Francesco Romani --- pkg/features/kube_features.go | 26 +++ pkg/kubelet/apis/podresources/server_v1.go | 13 +- .../apis/podresources/server_v1_test.go | 157 ++++++++++++++++++ .../podresources/testing/pods_provider.go | 47 ++++++ pkg/kubelet/apis/podresources/types.go | 1 + pkg/kubelet/kubelet.go | 18 +- .../reference/versioned_feature_list.yaml | 10 ++ 7 files changed, 270 insertions(+), 2 deletions(-) diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go index 9acaa6208ae..8ce128808bc 100644 --- a/pkg/features/kube_features.go +++ b/pkg/features/kube_features.go @@ -443,6 +443,27 @@ const ( // Enable POD resources API with Get method KubeletPodResourcesGet featuregate.Feature = "KubeletPodResourcesGet" + // owner: @ffromani + // Deprecated: v1.34 + // + // issue: https://github.com/kubernetes/kubernetes/issues/119423 + // Disables restricted output for the podresources API list endpoint. + // "Restricted" output only includes the pods which are actually running and thus they + // hold resources. Turns out this was originally the intended behavior, see: + // https://github.com/kubernetes/kubernetes/pull/79409#issuecomment-505975671 + // This behavior was lost over time and interaction with memory manager creates + // an unfixable bug because the endpoint returns spurious stale information the clients + // cannot filter out, because the API doesn't provide enough context. See: + // https://github.com/kubernetes/kubernetes/issues/132020 + // The endpoint has returning extra information for long time, but that information + // is also useless for the purpose of this API. Nevertheless, we are changing a long-established + // albeit buggy behavior, so users observing any regressions can use the + // KubeletPodResourcesListUseActivePods/ feature gate (default on) to restore the old behavior. + // Please file issues if you hit issues and have to use this Feature Gate. + // The Feature Gate will be locked to true in +4 releases (1.38) and then removed (1.39) + // if there are no bug reported. + KubeletPodResourcesListUseActivePods featuregate.Feature = "KubeletPodResourcesListUseActivePods" + // owner: @hoskeri // // Restores previous behavior where Kubelet fails self registration if node create returns 403 Forbidden. @@ -1275,6 +1296,11 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate {Version: version.MustParse("1.27"), Default: false, PreRelease: featuregate.Alpha}, }, + KubeletPodResourcesListUseActivePods: { + {Version: version.MustParse("1.0"), Default: false, PreRelease: featuregate.GA}, + {Version: version.MustParse("1.34"), Default: true, PreRelease: featuregate.Deprecated}, // lock to default in 1.38, remove in 1.39 + }, + KubeletRegistrationGetOnExistsOnly: { {Version: version.MustParse("1.0"), Default: true, PreRelease: featuregate.GA}, {Version: version.MustParse("1.32"), Default: false, PreRelease: featuregate.Deprecated}, diff --git a/pkg/kubelet/apis/podresources/server_v1.go b/pkg/kubelet/apis/podresources/server_v1.go index d69aba1e889..78c174f8466 100644 --- a/pkg/kubelet/apis/podresources/server_v1.go +++ b/pkg/kubelet/apis/podresources/server_v1.go @@ -22,6 +22,7 @@ import ( v1 "k8s.io/api/core/v1" utilfeature "k8s.io/apiserver/pkg/util/feature" + "k8s.io/klog/v2" podutil "k8s.io/kubernetes/pkg/api/v1/pod" kubefeatures "k8s.io/kubernetes/pkg/features" "k8s.io/kubernetes/pkg/kubelet/metrics" @@ -36,17 +37,21 @@ type v1PodResourcesServer struct { cpusProvider CPUsProvider memoryProvider MemoryProvider dynamicResourcesProvider DynamicResourcesProvider + useActivePods bool } // NewV1PodResourcesServer returns a PodResourcesListerServer which lists pods provided by the PodsProvider // with device information provided by the DevicesProvider func NewV1PodResourcesServer(providers PodResourcesProviders) podresourcesv1.PodResourcesListerServer { + useActivePods := utilfeature.DefaultFeatureGate.Enabled(kubefeatures.KubeletPodResourcesListUseActivePods) + klog.InfoS("podresources", "method", "list", "useActivePods", useActivePods) return &v1PodResourcesServer{ podsProvider: providers.Pods, devicesProvider: providers.Devices, cpusProvider: providers.Cpus, memoryProvider: providers.Memory, dynamicResourcesProvider: providers.DynamicResources, + useActivePods: useActivePods, } } @@ -55,7 +60,13 @@ func (p *v1PodResourcesServer) List(ctx context.Context, req *podresourcesv1.Lis metrics.PodResourcesEndpointRequestsTotalCount.WithLabelValues("v1").Inc() metrics.PodResourcesEndpointRequestsListCount.WithLabelValues("v1").Inc() - pods := p.podsProvider.GetPods() + var pods []*v1.Pod + if p.useActivePods { + pods = p.podsProvider.GetActivePods() + } else { + pods = p.podsProvider.GetPods() + } + podResources := make([]*podresourcesv1.PodResources, len(pods)) p.devicesProvider.UpdateAllocatedDevices() diff --git a/pkg/kubelet/apis/podresources/server_v1_test.go b/pkg/kubelet/apis/podresources/server_v1_test.go index aa1dffd4d03..9ed00a1fbfa 100644 --- a/pkg/kubelet/apis/podresources/server_v1_test.go +++ b/pkg/kubelet/apis/podresources/server_v1_test.go @@ -19,10 +19,12 @@ package podresources import ( "context" "fmt" + "sort" "testing" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" + "github.com/stretchr/testify/mock" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -221,6 +223,7 @@ func TestListPodResourcesV1(t *testing.T) { mockDynamicResourcesProvider := podresourcetest.NewMockDynamicResourcesProvider(t) mockPodsProvider.EXPECT().GetPods().Return(tc.pods).Maybe() + mockPodsProvider.EXPECT().GetActivePods().Return(tc.pods).Maybe() mockDevicesProvider.EXPECT().GetDevices(string(podUID), containerName).Return(tc.devices).Maybe() mockCPUsProvider.EXPECT().GetCPUs(string(podUID), containerName).Return(tc.cpus).Maybe() mockMemoryProvider.EXPECT().GetMemory(string(podUID), containerName).Return(tc.memory).Maybe() @@ -249,6 +252,159 @@ func TestListPodResourcesV1(t *testing.T) { } } +func makePod(idx int) *v1.Pod { + podNamespace := "pod-namespace" + podName := fmt.Sprintf("pod-name-%d", idx) + podUID := types.UID(fmt.Sprintf("pod-uid-%d", idx)) + containerName := fmt.Sprintf("container-name-%d", idx) + containers := []v1.Container{ + { + Name: containerName, + }, + } + return &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: podName, + Namespace: podNamespace, + UID: podUID, + }, + Spec: v1.PodSpec{ + Containers: containers, + }, + } +} + +func collectNamespacedNamesFromPods(pods []*v1.Pod) []string { + ret := make([]string, 0, len(pods)) + for _, pod := range pods { + ret = append(ret, pod.Namespace+"/"+pod.Name) + } + sort.Strings(ret) + return ret +} + +func collectNamespacedNamesFromPodResources(prs []*podresourcesapi.PodResources) []string { + ret := make([]string, 0, len(prs)) + for _, pr := range prs { + ret = append(ret, pr.Namespace+"/"+pr.Name) + } + sort.Strings(ret) + return ret +} + +func TestListPodResourcesUsesOnlyActivePodsV1(t *testing.T) { + numaID := int64(1) + + // we abuse the fact that we don't care about the assignments, + // so we reuse the same for all pods which is actually wrong. + devs := []*podresourcesapi.ContainerDevices{ + { + ResourceName: "resource", + DeviceIds: []string{"dev0"}, + Topology: &podresourcesapi.TopologyInfo{Nodes: []*podresourcesapi.NUMANode{{ID: numaID}}}, + }, + } + + cpus := []int64{1, 9} + + mems := []*podresourcesapi.ContainerMemory{ + { + MemoryType: "memory", + Size_: 1073741824, + Topology: &podresourcesapi.TopologyInfo{Nodes: []*podresourcesapi.NUMANode{{ID: numaID}}}, + }, + { + MemoryType: "hugepages-1Gi", + Size_: 1073741824, + Topology: &podresourcesapi.TopologyInfo{Nodes: []*podresourcesapi.NUMANode{{ID: numaID}}}, + }, + } + + for _, tc := range []struct { + desc string + pods []*v1.Pod + activePods []*v1.Pod + }{ + { + desc: "no pods", + pods: []*v1.Pod{}, + activePods: []*v1.Pod{}, + }, + { + desc: "no differences", + pods: []*v1.Pod{ + makePod(1), + makePod(2), + makePod(3), + makePod(4), + makePod(5), + }, + activePods: []*v1.Pod{ + makePod(1), + makePod(2), + makePod(3), + makePod(4), + makePod(5), + }, + }, + { + desc: "some terminated pods", + pods: []*v1.Pod{ + makePod(1), + makePod(2), + makePod(3), + makePod(4), + makePod(5), + makePod(6), + makePod(7), + }, + activePods: []*v1.Pod{ + makePod(1), + makePod(3), + makePod(4), + makePod(5), + makePod(6), + }, + }, + } { + t.Run(tc.desc, func(t *testing.T) { + mockDevicesProvider := podresourcetest.NewMockDevicesProvider(t) + mockPodsProvider := podresourcetest.NewMockPodsProvider(t) + mockCPUsProvider := podresourcetest.NewMockCPUsProvider(t) + mockMemoryProvider := podresourcetest.NewMockMemoryProvider(t) + mockDynamicResourcesProvider := podresourcetest.NewMockDynamicResourcesProvider(t) + + mockPodsProvider.EXPECT().GetPods().Return(tc.pods).Maybe() + mockPodsProvider.EXPECT().GetActivePods().Return(tc.activePods).Maybe() + mockDevicesProvider.EXPECT().GetDevices(mock.Anything, mock.Anything).Return(devs).Maybe() + mockCPUsProvider.EXPECT().GetCPUs(mock.Anything, mock.Anything).Return(cpus).Maybe() + mockMemoryProvider.EXPECT().GetMemory(mock.Anything, mock.Anything).Return(mems).Maybe() + mockDevicesProvider.EXPECT().UpdateAllocatedDevices().Return().Maybe() + mockCPUsProvider.EXPECT().GetAllocatableCPUs().Return([]int64{}).Maybe() + mockDevicesProvider.EXPECT().GetAllocatableDevices().Return([]*podresourcesapi.ContainerDevices{}).Maybe() + mockMemoryProvider.EXPECT().GetAllocatableMemory().Return([]*podresourcesapi.ContainerMemory{}).Maybe() + + providers := PodResourcesProviders{ + Pods: mockPodsProvider, + Devices: mockDevicesProvider, + Cpus: mockCPUsProvider, + Memory: mockMemoryProvider, + DynamicResources: mockDynamicResourcesProvider, + } + server := NewV1PodResourcesServer(providers) + resp, err := server.List(context.TODO(), &podresourcesapi.ListPodResourcesRequest{}) + if err != nil { + t.Errorf("want err = %v, got %q", nil, err) + } + expectedNames := collectNamespacedNamesFromPods(tc.activePods) + gotNames := collectNamespacedNamesFromPodResources(resp.GetPodResources()) + if diff := cmp.Diff(expectedNames, gotNames, cmpopts.EquateEmpty()); diff != "" { + t.Fatal(diff) + } + }) + } +} + func TestListPodResourcesWithInitContainersV1(t *testing.T) { featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, pkgfeatures.KubeletPodResourcesDynamicResources, true) @@ -423,6 +579,7 @@ func TestListPodResourcesWithInitContainersV1(t *testing.T) { mockDynamicResourcesProvider := podresourcetest.NewMockDynamicResourcesProvider(t) mockPodsProvider.EXPECT().GetPods().Return(tc.pods).Maybe() + mockPodsProvider.EXPECT().GetActivePods().Return(tc.pods).Maybe() tc.mockFunc(tc.pods, mockDevicesProvider, mockCPUsProvider, mockMemoryProvider, mockDynamicResourcesProvider) providers := PodResourcesProviders{ diff --git a/pkg/kubelet/apis/podresources/testing/pods_provider.go b/pkg/kubelet/apis/podresources/testing/pods_provider.go index 9cd9c3eef7f..d917e0c7cad 100644 --- a/pkg/kubelet/apis/podresources/testing/pods_provider.go +++ b/pkg/kubelet/apis/podresources/testing/pods_provider.go @@ -37,6 +37,53 @@ func (_m *MockPodsProvider) EXPECT() *MockPodsProvider_Expecter { return &MockPodsProvider_Expecter{mock: &_m.Mock} } +// GetActivePods provides a mock function with no fields +func (_m *MockPodsProvider) GetActivePods() []*v1.Pod { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for GetActivePods") + } + + var r0 []*v1.Pod + if rf, ok := ret.Get(0).(func() []*v1.Pod); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*v1.Pod) + } + } + + return r0 +} + +// MockPodsProvider_GetActivePods_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetActivePods' +type MockPodsProvider_GetActivePods_Call struct { + *mock.Call +} + +// GetActivePods is a helper method to define mock.On call +func (_e *MockPodsProvider_Expecter) GetActivePods() *MockPodsProvider_GetActivePods_Call { + return &MockPodsProvider_GetActivePods_Call{Call: _e.mock.On("GetActivePods")} +} + +func (_c *MockPodsProvider_GetActivePods_Call) Run(run func()) *MockPodsProvider_GetActivePods_Call { + _c.Call.Run(func(args mock.Arguments) { + run() + }) + return _c +} + +func (_c *MockPodsProvider_GetActivePods_Call) Return(_a0 []*v1.Pod) *MockPodsProvider_GetActivePods_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *MockPodsProvider_GetActivePods_Call) RunAndReturn(run func() []*v1.Pod) *MockPodsProvider_GetActivePods_Call { + _c.Call.Return(run) + return _c +} + // GetPodByName provides a mock function with given fields: namespace, name func (_m *MockPodsProvider) GetPodByName(namespace string, name string) (*v1.Pod, bool) { ret := _m.Called(namespace, name) diff --git a/pkg/kubelet/apis/podresources/types.go b/pkg/kubelet/apis/podresources/types.go index ee1269d969b..66d7c6cfda2 100644 --- a/pkg/kubelet/apis/podresources/types.go +++ b/pkg/kubelet/apis/podresources/types.go @@ -34,6 +34,7 @@ type DevicesProvider interface { // PodsProvider knows how to provide the pods admitted by the node type PodsProvider interface { + GetActivePods() []*v1.Pod GetPods() []*v1.Pod GetPodByName(namespace, name string) (*v1.Pod, bool) } diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index 0aa114b1770..7e334310ea3 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -2868,6 +2868,22 @@ func (kl *Kubelet) ListenAndServeReadOnly(address net.IP, port uint, tp trace.Tr server.ListenAndServeKubeletReadOnlyServer(kl, kl.resourceAnalyzer, kl.containerManager.GetHealthCheckers(), kl.flagz, address, port, tp) } +type kubeletPodsProvider struct { + kl *Kubelet +} + +func (pp *kubeletPodsProvider) GetActivePods() []*v1.Pod { + return pp.kl.GetActivePods() +} + +func (pp *kubeletPodsProvider) GetPods() []*v1.Pod { + return pp.kl.podManager.GetPods() +} + +func (pp *kubeletPodsProvider) GetPodByName(namespace, name string) (*v1.Pod, bool) { + return pp.kl.podManager.GetPodByName(namespace, name) +} + // ListenAndServePodResources runs the kubelet podresources grpc service func (kl *Kubelet) ListenAndServePodResources() { endpoint, err := util.LocalEndpoint(kl.getPodResourcesDir(), podresources.Socket) @@ -2877,7 +2893,7 @@ func (kl *Kubelet) ListenAndServePodResources() { } providers := podresources.PodResourcesProviders{ - Pods: kl.podManager, + Pods: &kubeletPodsProvider{kl: kl}, Devices: kl.containerManager, Cpus: kl.containerManager, Memory: kl.containerManager, diff --git a/test/compatibility_lifecycle/reference/versioned_feature_list.yaml b/test/compatibility_lifecycle/reference/versioned_feature_list.yaml index f6c5a3575f0..2c5b9a9e77f 100644 --- a/test/compatibility_lifecycle/reference/versioned_feature_list.yaml +++ b/test/compatibility_lifecycle/reference/versioned_feature_list.yaml @@ -727,6 +727,16 @@ lockToDefault: false preRelease: Alpha version: "1.27" +- name: KubeletPodResourcesListUseActivePods + versionedSpecs: + - default: false + lockToDefault: false + preRelease: GA + version: "1.0" + - default: true + lockToDefault: false + preRelease: Deprecated + version: "1.34" - name: KubeletPSI versionedSpecs: - default: false From 380ed8d9b3ad984138b9739de293f1ef08bd2cbf Mon Sep 17 00:00:00 2001 From: Francesco Romani Date: Sat, 31 May 2025 12:53:55 +0200 Subject: [PATCH 2/3] e2e: node: memory manager: build everywhere, run only on linux Since the KEP 4885 (https://github.com/kubernetes/enhancements/blob/master/keps/sig-windows/4885-windows-cpu-and-memory-affinity/README.md) memory manager is supported also on windows. Plus, we want to add podresources e2e tests which configure the memory manager. Both these facts suggest it's useful to build the e2e memory manager tests on all OSes, not just on linux; However, since we are not sure we are ready to run these tests everywhere, we tag them LinuxOnly to keep preserve most of the old behavior. Signed-off-by: Francesco Romani --- test/e2e_node/memory_manager_test.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/test/e2e_node/memory_manager_test.go b/test/e2e_node/memory_manager_test.go index 0f09ce48110..b7b8950e9e0 100644 --- a/test/e2e_node/memory_manager_test.go +++ b/test/e2e_node/memory_manager_test.go @@ -1,6 +1,3 @@ -//go:build linux -// +build linux - /* Copyright 2017 The Kubernetes Authors. @@ -242,7 +239,7 @@ func getAllNUMANodes() []int { } // Serial because the test updates kubelet configuration. -var _ = SIGDescribe("Memory Manager", framework.WithDisruptive(), framework.WithSerial(), feature.MemoryManager, func() { +var _ = SIGDescribe("Memory Manager", "[LinuxOnly]", framework.WithDisruptive(), framework.WithSerial(), feature.MemoryManager, func() { // TODO: add more complex tests that will include interaction between CPUManager, MemoryManager and TopologyManager var ( allNUMANodes []int From 8f92a81787e8daaa1a058a44b10c94d2067468d5 Mon Sep 17 00:00:00 2001 From: Francesco Romani Date: Sat, 31 May 2025 12:56:25 +0200 Subject: [PATCH 3/3] node: e2e: podresources: add more e2e tests add more e2e tests to cover the interaction with core resource managers (cpu, memory) and to ensure proper reporting. Signed-off-by: Francesco Romani --- test/e2e_node/podresources_test.go | 715 +++++++++++++++++++++++++++-- 1 file changed, 684 insertions(+), 31 deletions(-) diff --git a/test/e2e_node/podresources_test.go b/test/e2e_node/podresources_test.go index 779cf3b3ee7..b1e9c840125 100644 --- a/test/e2e_node/podresources_test.go +++ b/test/e2e_node/podresources_test.go @@ -24,6 +24,13 @@ import ( "strings" "time" + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" + "github.com/onsi/gomega/gstruct" + "github.com/onsi/gomega/types" + + "google.golang.org/grpc" + v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -39,10 +46,6 @@ import ( admissionapi "k8s.io/pod-security-admission/api" "k8s.io/utils/cpuset" - "github.com/onsi/ginkgo/v2" - "github.com/onsi/gomega" - "github.com/onsi/gomega/gstruct" - "github.com/onsi/gomega/types" "k8s.io/kubernetes/test/e2e/feature" "k8s.io/kubernetes/test/e2e/framework" e2emetrics "k8s.io/kubernetes/test/e2e/framework/metrics" @@ -61,14 +64,34 @@ type podDesc struct { resourceName string resourceAmount int cpuRequest int // cpuRequest is in millicores + memRequest int // memRequest is in megabytes + cpuLimit int // cpuLimit is in millicores + memLimit int // memLimit is in megabytes initContainers []initContainerDesc + mainCntCommand []string + restartPolicy *v1.RestartPolicy } -func (desc podDesc) CpuRequestQty() resource.Quantity { +func (desc podDesc) CPURequestQty() resource.Quantity { qty := resource.NewMilliQuantity(int64(desc.cpuRequest), resource.DecimalSI) return *qty } +func (desc podDesc) MemRequestQty() resource.Quantity { + qty := resource.NewQuantity(int64(desc.memRequest)*1024*1024, resource.DecimalSI) + return *qty +} + +func (desc podDesc) CPULimitQty() resource.Quantity { + qty := resource.NewMilliQuantity(int64(desc.cpuLimit), resource.DecimalSI) + return *qty +} + +func (desc podDesc) MemLimitQty() resource.Quantity { + qty := resource.NewQuantity(int64(desc.memLimit)*1024*1024, resource.DecimalSI) + return *qty +} + func (desc podDesc) CpuRequestExclusive() int { if (desc.cpuRequest % 1000) != 0 { // exclusive cpus are request only if the quantity is integral; @@ -116,6 +139,56 @@ func (desc initContainerDesc) RequiresDevices() bool { return desc.resourceName != "" && desc.resourceAmount > 0 } +// TODO: replace makePodResourcesTestPod reimplementing in terms of this function +func makeFullPodResourcesTestPod(desc podDesc) *v1.Pod { + cnt := v1.Container{ + Name: desc.cntName, + Image: busyboxImage, + Resources: v1.ResourceRequirements{ + Requests: v1.ResourceList{}, + Limits: v1.ResourceList{}, + }, + Command: []string{"sh", "-c", "sleep 1d"}, + } + if desc.cpuLimit > 0 { + cnt.Resources.Limits[v1.ResourceCPU] = desc.CPULimitQty() + } + if desc.cpuRequest > 0 { + cnt.Resources.Requests[v1.ResourceCPU] = desc.CPURequestQty() + } + if desc.memLimit > 0 { + cnt.Resources.Limits[v1.ResourceMemory] = desc.MemLimitQty() + } + if desc.memRequest > 0 { + cnt.Resources.Requests[v1.ResourceMemory] = desc.MemRequestQty() + } + if desc.RequiresDevices() { + devQty := resource.NewQuantity(int64(desc.resourceAmount), resource.DecimalSI) + cnt.Resources.Requests[v1.ResourceName(desc.resourceName)] = *devQty + cnt.Resources.Limits[v1.ResourceName(desc.resourceName)] = *devQty + } + if len(desc.mainCntCommand) > 0 { + cnt.Command = desc.mainCntCommand + } + restartPolicy := v1.RestartPolicyNever + if desc.restartPolicy != nil { + restartPolicy = *desc.restartPolicy + } + + // TODO: support for init containers - currently unneeded + return &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.podName, + }, + Spec: v1.PodSpec{ + RestartPolicy: restartPolicy, + Containers: []v1.Container{ + cnt, + }, + }, + } +} + func makePodResourcesTestPod(desc podDesc) *v1.Pod { cnt := v1.Container{ Name: desc.cntName, @@ -127,7 +200,7 @@ func makePodResourcesTestPod(desc podDesc) *v1.Pod { Command: []string{"sh", "-c", "sleep 1d"}, } if desc.RequiresCPU() { - cpuRequestQty := desc.CpuRequestQty() + cpuRequestQty := desc.CPURequestQty() cnt.Resources.Requests[v1.ResourceCPU] = cpuRequestQty cnt.Resources.Limits[v1.ResourceCPU] = cpuRequestQty // we don't really care, we only need to be in guaranteed QoS @@ -138,6 +211,13 @@ func makePodResourcesTestPod(desc podDesc) *v1.Pod { cnt.Resources.Requests[v1.ResourceName(desc.resourceName)] = resource.MustParse(fmt.Sprintf("%d", desc.resourceAmount)) cnt.Resources.Limits[v1.ResourceName(desc.resourceName)] = resource.MustParse(fmt.Sprintf("%d", desc.resourceAmount)) } + if len(desc.mainCntCommand) > 0 { + cnt.Command = desc.mainCntCommand + } + restartPolicy := v1.RestartPolicyNever + if desc.restartPolicy != nil { + restartPolicy = *desc.restartPolicy + } var initCnts []v1.Container for _, cntDesc := range desc.initContainers { @@ -174,7 +254,7 @@ func makePodResourcesTestPod(desc podDesc) *v1.Pod { Name: desc.podName, }, Spec: v1.PodSpec{ - RestartPolicy: v1.RestartPolicyNever, + RestartPolicy: restartPolicy, InitContainers: initCnts, Containers: []v1.Container{ cnt, @@ -272,6 +352,30 @@ func findContainerDeviceByName(devs []*kubeletpodresourcesv1.ContainerDevices, r return nil } +func matchPodDescWithResourcesNamesOnly(expected []podDesc, found podResMap) error { + framework.Logf("got %d pods expected %d", len(found), len(expected)) + if len(found) != len(expected) { + return fmt.Errorf("found %d items expected %d", len(found), len(expected)) + } + for _, podReq := range expected { + framework.Logf("matching: %#v", podReq) + + if _, ok := found[podReq.podName]; !ok { + return fmt.Errorf("no pod resources for pod %q", podReq.podName) + } + + } + return nil +} + +func matchPodDescAndCountWithResources(expected []podDesc, found podResMap) error { + framework.Logf("got %d pods expected %d", len(found), len(expected)) + if len(found) != len(expected) { + return fmt.Errorf("found %d items expected %d", len(found), len(expected)) + } + return matchPodDescWithResources(expected, found) +} + func matchPodDescWithResources(expected []podDesc, found podResMap) error { for _, podReq := range expected { framework.Logf("matching: %#v", podReq) @@ -847,7 +951,27 @@ var _ = SIGDescribe("POD Resources", framework.WithSerial(), feature.PodResource f := framework.NewDefaultFramework("podresources-test") f.NamespacePodSecurityLevel = admissionapi.LevelPrivileged - reservedSystemCPUs := cpuset.New(1) + var reservedSystemCPUs cpuset.CPUSet + var memoryQuantity resource.Quantity + var defaultKubeParams *memoryManagerKubeletParams + + ginkgo.BeforeEach(func() { + reservedSystemCPUs = cpuset.New(1) + memoryQuantity = resource.MustParse("1100Mi") + defaultKubeParams = &memoryManagerKubeletParams{ + systemReservedMemory: []kubeletconfig.MemoryReservation{ + { + NumaNode: 0, + Limits: v1.ResourceList{ + resourceMemory: memoryQuantity, + }, + }, + }, + systemReserved: map[string]string{resourceMemory: "500Mi"}, + kubeReserved: map[string]string{resourceMemory: "500Mi"}, + evictionHard: map[string]string{evictionHardMemory: "100Mi"}, + } + }) ginkgo.Context("with SRIOV devices in the system", func() { ginkgo.BeforeEach(func() { @@ -1024,27 +1148,6 @@ var _ = SIGDescribe("POD Resources", framework.WithSerial(), feature.PodResource podresourcesGetAllocatableResourcesTests(ctx, cli, nil, onlineCPUs, reservedSystemCPUs) podresourcesGetTests(ctx, f, cli, true) }) - ginkgo.It("should account for resources of pods in terminal phase", func(ctx context.Context) { - pd := podDesc{ - cntName: "e2e-test-cnt", - podName: "e2e-test-pod", - cpuRequest: 1000, - } - pod := makePodResourcesTestPod(pd) - pod.Spec.Containers[0].Command = []string{"sh", "-c", "/bin/true"} - pod = e2epod.NewPodClient(f).Create(ctx, pod) - defer e2epod.NewPodClient(f).DeleteSync(ctx, pod.Name, metav1.DeleteOptions{}, f.Timeouts.PodDelete) - err := e2epod.WaitForPodCondition(ctx, f.ClientSet, pod.Namespace, pod.Name, "Pod Succeeded", time.Minute*2, testutils.PodSucceeded) - framework.ExpectNoError(err) - endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) - framework.ExpectNoError(err) - cli, conn, err := podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) - framework.ExpectNoError(err) - defer conn.Close() - // although the pod moved into terminal state, PodResourcesAPI still list its cpus - expectPodResources(ctx, 1, cli, []podDesc{pd}) - - }) }) }) @@ -1081,6 +1184,556 @@ var _ = SIGDescribe("POD Resources", framework.WithSerial(), feature.PodResource }) }) + ginkgo.When("checking core resource managers assignments", func() { + var podresConn *grpc.ClientConn + var podMap map[string]*v1.Pod + var cpuAlloc int64 + + ginkgo.BeforeEach(func(ctx context.Context) { + podMap = make(map[string]*v1.Pod) + }) + + ginkgo.AfterEach(func(ctx context.Context) { + if podresConn != nil { + framework.ExpectNoError(podresConn.Close()) + } + deletePodsAsync(ctx, f, podMap) + }) + + ginkgo.JustBeforeEach(func(ctx context.Context) { + // this is a very rough check. We just want to rule out system that does NOT have enough resources + _, cpuAlloc, _ = getLocalNodeCPUDetails(ctx, f) + if cpuAlloc < minCoreCount { + e2eskipper.Skipf("Skipping CPU Manager tests since the CPU allocatable < %d", minCoreCount) + } + }) + + ginkgo.When("listing with restricted list output enabled", func() { + + tempSetCurrentKubeletConfig(f, func(ctx context.Context, initialConfig *kubeletconfig.KubeletConfiguration) { + kubeParams := *defaultKubeParams + kubeParams.policy = staticPolicy + updateKubeletConfigWithMemoryManagerParams(initialConfig, &kubeParams) + initialConfig.CPUManagerPolicy = string(cpumanager.PolicyStatic) + initialConfig.CPUManagerReconcilePeriod = metav1.Duration{Duration: 10 * time.Minute} // set it long enough it is practically disabled + cpus := reservedSystemCPUs.String() + framework.Logf("configurePodResourcesInKubelet: using reservedSystemCPUs=%q", cpus) + initialConfig.ReservedSystemCPUs = cpus + }) + + ginkgo.It("should report only the running pods regardless of the QoS", func(ctx context.Context) { + // TODO: compute automatically + if cpuAlloc < 8 { + e2eskipper.Skipf("Skipping CPU Manager tests since the CPU %d allocatable < %d", cpuAlloc, 8) + } + + restartNever := v1.RestartPolicyNever + restartAlways := v1.RestartPolicyAlways + descs := []podDesc{ + { + cntName: "e2e-test-cnt-gu-term-1", + podName: "e2e-test-pod-gu-term-1", + cpuLimit: 1000, + memLimit: 256, + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-gu-term-2", + podName: "e2e-test-pod-gu-term-2", + cpuLimit: 200, + memLimit: 256, + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-gu-run-1", + podName: "e2e-test-pod-gu-run-1", + cpuLimit: 200, + memLimit: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-gu-run-2", + podName: "e2e-test-pod-gu-run-2", + cpuLimit: 1000, + memLimit: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-bu-term-1", + podName: "e2e-test-pod-bu-term-1", + cpuRequest: 400, + memRequest: 256, + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-bu-run-1", + podName: "e2e-test-pod-bu-run-1", + cpuRequest: 400, + memRequest: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-be-term-1", + podName: "e2e-test-pod-be-term-1", + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-be-run-1", + podName: "e2e-test-pod-be-run-1", + restartPolicy: &restartAlways, + }, + } + expected := []podDesc{ + { + cntName: "e2e-test-cnt-gu-run-1", + podName: "e2e-test-pod-gu-run-1", + cpuLimit: 200, + memLimit: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-gu-run-2", + podName: "e2e-test-pod-gu-run-2", + cpuLimit: 1000, + memLimit: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-bu-run-1", + podName: "e2e-test-pod-bu-run-1", + cpuRequest: 400, + memRequest: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-be-run-1", + podName: "e2e-test-pod-be-run-1", + restartPolicy: &restartAlways, + }, + } + + for _, desc := range descs { + pod := makeFullPodResourcesTestPod(desc) + if desc.restartPolicy == &restartAlways { + pod = e2epod.NewPodClient(f).CreateSync(ctx, pod) + podMap[string(pod.UID)] = pod + } else { + pod = e2epod.NewPodClient(f).Create(ctx, pod) + pod, err := e2epod.NewPodClient(f).Get(ctx, pod.Name, metav1.GetOptions{}) + framework.ExpectNoError(err) + podMap[string(pod.UID)] = pod + err = e2epod.WaitForPodCondition(ctx, f.ClientSet, pod.Namespace, pod.Name, "Pod Succeeded", time.Minute*2, testutils.PodSucceeded) + framework.ExpectNoError(err) + } + } + + endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) + framework.ExpectNoError(err) + + var cli kubeletpodresourcesv1.PodResourcesListerClient + cli, podresConn, err = podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) + framework.ExpectNoError(err) + + gomega.Consistently(func(ctx context.Context) error { + found, err := getPodResourcesValues(ctx, cli) + if err != nil { + return err + } + return matchPodDescAndCountWithResources(expected, found) + }).WithContext(ctx).WithTimeout(30 * time.Second).WithPolling(2 * time.Second).Should(gomega.Succeed()) + }) + + ginkgo.It("should report only the running pods among many", func(ctx context.Context) { + // TODO: compute automatically + if cpuAlloc < 8 { + e2eskipper.Skipf("Skipping CPU Manager tests since the CPU %d allocatable < %d", cpuAlloc, 8) + } + restartNever := v1.RestartPolicyNever + restartAlways := v1.RestartPolicyAlways + descs := []podDesc{ + { + cntName: "e2e-test-cnt-payload-gu-1", + podName: "e2e-test-pod-payload-gu-1", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-mixed-cpu-mem", + podName: "e2e-test-pod-mixed-cpu-mem", + cpuRequest: 1000, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + restartPolicy: &restartNever, + }, + { + cntName: "e2e-test-cnt-payload-gu-2", + podName: "e2e-test-pod-payload-gu-2", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-payload-gu-3", + podName: "e2e-test-pod-payload-gu-3", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-mixed-mem", + podName: "e2e-test-pod-mixed-mem", + cpuRequest: 1000, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + restartPolicy: &restartNever, + }, + } + expected := []podDesc{ + { + cntName: "e2e-test-cnt-payload-gu-1", + podName: "e2e-test-pod-payload-gu-1", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-payload-gu-2", + podName: "e2e-test-pod-payload-gu-2", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-payload-gu-3", + podName: "e2e-test-pod-payload-gu-3", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + } + + for _, desc := range descs { + pod := makePodResourcesTestPod(desc) + if desc.restartPolicy == &restartAlways { + pod = e2epod.NewPodClient(f).CreateSync(ctx, pod) + podMap[string(pod.UID)] = pod + } else { + pod = e2epod.NewPodClient(f).Create(ctx, pod) + pod, err := e2epod.NewPodClient(f).Get(ctx, pod.Name, metav1.GetOptions{}) + framework.ExpectNoError(err) + podMap[string(pod.UID)] = pod + err = e2epod.WaitForPodCondition(ctx, f.ClientSet, pod.Namespace, pod.Name, "Pod Succeeded", time.Minute*2, testutils.PodSucceeded) + framework.ExpectNoError(err) + } + } + + endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) + framework.ExpectNoError(err) + + var cli kubeletpodresourcesv1.PodResourcesListerClient + cli, podresConn, err = podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) + framework.ExpectNoError(err) + + gomega.Consistently(func(ctx context.Context) error { + found, err := getPodResourcesValues(ctx, cli) + if err != nil { + return err + } + return matchPodDescAndCountWithResources(expected, found) + }).WithContext(ctx).WithTimeout(30 * time.Second).WithPolling(2 * time.Second).Should(gomega.Succeed()) + }) + + ginkgo.DescribeTable("all pods terminated and requiring", func(ctx context.Context, cpuReqMilli, podCount int) { + cpuReqTot := int64((cpuReqMilli * podCount) / 1000) + ginkgo.By(fmt.Sprintf("needed: %d cores", cpuReqTot)) + if cpuAlloc < cpuReqTot { + e2eskipper.Skipf("Skipping CPU Manager tests since the CPU %d allocatable < %d", cpuAlloc, cpuReqTot) + } + + restartNever := v1.RestartPolicyNever + + for idx := range podCount { + pd := podDesc{ + cntName: fmt.Sprintf("e2e-test-cnt-%d", idx), + podName: fmt.Sprintf("e2e-test-pod-%d", idx), + cpuRequest: cpuReqMilli, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + restartPolicy: &restartNever, + } + pod := makePodResourcesTestPod(pd) + pod = e2epod.NewPodClient(f).Create(ctx, pod) + pod, err := e2epod.NewPodClient(f).Get(ctx, pod.Name, metav1.GetOptions{}) + framework.ExpectNoError(err) + podMap[string(pod.UID)] = pod + err = e2epod.WaitForPodCondition(ctx, f.ClientSet, pod.Namespace, pod.Name, "Pod Succeeded", time.Minute*2, testutils.PodSucceeded) + framework.ExpectNoError(err) + } + endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) + framework.ExpectNoError(err) + + var cli kubeletpodresourcesv1.PodResourcesListerClient + cli, podresConn, err = podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) + framework.ExpectNoError(err) + + gomega.Consistently(func(ctx context.Context) error { + found, err := getPodResourcesValues(ctx, cli) + if err != nil { + return err + } + if len(found) > 0 { + return fmt.Errorf("returned unexpected pods: %v", found) + } + return nil + }).WithContext(ctx).WithTimeout(30 * time.Second).WithPolling(2 * time.Second).Should(gomega.Succeed()) + + }, + ginkgo.Entry("cpu and mem single", context.TODO(), 1000, 1), + ginkgo.Entry("cpu and mem multi", context.TODO(), 1000, 3), + ginkgo.Entry("mem only single", context.TODO(), 200, 1), + ginkgo.Entry("mem only multi", context.TODO(), 200, 3), + ) + }) + + ginkgo.When("listing with restricted list output disabled for backward compatible defaults", func() { + + tempSetCurrentKubeletConfig(f, func(ctx context.Context, initialConfig *kubeletconfig.KubeletConfiguration) { + kubeParams := *defaultKubeParams + kubeParams.policy = staticPolicy + updateKubeletConfigWithMemoryManagerParams(initialConfig, &kubeParams) + initialConfig.CPUManagerPolicy = string(cpumanager.PolicyStatic) + initialConfig.CPUManagerReconcilePeriod = metav1.Duration{Duration: 10 * time.Minute} // set it long enough it is practically disabled + cpus := reservedSystemCPUs.String() + framework.Logf("configurePodResourcesInKubelet: using reservedSystemCPUs=%q", cpus) + initialConfig.ReservedSystemCPUs = cpus + initialConfig.FeatureGates["KubeletPodResourcesListUseActivePods"] = false + }) + + ginkgo.It("should report all the known pods regardless of the QoS", func(ctx context.Context) { + // TODO: compute automatically + if cpuAlloc < 8 { + e2eskipper.Skipf("Skipping CPU Manager tests since the CPU %d allocatable < %d", cpuAlloc, 8) + } + + restartNever := v1.RestartPolicyNever + restartAlways := v1.RestartPolicyAlways + descs := []podDesc{ + { + cntName: "e2e-test-cnt-gu-term-1", + podName: "e2e-test-pod-gu-term-1", + cpuLimit: 1000, + memLimit: 256, + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-gu-term-2", + podName: "e2e-test-pod-gu-term-2", + cpuLimit: 200, + memLimit: 256, + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-gu-run-1", + podName: "e2e-test-pod-gu-run-1", + cpuLimit: 200, + memLimit: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-gu-run-2", + podName: "e2e-test-pod-gu-run-2", + cpuLimit: 1000, + memLimit: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-bu-term-1", + podName: "e2e-test-pod-bu-term-1", + cpuRequest: 400, + memRequest: 256, + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-bu-run-1", + podName: "e2e-test-pod-bu-run-1", + cpuRequest: 400, + memRequest: 256, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-be-term-1", + podName: "e2e-test-pod-be-term-1", + restartPolicy: &restartNever, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + }, + { + cntName: "e2e-test-cnt-be-run-1", + podName: "e2e-test-pod-be-run-1", + restartPolicy: &restartAlways, + }, + } + expected := descs + + for _, desc := range descs { + pod := makeFullPodResourcesTestPod(desc) + if desc.restartPolicy == &restartAlways { + pod = e2epod.NewPodClient(f).CreateSync(ctx, pod) + podMap[string(pod.UID)] = pod + } else { + pod = e2epod.NewPodClient(f).Create(ctx, pod) + pod, err := e2epod.NewPodClient(f).Get(ctx, pod.Name, metav1.GetOptions{}) + framework.ExpectNoError(err) + podMap[string(pod.UID)] = pod + err = e2epod.WaitForPodCondition(ctx, f.ClientSet, pod.Namespace, pod.Name, "Pod Succeeded", time.Minute*2, testutils.PodSucceeded) + framework.ExpectNoError(err) + } + } + + endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) + framework.ExpectNoError(err) + + var cli kubeletpodresourcesv1.PodResourcesListerClient + cli, podresConn, err = podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) + framework.ExpectNoError(err) + + gomega.Consistently(func(ctx context.Context) error { + found, err := getPodResourcesValues(ctx, cli) + if err != nil { + return err + } + return matchPodDescAndCountWithResources(expected, found) + }).WithContext(ctx).WithTimeout(30 * time.Second).WithPolling(2 * time.Second).Should(gomega.Succeed()) + }) + + ginkgo.It("should report all the known pods", func(ctx context.Context) { + // TODO: compute automatically + if cpuAlloc < 8 { + e2eskipper.Skipf("Skipping CPU Manager tests since the CPU %d allocatable < %d", cpuAlloc, 8) + } + restartNever := v1.RestartPolicyNever + restartAlways := v1.RestartPolicyAlways + descs := []podDesc{ + { + cntName: "e2e-test-cnt-payload-gu-1", + podName: "e2e-test-pod-payload-gu-1", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-mixed-cpu-mem", + podName: "e2e-test-pod-mixed-cpu-mem", + cpuRequest: 1000, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + restartPolicy: &restartNever, + }, + { + cntName: "e2e-test-cnt-payload-gu-2", + podName: "e2e-test-pod-payload-gu-2", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-payload-gu-3", + podName: "e2e-test-pod-payload-gu-3", + cpuRequest: 1000, + restartPolicy: &restartAlways, + }, + { + cntName: "e2e-test-cnt-mixed-mem", + podName: "e2e-test-pod-mixed-mem", + cpuRequest: 1000, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + restartPolicy: &restartNever, + }, + } + expected := descs + + for _, desc := range descs { + pod := makePodResourcesTestPod(desc) + if desc.restartPolicy == &restartAlways { + pod = e2epod.NewPodClient(f).CreateSync(ctx, pod) + podMap[string(pod.UID)] = pod + } else { + pod = e2epod.NewPodClient(f).Create(ctx, pod) + pod, err := e2epod.NewPodClient(f).Get(ctx, pod.Name, metav1.GetOptions{}) + framework.ExpectNoError(err) + podMap[string(pod.UID)] = pod + err = e2epod.WaitForPodCondition(ctx, f.ClientSet, pod.Namespace, pod.Name, "Pod Succeeded", time.Minute*2, testutils.PodSucceeded) + framework.ExpectNoError(err) + } + } + + endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) + framework.ExpectNoError(err) + + var cli kubeletpodresourcesv1.PodResourcesListerClient + cli, podresConn, err = podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) + framework.ExpectNoError(err) + + gomega.Consistently(func(ctx context.Context) error { + found, err := getPodResourcesValues(ctx, cli) + if err != nil { + return err + } + // all but the last pod will have inconsistent resource reporting. + // See: https://github.com/kubernetes/kubernetes/issues/132020#issuecomment-2921478368 + // this is why we introduced the FeatureGate flag in the first place + return matchPodDescWithResourcesNamesOnly(expected, found) + }).WithContext(ctx).WithTimeout(30 * time.Second).WithPolling(2 * time.Second).Should(gomega.Succeed()) + }) + + ginkgo.DescribeTable("all pods terminated and requiring", func(ctx context.Context, cpuReqMilli, podCount int) { + cpuReqTot := int64((cpuReqMilli * podCount) / 1000) + ginkgo.By(fmt.Sprintf("needed: %d cores", cpuReqTot)) + if cpuAlloc < cpuReqTot { + e2eskipper.Skipf("Skipping CPU Manager tests since the CPU %d allocatable < %d", cpuAlloc, cpuReqTot) + } + + restartNever := v1.RestartPolicyNever + + var expected []podDesc + for idx := range podCount { + pd := podDesc{ + cntName: fmt.Sprintf("e2e-test-cnt-%d", idx), + podName: fmt.Sprintf("e2e-test-pod-%d", idx), + cpuRequest: cpuReqMilli, + mainCntCommand: []string{"sh", "-c", "/bin/true"}, + restartPolicy: &restartNever, + } + expected = append(expected, pd) + pod := makePodResourcesTestPod(pd) + pod = e2epod.NewPodClient(f).Create(ctx, pod) + pod, err := e2epod.NewPodClient(f).Get(ctx, pod.Name, metav1.GetOptions{}) + framework.ExpectNoError(err) + podMap[string(pod.UID)] = pod + err = e2epod.WaitForPodCondition(ctx, f.ClientSet, pod.Namespace, pod.Name, "Pod Succeeded", time.Minute*2, testutils.PodSucceeded) + framework.ExpectNoError(err) + } + endpoint, err := util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket) + framework.ExpectNoError(err) + + var cli kubeletpodresourcesv1.PodResourcesListerClient + cli, podresConn, err = podresources.GetV1Client(endpoint, defaultPodResourcesTimeout, defaultPodResourcesMaxSize) + framework.ExpectNoError(err) + + gomega.Consistently(func(ctx context.Context) error { + found, err := getPodResourcesValues(ctx, cli) + if err != nil { + return err + } + // all but the last pod will have inconsistent resource reporting. + // See: https://github.com/kubernetes/kubernetes/issues/132020#issuecomment-2921478368 + // this is why we introduced the FeatureGate in the first place + return matchPodDescWithResourcesNamesOnly(expected, found) + }).WithContext(ctx).WithTimeout(30 * time.Second).WithPolling(2 * time.Second).Should(gomega.Succeed()) + }, + ginkgo.Entry("cpu and mem single", context.TODO(), 1000, 1), + ginkgo.Entry("cpu and mem multi", context.TODO(), 1000, 3), + ginkgo.Entry("mem only single", context.TODO(), 200, 1), + ginkgo.Entry("mem only multi", context.TODO(), 200, 3), + ) + }) + }) + ginkgo.Context("with a topology-unaware device plugin, which reports resources w/o hardware topology", func() { ginkgo.Context("with CPU manager Static policy", func() { ginkgo.BeforeEach(func(ctx context.Context) { @@ -1142,7 +1795,7 @@ var _ = SIGDescribe("POD Resources", framework.WithSerial(), feature.PodResource desc, }) - expectPodResources(ctx, 1, cli, []podDesc{desc}) + expectPodResources(ctx, 0, cli, []podDesc{desc}) ginkgo.By("Restarting Kubelet") restartKubelet(ctx, true) @@ -1152,7 +1805,7 @@ var _ = SIGDescribe("POD Resources", framework.WithSerial(), feature.PodResource ginkgo.By("Wait for node to be ready") waitForTopologyUnawareResources(ctx, f) - expectPodResources(ctx, 1, cli, []podDesc{desc}) + expectPodResources(ctx, 0, cli, []podDesc{desc}) tpd.deletePodsForTest(ctx, f) }) })