From 63ff7ef928cd556cae9c97599f4fb39c5a07e65a Mon Sep 17 00:00:00 2001 From: Michael Burman Date: Thu, 12 Dec 2024 18:00:48 +0200 Subject: [PATCH] Add a new indexer for matching Pod ClaimNames to PVCs. When deleting a PVC, verify that no Pods match this index search --- CHANGELOG.md | 2 + cmd/main.go | 24 ++++- pkg/reconciliation/reconcile_datacenter.go | 22 +++++ .../reconcile_datacenter_test.go | 89 +++++++++++++++---- 4 files changed, 121 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bb82d87..df1c98ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ Changelog for Cass Operator, new PRs should update the `main / unreleased` secti ## unreleased +* [ENHANCEMENT] [#737](https://github.com/k8ssandra/cass-operator/issues/737) Before issuing PVC deletion when deleting a datacenter, verify the PVCs that match the labels are not actually used by any pods. + ## v1.23.0 * [CHANGE] [#720](https://github.com/k8ssandra/cass-operator/issues/720) Always use ObjectMeta.Name for the PodDisruptionBudget resource name, not the DatacenterName diff --git a/cmd/main.go b/cmd/main.go index 58bf99d2..d8495c88 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -26,11 +26,13 @@ import ( "go.uber.org/zap/zapcore" _ "k8s.io/client-go/plugin/pkg/client/auth" + corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/cache" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/healthz" "sigs.k8s.io/controller-runtime/pkg/log/zap" @@ -119,6 +121,8 @@ func main() { os.Exit(1) } + ctx := ctrl.SetupSignalHandler() + if err = (&controllers.CassandraDatacenterReconciler{ Client: mgr.GetClient(), Log: ctrl.Log.WithName("controllers").WithName("CassandraDatacenter"), @@ -143,6 +147,24 @@ func main() { os.Exit(1) } + if err := mgr.GetFieldIndexer().IndexField(ctx, &corev1.Pod{}, "spec.volumes.persistentVolumeClaim.claimName", func(obj client.Object) []string { + pod, ok := obj.(*corev1.Pod) + if !ok { + return nil + } + + var pvcNames []string + for _, volume := range pod.Spec.Volumes { + if volume.PersistentVolumeClaim != nil { + pvcNames = append(pvcNames, volume.PersistentVolumeClaim.ClaimName) + } + } + return pvcNames + }); err != nil { + setupLog.Error(err, "unable to set up field indexer") + os.Exit(1) + } + if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { setupLog.Error(err, "unable to set up health check") os.Exit(1) @@ -153,7 +175,7 @@ func main() { } setupLog.Info("starting manager") - if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil { + if err := mgr.Start(ctx); err != nil { setupLog.Error(err, "problem running manager") os.Exit(1) } diff --git a/pkg/reconciliation/reconcile_datacenter.go b/pkg/reconciliation/reconcile_datacenter.go index ec8c11e8..3c45a0d7 100644 --- a/pkg/reconciliation/reconcile_datacenter.go +++ b/pkg/reconciliation/reconcile_datacenter.go @@ -13,6 +13,7 @@ import ( storagev1 "k8s.io/api/storage/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/fields" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" @@ -125,6 +126,14 @@ func (rc *ReconciliationContext) deletePVCs() error { "numPVCs", len(persistentVolumeClaimList)) for _, pvc := range persistentVolumeClaimList { + + if isBeingUsed, err := rc.isBeingUsed(pvc); err != nil { + logger.Error(err, "Failed to check if PVC is being used") + return err + } else if isBeingUsed { + return fmt.Errorf("PersistentVolumeClaim %s is still being used by a pod", pvc.Name) + } + if err := rc.Client.Delete(rc.Ctx, &pvc); err != nil { logger.Error(err, "Failed to delete PVCs for cassandraDatacenter") return err @@ -138,6 +147,19 @@ func (rc *ReconciliationContext) deletePVCs() error { return nil } +func (rc *ReconciliationContext) isBeingUsed(pvc corev1.PersistentVolumeClaim) (bool, error) { + rc.ReqLogger.Info("reconciler::isBeingUsed") + + pods := &corev1.PodList{} + + if err := rc.Client.List(rc.Ctx, pods, &client.ListOptions{Namespace: pvc.Namespace, FieldSelector: fields.SelectorFromSet(fields.Set{"spec.volumes.persistentVolumeClaim.claimName": pvc.Name})}); err != nil { + rc.ReqLogger.Error(err, "error getting pods for pvc", "pvc", pvc.Name) + return false, err + } + + return len(pods.Items) > 0, nil +} + func (rc *ReconciliationContext) listPVCs(selector map[string]string) ([]corev1.PersistentVolumeClaim, error) { rc.ReqLogger.Info("reconciler::listPVCs") diff --git a/pkg/reconciliation/reconcile_datacenter_test.go b/pkg/reconciliation/reconcile_datacenter_test.go index ca30425d..e7d5f462 100644 --- a/pkg/reconciliation/reconcile_datacenter_test.go +++ b/pkg/reconciliation/reconcile_datacenter_test.go @@ -5,18 +5,20 @@ package reconciliation import ( "fmt" + "strings" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" - v1 "k8s.io/api/core/v1" + corev1 "k8s.io/api/core/v1" storagev1 "k8s.io/api/storage/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" "github.com/k8ssandra/cass-operator/pkg/mocks" ) @@ -30,13 +32,23 @@ func TestDeletePVCs(t *testing.T) { k8sMockClientList(mockClient, nil). Run(func(args mock.Arguments) { - arg := args.Get(1).(*v1.PersistentVolumeClaimList) - arg.Items = []v1.PersistentVolumeClaim{{ + _, ok := args.Get(1).(*corev1.PodList) + if ok { + if strings.HasPrefix(args.Get(2).(*client.ListOptions).FieldSelector.String(), "spec.volumes.persistentVolumeClaim.claimName") { + arg := args.Get(1).(*corev1.PodList) + arg.Items = []corev1.Pod{} + } else { + t.Fail() + } + return + } + arg := args.Get(1).(*corev1.PersistentVolumeClaimList) + arg.Items = []corev1.PersistentVolumeClaim{{ ObjectMeta: metav1.ObjectMeta{ Name: "pvc-1", }, }} - }) + }).Twice() k8sMockClientDelete(mockClient, nil) @@ -55,8 +67,8 @@ func TestDeletePVCs_FailedToList(t *testing.T) { k8sMockClientList(mockClient, fmt.Errorf("failed to list PVCs for CassandraDatacenter")). Run(func(args mock.Arguments) { - arg := args.Get(1).(*v1.PersistentVolumeClaimList) - arg.Items = []v1.PersistentVolumeClaim{{ + arg := args.Get(1).(*corev1.PersistentVolumeClaimList) + arg.Items = []corev1.PersistentVolumeClaim{{ ObjectMeta: metav1.ObjectMeta{ Name: "pvc-1", }, @@ -73,24 +85,22 @@ func TestDeletePVCs_FailedToList(t *testing.T) { func TestDeletePVCs_PVCsNotFound(t *testing.T) { rc, _, cleanupMockScr := setupTest() defer cleanupMockScr() + assert := assert.New(t) mockClient := mocks.NewClient(t) rc.Client = mockClient k8sMockClientList(mockClient, errors.NewNotFound(schema.GroupResource{}, "name")). Run(func(args mock.Arguments) { - arg := args.Get(1).(*v1.PersistentVolumeClaimList) - arg.Items = []v1.PersistentVolumeClaim{{ + arg := args.Get(1).(*corev1.PersistentVolumeClaimList) + arg.Items = []corev1.PersistentVolumeClaim{{ ObjectMeta: metav1.ObjectMeta{ Name: "pvc-1", }, }} }) - err := rc.deletePVCs() - if err != nil { - t.Fatalf("deletePVCs should not have failed") - } + assert.NoError(rc.deletePVCs()) } func TestDeletePVCs_FailedToDelete(t *testing.T) { @@ -102,13 +112,23 @@ func TestDeletePVCs_FailedToDelete(t *testing.T) { k8sMockClientList(mockClient, nil). Run(func(args mock.Arguments) { - arg := args.Get(1).(*v1.PersistentVolumeClaimList) - arg.Items = []v1.PersistentVolumeClaim{{ + _, ok := args.Get(1).(*corev1.PodList) + if ok { + if strings.HasPrefix(args.Get(2).(*client.ListOptions).FieldSelector.String(), "spec.volumes.persistentVolumeClaim.claimName") { + arg := args.Get(1).(*corev1.PodList) + arg.Items = []corev1.Pod{} + } else { + t.Fail() + } + return + } + arg := args.Get(1).(*corev1.PersistentVolumeClaimList) + arg.Items = []corev1.PersistentVolumeClaim{{ ObjectMeta: metav1.ObjectMeta{ Name: "pvc-1", }, }} - }) + }).Twice() k8sMockClientDelete(mockClient, fmt.Errorf("failed to delete")) @@ -120,6 +140,45 @@ func TestDeletePVCs_FailedToDelete(t *testing.T) { assert.EqualError(t, err, "failed to delete") } +func TestDeletePVCs_FailedToDeleteBeingUsed(t *testing.T) { + rc, _, cleanupMockScr := setupTest() + defer cleanupMockScr() + assert := assert.New(t) + + mockClient := mocks.NewClient(t) + rc.Client = mockClient + + k8sMockClientList(mockClient, nil). + Run(func(args mock.Arguments) { + _, ok := args.Get(1).(*corev1.PodList) + if ok { + if strings.HasPrefix(args.Get(2).(*client.ListOptions).FieldSelector.String(), "spec.volumes.persistentVolumeClaim.claimName") { + arg := args.Get(1).(*corev1.PodList) + arg.Items = []corev1.Pod{ + { + ObjectMeta: metav1.ObjectMeta{ + Name: "pod-1", + }, + }, + } + } else { + t.Fail() + } + return + } + arg := args.Get(1).(*corev1.PersistentVolumeClaimList) + arg.Items = []corev1.PersistentVolumeClaim{{ + ObjectMeta: metav1.ObjectMeta{ + Name: "pvc-1", + }, + }} + }).Twice() + + err := rc.deletePVCs() + assert.Error(err) + assert.EqualError(err, "PersistentVolumeClaim pvc-1 is still being used by a pod") +} + func TestStorageExpansionNils(t *testing.T) { rc, _, cleanupMockScr := setupTest() defer cleanupMockScr()