diff --git a/pkg/kubelet/allocation/allocation_manager.go b/pkg/kubelet/allocation/allocation_manager.go index 41276a6081a..c229da41880 100644 --- a/pkg/kubelet/allocation/allocation_manager.go +++ b/pkg/kubelet/allocation/allocation_manager.go @@ -25,6 +25,7 @@ import ( "time" "encoding/json" + v1 "k8s.io/api/core/v1" apiequality "k8s.io/apimachinery/pkg/api/equality" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -57,6 +58,13 @@ const ( initialRetryDelay = 30 * time.Second retryDelay = 3 * time.Minute + + TriggerReasonPodResized = "pod_resized" + TriggerReasonPodUpdated = "pod_updated" + TriggerReasonPodsAdded = "pods_added" + TriggerReasonPodsRemoved = "pods_removed" + + triggerReasonPeriodic = "periodic_retry" ) // AllocationManager tracks pod resource allocations. @@ -109,8 +117,8 @@ type Manager interface { // Returns true if the pending resize was added to the queue, false if it was already present. PushPendingResize(uid types.UID) bool - // RetryPendingResizes retries all pending resizes. It returns a list of successful resizes. - RetryPendingResizes() []*v1.Pod + // RetryPendingResizes retries all pending resizes. + RetryPendingResizes(trigger string) // CheckPodResizeInProgress checks whether the actuated resizable resources differ from the allocated resources // for any running containers. @@ -222,7 +230,7 @@ func (m *manager) Run(ctx context.Context) { for { select { case <-m.ticker.C: - successfulResizes := m.RetryPendingResizes() + successfulResizes := m.retryPendingResizes(triggerReasonPeriodic) for _, po := range successfulResizes { klog.InfoS("Successfully retried resize after timeout", "pod", klog.KObj(po)) } @@ -261,7 +269,11 @@ func (m *manager) podResizeCompletionMsg(allocatedPod *v1.Pod) string { return podResizeCompletedMsg } -func (m *manager) RetryPendingResizes() []*v1.Pod { +func (m *manager) RetryPendingResizes(trigger string) { + m.retryPendingResizes(trigger) +} + +func (m *manager) retryPendingResizes(trigger string) []*v1.Pod { m.allocationMutex.Lock() defer m.allocationMutex.Unlock() @@ -285,6 +297,7 @@ func (m *manager) RetryPendingResizes() []*v1.Pod { } oldResizeStatus := m.statusManager.GetPodResizeConditions(uid) + isDeferred := m.statusManager.IsPodResizeDeferred(uid) resizeAllocated, err := m.handlePodResourcesResize(pod) switch { @@ -299,6 +312,9 @@ func (m *manager) RetryPendingResizes() []*v1.Pod { default: klog.V(4).InfoS("Pod resize successfully allocated", "pod", klog.KObj(pod)) successfulResizes = append(successfulResizes, pod) + if isDeferred { + metrics.PodDeferredAcceptedResizes.WithLabelValues(trigger).Inc() + } } // If the pod resize status has changed, we need to update the pod status. @@ -310,7 +326,6 @@ func (m *manager) RetryPendingResizes() []*v1.Pod { m.podsWithPendingResizes = newPendingResizes return successfulResizes - } func (m *manager) PushPendingResize(uid types.UID) bool { @@ -768,7 +783,7 @@ func (m *manager) CheckPodResizeInProgress(allocatedPod *v1.Pod, podStatus *kube m.recorder.Eventf(allocatedPod, v1.EventTypeNormal, events.ResizeCompleted, podResizeCompletedEventMsg) } // TODO(natasha41575): We only need to make this call if any of the resources were decreased. - m.RetryPendingResizes() + m.RetryPendingResizes(TriggerReasonPodResized) } } diff --git a/pkg/kubelet/allocation/allocation_manager_test.go b/pkg/kubelet/allocation/allocation_manager_test.go index 450b2a4dd88..1285cf49631 100644 --- a/pkg/kubelet/allocation/allocation_manager_test.go +++ b/pkg/kubelet/allocation/allocation_manager_test.go @@ -867,7 +867,7 @@ func TestHandlePodResourcesResize(t *testing.T) { return newPod, true } allocationManager.PushPendingResize(originalPod.UID) - allocationManager.RetryPendingResizes() + allocationManager.RetryPendingResizes(TriggerReasonPodUpdated) allocatedPod, _ := allocationManager.UpdatePodFromAllocation(newPod) allocationManager.CheckPodResizeInProgress(allocatedPod, podStatus) @@ -1058,7 +1058,7 @@ func TestHandlePodResourcesResizeWithSwap(t *testing.T) { return newPod, true } allocationManager.PushPendingResize(testPod.UID) - allocationManager.RetryPendingResizes() + allocationManager.RetryPendingResizes(TriggerReasonPodUpdated) allocatedPod, _ := allocationManager.UpdatePodFromAllocation(newPod) allocationManager.CheckPodResizeInProgress(allocatedPod, podStatus) @@ -1245,7 +1245,7 @@ func TestHandlePodResourcesResizeMultipleConditions(t *testing.T) { } allocationManager.PushPendingResize(testPod.UID) - allocationManager.RetryPendingResizes() + allocationManager.RetryPendingResizes(TriggerReasonPodUpdated) allocatedPod, _ := allocationManager.UpdatePodFromAllocation(testPod) allocationManager.CheckPodResizeInProgress(allocatedPod, podStatus) @@ -1709,6 +1709,108 @@ func TestSortPendingResizes(t *testing.T) { require.Equal(t, expected, allocationManager.(*manager).podsWithPendingResizes) } +func TestRecordPodDeferredAcceptedResizes(t *testing.T) { + if goruntime.GOOS == "windows" { + t.Skip("InPlacePodVerticalScaling is not currently supported for Windows") + } + + metrics.Register() + metrics.PodDeferredAcceptedResizes.Reset() + + cpu500m := resource.MustParse("500m") + cpu1000m := resource.MustParse("1") + mem500M := resource.MustParse("500Mi") + mem1000M := resource.MustParse("1Gi") + + for _, tc := range []struct { + name string + trigger string + hasPendingCondition bool + expectedMetrics string + }{ + { + name: "trigger reason: pod updated, no pending condition", + trigger: TriggerReasonPodUpdated, + }, + { + name: "trigger reason: pod resized, pending condition", + trigger: TriggerReasonPodResized, + hasPendingCondition: true, + expectedMetrics: ` + # HELP kubelet_pod_deferred_accepted_resizes_total [ALPHA] Cumulative number of resizes that were accepted after being deferred. + # TYPE kubelet_pod_deferred_accepted_resizes_total counter + kubelet_pod_deferred_accepted_resizes_total{retry_trigger="pod_resized"} 1 + `, + }, + { + name: "trigger reason: periodic retry, pending condition", + trigger: triggerReasonPeriodic, + hasPendingCondition: true, + expectedMetrics: ` + # HELP kubelet_pod_deferred_accepted_resizes_total [ALPHA] Cumulative number of resizes that were accepted after being deferred. + # TYPE kubelet_pod_deferred_accepted_resizes_total counter + kubelet_pod_deferred_accepted_resizes_total{retry_trigger="periodic_retry"} 1 + kubelet_pod_deferred_accepted_resizes_total{retry_trigger="pod_resized"} 1 + `, + }, + } { + t.Run(tc.name, func(t *testing.T) { + original := &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + UID: "1111", + Name: "pod1", + Namespace: "ns1", + }, + Spec: v1.PodSpec{ + Containers: []v1.Container{ + { + Name: "c1", + Image: "i1", + Resources: v1.ResourceRequirements{ + Requests: v1.ResourceList{v1.ResourceCPU: cpu1000m, v1.ResourceMemory: mem1000M}, + }, + }, + }, + }, + Status: v1.PodStatus{ + Phase: v1.PodRunning, + ContainerStatuses: []v1.ContainerStatus{ + { + Name: "c1", + AllocatedResources: v1.ResourceList{v1.ResourceCPU: cpu1000m, v1.ResourceMemory: mem1000M}, + Resources: &v1.ResourceRequirements{}, + }, + }, + }, + } + + resizedPod := original.DeepCopy() + resizedPod.Spec.Containers[0].Resources.Requests = v1.ResourceList{v1.ResourceCPU: cpu500m, v1.ResourceMemory: mem500M} + + am := makeAllocationManager(t, &containertest.FakeRuntime{}, []*v1.Pod{original}) + require.NoError(t, am.SetAllocatedResources(original)) + require.NoError(t, am.SetActuatedResources(original, nil)) + if tc.hasPendingCondition { + am.(*manager).statusManager.SetPodResizePendingCondition(original.UID, v1.PodReasonDeferred, "message", 1) + } + + am.(*manager).getPodByUID = func(uid types.UID) (*v1.Pod, bool) { + return resizedPod, true + } + am.PushPendingResize(original.UID) + resizedPods := am.(*manager).retryPendingResizes(tc.trigger) + + require.Len(t, resizedPods, 1) + require.Equal(t, original.UID, resizedPods[0].UID) + + require.NoError(t, testutil.GatherAndCompare( + legacyregistry.DefaultGatherer, strings.NewReader(tc.expectedMetrics), "kubelet_pod_deferred_accepted_resizes_total", + )) + }) + } + +} + func makeAllocationManager(t *testing.T, runtime *containertest.FakeRuntime, allocatedPods []*v1.Pod) Manager { t.Helper() statusManager := status.NewManager(&fake.Clientset{}, kubepod.NewBasicPodManager(), &statustest.FakePodDeletionSafetyProvider{}, kubeletutil.NewPodStartupLatencyTracker()) diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index 8a5f964b1e0..ea231d85f52 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -2654,7 +2654,7 @@ func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) { kl.allocationManager.PushPendingResize(uid) } if len(pendingResizes) > 0 { - kl.allocationManager.RetryPendingResizes() + kl.allocationManager.RetryPendingResizes(allocation.TriggerReasonPodsAdded) } } } @@ -2690,7 +2690,7 @@ func (kl *Kubelet) HandlePodUpdates(pods []*v1.Pod) { // TODO (natasha41575): If the resize is immediately actuated, it will trigger a pod sync // and we will end up calling UpdatePod twice. Figure out if there is a way to avoid this. - kl.allocationManager.RetryPendingResizes() + kl.allocationManager.RetryPendingResizes(allocation.TriggerReasonPodUpdated) } else { // We can hit this case if a pending resize has been reverted, // so we need to clear the pending resize condition. @@ -2789,7 +2789,7 @@ func (kl *Kubelet) HandlePodRemoves(pods []*v1.Pod) { } if utilfeature.DefaultFeatureGate.Enabled(features.InPlacePodVerticalScaling) { - kl.allocationManager.RetryPendingResizes() + kl.allocationManager.RetryPendingResizes(allocation.TriggerReasonPodsRemoved) } }