K8SPSMDB-1455: avoid repeated delete calls

cmd/manager/main.go

@@ -16,12 +16,14 @@ import (
 	volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	uzap "go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
+	corev1 "k8s.io/api/core/v1"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/discovery"
 	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"
 	metricsServer "sigs.k8s.io/controller-runtime/pkg/metrics/server"
@@ -104,6 +106,12 @@ func main() {
 		WebhookServer: webhook.NewServer(webhook.Options{
 			Port: 9443,
 		}),
+		Client: client.Options{
+			Scheme: scheme,
+			Cache: &client.CacheOptions{
+				DisableFor: []client.Object{&corev1.Node{}},
+			},
+		},
 	}
 
 	options.Controller.GroupKindConcurrency = map[string]int{

e2e-tests/pvc-resize/run

@@ -68,6 +68,28 @@ function wait_cluster_status() {
 	echo "psmdb/${cluster} status is ${expected}"
 }
 
+function wait_pvc_request_revert() {
+	local cluster=$1
+	local expected=$2
+	local retry=0
+
+	echo -n "Waiting for psmdb/${cluster} storage request to revert to ${expected}"
+	until [[ $(kubectl_bin get psmdb "${cluster}" -o jsonpath='{.spec.replsets[0].volumeSpec.persistentVolumeClaim.resources.requests.storage}') == "${expected}" ]]; do
+		if [[ $retry -ge 60 ]]; then
+			echo
+			echo "psmdb/${cluster} storage request did not revert to ${expected}, max retries exceeded"
+			exit 1
+		fi
+		echo -n "."
+		sleep 5
+
+		retry=$((retry + 1))
+	done
+
+	echo
+	echo "psmdb/${cluster} storage request reverted to ${expected}"
+}
+
 set_debug
 
 if [ "$EKS" == 1 ]; then
@@ -152,12 +174,10 @@ wait_all_pvc_resize "3Gi"
 
 desc "test downscale"
 
-# operator shouldn't try to downscale the PVCs and set status to error
+# operator shouldn't downscale the PVCs. instead it reverts the storage request
+# in the spec back to the configured size and keeps the cluster ready
 patch_pvc_request "${cluster}" "1G"
-wait_cluster_status ${cluster} "error"
-
-# user should be able to restore to the previous size and make the cluster ready
-patch_pvc_request "${cluster}" "3G"
+wait_pvc_request_revert "${cluster}" "3G"
 wait_cluster_status ${cluster} "ready"
 
 if [[ $EKS == 1 || -n ${OPENSHIFT} ]]; then

pkg/controller/perconaservermongodb/psmdb_controller.go

@@ -18,6 +18,7 @@ import (
 	"github.com/robfig/cron/v3"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -36,6 +37,7 @@ import (
 
 	"github.com/percona/percona-server-mongodb-operator/clientcmd"
 	api "github.com/percona/percona-server-mongodb-operator/pkg/apis/psmdb/v1"
+	k8sutils "github.com/percona/percona-server-mongodb-operator/pkg/k8s"
 	"github.com/percona/percona-server-mongodb-operator/pkg/naming"
 	"github.com/percona/percona-server-mongodb-operator/pkg/psmdb"
 	"github.com/percona/percona-server-mongodb-operator/pkg/psmdb/backup"
@@ -51,6 +53,19 @@ import (
 	"github.com/percona/percona-server-mongodb-operator/pkg/version"
 )
 
+const eventRegardingNameIndex = "regarding.name"
+
+func eventRegardingNameIndexer(o client.Object) []string {
+	evt, ok := o.(*eventsv1.Event)
+	if !ok {
+		return nil
+	}
+	if evt.Regarding.Name == "" {
+		return nil
+	}
+	return []string{evt.Regarding.Name}
+}
+
 // Add creates a new PerconaServerMongoDB Controller and adds it to the Manager. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
 func Add(mgr manager.Manager) error {
@@ -59,6 +74,13 @@ func Add(mgr manager.Manager) error {
 		return err
 	}
 
+	// the volume resize logic lists PVC events through the cached client,
+	// which requires the field to be indexed
+	err = mgr.GetFieldIndexer().IndexField(context.TODO(), &eventsv1.Event{}, eventRegardingNameIndex, eventRegardingNameIndexer)
+	if err != nil {
+		return errors.Wrapf(err, "index events by %s", eventRegardingNameIndex)
+	}
+
 	return add(mgr, r)
 }
 
@@ -81,22 +103,12 @@ func newReconciler(mgr manager.Manager) (reconcile.Reconciler, error) {
 		return nil, errors.Wrap(err, "failed to get operator pod image")
 	}
 
-	client, err := client.New(mgr.GetConfig(), client.Options{
-		Scheme: mgr.GetScheme(),
-		Cache: &client.CacheOptions{
-			DisableFor: []client.Object{&corev1.Node{}},
-		},
-	})
-	if err != nil {
-		return nil, errors.Wrap(err, "create client")
-	}
-
 	secretProviders := []pkgSecret.Provider{
 		new(vault.Provider),
 	}
 
 	return &ReconcilePerconaServerMongoDB{
-		client:                 client,
+		client:                 mgr.GetClient(),
 		scheme:                 mgr.GetScheme(),
 		serverVersion:          sv,
 		reconcileIn:            getReconcileInterval(),
@@ -532,10 +544,8 @@ func (r *ReconcilePerconaServerMongoDB) reconcileReplset(ctx context.Context, cr
 			return err
 		}
 	} else {
-		err := r.client.Delete(ctx, psmdb.NewStatefulSet(naming.ArbiterStatefulSetName(cr, replset), cr.Namespace))
-		if err != nil && !k8serrors.IsNotFound(err) {
-			err = errors.Errorf("delete arbiter in replset %s: %v", replset.Name, err)
-			return err
+		if err := k8sutils.DeleteIfExists(ctx, r.client, psmdb.NewStatefulSet(naming.ArbiterStatefulSetName(cr, replset), cr.Namespace)); err != nil {
+			return errors.Wrapf(err, "failed to delete arbiter statefulset: %s", naming.ArbiterStatefulSetName(cr, replset))
 		}
 	}
 
@@ -547,25 +557,21 @@ func (r *ReconcilePerconaServerMongoDB) reconcileReplset(ctx context.Context, cr
 			return err
 		}
 	} else {
-		err := r.client.Delete(ctx, psmdb.NewStatefulSet(naming.NonVotingStatefulSetName(cr, replset), cr.Namespace))
-		if err != nil && !k8serrors.IsNotFound(err) {
-			err = errors.Errorf("delete nonVoting statefulset %s: %v", replset.Name, err)
-			return err
+		if err := k8sutils.DeleteIfExists(ctx, r.client, psmdb.NewStatefulSet(naming.NonVotingStatefulSetName(cr, replset), cr.Namespace)); err != nil {
+			return errors.Wrapf(err, "failed to delete non voting statefulset: %s", naming.NonVotingStatefulSetName(cr, replset))
 		}
 	}
 
 	if replset.Hidden.Enabled {
 		matchLabels = naming.HiddenLabels(cr, replset)
 		_, err := r.reconcileStatefulSet(ctx, cr, replset, matchLabels)
 		if err != nil {
-			err = errors.Errorf("reconcile nonVoting StatefulSet for %s: %v", replset.Name, err)
+			err = errors.Errorf("reconcile hidden StatefulSet for %s: %v", replset.Name, err)
 			return err
 		}
 	} else {
-		err := r.client.Delete(ctx, psmdb.NewStatefulSet(naming.HiddenStatefulSetName(cr, replset), cr.Namespace))
-		if err != nil && !k8serrors.IsNotFound(err) {
-			err = errors.Errorf("delete hidden statefulset %s: %v", replset.Name, err)
-			return err
+		if err := k8sutils.DeleteIfExists(ctx, r.client, psmdb.NewStatefulSet(naming.HiddenStatefulSetName(cr, replset), cr.Namespace)); err != nil {
+			return errors.Wrapf(err, "failed to delete hidden statefulset: %s", naming.HiddenStatefulSetName(cr, replset))
 		}
 	}
 
@@ -1062,24 +1068,18 @@ func (r *ReconcilePerconaServerMongoDB) deleteCfgIfNeeded(ctx context.Context, c
 
 	sfsName := cr.Name + "-" + api.ConfigReplSetName
 	sfs := psmdb.NewStatefulSet(sfsName, cr.Namespace)
-
-	if err := r.client.Delete(ctx, sfs); err != nil && !k8serrors.IsNotFound(err) {
+	if err := k8sutils.DeleteIfExists(ctx, r.client, sfs); err != nil {
 		return errors.Wrapf(err, "failed to delete sfs: %s", sfs.Name)
 	}
 
-	svc := corev1.Service{}
-	err = r.client.Get(ctx, types.NamespacedName{Name: cr.Name + "-" + api.ConfigReplSetName, Namespace: cr.Namespace}, &svc)
-	if err != nil && !k8serrors.IsNotFound(err) {
-		return errors.Wrap(err, "failed to get config service")
-	}
-
-	if k8serrors.IsNotFound(err) {
-		return nil
+	svc := corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      cr.Name + "-" + api.ConfigReplSetName,
+			Namespace: cr.Namespace,
+		},
 	}
-
-	err = r.client.Delete(ctx, &svc)
-	if err != nil {
-		return errors.Wrap(err, "failed to delete config service")
+	if err := k8sutils.DeleteIfExists(ctx, r.client, &svc); err != nil {
+		return errors.Wrapf(err, "failed to delete config service: %s", svc.Name)
 	}
 
 	return nil
@@ -1117,15 +1117,6 @@ func (r *ReconcilePerconaServerMongoDB) upgradeFCVIfNeeded(ctx context.Context,
 	return errors.Wrap(err, "failed to set FCV")
 }
 
-func (r *ReconcilePerconaServerMongoDB) deleteMongos(ctx context.Context, cr *api.PerconaServerMongoDB) error {
-	err := r.client.Delete(ctx, psmdb.MongosStatefulset(cr))
-	if err != nil && !k8serrors.IsNotFound(err) {
-		return errors.Wrap(err, "failed to delete mongos statefulset")
-	}
-
-	return nil
-}
-
 func (r *ReconcilePerconaServerMongoDB) deleteMongosIfNeeded(ctx context.Context, cr *api.PerconaServerMongoDB) error {
 	if cr.Spec.Sharding.Enabled {
 		return nil
@@ -1152,7 +1143,10 @@ func (r *ReconcilePerconaServerMongoDB) deleteMongosIfNeeded(ctx context.Context
 		}
 	}
 
-	return r.deleteMongos(ctx, cr)
+	if err := k8sutils.DeleteIfExists(ctx, r.client, psmdb.MongosStatefulset(cr)); err != nil {
+		return errors.Wrap(err, "failed to delete mongos statefulset")
+	}
+	return nil
 }
 
 func (r *ReconcilePerconaServerMongoDB) reconcileMongodConfigMaps(ctx context.Context, cr *api.PerconaServerMongoDB, repls []*api.ReplsetSpec) error {

pkg/k8s/utils.go

@@ -1,11 +1,15 @@
 package k8s
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"strings"
 
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 const WatchNamespaceEnvVar = "WATCH_NAMESPACE"
@@ -41,3 +45,21 @@ func IsPodReady(pod corev1.Pod) bool {
 
 	return false
 }
+
+func DeleteIfExists(ctx context.Context, c client.Client, obj client.Object) error {
+	if err := c.Get(ctx, client.ObjectKeyFromObject(obj), obj); k8serrors.IsNotFound(err) {
+		return nil
+	} else if err != nil {
+		return errors.Wrapf(err, "failed to get %T %s/%s", obj, obj.GetNamespace(), obj.GetName())
+	}
+
+	// Deletion was already requested
+	if obj.GetDeletionTimestamp() != nil {
+		return nil
+	}
+
+	if err := c.Delete(ctx, obj); client.IgnoreNotFound(err) != nil {
+		return errors.Wrapf(err, "failed to delete %T %s/%s", obj, obj.GetNamespace(), obj.GetName())
+	}
+	return nil
+}

pkg/psmdb/getters.go

@@ -44,6 +44,12 @@ func getRSPods(ctx context.Context, k8sclient client.Client, cr *api.PerconaServ
 		return rsPods, errors.Wrapf(err, "failed to get statefulset list related to replset %s", rsName)
 	}
 
+	// `client.List` doesn't guarantee ordering (the cache-backed client returns items
+	// in map iteration order). Sort StatefulSets by name so iteration order is deterministic.
+	sort.Slice(stsList.Items, func(i, j int) bool {
+		return stsList.Items[i].Name < stsList.Items[j].Name
+	})
+
 	for _, sts := range stsList.Items {
 		rs := cr.Spec.Replset(rsName)