From 7279ae52939c2b9e26e93c7534148f92ecb36fa6 Mon Sep 17 00:00:00 2001 From: Simon Lauger Date: Sun, 6 Sep 2026 11:41:59 +0200 Subject: [PATCH] feat(controller): change replicas through the scale subresource A full-object update is the wrong tool for a replica change on a workload something else also manages. A HorizontalPodAutoscaler writes through /scale, so a narrow write competes with it cleanly, whereas a full update replays the entire spec on a conflict retry. Deployment and StatefulSet now go through client.SubResource("scale"). CronJob is unchanged: it has no scale subresource and spec.suspend remains the mechanism. The two branches were near-identical, so they collapse into one helper. The annotations are written before the scale call on purpose: if the scale then fails the workload is still running and the next evaluation retries it, whereas the other order could leave a stopped workload with no record of its previous replica count, and the zero-replica guard would stop the operator ever revisiting it. Adds deployments/scale and statefulsets/scale with get and update. Narrow and explicit; the generic variant was considered and rejected, with the reasoning recorded on the issue. Closes #68 Signed-off-by: Simon Lauger --- .../templates/clusterrole.yaml | 5 + config/rbac/role.yaml | 8 + .../controller/crashlooppolicy_controller.go | 1 + .../crashlooppolicy_controller_test.go | 78 ++++++++ internal/controller/helpers.go | 171 +++++++++++------- internal/controller/testutil_test.go | 21 +++ 6 files changed, 214 insertions(+), 70 deletions(-) diff --git a/charts/crashloop-operator/templates/clusterrole.yaml b/charts/crashloop-operator/templates/clusterrole.yaml index 85f97cc..84a1b4d 100644 --- a/charts/crashloop-operator/templates/clusterrole.yaml +++ b/charts/crashloop-operator/templates/clusterrole.yaml @@ -57,6 +57,11 @@ rules: - apiGroups: ["apps"] resources: ["statefulsets"] verbs: ["get", "list", "watch", "update", "patch"] + # Replica changes go through the scale subresource so they do not compete + # with anything else managing the count, such as a HorizontalPodAutoscaler. + - apiGroups: ["apps"] + resources: ["deployments/scale", "statefulsets/scale"] + verbs: ["get", "update"] - apiGroups: ["apps"] resources: ["replicasets"] verbs: ["get", "list", "watch"] diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 2a80e3a..6e9b078 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -32,6 +32,14 @@ rules: - patch - update - watch +- apiGroups: + - apps + resources: + - deployments/scale + - statefulsets/scale + verbs: + - get + - update - apiGroups: - apps resources: diff --git a/internal/controller/crashlooppolicy_controller.go b/internal/controller/crashlooppolicy_controller.go index aa74888..1d6ca04 100644 --- a/internal/controller/crashlooppolicy_controller.go +++ b/internal/controller/crashlooppolicy_controller.go @@ -37,6 +37,7 @@ type CrashLoopPolicyReconciler struct { // +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;update;patch // +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;update;patch +// +kubebuilder:rbac:groups=apps,resources=deployments/scale;statefulsets/scale,verbs=get;update // +kubebuilder:rbac:groups=apps,resources=replicasets,verbs=get;list;watch // +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch // +kubebuilder:rbac:groups=batch,resources=cronjobs,verbs=get;list;watch;update;patch diff --git a/internal/controller/crashlooppolicy_controller_test.go b/internal/controller/crashlooppolicy_controller_test.go index b7a4bea..267944c 100644 --- a/internal/controller/crashlooppolicy_controller_test.go +++ b/internal/controller/crashlooppolicy_controller_test.go @@ -1415,3 +1415,81 @@ func TestReconcile_IgnoresRestartLoopWithoutOptIn(t *testing.T) { t.Error("expected no action without watchTerminationReasons set") } } + +func TestScaleWorkloadToZero_RecordsBeforeScaling(t *testing.T) { + // The annotations must be written before the replica change. If the scale + // call then fails the workload is still running and the next evaluation + // retries it; in the other order a failed annotation write would leave a + // stopped workload with no record of its previous replica count, and the + // zero-replica guard would stop the operator ever revisiting it. + deploy := newDeployment("my-app", testNamespace, 4) + c := &scaleFailingClient{Client: setupTestClient(deploy)} + key := types.NamespacedName{Name: "my-app", Namespace: testNamespace} + + acted, err := scaleWorkloadToZero(testCtx(), c, &appsv1.Deployment{}, key, + "because", "policy-a", "2026-01-01T00:00:00Z", false) + if err == nil { + t.Fatal("expected the injected scale failure to surface") + } + if acted { + t.Error("expected acted=false when the scale call failed") + } + + updated := &appsv1.Deployment{} + if err := c.Get(testCtx(), key, updated); err != nil { + t.Fatalf("failed to get deployment: %v", err) + } + if updated.Spec.Replicas == nil || *updated.Spec.Replicas != 4 { + t.Error("expected the workload to still be running after a failed scale") + } + if got := updated.Annotations[AnnotationPreviousReplicas]; got != "4" { + t.Errorf("expected the previous replica count to be recorded first, got %q", got) + } +} + +func TestScaleWorkloadToZero_SkipsAnAlreadyStoppedWorkload(t *testing.T) { + deploy := newDeployment("my-app", testNamespace, 0) + c := setupTestClient(deploy) + key := types.NamespacedName{Name: "my-app", Namespace: testNamespace} + + acted, err := scaleWorkloadToZero(testCtx(), c, &appsv1.Deployment{}, key, + "because", "policy-a", "2026-01-01T00:00:00Z", false) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if acted { + t.Error("expected no action on a workload already at zero") + } + updated := &appsv1.Deployment{} + if err := c.Get(testCtx(), key, updated); err != nil { + t.Fatalf("failed to get deployment: %v", err) + } + if _, ok := updated.Annotations[AnnotationScaledDownBy]; ok { + t.Error("expected no annotations on a workload the operator did not act on") + } +} + +func TestScaleWorkloadToZero_DryRunTouchesNothing(t *testing.T) { + deploy := newDeployment("my-app", testNamespace, 3) + c := setupTestClient(deploy) + key := types.NamespacedName{Name: "my-app", Namespace: testNamespace} + + acted, err := scaleWorkloadToZero(testCtx(), c, &appsv1.Deployment{}, key, + "because", "policy-a", "2026-01-01T00:00:00Z", true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !acted { + t.Error("expected dry run to report that it would have acted") + } + updated := &appsv1.Deployment{} + if err := c.Get(testCtx(), key, updated); err != nil { + t.Fatalf("failed to get deployment: %v", err) + } + if updated.Spec.Replicas == nil || *updated.Spec.Replicas != 3 { + t.Error("dry run must not change replicas") + } + if len(updated.Annotations) != 0 { + t.Errorf("dry run must not write annotations, got %v", updated.Annotations) + } +} diff --git a/internal/controller/helpers.go b/internal/controller/helpers.go index 88586f6..cfb8c80 100644 --- a/internal/controller/helpers.go +++ b/internal/controller/helpers.go @@ -7,6 +7,7 @@ import ( "time" appsv1 "k8s.io/api/apps/v1" + autoscalingv1 "k8s.io/api/autoscaling/v1" batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/equality" @@ -444,6 +445,104 @@ func allReplicasFailing(ctx context.Context, c client.Client, owner *ownerWorklo return false, nil } +// currentReplicas reads spec.replicas from a Deployment or StatefulSet. +func currentReplicas(obj client.Object) (int32, bool) { + switch o := obj.(type) { + case *appsv1.Deployment: + if o.Spec.Replicas == nil { + return 1, true + } + return *o.Spec.Replicas, true + case *appsv1.StatefulSet: + if o.Spec.Replicas == nil { + return 1, true + } + return *o.Spec.Replicas, true + default: + return 0, false + } +} + +// scaleWorkloadToZero records why the workload is being stopped and then sets +// its replica count to zero through the scale subresource. +// +// The replica change goes through /scale rather than a full object update +// because something else may also be managing the count. A HorizontalPodAutoscaler +// writes through the same subresource, so a narrow write competes with it +// cleanly, whereas a full-object update replays the entire spec on a conflict +// retry. +// +// The annotations are written first on purpose. If the scale call then fails, +// the workload is still running and the next evaluation retries it, so the +// mistake corrects itself. In the other order a failed annotation write would +// leave a workload stopped with no record of its previous replica count, and +// the guard above would stop the operator ever revisiting it. +func scaleWorkloadToZero( + ctx context.Context, + c client.Client, + obj client.Object, + key types.NamespacedName, + reason, policyName, now string, + dryRun bool, +) (bool, error) { + if err := c.Get(ctx, key, obj); err != nil { + return false, err + } + replicas, ok := currentReplicas(obj) + if !ok { + return false, fmt.Errorf("unsupported workload type %T", obj) + } + if replicas == 0 { + return false, nil + } + if dryRun { + return true, nil + } + + var prevReplicas int32 + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := c.Get(ctx, key, obj); err != nil { + return err + } + current, _ := currentReplicas(obj) + if current == 0 { + return nil + } + prevReplicas = current + + annotations := obj.GetAnnotations() + if annotations == nil { + annotations = make(map[string]string) + } + annotations[AnnotationScaledDownReason] = reason + annotations[AnnotationScaledDownAt] = now + annotations[AnnotationScaledDownBy] = policyName + annotations[AnnotationPreviousReplicas] = fmt.Sprintf("%d", prevReplicas) + obj.SetAnnotations(annotations) + return c.Update(ctx, obj) + }); err != nil { + return false, err + } + if prevReplicas == 0 { + return false, nil + } + + scale := &autoscalingv1.Scale{} + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := c.SubResource("scale").Get(ctx, obj, scale); err != nil { + return err + } + if scale.Spec.Replicas == 0 { + return nil + } + scale.Spec.Replicas = 0 + return c.SubResource("scale").Update(ctx, obj, client.WithSubResourceBody(scale)) + }); err != nil { + return false, err + } + return true, nil +} + // scaleDownWorkload scales a workload to zero or suspends it. // It uses RetryOnConflict to handle concurrent updates safely. func scaleDownWorkload(ctx context.Context, c client.Client, owner *ownerWorkload, reason, policyName string, dryRun bool) (bool, error) { @@ -452,78 +551,10 @@ func scaleDownWorkload(ctx context.Context, c client.Client, owner *ownerWorkloa switch owner.Kind { case "Deployment": - deploy := &appsv1.Deployment{} - if err := c.Get(ctx, key, deploy); err != nil { - return false, err - } - if deploy.Spec.Replicas != nil && *deploy.Spec.Replicas == 0 { - return false, nil - } - if dryRun { - return true, nil - } - err := retry.RetryOnConflict(retry.DefaultRetry, func() error { - if err := c.Get(ctx, key, deploy); err != nil { - return err - } - prevReplicas := int32(1) - if deploy.Spec.Replicas != nil { - prevReplicas = *deploy.Spec.Replicas - } - if prevReplicas == 0 { - return nil - } - deploy.Spec.Replicas = new(int32(0)) - if deploy.Annotations == nil { - deploy.Annotations = make(map[string]string) - } - deploy.Annotations[AnnotationScaledDownReason] = reason - deploy.Annotations[AnnotationScaledDownAt] = now - deploy.Annotations[AnnotationScaledDownBy] = policyName - deploy.Annotations[AnnotationPreviousReplicas] = fmt.Sprintf("%d", prevReplicas) - return c.Update(ctx, deploy) - }) - if err != nil { - return false, err - } - return true, nil + return scaleWorkloadToZero(ctx, c, &appsv1.Deployment{}, key, reason, policyName, now, dryRun) case "StatefulSet": - sts := &appsv1.StatefulSet{} - if err := c.Get(ctx, key, sts); err != nil { - return false, err - } - if sts.Spec.Replicas != nil && *sts.Spec.Replicas == 0 { - return false, nil - } - if dryRun { - return true, nil - } - err := retry.RetryOnConflict(retry.DefaultRetry, func() error { - if err := c.Get(ctx, key, sts); err != nil { - return err - } - prevReplicas := int32(1) - if sts.Spec.Replicas != nil { - prevReplicas = *sts.Spec.Replicas - } - if prevReplicas == 0 { - return nil - } - sts.Spec.Replicas = new(int32(0)) - if sts.Annotations == nil { - sts.Annotations = make(map[string]string) - } - sts.Annotations[AnnotationScaledDownReason] = reason - sts.Annotations[AnnotationScaledDownAt] = now - sts.Annotations[AnnotationScaledDownBy] = policyName - sts.Annotations[AnnotationPreviousReplicas] = fmt.Sprintf("%d", prevReplicas) - return c.Update(ctx, sts) - }) - if err != nil { - return false, err - } - return true, nil + return scaleWorkloadToZero(ctx, c, &appsv1.StatefulSet{}, key, reason, policyName, now, dryRun) case "CronJob": cj := &batchv1.CronJob{} diff --git a/internal/controller/testutil_test.go b/internal/controller/testutil_test.go index 617c348..792314d 100644 --- a/internal/controller/testutil_test.go +++ b/internal/controller/testutil_test.go @@ -382,3 +382,24 @@ func (f *failingOwnerClient) Get(ctx context.Context, key client.ObjectKey, obj } return f.Client.Get(ctx, key, obj, opts...) } + +// scaleFailingClient lets the annotation update succeed but fails the scale +// subresource write, so the ordering guarantee can be tested. +type scaleFailingClient struct { + client.Client +} + +func (f *scaleFailingClient) SubResource(subResource string) client.SubResourceClient { + if subResource == "scale" { + return &failingSubResourceClient{SubResourceClient: f.Client.SubResource(subResource)} + } + return f.Client.SubResource(subResource) +} + +type failingSubResourceClient struct { + client.SubResourceClient +} + +func (f *failingSubResourceClient) Update(ctx context.Context, obj client.Object, opts ...client.SubResourceUpdateOption) error { + return apierrors.NewInternalError(errors.New("simulated scale failure")) +}