From d3248e120a19e9e1e02a07a5f3bc94e5ec6bd025 Mon Sep 17 00:00:00 2001 From: Morten Zwarenstein Date: Tue, 8 Sep 2026 11:22:21 +0200 Subject: [PATCH 1/4] test: added actual tests --- internal/controller/util_test.go | 156 +++++++ internal/controller/volume_controller_test.go | 409 +++++++++++++++++- 2 files changed, 562 insertions(+), 3 deletions(-) create mode 100644 internal/controller/util_test.go diff --git a/internal/controller/util_test.go b/internal/controller/util_test.go new file mode 100644 index 0000000..4501a93 --- /dev/null +++ b/internal/controller/util_test.go @@ -0,0 +1,156 @@ +/* +Copyright 2025. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/PDOK/volume-operator/internal/config" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +func helperReplicaSet(name, revision, suffix string, replicas *int32, available int32) appsv1.ReplicaSet { + return appsv1.ReplicaSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: testNamespace, + Annotations: map[string]string{ + config.RevisionAnnotation: revision, + config.ResourceSuffixAnnotation: suffix, + }, + }, + Spec: appsv1.ReplicaSetSpec{Replicas: replicas}, + Status: appsv1.ReplicaSetStatus{AvailableReplicas: available}, + } +} + +func int32Ptr(i int32) *int32 { return &i } + +var _ = Describe("Controller helpers", func() { + Describe("hasReplicas", func() { + It("is false when Replicas is nil", func() { + Expect(hasReplicas(helperReplicaSet("rs", "1", "s", nil, 0))).To(BeFalse()) + }) + + It("is false when Replicas is zero", func() { + Expect(hasReplicas(helperReplicaSet("rs", "1", "s", int32Ptr(0), 0))).To(BeFalse()) + }) + + It("is true when Replicas is greater than zero", func() { + Expect(hasReplicas(helperReplicaSet("rs", "1", "s", int32Ptr(2), 0))).To(BeTrue()) + }) + }) + + Describe("currentReplicaSetIsAvailable", func() { + It("is false when the deployment revision is empty", func() { + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("rs", "1", "s", int32Ptr(1), 1), + }} + Expect(currentReplicaSetIsAvailable(list, "")).To(BeFalse()) + }) + + It("is false when no ReplicaSet carries the deployment revision", func() { + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("rs", "1", "s", int32Ptr(1), 1), + }} + Expect(currentReplicaSetIsAvailable(list, "2")).To(BeFalse()) + }) + + It("is false when the matching ReplicaSet has no available replicas", func() { + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("rs", "2", "s", int32Ptr(1), 0), + }} + Expect(currentReplicaSetIsAvailable(list, "2")).To(BeFalse()) + }) + + It("is true when the matching ReplicaSet has available replicas", func() { + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("rs", "2", "s", int32Ptr(1), 3), + }} + Expect(currentReplicaSetIsAvailable(list, "2")).To(BeTrue()) + }) + }) + + Describe("resourceIsUsedByOtherReplicaSet", func() { + current := helperReplicaSet("current", "1", "shared", int32Ptr(0), 0) + + It("ignores the ReplicaSet with the same name", func() { + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("current", "1", "shared", int32Ptr(5), 5), + }} + Expect(resourceIsUsedByOtherReplicaSet(list, current)).To(BeFalse()) + }) + + It("is false when the shared suffix is empty", func() { + empty := helperReplicaSet("current", "1", "", int32Ptr(0), 0) + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("other", "1", "", int32Ptr(5), 5), + }} + Expect(resourceIsUsedByOtherReplicaSet(list, empty)).To(BeFalse()) + }) + + It("is false when the other ReplicaSet with the same suffix has no replicas", func() { + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("other", "2", "shared", int32Ptr(0), 0), + }} + Expect(resourceIsUsedByOtherReplicaSet(list, current)).To(BeFalse()) + }) + + It("is true when another ReplicaSet with the same suffix still has replicas", func() { + list := appsv1.ReplicaSetList{Items: []appsv1.ReplicaSet{ + helperReplicaSet("other", "2", "shared", int32Ptr(1), 1), + }} + Expect(resourceIsUsedByOtherReplicaSet(list, current)).To(BeTrue()) + }) + }) + + Describe("pvcIsBeingPopulated", func() { + var ( + k8sClient client.Client + conf config.Config + ) + + BeforeEach(func() { + scheme := runtime.NewScheme() + Expect(corev1.AddToScheme(scheme)).To(Succeed()) + k8sClient = fake.NewClientBuilder().WithScheme(scheme).Build() + conf = config.Config{ResourceName: "pvc", ResourceNamespace: testNamespace} + }) + + It("is false when the PVC does not exist", func() { + Expect(pvcIsBeingPopulated(context.Background(), k8sClient, conf)).To(BeFalse()) + }) + + It("is true when the PVC is pending", func() { + Expect(k8sClient.Create(context.Background(), newPVC("pvc", corev1.ClaimPending))).To(Succeed()) + Expect(pvcIsBeingPopulated(context.Background(), k8sClient, conf)).To(BeTrue()) + }) + + It("is false when the PVC is bound", func() { + Expect(k8sClient.Create(context.Background(), newPVC("pvc", corev1.ClaimBound))).To(Succeed()) + Expect(pvcIsBeingPopulated(context.Background(), k8sClient, conf)).To(BeFalse()) + }) + }) +}) diff --git a/internal/controller/volume_controller_test.go b/internal/controller/volume_controller_test.go index b5371f5..2773800 100644 --- a/internal/controller/volume_controller_test.go +++ b/internal/controller/volume_controller_test.go @@ -27,6 +27,7 @@ import ( avp "github.com/pdok/azure-volume-populator/api/v1alpha1" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -36,9 +37,11 @@ import ( ) const ( - testNamespace = "default" - blobPrefixAnnotation = "volume-operator.pdok.nl/blob-prefix" - volumePathAnnotation = "volume-operator.pdok.nl/volume-path" + testNamespace = "default" + blobPrefixAnnotation = "volume-operator.pdok.nl/blob-prefix" + volumePathAnnotation = "volume-operator.pdok.nl/volume-path" + storageClassAnnotation = "volume-operator.pdok.nl/storage-class" + storageCapacityAnnotation = "volume-operator.pdok.nl/storage-capacity" ) func newDeployment(name, revision string, annotations map[string]string) *appsv1.Deployment { @@ -99,6 +102,22 @@ func newReplicaSet(name string, owner *appsv1.Deployment, revision string, repli } } +func newAVP(name string) *avp.AzureVolumePopulator { + return &avp.AzureVolumePopulator{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNamespace}, + } +} + +func newPVC(name string, phase corev1.PersistentVolumeClaimPhase) *corev1.PersistentVolumeClaim { + pvc := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNamespace}, + } + if phase != "" { + pvc.Status.Phase = phase + } + return pvc +} + var _ = Describe("Volume Controller", func() { var ( ctx context.Context @@ -293,4 +312,388 @@ var _ = Describe("Volume Controller", func() { Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-old-deferred", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-old-deferred", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) }) + + Context("creating resources", func() { + It("uses the default storage class and capacity when the storage annotations are omitted", func() { + deployment := newDeployment("dep-defaults", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-defaults", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-defaults", deployment, "1", 1, 1, "suffix-defaults") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + pvc := &corev1.PersistentVolumeClaim{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-defaults", Namespace: testNamespace}, pvc)).To(Succeed()) + Expect(pvc.Spec.StorageClassName).NotTo(BeNil()) + Expect(*pvc.Spec.StorageClassName).To(Equal("managed-premium-zrs")) + request := pvc.Spec.Resources.Requests[corev1.ResourceStorage] + Expect(request.Cmp(resource.MustParse("1Gi"))).To(Equal(0)) + }) + + It("honours custom storage-class and storage-capacity annotations", func() { + deployment := newDeployment("dep-custom-storage", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-custom-storage", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + storageClassAnnotation: "fast-ssd", + storageCapacityAnnotation: "5Gi", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-custom-storage", deployment, "1", 1, 1, "suffix-custom-storage") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + pvc := &corev1.PersistentVolumeClaim{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-custom-storage", Namespace: testNamespace}, pvc)).To(Succeed()) + Expect(*pvc.Spec.StorageClassName).To(Equal("fast-ssd")) + request := pvc.Spec.Resources.Requests[corev1.ResourceStorage] + Expect(request.Cmp(resource.MustParse("5Gi"))).To(Equal(0)) + }) + + It("sets the PVC access mode and data source reference", func() { + deployment := newDeployment("dep-pvc-fields", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-pvc-fields", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-pvc-fields", deployment, "1", 1, 1, "suffix-pvc-fields") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + pvc := &corev1.PersistentVolumeClaim{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-pvc-fields", Namespace: testNamespace}, pvc)).To(Succeed()) + Expect(pvc.Spec.AccessModes).To(Equal([]corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce})) + Expect(pvc.Spec.DataSourceRef).NotTo(BeNil()) + Expect(pvc.Spec.DataSourceRef.APIGroup).NotTo(BeNil()) + Expect(*pvc.Spec.DataSourceRef.APIGroup).To(Equal("volume.pdok.nl")) + Expect(pvc.Spec.DataSourceRef.Kind).To(Equal("AzureVolumePopulator")) + Expect(pvc.Spec.DataSourceRef.Name).To(Equal("suffix-pvc-fields")) + }) + + It("sets the AVP spec from the Deployment annotations", func() { + deployment := newDeployment("dep-avp-fields", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-avp-fields", + blobPrefixAnnotation: "some/blob/prefix", + volumePathAnnotation: "/mnt/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-avp-fields", deployment, "1", 1, 1, "suffix-avp-fields") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + createdAvp := &avp.AzureVolumePopulator{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-avp-fields", Namespace: testNamespace}, createdAvp)).To(Succeed()) + Expect(createdAvp.Spec.BlobPrefix).To(Equal("some/blob/prefix")) + Expect(createdAvp.Spec.VolumePath).To(Equal("/mnt/data")) + Expect(createdAvp.Spec.BlobDownloadOptions).NotTo(BeNil()) + }) + + It("creates the PVC referencing an already-existing AVP", func() { + deployment := newDeployment("dep-partial", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-partial", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-partial", deployment, "1", 1, 1, "suffix-partial") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + Expect(k8sClient.Create(ctx, newAVP("suffix-partial"))).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + pvc := &corev1.PersistentVolumeClaim{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-partial", Namespace: testNamespace}, pvc)).To(Succeed()) + Expect(pvc.Spec.DataSourceRef).NotTo(BeNil()) + Expect(pvc.Spec.DataSourceRef.Name).To(Equal("suffix-partial")) + + avpList := &avp.AzureVolumePopulatorList{} + Expect(k8sClient.List(ctx, avpList, client.InNamespace(testNamespace))).To(Succeed()) + Expect(avpList.Items).To(HaveLen(1)) + }) + + It("returns an empty result without requeue on the happy path", func() { + deployment := newDeployment("dep-result", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-result", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-result", deployment, "1", 1, 1, "suffix-result") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + result, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(reconcile.Result{})) + }) + + It("returns an error when the owning Deployment does not exist", func() { + // Deployment is built to wire the owner reference but is never created. + deployment := newDeployment("dep-missing", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-missing-dep", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + + rs := newReplicaSet("rs-dangling-owner", deployment, "1", 1, 1, "suffix-missing-dep") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).To(HaveOccurred()) + }) + }) + + Context("updating resources", func() { + It("does not mutate an existing AVP when the Deployment's blob-prefix changes", func() { + deployment := newDeployment("dep-update-avp", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-update-avp", + blobPrefixAnnotation: "blob/v1", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-update-avp", deployment, "1", 1, 1, "suffix-update-avp") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + // Change the annotation on the Deployment (suffix and revision unchanged). + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: deployment.Name, Namespace: testNamespace}, deployment)).To(Succeed()) + deployment.Annotations[blobPrefixAnnotation] = "blob/v2" + Expect(k8sClient.Update(ctx, deployment)).To(Succeed()) + + _, err = reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + createdAvp := &avp.AzureVolumePopulator{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-update-avp", Namespace: testNamespace}, createdAvp)).To(Succeed()) + Expect(createdAvp.Spec.BlobPrefix).To(Equal("blob/v1")) + + avpList := &avp.AzureVolumePopulatorList{} + Expect(k8sClient.List(ctx, avpList, client.InNamespace(testNamespace))).To(Succeed()) + Expect(avpList.Items).To(HaveLen(1)) + }) + + It("does not mutate an existing PVC when the Deployment's storage annotations change", func() { + deployment := newDeployment("dep-update-pvc", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-update-pvc", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-update-pvc", deployment, "1", 1, 1, "suffix-update-pvc") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: deployment.Name, Namespace: testNamespace}, deployment)).To(Succeed()) + deployment.Annotations[storageCapacityAnnotation] = "10Gi" + deployment.Annotations[storageClassAnnotation] = "fast-ssd" + Expect(k8sClient.Update(ctx, deployment)).To(Succeed()) + + _, err = reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + pvc := &corev1.PersistentVolumeClaim{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-update-pvc", Namespace: testNamespace}, pvc)).To(Succeed()) + request := pvc.Spec.Resources.Requests[corev1.ResourceStorage] + Expect(request.Cmp(resource.MustParse("1Gi"))).To(Equal(0)) + Expect(*pvc.Spec.StorageClassName).To(Equal("managed-premium-zrs")) + }) + + It("provisions new resources and removes the old ones when the resource-suffix changes", func() { + deployment := newDeployment("dep-rollout", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-rollout-new", + blobPrefixAnnotation: "blob/v2", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-rollout-old", deployment, "1", 0, 0, "suffix-rollout-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + newRs := newReplicaSet("rs-rollout-new", deployment, "2", 1, 1, "suffix-rollout-new") + Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-rollout-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-rollout-old", ""))).To(Succeed()) + + _, err := reconcileRS(newRs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-rollout-new", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-rollout-new", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-rollout-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{}) + Expect(err).To(HaveOccurred()) + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-rollout-old", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{}) + Expect(err).To(HaveOccurred()) + }) + }) + + Context("deleting resources", func() { + It("keeps an old ReplicaSet's resources while another ReplicaSet still uses the same suffix", func() { + deployment := newDeployment("dep-shared", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-current", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + scaledDownRs := newReplicaSet("rs-shared-down", deployment, "1", 0, 0, "suffix-shared") + Expect(k8sClient.Create(ctx, scaledDownRs)).To(Succeed()) + scaledUpRs := newReplicaSet("rs-shared-up", deployment, "1", 2, 2, "suffix-shared") + Expect(k8sClient.Create(ctx, scaledUpRs)).To(Succeed()) + currentRs := newReplicaSet("rs-shared-current", deployment, "2", 1, 1, "suffix-current") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-shared"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-shared", ""))).To(Succeed()) + + _, err := reconcileRS(currentRs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + }) + + It("keeps old resources while the current PVC is still being populated", func() { + deployment := newDeployment("dep-populating", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-populating-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-populating-old", deployment, "1", 0, 0, "suffix-populating-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + newRs := newReplicaSet("rs-populating-new", deployment, "2", 1, 1, "suffix-populating-new") + Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-populating-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-populating-old", ""))).To(Succeed()) + // The current PVC already exists and is still pending (being populated). + Expect(k8sClient.Create(ctx, newPVC("suffix-populating-new", corev1.ClaimPending))).To(Succeed()) + + _, err := reconcileRS(newRs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-populating-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-populating-old", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + }) + + It("keeps an old ReplicaSet's resources while it still has replicas", func() { + deployment := newDeployment("dep-oldscaled", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-oldscaled-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-oldscaled-old", deployment, "1", 1, 1, "suffix-oldscaled-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + newRs := newReplicaSet("rs-oldscaled-new", deployment, "2", 1, 1, "suffix-oldscaled-new") + Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-oldscaled-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-oldscaled-old", ""))).To(Succeed()) + + _, err := reconcileRS(newRs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-oldscaled-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-oldscaled-old", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + }) + + It("deletes the resources of every scaled-down old ReplicaSet in one reconcile", func() { + deployment := newDeployment("dep-multi", "3", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-multi-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs1 := newReplicaSet("rs-multi-old1", deployment, "1", 0, 0, "suffix-multi-old1") + Expect(k8sClient.Create(ctx, oldRs1)).To(Succeed()) + oldRs2 := newReplicaSet("rs-multi-old2", deployment, "2", 0, 0, "suffix-multi-old2") + Expect(k8sClient.Create(ctx, oldRs2)).To(Succeed()) + newRs := newReplicaSet("rs-multi-new", deployment, "3", 1, 1, "suffix-multi-new") + Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) + + for _, name := range []string{"suffix-multi-old1", "suffix-multi-old2"} { + Expect(k8sClient.Create(ctx, newAVP(name))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC(name, ""))).To(Succeed()) + } + + _, err := reconcileRS(newRs.Name) + Expect(err).NotTo(HaveOccurred()) + + for _, name := range []string{"suffix-multi-old1", "suffix-multi-old2"} { + err = k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: testNamespace}, &avp.AzureVolumePopulator{}) + Expect(err).To(HaveOccurred()) + err = k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: testNamespace}, &corev1.PersistentVolumeClaim{}) + Expect(err).To(HaveOccurred()) + } + }) + + It("does not error when a scaled-down old ReplicaSet has no resources to delete", func() { + deployment := newDeployment("dep-noresources", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-noresources-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-noresources-old", deployment, "1", 0, 0, "suffix-noresources-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + newRs := newReplicaSet("rs-noresources-new", deployment, "2", 1, 1, "suffix-noresources-new") + Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) + + _, err := reconcileRS(newRs.Name) + Expect(err).NotTo(HaveOccurred()) + + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-noresources-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{}) + Expect(err).To(HaveOccurred()) + }) + + It("keeps the shared resources when a new ReplicaSet reuses the old resource-suffix", func() { + deployment := newDeployment("dep-samehash", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-samehash", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-samehash-old", deployment, "1", 0, 0, "suffix-samehash") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + newRs := newReplicaSet("rs-samehash-new", deployment, "2", 1, 1, "suffix-samehash") + Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) + + _, err := reconcileRS(newRs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-samehash", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-samehash", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + }) + }) }) From aec7b4c106a8d409978af7dfba69844063481a35 Mon Sep 17 00:00:00 2001 From: Morten Zwarenstein Date: Mon, 21 Sep 2026 08:42:14 +0200 Subject: [PATCH 2/4] feat: safer deletes based on available pods --- AGENTS.md | 19 +- internal/controller/volume_controller.go | 250 +++++++++++++++-------- 2 files changed, 185 insertions(+), 84 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index c8f4130..5ae23b7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,6 +31,7 @@ This file is the constitution for this project. It captures persistent principle 4. **volume-operator** hooks into the ReplicaSet lifecycle. For the active ReplicaSet (revision matches Deployment), it creates: - An **`AzureVolumePopulator` CR** (the AVP) - A **source PVC** with `DataSourceRef` pointing to the AVP — both named by the hash + - Both carry an **`ownerReference` to the ReplicaSet** (see "Cleanup on full deletion" below) 5. **azure-volume-populator** sees the AVP CR and spawns a **populator pod** that copies data from Azure Blob Storage into the source PVC. 6. The source PVC reaches **`Bound`** state. 7. Kubernetes detects that the ephemeral volume's `DataSource` PVC is now Bound and triggers native **volume cloning** — each pod gets its own ephemeral clone. @@ -41,9 +42,23 @@ This file is the constitution for this project. It captures persistent principle When a Deployment rolls out a new ReplicaSet (new revision), on every reconcile of any ReplicaSet in the set: - volume-operator deletes the **AVP and PVC** (not the ReplicaSet itself — ReplicaSet lifecycle belongs to Kubernetes/the Deployment controller, not this operator) for old ReplicaSets that are scaled to 0 replicas. - A resource is only deleted if no other ReplicaSet still references the same `resource-suffix` hash with `replicas > 0` (the hash can be shared across revisions when `blobPrefix`/`volumeMountPath`/`storageCapacity` are unchanged — see "Hash-based deduplication" below). -- **Safety gate:** cleanup is skipped entirely unless the ReplicaSet matching the Deployment's *current* revision has `AvailableReplicas > 0` — i.e. the new rollout must actually be serving before old storage is torn down (`currentReplicaSetIsAvailable` in `internal/controller/util.go`). This guards against deleting the old PVC/AVP while the new one is still failing to come up. +- **Cleanup runs on a reconcile of *any* ReplicaSet in the set**, not just the current-revision one — an old ReplicaSet scaling to 0 is itself the trigger that makes it eligible. Resource *provisioning* still only happens on the current-revision ReplicaSet (revision match + required annotations); the `else` branch falls through to cleanup. +- **Safety gate:** deletion is skipped unless the ReplicaSet matching the Deployment's *current* revision has `AvailableReplicas > 0` — i.e. the new rollout must actually be serving before old storage is torn down (`currentReplicaSetIsAvailable` in `internal/controller/util.go`). This guards against deleting the old PVC/AVP while the new one is still failing to come up. +- **Requeue while blocked:** when the safety gate holds cleanup back but eligible old resources still exist, `Reconcile` returns `RequeueAfter: cleanupRetryInterval` (1 min) so it converges once the new rollout serves, instead of waiting for an incidental later event. `cleanUpOldReplicaSets` returns `(requeue bool, err error)` for this; `replicaSetHasResources` checks whether the AVP/PVC still exist, and fails safe (reports `true`/still-pending) on any unexpected, non-`NotFound` `Get` error rather than silently treating it as "resource gone". +- **The Deployment's current hash is never deleted:** an old ReplicaSet whose `resource-suffix` equals the Deployment's current one is skipped, regardless of replica counts. `resourceIsUsedByOtherReplicaSet` only counts *other* ReplicaSets with `replicas > 0`, so without this a Deployment scaled to 0 (current ReplicaSet `Spec.Replicas == 0` while `AvailableReplicas` still lags at >0 as pods terminate) would have its live source PVC/AVP deleted and re-populated from blob on scale-up. It also keeps the requeue from firing forever for a scaled-to-0 Deployment. - The new and old PVCs are **independent** (different names if the hash changed), so there is no conflict during transition. -- Cleanup logic lives in `cleanUpOldReplicaSets` / `deleteResourcesForReplicaSet` in `internal/controller/volume_controller.go`, backed by `hasReplicas`, `resourceIsUsedByOtherReplicaSet`, and `currentReplicaSetIsAvailable` in `internal/controller/util.go`. +- Cleanup logic lives in `cleanUpOldReplicaSets` / `deleteResourcesForReplicaSet` / `replicaSetHasResources` in `internal/controller/volume_controller.go`, backed by `hasReplicas`, `resourceIsUsedByOtherReplicaSet`, and `currentReplicaSetIsAvailable` in `internal/controller/util.go`. + +### Cleanup on full deletion (owner references) + +The rollout cleanup above only fires on a live reconcile of *some* ReplicaSet in the set — but when the whole Deployment (and therefore every one of its ReplicaSets) is deleted outright (e.g. the owning OGCAPI CR itself is deleted), there is no ReplicaSet left to reconcile at all: `Reconcile` sees `Get` return `NotFound` and returns immediately, before `cleanUpOldReplicaSets` ever runs. This used to leave the AVP and PVC permanently orphaned (confirmed live against a test AKS cluster: deploy → delete the OGCAPI → Deployment/ReplicaSet gone in seconds, AVP/PVC left `Bound` indefinitely). + +The fix is a safety net independent of any reconcile: `createAvpIfNotExists` / `createPvcIfNotExists` set (or, if the resource already exists via hash-reuse, add to) an `ownerReference` from the AVP/PVC to the owning ReplicaSet, via `ensureOwnerReference` in `internal/controller/volume_controller.go`. Kubernetes' own garbage collector then deletes the AVP/PVC once **every** owning ReplicaSet is gone — no reconcile required. + +- Ownership is **additive, not exclusive**: because the resource-suffix hash can be shared across ReplicaSet revisions (see "Hash-based deduplication"), a reused AVP/PVC can carry owner references to more than one live ReplicaSet at once. `ensureOwnerReference` relies on `controllerutil.SetOwnerReference`, which upserts by (group, kind, name) rather than replacing the whole list, so an older revision's ownership survives until that specific ReplicaSet is actually gone. +- `ensureOwnerReference` re-reads the object and sends a **merge patch with an optimistic lock** (`MergeFromWithOptimisticLock`), only when the owner list actually changed, retrying on `Conflict` (`retry.RetryOnConflict`). The lock is required, not cosmetic: a merge patch replaces the whole `ownerReferences` list, so without a `resourceVersion` precondition a stale cache read could drop another ReplicaSet's ownership. Patching rather than `Update` also avoids writing back the whole object, which would prune CRD fields unknown to the vendored AVP types. The client needs `patch` on PVCs and AVPs (this repo has no RBAC markers and `config/rbac/role.yaml` has no PVC rules, so check the ClusterRole where it is actually defined). +- This is a *complement* to the explicit rollout cleanup above, not a replacement — the explicit path still deletes eagerly on scale-to-0 rather than waiting for a ReplicaSet object to eventually disappear (e.g. via `revisionHistoryLimit` trimming). +- `createAvpIfNotExists` / `createPvcIfNotExists` also tolerate `AlreadyExists` on `Create` as a non-error: two near-simultaneous reconciles for the same new ReplicaSet can both read a not-yet-cached AVP/PVC as `NotFound` (informer cache lag) and race to create it — the loser's `Create` legitimately fails with `AlreadyExists`, which isn't a real failure since the desired object already exists. Without this, that race surfaced as noisy `ERROR Reconciler error` log lines (with stack traces) on effectively every normal rollout — confirmed live during the same debugging session. ### Hash-based deduplication diff --git a/internal/controller/volume_controller.go b/internal/controller/volume_controller.go index 618351e..51caa82 100644 --- a/internal/controller/volume_controller.go +++ b/internal/controller/volume_controller.go @@ -19,6 +19,9 @@ package controller import ( "context" "errors" + "fmt" + "slices" + "time" "github.com/PDOK/volume-operator/internal/config" appsv1 "k8s.io/api/apps/v1" @@ -27,8 +30,10 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" "sigs.k8s.io/controller-runtime/pkg/predicate" k8serrors "k8s.io/apimachinery/pkg/api/errors" @@ -39,6 +44,8 @@ import ( smoothoperator "github.com/pdok/smooth-operator/pkg/util" ) +const cleanupRetryInterval = time.Minute + type VolumeReconciler struct { client.Client Scheme *runtime.Scheme @@ -71,116 +78,164 @@ func (r *VolumeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr return ctrl.Result{}, nil } - if !conf.RevisionsMatch() { + if conf.RevisionsMatch() { + if !conf.HasRequiredAnnotations() { + logger.Info("Missing required volume annotations") + return ctrl.Result{}, nil + } + + volumepopulator, err := createAvpIfNotExists(ctx, r.Client, r.Scheme, conf, &rs) + if err != nil { + logger.Error(err, "Failed to create AzureVolumePopulator") + return ctrl.Result{}, err + } + + if err = createPvcIfNotExists(ctx, r.Client, r.Scheme, conf, volumepopulator, &rs); err != nil { + logger.Error(err, "Failed to create PVC") + return ctrl.Result{}, err + } + } else { logger.Info( - "Revision mismatch, skipping reconciliation", + "Revision mismatch, skipping resource creation", "rsRevision", conf.ReplicaSetRevision, "deploymentRevision", conf.DeploymentRevision, ) - return ctrl.Result{}, nil } - if !conf.HasRequiredAnnotations() { - logger.Info("Missing required volume annotations") - return ctrl.Result{}, nil - } - - volumepopulator, err := createAvpIfNotExists(ctx, r.Client, conf) - if err != nil { - logger.Error(err, "Failed to create AzureVolumePopulator") - return ctrl.Result{}, err - } - - err = createPvcIfNotExists(ctx, r.Client, conf, volumepopulator) - if err != nil { - logger.Error(err, "Failed to create PVC") - return ctrl.Result{}, err - } - - err = cleanUpOldReplicaSets(ctx, r.Client, &rs, deployment, conf) + requeue, err := cleanUpOldReplicaSets(ctx, r.Client, &rs, deployment, conf) if err != nil { return ctrl.Result{}, client.IgnoreNotFound(err) } + if requeue { + return ctrl.Result{RequeueAfter: cleanupRetryInterval}, nil + } - return ctrl.Result{}, err + return ctrl.Result{}, nil } -func createPvcIfNotExists(ctx context.Context, obj client.Client, conf config.Config, populator avp.AzureVolumePopulator) error { +func createPvcIfNotExists(ctx context.Context, c client.Client, scheme *runtime.Scheme, conf config.Config, populator avp.AzureVolumePopulator, rs *appsv1.ReplicaSet) error { typeMeta := metav1.TypeMeta{ Kind: "PersistentVolumeClaim", } pvc := corev1.PersistentVolumeClaim{} - err := obj.Get(ctx, types.NamespacedName{ + err := c.Get(ctx, types.NamespacedName{ Name: conf.ResourceName, Namespace: conf.ResourceNamespace, }, &pvc) - if k8serrors.IsNotFound(err) { - pvc = corev1.PersistentVolumeClaim{ - ObjectMeta: metav1.ObjectMeta{ - Name: conf.ResourceName, - Namespace: conf.ResourceNamespace, + if err == nil { + return ensureOwnerReference(ctx, c, &pvc, rs, scheme) + } + if !k8serrors.IsNotFound(err) { + return err + } + + pvc = corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: conf.ResourceName, + Namespace: conf.ResourceNamespace, + }, + TypeMeta: typeMeta, + Spec: corev1.PersistentVolumeClaimSpec{ + StorageClassName: &conf.StorageClassName, + AccessModes: []corev1.PersistentVolumeAccessMode{ + corev1.ReadWriteOnce, }, - TypeMeta: typeMeta, - Spec: corev1.PersistentVolumeClaimSpec{ - StorageClassName: &conf.StorageClassName, - AccessModes: []corev1.PersistentVolumeAccessMode{ - corev1.ReadWriteOnce, - }, - Resources: corev1.VolumeResourceRequirements{ - Requests: corev1.ResourceList{ - corev1.ResourceStorage: resource.MustParse(conf.StorageCapacity), - }, - }, - DataSourceRef: &corev1.TypedObjectReference{ - APIGroup: new(avp.GroupVersion.Group), - Kind: populator.Kind, - Name: populator.Name, + Resources: corev1.VolumeResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceStorage: resource.MustParse(conf.StorageCapacity), }, }, - } + DataSourceRef: &corev1.TypedObjectReference{ + APIGroup: new(avp.GroupVersion.Group), + Kind: populator.Kind, + Name: populator.Name, + }, + }, + } - err = obj.Create(ctx, &pvc) - pvc.TypeMeta = typeMeta + if err := controllerutil.SetOwnerReference(rs, &pvc, scheme); err != nil { + return err + } + + if err := c.Create(ctx, &pvc); err != nil && !k8serrors.IsAlreadyExists(err) { + return err } - return err + return nil } -func createAvpIfNotExists(ctx context.Context, obj client.Client, conf config.Config) (avp.AzureVolumePopulator, error) { +func createAvpIfNotExists(ctx context.Context, c client.Client, scheme *runtime.Scheme, conf config.Config, rs *appsv1.ReplicaSet) (avp.AzureVolumePopulator, error) { typeMeta := metav1.TypeMeta{ Kind: "AzureVolumePopulator", APIVersion: avp.GroupVersion.Group, } populator := avp.AzureVolumePopulator{} - err := obj.Get(ctx, types.NamespacedName{ + err := c.Get(ctx, types.NamespacedName{ Name: conf.ResourceName, Namespace: conf.ResourceNamespace, }, &populator) populator.TypeMeta = typeMeta - if k8serrors.IsNotFound(err) { - populator = avp.AzureVolumePopulator{ - ObjectMeta: metav1.ObjectMeta{ - Name: conf.ResourceName, - Namespace: conf.ResourceNamespace, - }, - Spec: avp.AzureVolumePopulatorSpec{ - BlobPrefix: conf.BlobPrefix, - VolumePath: conf.VolumePath, - BlobDownloadOptions: &avp.BlobDownloadOptions{}, - }, - } - - err = obj.Create(ctx, &populator) + if err == nil { + err = ensureOwnerReference(ctx, c, &populator, rs, scheme) populator.TypeMeta = typeMeta return populator, err } + if !k8serrors.IsNotFound(err) { + return populator, err + } + + populator = avp.AzureVolumePopulator{ + ObjectMeta: metav1.ObjectMeta{ + Name: conf.ResourceName, + Namespace: conf.ResourceNamespace, + }, + Spec: avp.AzureVolumePopulatorSpec{ + BlobPrefix: conf.BlobPrefix, + VolumePath: conf.VolumePath, + BlobDownloadOptions: &avp.BlobDownloadOptions{}, + }, + } - return populator, err + if err := controllerutil.SetOwnerReference(rs, &populator, scheme); err != nil { + return populator, err + } + + if err := c.Create(ctx, &populator); err != nil && !k8serrors.IsAlreadyExists(err) { + return populator, err + } + populator.TypeMeta = typeMeta + + return populator, nil +} + +func ensureOwnerReference(ctx context.Context, c client.Client, obj client.Object, rs *appsv1.ReplicaSet, scheme *runtime.Scheme) error { + key := client.ObjectKeyFromObject(obj) + + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := c.Get(ctx, key, obj); err != nil { + return err + } + + orig, ok := obj.DeepCopyObject().(client.Object) + if !ok { + return fmt.Errorf("unable to copy %T", obj) + } + + if err := controllerutil.SetOwnerReference(rs, obj, scheme); err != nil { + return err + } + + if slices.Equal(orig.GetOwnerReferences(), obj.GetOwnerReferences()) { + return nil + } + + return c.Patch(ctx, obj, client.MergeFromWithOptions(orig, client.MergeFromWithOptimisticLock{})) + }) } func getOwningDeploymentFromReplicaSet(ctx context.Context, c client.Client, rs appsv1.ReplicaSet) (*appsv1.Deployment, error) { @@ -205,37 +260,68 @@ func getOwningDeploymentFromReplicaSet(ctx context.Context, c client.Client, rs return deployment, nil } -func cleanUpOldReplicaSets(ctx context.Context, c client.Client, obj client.Object, deployment *appsv1.Deployment, conf config.Config) error { +func cleanUpOldReplicaSets(ctx context.Context, c client.Client, obj client.Object, deployment *appsv1.Deployment, conf config.Config) (requeue bool, err error) { var rsList appsv1.ReplicaSetList selector, _ := metav1.LabelSelectorAsSelector(deployment.Spec.Selector) - err := c.List(ctx, &rsList, client.InNamespace(obj.GetNamespace()), client.MatchingLabelsSelector{ + if err = c.List(ctx, &rsList, client.InNamespace(obj.GetNamespace()), client.MatchingLabelsSelector{ Selector: selector, - }) - - if err != nil { - return err + }); err != nil { + return false, err } - if !currentReplicaSetIsAvailable(rsList, conf.DeploymentRevision) { - return nil - } + currentIsAvailable := currentReplicaSetIsAvailable(rsList, conf.DeploymentRevision) var errs []error + pending := false for _, rs := range rsList.Items { - rsRevision := rs.Annotations[config.RevisionAnnotation] - if rsRevision == conf.DeploymentRevision { + if rs.Annotations[config.RevisionAnnotation] == conf.DeploymentRevision { + continue + } + if hasReplicas(rs) || resourceIsUsedByOtherReplicaSet(rsList, rs) { + continue + } + if rs.Annotations[config.ResourceSuffixAnnotation] == conf.ResourceName { continue } - if !hasReplicas(rs) && !resourceIsUsedByOtherReplicaSet(rsList, rs) { - if err := deleteResourcesForReplicaSet(ctx, c, &rs); err != nil { - // don't stop loop if deletion fails, collect errors and join them later - errs = append(errs, err) - } + if !currentIsAvailable { + pending = pending || replicaSetHasResources(ctx, c, &rs) + continue } + + if err := deleteResourcesForReplicaSet(ctx, c, &rs); err != nil { + errs = append(errs, err) + } + } + + return pending, errors.Join(errs...) +} + +func replicaSetHasResources(ctx context.Context, c client.Client, rs *appsv1.ReplicaSet) bool { + logger := logf.FromContext(ctx) + name := types.NamespacedName{ + Name: rs.Annotations[config.ResourceSuffixAnnotation], + Namespace: rs.Namespace, + } + if name.Name == "" { + return false + } + + if err := c.Get(ctx, name, &avp.AzureVolumePopulator{}); err == nil { + return true + } else if !k8serrors.IsNotFound(err) { + logger.Error(err, "Failed to check AzureVolumePopulator existence during cleanup gating", "replicaSet", rs.Name) + return true + } + + if err := c.Get(ctx, name, &corev1.PersistentVolumeClaim{}); err == nil { + return true + } else if !k8serrors.IsNotFound(err) { + logger.Error(err, "Failed to check PVC existence during cleanup gating", "replicaSet", rs.Name) + return true } - return errors.Join(errs...) + return false } func deleteResourcesForReplicaSet(ctx context.Context, c client.Client, rs *appsv1.ReplicaSet) error { From 34909ab54173206faa7e71214ae28008e344b6ba Mon Sep 17 00:00:00 2001 From: Morten Zwarenstein Date: Mon, 21 Sep 2026 08:42:36 +0200 Subject: [PATCH 3/4] tests: added and improved tests to prevent regression --- internal/controller/util_test.go | 34 -- internal/controller/volume_controller_test.go | 469 ++++++++++++++++-- 2 files changed, 434 insertions(+), 69 deletions(-) diff --git a/internal/controller/util_test.go b/internal/controller/util_test.go index 4501a93..8863b99 100644 --- a/internal/controller/util_test.go +++ b/internal/controller/util_test.go @@ -17,18 +17,12 @@ limitations under the License. package controller import ( - "context" - . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "github.com/PDOK/volume-operator/internal/config" appsv1 "k8s.io/api/apps/v1" - corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "sigs.k8s.io/controller-runtime/pkg/client" - "sigs.k8s.io/controller-runtime/pkg/client/fake" ) func helperReplicaSet(name, revision, suffix string, replicas *int32, available int32) appsv1.ReplicaSet { @@ -125,32 +119,4 @@ var _ = Describe("Controller helpers", func() { Expect(resourceIsUsedByOtherReplicaSet(list, current)).To(BeTrue()) }) }) - - Describe("pvcIsBeingPopulated", func() { - var ( - k8sClient client.Client - conf config.Config - ) - - BeforeEach(func() { - scheme := runtime.NewScheme() - Expect(corev1.AddToScheme(scheme)).To(Succeed()) - k8sClient = fake.NewClientBuilder().WithScheme(scheme).Build() - conf = config.Config{ResourceName: "pvc", ResourceNamespace: testNamespace} - }) - - It("is false when the PVC does not exist", func() { - Expect(pvcIsBeingPopulated(context.Background(), k8sClient, conf)).To(BeFalse()) - }) - - It("is true when the PVC is pending", func() { - Expect(k8sClient.Create(context.Background(), newPVC("pvc", corev1.ClaimPending))).To(Succeed()) - Expect(pvcIsBeingPopulated(context.Background(), k8sClient, conf)).To(BeTrue()) - }) - - It("is false when the PVC is bound", func() { - Expect(k8sClient.Create(context.Background(), newPVC("pvc", corev1.ClaimBound))).To(Succeed()) - Expect(pvcIsBeingPopulated(context.Background(), k8sClient, conf)).To(BeFalse()) - }) - }) }) diff --git a/internal/controller/volume_controller_test.go b/internal/controller/volume_controller_test.go index 2773800..389113c 100644 --- a/internal/controller/volume_controller_test.go +++ b/internal/controller/volume_controller_test.go @@ -18,6 +18,7 @@ package controller import ( "context" + "fmt" "maps" . "github.com/onsi/ginkgo/v2" @@ -27,12 +28,15 @@ import ( avp "github.com/pdok/azure-volume-populator/api/v1alpha1" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + k8serrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" "sigs.k8s.io/controller-runtime/pkg/reconcile" ) @@ -108,14 +112,10 @@ func newAVP(name string) *avp.AzureVolumePopulator { } } -func newPVC(name string, phase corev1.PersistentVolumeClaimPhase) *corev1.PersistentVolumeClaim { - pvc := &corev1.PersistentVolumeClaim{ +func newPVC(name string) *corev1.PersistentVolumeClaim { + return &corev1.PersistentVolumeClaim{ ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: testNamespace}, } - if phase != "" { - pvc.Status.Phase = phase - } - return pvc } var _ = Describe("Volume Controller", func() { @@ -461,6 +461,187 @@ var _ = Describe("Volume Controller", func() { }) }) + Context("owner references", func() { + It("sets an owner reference on the AVP and PVC pointing at the creating ReplicaSet", func() { + deployment := newDeployment("dep-owner-ref", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-owner-ref", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-owner-ref", deployment, "1", 1, 1, "suffix-owner-ref") + Expect(k8sClient.Create(ctx, rs)).To(Succeed()) + + _, err := reconcileRS(rs.Name) + Expect(err).NotTo(HaveOccurred()) + + createdAvp := &avp.AzureVolumePopulator{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-owner-ref", Namespace: testNamespace}, createdAvp)).To(Succeed()) + Expect(createdAvp.OwnerReferences).To(ConsistOf(HaveField("UID", rs.UID))) + + createdPvc := &corev1.PersistentVolumeClaim{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-owner-ref", Namespace: testNamespace}, createdPvc)).To(Succeed()) + Expect(createdPvc.OwnerReferences).To(ConsistOf(HaveField("UID", rs.UID))) + }) + + It("adds a new ReplicaSet as an additional owner of a reused AVP/PVC, without dropping the previous owner", func() { + deployment := newDeployment("dep-shared-owner", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-shared-owner", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-shared-owner-old", deployment, "1", 1, 1, "suffix-shared-owner") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + + _, err := reconcileRS(oldRs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: deployment.Name, Namespace: testNamespace}, deployment)).To(Succeed()) + deployment.Annotations[config.RevisionAnnotation] = "2" + Expect(k8sClient.Update(ctx, deployment)).To(Succeed()) + + newRs := newReplicaSet("rs-shared-owner-new", deployment, "2", 1, 1, "suffix-shared-owner") + Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) + + _, err = reconcileRS(newRs.Name) + Expect(err).NotTo(HaveOccurred()) + + createdAvp := &avp.AzureVolumePopulator{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared-owner", Namespace: testNamespace}, createdAvp)).To(Succeed()) + Expect(createdAvp.OwnerReferences).To(ConsistOf(HaveField("UID", oldRs.UID), HaveField("UID", newRs.UID))) + + createdPvc := &corev1.PersistentVolumeClaim{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared-owner", Namespace: testNamespace}, createdPvc)).To(Succeed()) + Expect(createdPvc.OwnerReferences).To(ConsistOf(HaveField("UID", oldRs.UID), HaveField("UID", newRs.UID))) + }) + + It("retries when adding an owner reference hits a conflict instead of surfacing an error", func() { + scheme := k8sClient.Scheme() + conflicts := map[string]int{} + conflictingClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + kind := fmt.Sprintf("%T", obj) + if conflicts[kind] == 0 { + conflicts[kind]++ + return k8serrors.NewConflict(schema.GroupResource{}, obj.GetName(), nil) + } + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + deployment := newDeployment("dep-conflict", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-conflict", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(conflictingClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-conflict", deployment, "1", 1, 1, "suffix-conflict") + Expect(conflictingClient.Create(ctx, rs)).To(Succeed()) + + Expect(conflictingClient.Create(ctx, newAVP("suffix-conflict"))).To(Succeed()) + Expect(conflictingClient.Create(ctx, newPVC("suffix-conflict"))).To(Succeed()) + + conflictingReconciler := &VolumeReconciler{Client: conflictingClient, Scheme: scheme} + _, err := conflictingReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: rs.Name, Namespace: testNamespace}, + }) + Expect(err).NotTo(HaveOccurred()) + Expect(conflicts).To(HaveLen(2), "expected one conflict each for the AVP and the PVC") + + createdAvp := &avp.AzureVolumePopulator{} + Expect(conflictingClient.Get(ctx, types.NamespacedName{Name: "suffix-conflict", Namespace: testNamespace}, createdAvp)).To(Succeed()) + Expect(createdAvp.OwnerReferences).To(ConsistOf(HaveField("UID", rs.UID))) + + createdPvc := &corev1.PersistentVolumeClaim{} + Expect(conflictingClient.Get(ctx, types.NamespacedName{Name: "suffix-conflict", Namespace: testNamespace}, createdPvc)).To(Succeed()) + Expect(createdPvc.OwnerReferences).To(ConsistOf(HaveField("UID", rs.UID))) + }) + + It("does not patch a reused AVP/PVC that is already owned by the ReplicaSet", func() { + scheme := k8sClient.Scheme() + patches := 0 + countingClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + patches++ + return c.Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + deployment := newDeployment("dep-idempotent", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-idempotent", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(countingClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-idempotent", deployment, "1", 1, 1, "suffix-idempotent") + Expect(countingClient.Create(ctx, rs)).To(Succeed()) + + countingReconciler := &VolumeReconciler{Client: countingClient, Scheme: scheme} + for range 3 { + _, err := countingReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: rs.Name, Namespace: testNamespace}, + }) + Expect(err).NotTo(HaveOccurred()) + } + Expect(patches).To(BeZero()) + }) + + It("treats AlreadyExists on create as success instead of surfacing an error", func() { + scheme := k8sClient.Scheme() + racyClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + switch obj.(type) { + case *avp.AzureVolumePopulator, *corev1.PersistentVolumeClaim: + return k8serrors.NewNotFound(schema.GroupResource{}, key.Name) + default: + return c.Get(ctx, key, obj, opts...) + } + }, + }). + Build() + + deployment := newDeployment("dep-create-race", "1", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-create-race", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(racyClient.Create(ctx, deployment)).To(Succeed()) + + rs := newReplicaSet("rs-create-race", deployment, "1", 1, 1, "suffix-create-race") + Expect(racyClient.Create(ctx, rs)).To(Succeed()) + + Expect(racyClient.Create(ctx, newAVP("suffix-create-race"))).To(Succeed()) + Expect(racyClient.Create(ctx, newPVC("suffix-create-race"))).To(Succeed()) + + racyReconciler := &VolumeReconciler{Client: racyClient, Scheme: scheme} + _, err := racyReconciler.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: rs.Name, Namespace: testNamespace}, + }) + Expect(err).NotTo(HaveOccurred()) + + avpList := &avp.AzureVolumePopulatorList{} + Expect(racyClient.List(ctx, avpList, client.InNamespace(testNamespace))).To(Succeed()) + Expect(avpList.Items).To(HaveLen(1)) + + pvcList := &corev1.PersistentVolumeClaimList{} + Expect(racyClient.List(ctx, pvcList, client.InNamespace(testNamespace))).To(Succeed()) + Expect(pvcList.Items).To(HaveLen(1)) + }) + }) + Context("updating resources", func() { It("does not mutate an existing AVP when the Deployment's blob-prefix changes", func() { deployment := newDeployment("dep-update-avp", "1", map[string]string{ @@ -536,7 +717,7 @@ var _ = Describe("Volume Controller", func() { Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) Expect(k8sClient.Create(ctx, newAVP("suffix-rollout-old"))).To(Succeed()) - Expect(k8sClient.Create(ctx, newPVC("suffix-rollout-old", ""))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-rollout-old"))).To(Succeed()) _, err := reconcileRS(newRs.Name) Expect(err).NotTo(HaveOccurred()) @@ -568,7 +749,7 @@ var _ = Describe("Volume Controller", func() { Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) Expect(k8sClient.Create(ctx, newAVP("suffix-shared"))).To(Succeed()) - Expect(k8sClient.Create(ctx, newPVC("suffix-shared", ""))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-shared"))).To(Succeed()) _, err := reconcileRS(currentRs.Name) Expect(err).NotTo(HaveOccurred()) @@ -577,31 +758,6 @@ var _ = Describe("Volume Controller", func() { Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) }) - It("keeps old resources while the current PVC is still being populated", func() { - deployment := newDeployment("dep-populating", "2", map[string]string{ - config.ResourceSuffixAnnotation: "suffix-populating-new", - blobPrefixAnnotation: "blob/prefix", - volumePathAnnotation: "/data", - }) - Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) - - oldRs := newReplicaSet("rs-populating-old", deployment, "1", 0, 0, "suffix-populating-old") - Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) - newRs := newReplicaSet("rs-populating-new", deployment, "2", 1, 1, "suffix-populating-new") - Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) - - Expect(k8sClient.Create(ctx, newAVP("suffix-populating-old"))).To(Succeed()) - Expect(k8sClient.Create(ctx, newPVC("suffix-populating-old", ""))).To(Succeed()) - // The current PVC already exists and is still pending (being populated). - Expect(k8sClient.Create(ctx, newPVC("suffix-populating-new", corev1.ClaimPending))).To(Succeed()) - - _, err := reconcileRS(newRs.Name) - Expect(err).NotTo(HaveOccurred()) - - Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-populating-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) - Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-populating-old", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) - }) - It("keeps an old ReplicaSet's resources while it still has replicas", func() { deployment := newDeployment("dep-oldscaled", "2", map[string]string{ config.ResourceSuffixAnnotation: "suffix-oldscaled-new", @@ -616,7 +772,7 @@ var _ = Describe("Volume Controller", func() { Expect(k8sClient.Create(ctx, newRs)).To(Succeed()) Expect(k8sClient.Create(ctx, newAVP("suffix-oldscaled-old"))).To(Succeed()) - Expect(k8sClient.Create(ctx, newPVC("suffix-oldscaled-old", ""))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-oldscaled-old"))).To(Succeed()) _, err := reconcileRS(newRs.Name) Expect(err).NotTo(HaveOccurred()) @@ -642,7 +798,7 @@ var _ = Describe("Volume Controller", func() { for _, name := range []string{"suffix-multi-old1", "suffix-multi-old2"} { Expect(k8sClient.Create(ctx, newAVP(name))).To(Succeed()) - Expect(k8sClient.Create(ctx, newPVC(name, ""))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC(name))).To(Succeed()) } _, err := reconcileRS(newRs.Name) @@ -695,5 +851,248 @@ var _ = Describe("Volume Controller", func() { Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-samehash", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-samehash", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) }) + + It("keeps the current hash's resources when the current ReplicaSet is scaled to 0 but still reports available pods", func() { + deployment := newDeployment("dep-scaledown", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-scaledown", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-scaledown-old", deployment, "1", 0, 0, "suffix-scaledown") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + // desired 0, but status still lags: the pods are terminating + currentRs := newReplicaSet("rs-scaledown-current", deployment, "2", 0, 1, "suffix-scaledown") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-scaledown"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-scaledown"))).To(Succeed()) + + for _, name := range []string{oldRs.Name, currentRs.Name} { + _, err := reconcileRS(name) + Expect(err).NotTo(HaveOccurred()) + } + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-scaledown", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-scaledown", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + }) + + It("does not requeue for the current hash's resources while the Deployment is scaled to 0", func() { + deployment := newDeployment("dep-scaledown-requeue", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-scaledown-requeue", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-scaledown-requeue-old", deployment, "1", 0, 0, "suffix-scaledown-requeue") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + currentRs := newReplicaSet("rs-scaledown-requeue-current", deployment, "2", 0, 0, "suffix-scaledown-requeue") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-scaledown-requeue"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-scaledown-requeue"))).To(Succeed()) + + result, err := reconcileRS(oldRs.Name) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RequeueAfter).To(BeZero()) + }) + }) + + Context("cleanup is driven by every ReplicaSet in the set", func() { + It("cleans up a superseded ReplicaSet's resources when that ReplicaSet is the one reconciled", func() { + deployment := newDeployment("dep-trigger-old", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-trigger-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-trigger-old", deployment, "1", 0, 0, "suffix-trigger-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + currentRs := newReplicaSet("rs-trigger-new", deployment, "2", 1, 1, "suffix-trigger-new") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-trigger-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-trigger-old"))).To(Succeed()) + + _, err := reconcileRS(oldRs.Name) + Expect(err).NotTo(HaveOccurred()) + + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-trigger-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{}) + Expect(err).To(HaveOccurred()) + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-trigger-old", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{}) + Expect(err).To(HaveOccurred()) + }) + + // test bgt specific case of quick updates + It("cleans up the oldest resources when a superseded ReplicaSet that shares the current suffix is reconciled", func() { + deployment := newDeployment("dep-trigger-shared", "3", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-shared-b", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + rs1 := newReplicaSet("rs-shared-1", deployment, "1", 0, 0, "suffix-shared-a") + Expect(k8sClient.Create(ctx, rs1)).To(Succeed()) + rs2 := newReplicaSet("rs-shared-2", deployment, "2", 0, 0, "suffix-shared-b") + Expect(k8sClient.Create(ctx, rs2)).To(Succeed()) + rs3 := newReplicaSet("rs-shared-3", deployment, "3", 1, 1, "suffix-shared-b") + Expect(k8sClient.Create(ctx, rs3)).To(Succeed()) + + for _, name := range []string{"suffix-shared-a", "suffix-shared-b"} { + Expect(k8sClient.Create(ctx, newAVP(name))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC(name))).To(Succeed()) + } + + _, err := reconcileRS(rs2.Name) + Expect(err).NotTo(HaveOccurred()) + + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared-a", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{}) + Expect(err).To(HaveOccurred()) + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared-a", Namespace: testNamespace}, &avp.AzureVolumePopulator{}) + Expect(err).To(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared-b", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-shared-b", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + }) + + // test bgt specific case of quick updates, latest rs not being available when new ones are preparing + It("does not provision resources for a superseded revision when its ReplicaSet is reconciled", func() { + deployment := newDeployment("dep-trigger-nocreate", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-nocreate-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-nocreate-old", deployment, "1", 0, 0, "suffix-nocreate-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + currentRs := newReplicaSet("rs-nocreate-new", deployment, "2", 1, 1, "suffix-nocreate-new") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + _, err := reconcileRS(oldRs.Name) + Expect(err).NotTo(HaveOccurred()) + + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-nocreate-new", Namespace: testNamespace}, &avp.AzureVolumePopulator{}) + Expect(err).To(HaveOccurred()) + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-nocreate-new", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{}) + Expect(err).To(HaveOccurred()) + }) + + It("respects the availability gate when cleanup is triggered by an old ReplicaSet", func() { + deployment := newDeployment("dep-trigger-gate", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-gate-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-gate-old", deployment, "1", 0, 0, "suffix-gate-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + // current revision exists but is not serving yet. + currentRs := newReplicaSet("rs-gate-new", deployment, "2", 1, 0, "suffix-gate-new") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-gate-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-gate-old"))).To(Succeed()) + + _, err := reconcileRS(oldRs.Name) + Expect(err).NotTo(HaveOccurred()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-gate-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{})).To(Succeed()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-gate-old", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{})).To(Succeed()) + }) + + It("still cleans up superseded resources when the Deployment is missing volume annotations", func() { + deployment := newDeployment("dep-trigger-noann", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-noann-new", + // blob-prefix / volume-path deliberately omitted + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-noann-old", deployment, "1", 0, 0, "suffix-noann-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + currentRs := newReplicaSet("rs-noann-new", deployment, "2", 1, 1, "suffix-noann-new") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-noann-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-noann-old"))).To(Succeed()) + + _, err := reconcileRS(oldRs.Name) + Expect(err).NotTo(HaveOccurred()) + + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-noann-old", Namespace: testNamespace}, &corev1.PersistentVolumeClaim{}) + Expect(err).To(HaveOccurred()) + }) + }) + + // Fix 2: while the availability gate holds cleanup back, the reconcile must ask to + // be requeued so it converges without depending on an incidental later event. + Context("requeueing while cleanup is blocked", func() { + It("requeues when the current ReplicaSet is not yet available and old resources remain", func() { + deployment := newDeployment("dep-requeue", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-requeue-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-requeue-old", deployment, "1", 0, 0, "suffix-requeue-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + currentRs := newReplicaSet("rs-requeue-new", deployment, "2", 1, 0, "suffix-requeue-new") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-requeue-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-requeue-old"))).To(Succeed()) + + result, err := reconcileRS(currentRs.Name) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RequeueAfter).To(BeNumerically(">", 0)) + }) + + It("does not requeue once the old resources have been cleaned up", func() { + deployment := newDeployment("dep-requeue-settled", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-settled-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-settled-old", deployment, "1", 0, 0, "suffix-settled-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + currentRs := newReplicaSet("rs-settled-new", deployment, "2", 1, 1, "suffix-settled-new") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + Expect(k8sClient.Create(ctx, newAVP("suffix-settled-old"))).To(Succeed()) + Expect(k8sClient.Create(ctx, newPVC("suffix-settled-old"))).To(Succeed()) + + result, err := reconcileRS(currentRs.Name) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(reconcile.Result{})) + + err = k8sClient.Get(ctx, types.NamespacedName{Name: "suffix-settled-old", Namespace: testNamespace}, &avp.AzureVolumePopulator{}) + Expect(err).To(HaveOccurred()) + }) + + It("does not requeue when the current ReplicaSet is unavailable but there is nothing to clean up", func() { + deployment := newDeployment("dep-requeue-idle", "2", map[string]string{ + config.ResourceSuffixAnnotation: "suffix-idle-new", + blobPrefixAnnotation: "blob/prefix", + volumePathAnnotation: "/data", + }) + Expect(k8sClient.Create(ctx, deployment)).To(Succeed()) + + oldRs := newReplicaSet("rs-idle-old", deployment, "1", 0, 0, "suffix-idle-old") + Expect(k8sClient.Create(ctx, oldRs)).To(Succeed()) + currentRs := newReplicaSet("rs-idle-new", deployment, "2", 1, 0, "suffix-idle-new") + Expect(k8sClient.Create(ctx, currentRs)).To(Succeed()) + + result, err := reconcileRS(currentRs.Name) + Expect(err).NotTo(HaveOccurred()) + Expect(result).To(Equal(reconcile.Result{})) + }) }) }) From 03bbcd375c1a6d949e3fd5a287a8137681f0f006 Mon Sep 17 00:00:00 2001 From: Morten Zwarenstein Date: Mon, 21 Sep 2026 10:58:37 +0200 Subject: [PATCH 4/4] chore: removed Apache License in favor of LICENSE.md --- cmd/main.go | 16 ---------------- internal/controller/suite_test.go | 16 ---------------- internal/controller/util_test.go | 16 ---------------- internal/controller/volume_controller.go | 16 ---------------- internal/controller/volume_controller_test.go | 16 ---------------- 5 files changed, 80 deletions(-) diff --git a/cmd/main.go b/cmd/main.go index d7e70ee..276faef 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -1,19 +1,3 @@ -/* -Copyright 2025. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - package main import ( diff --git a/internal/controller/suite_test.go b/internal/controller/suite_test.go index 62d4348..6dcd7f2 100644 --- a/internal/controller/suite_test.go +++ b/internal/controller/suite_test.go @@ -1,19 +1,3 @@ -/* -Copyright 2025. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - package controller import ( diff --git a/internal/controller/util_test.go b/internal/controller/util_test.go index 8863b99..991ad43 100644 --- a/internal/controller/util_test.go +++ b/internal/controller/util_test.go @@ -1,19 +1,3 @@ -/* -Copyright 2025. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - package controller import ( diff --git a/internal/controller/volume_controller.go b/internal/controller/volume_controller.go index 51caa82..3fe0dc6 100644 --- a/internal/controller/volume_controller.go +++ b/internal/controller/volume_controller.go @@ -1,19 +1,3 @@ -/* -Copyright 2025. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - package controller import ( diff --git a/internal/controller/volume_controller_test.go b/internal/controller/volume_controller_test.go index 389113c..4383554 100644 --- a/internal/controller/volume_controller_test.go +++ b/internal/controller/volume_controller_test.go @@ -1,19 +1,3 @@ -/* -Copyright 2025. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - package controller import (