operator: Clean up data PVCs on cleanupMetadata, regardless of cleanupData option

Signed-off-by: Aaron Wilson <aawilson@nvidia.com>

Author: aaronnw

operator/CHANGELOG.md

@@ -15,6 +15,10 @@ We structure this changelog in accordance with [Keep a Changelog](https://keepac
 - Fixed syncPodTemplate not syncing env var removals when the desired env slice was a prefix of the current, causing rollout loops on config changes such as clearing `authNSecretName`.
 - Invalidate the AIStore API client on authentication failure, not just on failure to acquire initial token.
 - Mount path size is now optional to better support hostPath data mounts.
+- On cluster decommission, data PVCs are now deleted whenever `cleanupMetadata` is enabled, regardless of `cleanupData`.
+    - Clarifies the purpose of `cleanupData` as an API option to AIS for cleaning data on disk.
+    - Use node selectors for host path cleanup jobs rather than querying existing pods by label.
+    - Validate CRD only allows `cleanupData` if `cleanupMetadata` is enabled.
 
 ## v2.15.0
 

operator/api/v1beta1/aistore_types.go

@@ -301,6 +301,7 @@ type CertIssuerRef struct {
 // AIStoreSpec defines the desired state of AIStore
 // +kubebuilder:validation:XValidation:rule="(has(self.targetSpec.size) && has(self.proxySpec.size)) || has(self.size)",message="Invalid cluster size, it is either not specified or value is not valid"
 // +kubebuilder:validation:XValidation:rule="[has(self.tls), has(self.tlsCertificate), has(self.tlsCertManagerIssuerName), has(self.tlsSecretName)].filter(x, x).size() <= 1",message="specify only one TLS option: tls, tlsCertificate, tlsCertManagerIssuerName, or tlsSecretName"
+// +kubebuilder:validation:XValidation:rule="!has(self.cleanupData) || !self.cleanupData || (has(self.cleanupMetadata) && self.cleanupMetadata)",message="cleanupData requires cleanupMetadata to be enabled"
 type AIStoreSpec struct {
 	// Size of the cluster i.e. number of proxies and number of targets.
 	// This can be changed by specifying size in either `proxySpec` or `targetSpec`.
@@ -363,15 +364,21 @@ type AIStoreSpec struct {
 	ShutdownCluster *bool `json:"shutdownCluster,omitempty"`
 
 	// CleanupMetadata determines whether to clean up cluster and bucket metadata when the cluster is decommissioned.
-	// When enabled, the cluster will fully decommission, removing metadata and optionally deleting user data.
+	// When enabled, the cluster will fully decommission, removing metadata and optionally deleting user data according to the CleanupData option.
 	// When disabled, the operator will call the AIS shutdown API to preserve metadata before deleting other k8s resources.
-	// The metadata stored in the state PVCs will be preserved to be usable in a future AIS deployment.
+	// State cleanup:
+	//  - All state PVCs will be deleted and cleaned up according to the storage class
+	//  - Any state host paths will be cleaned with operator-managed jobs
+	// Data cleanup:
+	//  - Data PVCs will be deleted but data on disk will remain unless specified by CleanupData
+	//  - The reclamation behavior for PVs bound to the PVCs depends on the PVC's reclaim policy, or, if the PV was dynamically provisioned, on the reclaim policy of the associated StorageClass.
 	// +optional
 	CleanupMetadata *bool `json:"cleanupMetadata,omitempty"`
 
-	// CleanupData determines whether to clean up PVCs and user data (including buckets and objects) when the cluster is decommissioned.
-	// The reclamation of PVs linked to the PVCs depends on the PV reclaim policy or the default policy of the associated StorageClass.
-	// This field is relevant only if you are deleting the CR (leading to decommissioning of the cluster).
+	// CleanupData determines whether to clean up user data (including buckets and objects) when the cluster is decommissioned.
+	// This field is relevant only if CleanupMetadata is true when deleting the AIStore CR (leading to decommissioning of the cluster).
+	// If this option is enabled, AIStore itself will delete user data on cluster decommission.
+	// If not enabled, user data cleanup will be left to the PV reclaim policy.
 	// +optional
 	CleanupData *bool `json:"cleanupData,omitempty"`
 

operator/api/v1beta1/aistore_webhook.go

@@ -132,13 +132,29 @@ func (aisw *AIStoreWebhook) validateSpec(ctx context.Context, ais *AIStore) (adm
 	)
 }
 
+func (ais *AIStore) validateCleanupConfig() (admission.Warnings, error) {
+	if !ais.ShouldCleanupMetadata() {
+		return nil, nil
+	}
+	if ais.Spec.StateStorageClass != nil {
+		return nil, nil
+	}
+	if len(ais.Spec.TargetSpec.NodeSelector) == 0 || len(ais.Spec.ProxySpec.NodeSelector) == 0 {
+		return admission.Warnings{
+			"cleanupMetadata is enabled with hostpath state and empty nodeSelector; host cleanup jobs will run on ALL nodes in the cluster",
+		}, nil
+	}
+	return nil, nil
+}
+
 func (ais *AIStore) ValidateSpec(_ context.Context, extraValidations ...func() (admission.Warnings, error)) (admission.Warnings, error) {
 	var allWarnings admission.Warnings
 	base := []func() (admission.Warnings, error){
 		ais.validateSize,
 		ais.validateStateStorage,
 		ais.validateAutoScaling,
 		ais.validateServiceSpec,
+		ais.validateCleanupConfig,
 	}
 
 	validations := make(

operator/config/base/crd/ais.nvidia.com_aistores.yaml

@@ -1374,16 +1374,22 @@ spec:
                 type: string
               cleanupData:
                 description: |-
-                  CleanupData determines whether to clean up PVCs and user data (including buckets and objects) when the cluster is decommissioned.
-                  The reclamation of PVs linked to the PVCs depends on the PV reclaim policy or the default policy of the associated StorageClass.
-                  This field is relevant only if you are deleting the CR (leading to decommissioning of the cluster).
+                  CleanupData determines whether to clean up user data (including buckets and objects) when the cluster is decommissioned.
+                  This field is relevant only if CleanupMetadata is true when deleting the AIStore CR (leading to decommissioning of the cluster).
+                  If this option is enabled, AIStore itself will delete user data on cluster decommission.
+                  If not enabled, user data cleanup will be left to the PV reclaim policy.
                 type: boolean
               cleanupMetadata:
                 description: |-
                   CleanupMetadata determines whether to clean up cluster and bucket metadata when the cluster is decommissioned.
-                  When enabled, the cluster will fully decommission, removing metadata and optionally deleting user data.
+                  When enabled, the cluster will fully decommission, removing metadata and optionally deleting user data according to the CleanupData option.
                   When disabled, the operator will call the AIS shutdown API to preserve metadata before deleting other k8s resources.
-                  The metadata stored in the state PVCs will be preserved to be usable in a future AIS deployment.
+                  State cleanup:
+                   - All state PVCs will be deleted and cleaned up according to the storage class
+                   - Any state host paths will be cleaned with operator-managed jobs
+                  Data cleanup:
+                   - Data PVCs will be deleted but data on disk will remain unless specified by CleanupData
+                   - The reclamation behavior for PVs bound to the PVCs depends on the PVC's reclaim policy, or, if the PV was dynamically provisioned, on the reclaim policy of the associated StorageClass.
                 type: boolean
               clusterDomain:
                 description: 'Defines the cluster domain name for DNS. Default: cluster.local.'
@@ -5902,6 +5908,9 @@ spec:
                 or tlsSecretName'
               rule: '[has(self.tls), has(self.tlsCertificate), has(self.tlsCertManagerIssuerName),
                 has(self.tlsSecretName)].filter(x, x).size() <= 1'
+            - message: cleanupData requires cleanupMetadata to be enabled
+              rule: '!has(self.cleanupData) || !self.cleanupData || (has(self.cleanupMetadata)
+                && self.cleanupMetadata)'
           status:
             description: AIStoreStatus defines the observed state of AIStore
             properties:

operator/helm/ais-operator/templates/aistore-crd.yaml

@@ -1376,16 +1376,22 @@ spec:
                 type: string
               cleanupData:
                 description: |-
-                  CleanupData determines whether to clean up PVCs and user data (including buckets and objects) when the cluster is decommissioned.
-                  The reclamation of PVs linked to the PVCs depends on the PV reclaim policy or the default policy of the associated StorageClass.
-                  This field is relevant only if you are deleting the CR (leading to decommissioning of the cluster).
+                  CleanupData determines whether to clean up user data (including buckets and objects) when the cluster is decommissioned.
+                  This field is relevant only if CleanupMetadata is true when deleting the AIStore CR (leading to decommissioning of the cluster).
+                  If this option is enabled, AIStore itself will delete user data on cluster decommission.
+                  If not enabled, user data cleanup will be left to the PV reclaim policy.
                 type: boolean
               cleanupMetadata:
                 description: |-
                   CleanupMetadata determines whether to clean up cluster and bucket metadata when the cluster is decommissioned.
-                  When enabled, the cluster will fully decommission, removing metadata and optionally deleting user data.
+                  When enabled, the cluster will fully decommission, removing metadata and optionally deleting user data according to the CleanupData option.
                   When disabled, the operator will call the AIS shutdown API to preserve metadata before deleting other k8s resources.
-                  The metadata stored in the state PVCs will be preserved to be usable in a future AIS deployment.
+                  State cleanup:
+                   - All state PVCs will be deleted and cleaned up according to the storage class
+                   - Any state host paths will be cleaned with operator-managed jobs
+                  Data cleanup:
+                   - Data PVCs will be deleted but data on disk will remain unless specified by CleanupData
+                   - The reclamation behavior for PVs bound to the PVCs depends on the PVC's reclaim policy, or, if the PV was dynamically provisioned, on the reclaim policy of the associated StorageClass.
                 type: boolean
               clusterDomain:
                 description: 'Defines the cluster domain name for DNS. Default: cluster.local.'
@@ -5890,6 +5896,9 @@ spec:
                 or tlsSecretName'
               rule: '[has(self.tls), has(self.tlsCertificate), has(self.tlsCertManagerIssuerName),
                 has(self.tlsSecretName)].filter(x, x).size() <= 1'
+            - message: cleanupData requires cleanupMetadata to be enabled
+              rule: '!has(self.cleanupData) || !self.cleanupData || (has(self.cleanupMetadata)
+                && self.cleanupMetadata)'
           status:
             description: AIStoreStatus defines the observed state of AIStore
             properties:

operator/pkg/client/api.go

@@ -10,8 +10,6 @@ import (
 	"reflect"
 
 	aisv1 "github.com/ais-operator/api/v1beta1"
-	"github.com/ais-operator/pkg/resources/proxy"
-	"github.com/ais-operator/pkg/resources/target"
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -137,22 +135,6 @@ func (c *K8sClient) GetPod(ctx context.Context, name types.NamespacedName) (*cor
 
 func (c *K8sClient) Status() client.StatusWriter { return c.client.Status() }
 
-// listPodsAndUpdateNodeNames lists pods based on the provided label selector and updates the uniqueNodeNames map with the node names of those pods
-func (c *K8sClient) listPodsAndUpdateNodeNames(ctx context.Context, ais *aisv1.AIStore, labelSelector map[string]string, uniqueNodeNames sets.Set[string]) error {
-	pods, err := c.ListPods(ctx, ais, labelSelector)
-	if err != nil {
-		return err
-	}
-	for i := range pods.Items {
-		pod := &pods.Items[i]
-		// check if the pod is running on a node (not failed or pending)
-		if pod.Spec.NodeName != "" {
-			uniqueNodeNames.Insert(pod.Spec.NodeName)
-		}
-	}
-	return nil
-}
-
 // ListNodesMatchingSelector returns a NodeList matching the given node selector
 func (c *K8sClient) ListNodesMatchingSelector(ctx context.Context, nodeSelector map[string]string) (*corev1.NodeList, error) {
 	nodeList := &corev1.NodeList{}
@@ -161,16 +143,23 @@ func (c *K8sClient) ListNodesMatchingSelector(ctx context.Context, nodeSelector
 	return nodeList, err
 }
 
-// ListNodesRunningAIS returns a map of unique node names where AIS pods are running
-func (c *K8sClient) ListNodesRunningAIS(ctx context.Context, ais *aisv1.AIStore) ([]string, error) {
-	uniqueNodeNames := sets.New[string]()
-	if err := c.listPodsAndUpdateNodeNames(ctx, ais, proxy.RequiredPodLabels(ais), uniqueNodeNames); err != nil {
-		return nil, err
-	}
-	if err := c.listPodsAndUpdateNodeNames(ctx, ais, target.RequiredPodLabels(ais), uniqueNodeNames); err != nil {
-		return nil, err
+// ListNodesMatchingAISSelectors returns the union of node names matching
+// the proxy or target NodeSelector specs. A nil/empty selector matches all nodes.
+func (c *K8sClient) ListNodesMatchingAISSelectors(ctx context.Context, ais *aisv1.AIStore) ([]string, error) {
+	nodeNames := sets.New[string]()
+	for _, selector := range []map[string]string{
+		ais.Spec.TargetSpec.NodeSelector,
+		ais.Spec.ProxySpec.NodeSelector,
+	} {
+		nodes, err := c.ListNodesMatchingSelector(ctx, selector)
+		if err != nil {
+			return nil, err
+		}
+		for i := range nodes.Items {
+			nodeNames.Insert(nodes.Items[i].Name)
+		}
 	}
-	return uniqueNodeNames.UnsortedList(), nil
+	return nodeNames.UnsortedList(), nil
 }
 
 //////////////////////////////////////

operator/pkg/controllers/cluster_cleanup.go

@@ -16,14 +16,11 @@ import (
 	"github.com/ais-operator/pkg/resources/statsd"
 	"github.com/ais-operator/pkg/resources/target"
 	batchv1 "k8s.io/api/batch/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 func (r *AIStoreReconciler) cleanup(ctx context.Context, ais *aisv1.AIStore) (updated bool, err error) {
-	nodeNames, err := r.k8sClient.ListNodesRunningAIS(ctx, ais)
-	if err != nil {
-		r.log.Error(err, "Failed to list nodes running AIS")
-	}
 	updated, err = cmn.AnyFunc(
 		func() (bool, error) { return r.cleanupTarget(ctx, ais) },
 		func() (bool, error) { return r.cleanupProxy(ctx, ais) },
@@ -33,28 +30,26 @@ func (r *AIStoreReconciler) cleanup(ctx context.Context, ais *aisv1.AIStore) (up
 		func() (bool, error) { return r.cleanupPVC(ctx, ais) },
 		func() (bool, error) { return r.cleanupTLS(ctx, ais) },
 	)
-	if updated && ais.ShouldCleanupMetadata() {
-		err = r.createCleanupJobs(ctx, ais, nodeNames)
-		if err != nil {
-			return
-		}
-		err = r.updateStatusWithState(ctx, ais, aisv1.HostCleanup)
-		if err != nil {
-			return
-		}
+	if err != nil {
+		return
+	}
+	// If not using storage class, trigger jobs to clean up host mounts
+	// Note this is not part of the above list to avoid host cleanup jobs blocking any K8s resource cleanup
+	if ais.ShouldCleanupMetadata() && ais.Spec.StateStorageClass == nil {
+		err = r.cleanupHostStateMounts(ctx, ais)
 	}
 	return
 }
 
 func (r *AIStoreReconciler) createCleanupJobs(ctx context.Context, ais *aisv1.AIStore, nodes []string) error {
-	if ais.Spec.StateStorageClass != nil {
-		return nil
-	}
 	logger := logf.FromContext(ctx)
 	logger.Info("Creating manual cleanup jobs", "nodes", nodes)
 	for _, nodeName := range nodes {
 		jobDef := cmn.NewCleanupJob(ais, nodeName)
 		if err := r.k8sClient.Create(ctx, jobDef); err != nil {
+			if apierrors.IsAlreadyExists(err) {
+				continue
+			}
 			logger.Error(err, "Failed to create cleanup job", "name", jobDef.Name, "node", nodeName)
 			return err
 		}
@@ -111,17 +106,27 @@ func (r *AIStoreReconciler) deleteFinishedJobs(ctx context.Context, jobs *batchv
 	return remaining, nil
 }
 
+func (r *AIStoreReconciler) cleanupHostStateMounts(ctx context.Context, ais *aisv1.AIStore) error {
+	nodeNames, err := r.k8sClient.ListNodesMatchingAISSelectors(ctx, ais)
+	if err != nil {
+		return err
+	}
+	err = r.createCleanupJobs(ctx, ais, nodeNames)
+	if err != nil {
+		return err
+	}
+	return r.updateStatusWithState(ctx, ais, aisv1.HostCleanup)
+}
+
 func (r *AIStoreReconciler) cleanupPVC(ctx context.Context, ais *aisv1.AIStore) (bool, error) {
+	// Note: Data PVCs must be deleted in-sync with state PVCs
+	// Otherwise we risk new PVs being created on nodes with conflicting pre-existing state,
+	// including cluster maps with ordinal-based addresses, e.g. ais-target-0.ais.svc.cluster.local
 	if !ais.ShouldCleanupMetadata() {
 		return false, nil
 	}
-	if ais.Spec.CleanupData != nil && *ais.Spec.CleanupData {
-		return r.deleteAllPVCs(ctx, ais)
-	}
-	if ais.Spec.StateStorageClass != nil {
-		return r.deleteStatePVCs(ctx, ais)
-	}
-	return false, nil
+	// Includes all PVCs (both data and state)
+	return r.deleteAllPVCs(ctx, ais)
 }
 
 func (r *AIStoreReconciler) deleteAllPVCs(ctx context.Context, ais *aisv1.AIStore) (bool, error) {
@@ -134,17 +139,6 @@ func (r *AIStoreReconciler) deleteAllPVCs(ctx context.Context, ais *aisv1.AIStor
 	return r.k8sClient.DeletePVCs(ctx, ais.Namespace, proxy.RequiredPodLabels(ais), nil)
 }
 
-// Cleans up only dynamically created volumes by adding a filter by the defined state storage class
-func (r *AIStoreReconciler) deleteStatePVCs(ctx context.Context, ais *aisv1.AIStore) (bool, error) {
-	r.log.Info("Cleaning up dynamic target PVCs")
-	updated, err := r.k8sClient.DeletePVCs(ctx, ais.Namespace, target.RequiredPodLabels(ais), ais.Spec.StateStorageClass)
-	if err != nil {
-		return updated, err
-	}
-	r.log.Info("Cleaning up dynamic proxy PVCs")
-	return r.k8sClient.DeletePVCs(ctx, ais.Namespace, proxy.RequiredPodLabels(ais), ais.Spec.StateStorageClass)
-}
-
 func (r *AIStoreReconciler) cleanupRBAC(ctx context.Context, ais *aisv1.AIStore) (anyUpdated bool, err error) {
 	return cmn.AnyFunc(
 		func() (bool, error) {

operator/tests/e2e/cluster_test.go

@@ -509,6 +509,7 @@ var _ = Describe("Run Controller", func() {
 
 		It("Re-deploying with CleanupMetadata disabled should recover cluster", func(ctx context.Context) {
 			cluArgs.CleanupMetadata = false
+			cluArgs.CleanupData = false
 			By("Deploy with cleanupMetadata false")
 			cc := newClientCluster(ctx, AISTestCfg, WorkerCfg.K8sClient, cluArgs)
 			defer func() {
@@ -537,6 +538,7 @@ var _ = Describe("Run Controller", func() {
 			cc.destroyClusterOnly()
 			// Cleanup metadata to remove PVCs so we can destroyAndCleanup PVs at the end
 			cluArgs.CleanupMetadata = true
+			cluArgs.CleanupData = true
 			// Same cluster should recover all the same data and metadata
 			By("Redeploy with cleanupMetadata true")
 			cc.recreate(ctx, cluArgs)
@@ -546,12 +548,14 @@ var _ = Describe("Run Controller", func() {
 
 		It("Should detect port change when cluster is redeployed with different port", func(ctx context.Context) {
 			cluArgs.CleanupMetadata = false
+			cluArgs.CleanupData = false
 			By("Deploy initial cluster with default ports")
 			cc := newClientCluster(ctx, AISTestCfg, WorkerCfg.K8sClient, cluArgs)
 			defer func() {
 				cc.printLogs(ctx)
 				// Ensure final cleanup has CleanupMetadata enabled
 				cluArgs.CleanupMetadata = true
+				cluArgs.CleanupData = true
 				cc.destroyAndCleanup()
 			}()
 			cc.create(ctx)