From e9bedd3607a45a151b11901808e65f290dce14f3 Mon Sep 17 00:00:00 2001 From: Scott Holodak Date: Wed, 28 Oct 2026 02:42:23 -0400 Subject: [PATCH] Add PVC retention policy to WorkerPool WorkerPool StatefulSets were created without a persistentVolumeClaimRetentionPolicy, so Kubernetes retained every worker PVC when a pool was scaled down or deleted, leaving the claims and their backing disks behind indefinitely. Add an optional spec.persistentVolumeClaimRetentionPolicy with whenDeleted and whenScaled (Retain or Delete). The controller sets it on the StatefulSet at creation and updates existing StatefulSets when it changes. Leaving it unset keeps the current Retain behavior. Co-Authored-By: Claude Opus 5.5 --- api/v1alpha2/workerpool_types.go | 11 +++ api/v1alpha2/zz_generated.deepcopy.go | 6 ++ docs/reference.md | 2 + internal/controller/workerpool_controller.go | 35 +++++++- .../controller/workerpool_controller_test.go | 86 +++++++++++++++++++ .../kelos-crds/templates/workerpool-crd.yaml | 31 +++++++ internal/manifests/install-crd.yaml | 31 +++++++ test/integration/workerpool_test.go | 86 +++++++++++++++++++ 8 files changed, 284 insertions(+), 4 deletions(-) diff --git a/api/v1alpha2/workerpool_types.go b/api/v1alpha2/workerpool_types.go index 90585c05b..c2761bb07 100644 --- a/api/v1alpha2/workerpool_types.go +++ b/api/v1alpha2/workerpool_types.go @@ -17,6 +17,7 @@ limitations under the License. package v1alpha2 import ( + appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -74,6 +75,16 @@ type WorkerPoolSpec struct { // +kubebuilder:validation:Required // +kubebuilder:validation:XValidation:rule="self == oldSelf",message="volumeClaimTemplate is immutable" VolumeClaimTemplate corev1.PersistentVolumeClaimSpec `json:"volumeClaimTemplate"` + + // PersistentVolumeClaimRetentionPolicy controls whether the worker PVCs + // created from volumeClaimTemplate are deleted when the WorkerPool is + // deleted (whenDeleted) or scaled down (whenScaled). Each field accepts + // Retain or Delete. When unset, or when a field is omitted, PVCs are + // retained. + // +optional + // +kubebuilder:validation:XValidation:rule="!has(self.whenDeleted) || self.whenDeleted in ['Retain', 'Delete']",message="whenDeleted must be Retain or Delete" + // +kubebuilder:validation:XValidation:rule="!has(self.whenScaled) || self.whenScaled in ['Retain', 'Delete']",message="whenScaled must be Retain or Delete" + PersistentVolumeClaimRetentionPolicy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy `json:"persistentVolumeClaimRetentionPolicy,omitempty"` } // WorkerPoolStatus defines the observed state of WorkerPool. diff --git a/api/v1alpha2/zz_generated.deepcopy.go b/api/v1alpha2/zz_generated.deepcopy.go index bbfad5cb7..45b27284d 100644 --- a/api/v1alpha2/zz_generated.deepcopy.go +++ b/api/v1alpha2/zz_generated.deepcopy.go @@ -21,6 +21,7 @@ limitations under the License. package v1alpha2 import ( + appsv1 "k8s.io/api/apps/v1" batchv1 "k8s.io/api/batch/v1" "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -2634,6 +2635,11 @@ func (in *WorkerPoolSpec) DeepCopyInto(out *WorkerPoolSpec) { **out = **in } in.VolumeClaimTemplate.DeepCopyInto(&out.VolumeClaimTemplate) + if in.PersistentVolumeClaimRetentionPolicy != nil { + in, out := &in.PersistentVolumeClaimRetentionPolicy, &out.PersistentVolumeClaimRetentionPolicy + *out = new(appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerPoolSpec. diff --git a/docs/reference.md b/docs/reference.md index 8a901f546..e8d87e615 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -1134,6 +1134,8 @@ keep repository access without restarting (see [Workspace authentication](#works | `spec.worker.podOverrides` | Pod customization for worker pods | No | | `spec.replicas` | Number of persistent worker pods (defaults to 1) | No | | `spec.volumeClaimTemplate` | PersistentVolumeClaimSpec for each worker pod's storage | Yes | +| `spec.persistentVolumeClaimRetentionPolicy.whenDeleted` | `Retain` or `Delete`. Whether worker PVCs are deleted when the WorkerPool is deleted (defaults to `Retain`) | No | +| `spec.persistentVolumeClaimRetentionPolicy.whenScaled` | `Retain` or `Delete`. Whether the PVCs of removed worker pods are deleted when `spec.replicas` is reduced (defaults to `Retain`) | No | ## Workspace diff --git a/internal/controller/workerpool_controller.go b/internal/controller/workerpool_controller.go index 45f2654e2..01790d7a6 100644 --- a/internal/controller/workerpool_controller.go +++ b/internal/controller/workerpool_controller.go @@ -492,7 +492,7 @@ func (r *WorkerPoolReconciler) reconcileStatefulSet(ctx context.Context, pool *k return ctrl.Result{RequeueAfter: 5 * time.Second}, nil } - // Update StatefulSet if replicas or pod template changed + // Update StatefulSet if replicas, pod template, or PVC retention policy changed needsUpdate := false if sts.Spec.Replicas == nil || *sts.Spec.Replicas != ptr.Deref(pool.Spec.Replicas, 1) { sts.Spec.Replicas = pool.Spec.Replicas @@ -506,6 +506,11 @@ func (r *WorkerPoolReconciler) reconcileStatefulSet(ctx context.Context, pool *k needsUpdate = true } + if !pvcRetentionPolicyEqual(sts.Spec.PersistentVolumeClaimRetentionPolicy, desired.Spec.PersistentVolumeClaimRetentionPolicy) { + sts.Spec.PersistentVolumeClaimRetentionPolicy = desired.Spec.PersistentVolumeClaimRetentionPolicy + needsUpdate = true + } + if needsUpdate { if err := r.Update(ctx, &sts); err != nil { return ctrl.Result{}, fmt.Errorf("workerpool %s: updating StatefulSet: %w", pool.Name, err) @@ -1036,9 +1041,10 @@ func (r *WorkerPoolReconciler) buildStatefulSet(pool *kelos.WorkerPool, stsName, Labels: labels, }, Spec: appsv1.StatefulSetSpec{ - Replicas: pool.Spec.Replicas, - ServiceName: svcName, - PodManagementPolicy: parallel, + Replicas: pool.Spec.Replicas, + ServiceName: svcName, + PodManagementPolicy: parallel, + PersistentVolumeClaimRetentionPolicy: pool.Spec.PersistentVolumeClaimRetentionPolicy.DeepCopy(), Selector: &metav1.LabelSelector{ MatchLabels: labels, }, @@ -1889,6 +1895,27 @@ func podTemplateSpecEqual(a, b corev1.PodSpec) bool { apiequality.Semantic.DeepEqual(a.ImagePullSecrets, b.ImagePullSecrets) } +// pvcRetentionPolicyEqual reports whether two StatefulSet PVC retention +// policies are equivalent, treating a nil policy or an empty field as Retain +// to match the API server default. +func pvcRetentionPolicyEqual(a, b *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy) bool { + effective := func(p *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy) (appsv1.PersistentVolumeClaimRetentionPolicyType, appsv1.PersistentVolumeClaimRetentionPolicyType) { + whenDeleted, whenScaled := appsv1.RetainPersistentVolumeClaimRetentionPolicyType, appsv1.RetainPersistentVolumeClaimRetentionPolicyType + if p != nil { + if p.WhenDeleted != "" { + whenDeleted = p.WhenDeleted + } + if p.WhenScaled != "" { + whenScaled = p.WhenScaled + } + } + return whenDeleted, whenScaled + } + aDeleted, aScaled := effective(a) + bDeleted, bScaled := effective(b) + return aDeleted == bDeleted && aScaled == bScaled +} + func mergeReconcileResults(a, b ctrl.Result) ctrl.Result { a.Requeue = a.Requeue || b.Requeue if b.RequeueAfter != 0 && (a.RequeueAfter == 0 || b.RequeueAfter < a.RequeueAfter) { diff --git a/internal/controller/workerpool_controller_test.go b/internal/controller/workerpool_controller_test.go index 61720d95a..73030b5ff 100644 --- a/internal/controller/workerpool_controller_test.go +++ b/internal/controller/workerpool_controller_test.go @@ -409,6 +409,92 @@ func TestWorkerPoolReconciler_UpdatesReplicas(t *testing.T) { assert.Equal(t, int32(3), *sts.Spec.Replicas) } +func reconcileWorkerPoolStatefulSet(t *testing.T, objs ...client.Object) appsv1.StatefulSet { + t.Helper() + scheme := newWorkerPoolTestScheme() + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(&kelos.WorkerPool{}). + WithObjects(objs...). + Build() + + r := newWorkerPoolReconciler(cl, scheme) + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "my-pool", Namespace: "default"}, + }) + require.NoError(t, err) + + var sts appsv1.StatefulSet + require.NoError(t, cl.Get(context.Background(), types.NamespacedName{Name: "wp-my-pool", Namespace: "default"}, &sts)) + return sts +} + +func TestWorkerPoolReconciler_PVCRetentionPolicyUnsetOnCreate(t *testing.T) { + pool := newTestWorkerPool("my-pool", "default", 1) + + sts := reconcileWorkerPoolStatefulSet(t, pool, newTestWorkspace("default")) + + assert.Nil(t, sts.Spec.PersistentVolumeClaimRetentionPolicy) +} + +func TestWorkerPoolReconciler_PVCRetentionPolicyPropagatedOnCreate(t *testing.T) { + pool := newTestWorkerPool("my-pool", "default", 1) + pool.Spec.PersistentVolumeClaimRetentionPolicy = &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + WhenScaled: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + } + + sts := reconcileWorkerPoolStatefulSet(t, pool, newTestWorkspace("default")) + + assert.Equal(t, &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + WhenScaled: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + }, sts.Spec.PersistentVolumeClaimRetentionPolicy) +} + +func TestWorkerPoolReconciler_PVCRetentionPolicyUpdatesExistingStatefulSet(t *testing.T) { + pool := newTestWorkerPool("my-pool", "default", 1) + pool.Spec.PersistentVolumeClaimRetentionPolicy = &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + WhenScaled: appsv1.RetainPersistentVolumeClaimRetentionPolicyType, + } + + // Build the StatefulSet as the controller would for the same pool, but + // with the API server's Retain/Retain default, so the retention policy is + // the only difference. + scheme := newWorkerPoolTestScheme() + r := newWorkerPoolReconciler(fake.NewClientBuilder().WithScheme(scheme).Build(), scheme) + ws := newTestWorkspace("default") + existingSTS, err := r.buildStatefulSet(pool, "wp-my-pool", "wp-my-pool", &ws.Spec, nil) + require.NoError(t, err) + existingSTS.Spec.PersistentVolumeClaimRetentionPolicy = &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: appsv1.RetainPersistentVolumeClaimRetentionPolicyType, + WhenScaled: appsv1.RetainPersistentVolumeClaimRetentionPolicyType, + } + + sts := reconcileWorkerPoolStatefulSet(t, pool, ws, existingSTS) + + assert.Equal(t, &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + WhenScaled: appsv1.RetainPersistentVolumeClaimRetentionPolicyType, + }, sts.Spec.PersistentVolumeClaimRetentionPolicy) +} + +func TestPVCRetentionPolicyEqual_TreatsUnsetAsRetain(t *testing.T) { + retainAll := &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: appsv1.RetainPersistentVolumeClaimRetentionPolicyType, + WhenScaled: appsv1.RetainPersistentVolumeClaimRetentionPolicyType, + } + deleteOnScale := &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenScaled: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + } + + assert.True(t, pvcRetentionPolicyEqual(nil, retainAll)) + assert.True(t, pvcRetentionPolicyEqual(&appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{}, retainAll)) + assert.False(t, pvcRetentionPolicyEqual(nil, deleteOnScale)) + assert.False(t, pvcRetentionPolicyEqual(retainAll, deleteOnScale)) +} + func TestWorkerPoolReconciler_StatusPhases(t *testing.T) { tests := []struct { name string diff --git a/internal/manifests/charts/kelos/charts/kelos-crds/templates/workerpool-crd.yaml b/internal/manifests/charts/kelos/charts/kelos-crds/templates/workerpool-crd.yaml index c129aa090..47d530e03 100644 --- a/internal/manifests/charts/kelos/charts/kelos-crds/templates/workerpool-crd.yaml +++ b/internal/manifests/charts/kelos/charts/kelos-crds/templates/workerpool-crd.yaml @@ -66,6 +66,37 @@ spec: spec: description: WorkerPoolSpec defines the desired state of WorkerPool. properties: + persistentVolumeClaimRetentionPolicy: + description: |- + PersistentVolumeClaimRetentionPolicy controls whether the worker PVCs + created from volumeClaimTemplate are deleted when the WorkerPool is + deleted (whenDeleted) or scaled down (whenScaled). Each field accepts + Retain or Delete. When unset, or when a field is omitted, PVCs are + retained. + properties: + whenDeleted: + description: |- + WhenDeleted specifies what happens to PVCs created from StatefulSet + VolumeClaimTemplates when the StatefulSet is deleted. The default policy + of `Retain` causes PVCs to not be affected by StatefulSet deletion. The + `Delete` policy causes those PVCs to be deleted. + type: string + whenScaled: + description: |- + WhenScaled specifies what happens to PVCs created from StatefulSet + VolumeClaimTemplates when the StatefulSet is scaled down. The default + policy of `Retain` causes PVCs to not be affected by a scaledown. The + `Delete` policy causes the associated PVCs for any excess pods above + the replica count to be deleted. + type: string + type: object + x-kubernetes-validations: + - message: whenDeleted must be Retain or Delete + rule: '!has(self.whenDeleted) || self.whenDeleted in [''Retain'', + ''Delete'']' + - message: whenScaled must be Retain or Delete + rule: '!has(self.whenScaled) || self.whenScaled in [''Retain'', + ''Delete'']' replicas: default: 1 description: |- diff --git a/internal/manifests/install-crd.yaml b/internal/manifests/install-crd.yaml index c72743955..7d5d463a5 100644 --- a/internal/manifests/install-crd.yaml +++ b/internal/manifests/install-crd.yaml @@ -69017,6 +69017,37 @@ spec: spec: description: WorkerPoolSpec defines the desired state of WorkerPool. properties: + persistentVolumeClaimRetentionPolicy: + description: |- + PersistentVolumeClaimRetentionPolicy controls whether the worker PVCs + created from volumeClaimTemplate are deleted when the WorkerPool is + deleted (whenDeleted) or scaled down (whenScaled). Each field accepts + Retain or Delete. When unset, or when a field is omitted, PVCs are + retained. + properties: + whenDeleted: + description: |- + WhenDeleted specifies what happens to PVCs created from StatefulSet + VolumeClaimTemplates when the StatefulSet is deleted. The default policy + of `Retain` causes PVCs to not be affected by StatefulSet deletion. The + `Delete` policy causes those PVCs to be deleted. + type: string + whenScaled: + description: |- + WhenScaled specifies what happens to PVCs created from StatefulSet + VolumeClaimTemplates when the StatefulSet is scaled down. The default + policy of `Retain` causes PVCs to not be affected by a scaledown. The + `Delete` policy causes the associated PVCs for any excess pods above + the replica count to be deleted. + type: string + type: object + x-kubernetes-validations: + - message: whenDeleted must be Retain or Delete + rule: '!has(self.whenDeleted) || self.whenDeleted in [''Retain'', + ''Delete'']' + - message: whenScaled must be Retain or Delete + rule: '!has(self.whenScaled) || self.whenScaled in [''Retain'', + ''Delete'']' replicas: default: 1 description: |- diff --git a/test/integration/workerpool_test.go b/test/integration/workerpool_test.go index a2399ab71..b553e27a4 100644 --- a/test/integration/workerpool_test.go +++ b/test/integration/workerpool_test.go @@ -116,4 +116,90 @@ var _ = Describe("WorkerPool Controller", func() { Expect(updatedPool.Status.Phase).To(Equal(kelos.WorkerPoolPhasePending)) }) }) + + Context("When setting a PVC retention policy", func() { + newPool := func(name, namespace string, policy *appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy) *kelos.WorkerPool { + return &kelos.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + }, + Spec: kelos.WorkerPoolSpec{ + Worker: kelos.WorkerSpec{ + Type: "claude-code", + Credentials: &kelos.Credentials{ + Type: kelos.CredentialTypeNone, + }, + WorkspaceRef: &kelos.WorkspaceReference{ + Name: "pool-workspace", + }, + }, + Replicas: ptr.To(int32(1)), + VolumeClaimTemplate: corev1.PersistentVolumeClaimSpec{ + AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce}, + Resources: corev1.VolumeResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceStorage: resource.MustParse("1Gi"), + }, + }, + }, + PersistentVolumeClaimRetentionPolicy: policy, + }, + } + } + + It("Should propagate the policy to the StatefulSet and update it when changed", func() { + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "test-workerpool-pvc-retention"}} + Expect(k8sClient.Create(ctx, ns)).Should(Succeed()) + + ws := &kelos.Workspace{ + ObjectMeta: metav1.ObjectMeta{Name: "pool-workspace", Namespace: ns.Name}, + Spec: kelos.WorkspaceSpec{Repo: "https://github.com/example/repo.git", Ref: "main"}, + } + Expect(k8sClient.Create(ctx, ws)).Should(Succeed()) + + By("Creating a WorkerPool that deletes PVCs on scale down") + pool := newPool("retention-pool", ns.Name, &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenScaled: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + }) + Expect(k8sClient.Create(ctx, pool)).Should(Succeed()) + + stsKey := types.NamespacedName{Name: "wp-retention-pool", Namespace: ns.Name} + Eventually(func(g Gomega) { + sts := &appsv1.StatefulSet{} + g.Expect(k8sClient.Get(ctx, stsKey, sts)).To(Succeed()) + g.Expect(sts.Spec.PersistentVolumeClaimRetentionPolicy).NotTo(BeNil()) + g.Expect(sts.Spec.PersistentVolumeClaimRetentionPolicy.WhenScaled).To(Equal(appsv1.DeletePersistentVolumeClaimRetentionPolicyType)) + }, timeout, interval).Should(Succeed()) + + By("Changing the WorkerPool to also delete PVCs on deletion") + poolKey := types.NamespacedName{Name: "retention-pool", Namespace: ns.Name} + Eventually(func(g Gomega) { + latest := &kelos.WorkerPool{} + g.Expect(k8sClient.Get(ctx, poolKey, latest)).To(Succeed()) + latest.Spec.PersistentVolumeClaimRetentionPolicy = &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + WhenScaled: appsv1.DeletePersistentVolumeClaimRetentionPolicyType, + } + g.Expect(k8sClient.Update(ctx, latest)).To(Succeed()) + }, timeout, interval).Should(Succeed()) + + Eventually(func(g Gomega) { + sts := &appsv1.StatefulSet{} + g.Expect(k8sClient.Get(ctx, stsKey, sts)).To(Succeed()) + g.Expect(sts.Spec.PersistentVolumeClaimRetentionPolicy).NotTo(BeNil()) + g.Expect(sts.Spec.PersistentVolumeClaimRetentionPolicy.WhenDeleted).To(Equal(appsv1.DeletePersistentVolumeClaimRetentionPolicyType)) + g.Expect(sts.Spec.PersistentVolumeClaimRetentionPolicy.WhenScaled).To(Equal(appsv1.DeletePersistentVolumeClaimRetentionPolicyType)) + }, timeout, interval).Should(Succeed()) + }) + + It("Should reject an unknown policy value", func() { + pool := newPool("invalid-retention-pool", "default", &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenScaled: "Archive", + }) + err := k8sClient.Create(ctx, pool) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("whenScaled must be Retain or Delete")) + }) + }) })