add flag for PVC deletion

src/go/k8s/config/e2e-tests/manager.yaml

@@ -17,3 +17,4 @@ spec:
         - "--configurator-base-image=localhost/configurator"
         - "--configurator-tag=dev"
         - "--configurator-image-pull-policy=Never"
+        - "--allow-pvc-deletion=true"

src/go/k8s/controllers/redpanda/cluster_controller.go

@@ -70,6 +70,7 @@ type ClusterReconciler struct {
 	AdminAPIClientFactory     adminutils.AdminAPIClientFactory
 	DecommissionWaitInterval  time.Duration
 	RestrictToRedpandaVersion string
+	allowPVCDeletion          bool
 }
 
 //+kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
@@ -453,6 +454,14 @@ func (r *ClusterReconciler) handlePodFinalizer(
 				return fmt.Errorf(`unable to decommission node "%d": %w`, nodeID, err)
 			}
 		}
+
+		if !r.allowPVCDeletion {
+			//   remove the finalizer
+			if err = r.removePodFinalizer(ctx, pod, log); err != nil {
+				return fmt.Errorf(`unable to remove finalizer from pod "%s/%s: %w"`, pod.GetNamespace(), pod.GetName(), err)
+			}
+			return nil
+		}
 		//   delete the associated pvc
 		pvc := corev1.PersistentVolumeClaim{}
 		//nolint: gocritic // 248 bytes 6 times is not worth decreasing the readability over
@@ -467,6 +476,7 @@ func (r *ClusterReconciler) handlePodFinalizer(
 					if !apierrors.IsNotFound(err) {
 						return fmt.Errorf(`unable to fetch PersistentVolumeClaim "%s/%s": %w`, key.Namespace, key.Name, err)
 					}
+					continue
 				}
 				log.WithValues(key).Info("deleting PersistentVolumeClaim")
 				if err := r.Delete(ctx, &pvc, &client.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
@@ -688,6 +698,13 @@ func (r *ClusterReconciler) WithClusterDomain(
 	return r
 }
 
+func (r *ClusterReconciler) WithAllowPVCDeletion(
+	allowPVCDeletion bool,
+) *ClusterReconciler {
+	r.allowPVCDeletion = allowPVCDeletion
+	return r
+}
+
 //nolint:funlen,gocyclo // External nodes list should be refactored
 func (r *ClusterReconciler) createExternalNodesList(
 	ctx context.Context,

src/go/k8s/main.go

@@ -61,6 +61,12 @@ func main() {
 		configuratorImagePullPolicy string
 		decommissionWaitInterval    time.Duration
 		restrictToRedpandaVersion   string
+
+		// allowPVCDeletion controls the PVC deletion feature in the Cluster custom resource.
+		// PVCs will be deleted when its Pod has been deleted and the Node that Pod is assigned to
+		// does not exist, or has the NoExecute taint. This is intended to support the rancher.io/local-path
+		// storage driver.
+		allowPVCDeletion bool
 	)
 
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
@@ -75,6 +81,7 @@ func main() {
 	flag.StringVar(&configuratorImagePullPolicy, "configurator-image-pull-policy", "Always", "Set the configurator image pull policy")
 	flag.DurationVar(&decommissionWaitInterval, "decommission-wait-interval", 8*time.Second, "Set the time to wait for a node decommission to happen in the cluster")
 	flag.BoolVar(&redpandav1alpha1.AllowDownscalingInWebhook, "allow-downscaling", false, "Allow to reduce the number of replicas in existing clusters (alpha feature)")
+	flag.BoolVar(&allowPVCDeletion, "allow-pvc-deletion", false, "Allow the operator to delete PVCs for Pods assigned to failed or missing Nodes (alpha feature)")
 	flag.BoolVar(&redpandav1alpha1.AllowConsoleAnyNamespace, "allow-console-any-ns", false, "Allow to create Console in any namespace. Allowing this copies Redpanda SchemaRegistry TLS Secret to namespace (alpha feature)")
 	flag.StringVar(&restrictToRedpandaVersion, "restrict-redpanda-version", "", "Restrict management of clusters to those with this version")
 
@@ -114,7 +121,7 @@ func main() {
 		AdminAPIClientFactory:     adminutils.NewInternalAdminAPI,
 		DecommissionWaitInterval:  decommissionWaitInterval,
 		RestrictToRedpandaVersion: restrictToRedpandaVersion,
-	}).WithClusterDomain(clusterDomain).WithConfiguratorSettings(configurator).SetupWithManager(mgr); err != nil {
+	}).WithClusterDomain(clusterDomain).WithConfiguratorSettings(configurator).WithAllowPVCDeletion(allowPVCDeletion).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "Unable to create controller", "controller", "Cluster")
 		os.Exit(1)
 	}