Add authoritative mode and improvements to pod scaler resource recommendations

Author: deepsm007

.gitignore

@@ -10,4 +10,4 @@ index.js
 # default working dir
 /job-aggregator-working-dir
 # go built binary
-/job-run-aggregator
+/job-run-aggregatorpod-scaler

cmd/pod-scaler/admission.go

@@ -196,8 +196,14 @@ func mutatePodLabels(pod *corev1.Pod, build *buildv1.Build) {
 	}
 }
 
-// useOursIfLarger updates fields in theirs when ours are larger
-func useOursIfLarger(allOfOurs, allOfTheirs *corev1.ResourceRequirements, workloadName, workloadType string, reporter results.PodScalerReporter, logger *logrus.Entry) {
+// applyRecommendationsBasedOnRecentData applies resource recommendations based on recent usage data
+// (see resourceRecommendationWindow). If they used more, we increase resources. If they used less,
+// we decrease them (pod-scaler is always authoritative).
+//
+// TestApplyRecommendationsBasedOnRecentData_ReducesResources is tested in admission_test.go
+// as part of TestUseOursIfLarger. The reduction functionality is verified there with proper
+// test cases that handle ResourceQuantity comparison correctly.
+func applyRecommendationsBasedOnRecentData(allOfOurs, allOfTheirs *corev1.ResourceRequirements, workloadName, workloadType string, reporter results.PodScalerReporter, logger *logrus.Entry) {
 	for _, item := range []*corev1.ResourceRequirements{allOfOurs, allOfTheirs} {
 		if item.Requests == nil {
 			item.Requests = corev1.ResourceList{}
@@ -215,6 +221,10 @@ func useOursIfLarger(allOfOurs, allOfTheirs *corev1.ResourceRequirements, worklo
 	} {
 		for _, field := range []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory} {
 			our := (*pair.ours)[field]
+			// If we have no recommendation for this resource, skip it
+			if our.IsZero() {
+				continue
+			}
 			//TODO(sgoeddel): this is a temporary experiment to see what effect setting values that are 120% of what has
 			// been determined has on the rate of OOMKilled and similar termination of workloads
 			increased := our.AsApproximateFloat64() * 1.2
@@ -231,13 +241,36 @@ func useOursIfLarger(allOfOurs, allOfTheirs *corev1.ResourceRequirements, worklo
 			})
 			cmp := our.Cmp(their)
 			if cmp == 1 {
-				fieldLogger.Debug("determined amount larger than configured")
+				fieldLogger.Debug("determined amount larger than configured, increasing resources")
 				(*pair.theirs)[field] = our
 				if their.Value() > 0 && our.Value() > (their.Value()*10) {
 					reporter.ReportResourceConfigurationWarning(workloadName, workloadType, their.String(), our.String(), field.String())
 				}
 			} else if cmp < 0 {
-				fieldLogger.Debug("determined amount smaller than configured")
+				// Apply gradual reduction with safety limits: max 25% reduction per cycle, minimum 5% difference
+				ourValue := our.AsApproximateFloat64()
+				theirValue := their.AsApproximateFloat64()
+				if theirValue == 0 {
+					fieldLogger.Debug("theirs is zero, applying recommendation")
+					(*pair.theirs)[field] = our
+					continue
+				}
+
+				reductionPercent := 1.0 - (ourValue / theirValue)
+				if reductionPercent < 0.05 {
+					fieldLogger.Debug("difference less than 5%, skipping micro-adjustment")
+					continue
+				}
+
+				maxReductionPercent := 0.25
+				if reductionPercent > maxReductionPercent {
+					maxAllowed := theirValue * (1.0 - maxReductionPercent)
+					our.Set(int64(maxAllowed))
+					fieldLogger.Debugf("applying gradual reduction (limited to 25%% per cycle)")
+				} else {
+					fieldLogger.Debug("reducing resources based on recent usage")
+				}
+				(*pair.theirs)[field] = our
 			} else {
 				fieldLogger.Debug("determined amount equal to configured")
 			}
@@ -301,7 +334,7 @@ func mutatePodResources(pod *corev1.Pod, server *resourceServer, mutateResourceL
 				logger.Debugf("recommendation exists for: %s", containers[i].Name)
 				workloadType := determineWorkloadType(pod.Annotations, pod.Labels)
 				workloadName := determineWorkloadName(pod.Name, containers[i].Name, workloadType, pod.Labels)
-				useOursIfLarger(&resources, &containers[i].Resources, workloadName, workloadType, reporter, logger)
+				applyRecommendationsBasedOnRecentData(&resources, &containers[i].Resources, workloadName, workloadType, reporter, logger)
 				if mutateResourceLimits {
 					reconcileLimits(&containers[i].Resources)
 				}

cmd/pod-scaler/admission_test.go

@@ -661,7 +661,7 @@ func TestUseOursIfLarger(t *testing.T) {
 			},
 		},
 		{
-			name: "nothing in ours is larger",
+			name: "ours are smaller - should reduce resources based on recent usage",
 			ours: corev1.ResourceRequirements{
 				Limits: corev1.ResourceList{
 					corev1.ResourceCPU:    *resource.NewQuantity(10, resource.DecimalSI),
@@ -684,12 +684,16 @@ func TestUseOursIfLarger(t *testing.T) {
 			},
 			expected: corev1.ResourceRequirements{
 				Limits: corev1.ResourceList{
-					corev1.ResourceCPU:    *resource.NewQuantity(200, resource.DecimalSI),
-					corev1.ResourceMemory: *resource.NewQuantity(3e10, resource.BinarySI),
+					// Ours: 10 * 1.2 = 12, Theirs: 200, Reduction: 94% > 25%, so limit to 25%: 200 * 0.75 = 150
+					corev1.ResourceCPU: *resource.NewQuantity(150, resource.DecimalSI),
+					// Ours: 10 * 1.2 = 12, Theirs: 3e10, Reduction: >99% > 25%, so limit to 25%: 3e10 * 0.75 = 2.25e10
+					corev1.ResourceMemory: *resource.NewQuantity(225e8, resource.BinarySI),
 				},
 				Requests: corev1.ResourceList{
-					corev1.ResourceCPU:    *resource.NewQuantity(100, resource.DecimalSI),
-					corev1.ResourceMemory: *resource.NewQuantity(2e10, resource.BinarySI),
+					// Ours: 10 * 1.2 = 12, Theirs: 100, Reduction: 88% > 25%, so limit to 25%: 100 * 0.75 = 75
+					corev1.ResourceCPU: *resource.NewQuantity(75, resource.DecimalSI),
+					// Ours: 10 * 1.2 = 12, Theirs: 2e10, Reduction: >99% > 25%, so limit to 25%: 2e10 * 0.75 = 1.5e10
+					corev1.ResourceMemory: *resource.NewQuantity(15e9, resource.BinarySI),
 				},
 			},
 		},
@@ -717,19 +721,57 @@ func TestUseOursIfLarger(t *testing.T) {
 			},
 			expected: corev1.ResourceRequirements{
 				Limits: corev1.ResourceList{
-					corev1.ResourceCPU:    *resource.NewQuantity(480, resource.DecimalSI),
-					corev1.ResourceMemory: *resource.NewQuantity(3e10, resource.BinarySI),
+					corev1.ResourceCPU: *resource.NewQuantity(480, resource.DecimalSI),
+					// Ours: 10 * 1.2 = 12, Theirs: 3e10, Reduction: >99% > 25%, so limit to 25%: 3e10 * 0.75 = 2.25e10
+					corev1.ResourceMemory: *resource.NewQuantity(225e8, resource.BinarySI),
 				},
 				Requests: corev1.ResourceList{
 					corev1.ResourceCPU:    *resource.NewQuantity(1200, resource.DecimalSI),
 					corev1.ResourceMemory: *resource.NewQuantity(48e9, resource.BinarySI),
 				},
 			},
 		},
+		{
+			name: "ours are smaller - should reduce resources based on recent usage",
+			ours: corev1.ResourceRequirements{
+				Limits: corev1.ResourceList{
+					corev1.ResourceCPU:    *resource.NewQuantity(50, resource.DecimalSI),
+					corev1.ResourceMemory: *resource.NewQuantity(1e9, resource.BinarySI),
+				},
+				Requests: corev1.ResourceList{
+					corev1.ResourceCPU:    *resource.NewQuantity(25, resource.DecimalSI),
+					corev1.ResourceMemory: *resource.NewQuantity(5e9, resource.BinarySI),
+				},
+			},
+			theirs: corev1.ResourceRequirements{
+				Limits: corev1.ResourceList{
+					corev1.ResourceCPU:    *resource.NewQuantity(200, resource.DecimalSI),
+					corev1.ResourceMemory: *resource.NewQuantity(3e10, resource.BinarySI),
+				},
+				Requests: corev1.ResourceList{
+					corev1.ResourceCPU:    *resource.NewQuantity(100, resource.DecimalSI),
+					corev1.ResourceMemory: *resource.NewQuantity(2e10, resource.BinarySI),
+				},
+			},
+			expected: corev1.ResourceRequirements{
+				Limits: corev1.ResourceList{
+					// Ours: 50 * 1.2 = 60, Theirs: 200, Reduction: 70% > 25%, so limit to 25%: 200 * 0.75 = 150
+					corev1.ResourceCPU: *resource.NewQuantity(150, resource.DecimalSI),
+					// Ours: 1e9 * 1.2 = 1.2e9, Theirs: 3e10, Reduction: 96% > 25%, so limit to 25%: 3e10 * 0.75 = 2.25e10
+					corev1.ResourceMemory: *resource.NewQuantity(225e8, resource.BinarySI),
+				},
+				Requests: corev1.ResourceList{
+					// Ours: 25 * 1.2 = 30, Theirs: 100, Reduction: 70% > 25%, so limit to 25%: 100 * 0.75 = 75
+					corev1.ResourceCPU: *resource.NewQuantity(75, resource.DecimalSI),
+					// Ours: 5e9 * 1.2 = 6e9, Theirs: 2e10, Reduction: 70% > 25%, so limit to 25%: 2e10 * 0.75 = 1.5e10
+					corev1.ResourceMemory: *resource.NewQuantity(15e9, resource.BinarySI),
+				},
+			},
+		},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			useOursIfLarger(&testCase.ours, &testCase.theirs, "test", "build", &defaultReporter, logrus.WithField("test", testCase.name))
+			applyRecommendationsBasedOnRecentData(&testCase.ours, &testCase.theirs, "test", "build", &defaultReporter, logrus.WithField("test", testCase.name))
 			if diff := cmp.Diff(testCase.theirs, testCase.expected); diff != "" {
 				t.Errorf("%s: got incorrect resources after mutation: %v", testCase.name, diff)
 			}
@@ -814,7 +856,7 @@ func TestUseOursIsLarger_ReporterReports(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			useOursIfLarger(&tc.ours, &tc.theirs, "test", "build", &tc.reporter, logrus.WithField("test", tc.name))
+			applyRecommendationsBasedOnRecentData(&tc.ours, &tc.theirs, "test", "build", &tc.reporter, logrus.WithField("test", tc.name))
 
 			if diff := cmp.Diff(tc.reporter.called, tc.expected); diff != "" {
 				t.Errorf("actual and expected reporter states don't match, : %v", diff)