fix: harden PVC and disaggregated scale reconciliation

pkg/common/utils/k8s/client.go

@@ -30,7 +30,6 @@ import (
 	v2 "k8s.io/api/autoscaling/v2"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -378,14 +377,21 @@ func GetFoundationDBCluster(ctx context.Context, k8sclient client.Client, namesp
 
 // DeletePVC clean up existing pvc by pvc name, namespace and labels
 func DeletePVC(ctx context.Context, k8sclient client.Client, namespace, pvcName string, labels map[string]string) error {
-	pvc := corev1.PersistentVolumeClaim{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      pvcName,
-			Namespace: namespace,
-			Labels:    labels,
-		},
-	}
-	err := k8sclient.Delete(ctx, &pvc)
+	pvc, err := GetPVC(ctx, k8sclient, pvcName, namespace)
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	original := pvc.DeepCopy()
+	pvc.Finalizers = resource.RemoveOperatorPVCFinalizers(pvc.Finalizers)
+	if len(original.Finalizers) != len(pvc.Finalizers) {
+		if err := k8sclient.Patch(ctx, pvc, client.MergeFrom(original)); err != nil {
+			return err
+		}
+	}
+	err = k8sclient.Delete(ctx, pvc)
 	if err != nil && !apierrors.IsNotFound(err) {
 		return err
 	}

pkg/common/utils/k8s/client_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/apache/doris-operator/pkg/common/utils/resource"
 	appv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -217,6 +218,9 @@ func Test_DeletePVC(t *testing.T) {
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "test1",
 				Namespace: "test",
+				Finalizers: []string{
+					"selectdb.doris.com/pvc-finalizer",
+				},
 			},
 			Spec: corev1.PersistentVolumeClaimSpec{},
 		},
@@ -239,4 +243,8 @@ func Test_DeletePVC(t *testing.T) {
 			t.Errorf("delete pvc failed, pvc name=%s, err=%s", nn.Name, err.Error())
 		}
 	}
+	var pvc corev1.PersistentVolumeClaim
+	if err := fakeClient.Get(context.Background(), types.NamespacedName{Namespace: "test", Name: "test1"}, &pvc); !apierrors.IsNotFound(err) {
+		t.Fatalf("pvc test1 still exists after delete: %v", err)
+	}
 }

pkg/common/utils/mysql/mysql.go

@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/go-sql-driver/mysql"
 	_ "github.com/go-sql-driver/mysql"
@@ -210,6 +211,13 @@ func (db *DB) DropObserver(nodes []*Frontend) error {
 	return nil
 }
 
+func IsRetryableDropObserverError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(strings.ToLower(err.Error()), "drop fe node not in safe time")
+}
+
 func (db *DB) GetObservers() ([]*Frontend, error) {
 	frontends, err := db.ShowFrontends()
 	if err != nil {

pkg/common/utils/mysql/mysql_test.go

@@ -189,6 +189,21 @@ func Test_DropObserver(t *testing.T) {
 	}
 }
 
+func Test_IsRetryableDropObserverError(t *testing.T) {
+	retryable := fmt.Errorf(
+		"drop observer fe-2:9010 failed: %w",
+		errors.New("drop fe node not in safe time, try later"))
+	if !IsRetryableDropObserverError(retryable) {
+		t.Fatal("expected Doris safe-time refusal to be retryable")
+	}
+	if IsRetryableDropObserverError(errors.New("access denied")) {
+		t.Fatal("expected unrelated SQL error to remain non-retryable")
+	}
+	if IsRetryableDropObserverError(nil) {
+		t.Fatal("expected nil error to be non-retryable")
+	}
+}
+
 func Test_GetObservers(t *testing.T) {
 	mysql_db, mock, err := sqlmock.New()
 	if err != nil {

pkg/common/utils/resource/persistent_volume_claim.go

@@ -57,7 +57,6 @@ func BuildPVC(volume dorisv1.PersistentVolume, labels map[string]string, namespa
 			Namespace:   namespace,
 			Labels:      labels,
 			Annotations: annotations,
-			Finalizers:  []string{pvc_finalizer},
 		},
 		Spec: volume.PersistentVolumeClaimSpec,
 	}
@@ -75,13 +74,22 @@ func BuildDisaggregatedPVC(
 			Namespace:   namespace,
 			Labels:      labels,
 			Annotations: pvcTemplate.Annotations,
-			Finalizers:  []string{pvcFinalizerApache},
 		},
 		Spec: pvcTemplate.Spec,
 	}
 	return pvc
 }
 
+func RemoveOperatorPVCFinalizers(finalizers []string) []string {
+	result := make([]string, 0, len(finalizers))
+	for _, finalizer := range finalizers {
+		if finalizer != pvc_finalizer && finalizer != pvcFinalizerApache {
+			result = append(result, finalizer)
+		}
+	}
+	return result
+}
+
 // finalAnnotations is a combination of user annotations and operator default annotations
 func buildPVCAnnotations(volume dorisv1.PersistentVolume) Annotations {
 	annotations := Annotations{}

pkg/common/utils/resource/persistent_volume_claim_test.go

@@ -19,7 +19,9 @@ package resource
 
 import (
 	"fmt"
+
 	dorisv1 "github.com/apache/doris-operator/api/doris/v1"
+	corev1 "k8s.io/api/core/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"testing"
 )
@@ -40,6 +42,28 @@ func Test_BuildPVCAnnotations(t *testing.T) {
 	}
 }
 
+func TestBuildPVCDoesNotAddOperatorFinalizer(t *testing.T) {
+	pvc := BuildPVC(
+		dorisv1.PersistentVolume{}, map[string]string{"app": "doris"}, "default", "doris-fe", "0")
+	if len(pvc.Finalizers) != 0 {
+		t.Fatalf("BuildPVC finalizers = %v, want none", pvc.Finalizers)
+	}
+
+	pvc = BuildDisaggregatedPVC(
+		corev1.PersistentVolumeClaim{}, map[string]string{"app": "doris"}, "default", "doris-cg", "0")
+	if len(pvc.Finalizers) != 0 {
+		t.Fatalf("BuildDisaggregatedPVC finalizers = %v, want none", pvc.Finalizers)
+	}
+}
+
+func TestRemoveOperatorPVCFinalizersPreservesKubernetesFinalizers(t *testing.T) {
+	got := RemoveOperatorPVCFinalizers(
+		[]string{pvc_finalizer, pvcFinalizerApache, "kubernetes.io/pvc-protection"})
+	if len(got) != 1 || got[0] != "kubernetes.io/pvc-protection" {
+		t.Fatalf("RemoveOperatorPVCFinalizers = %v", got)
+	}
+}
+
 func Test_Result(t *testing.T) {
 	res := ctrl.Result{}
 	if res.IsZero() {

pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go

@@ -28,7 +28,7 @@ import (
 	"sync"
 
 	dv1 "github.com/apache/doris-operator/api/disaggregated/v1"
-	"github.com/apache/doris-operator/pkg/common/utils"
+	dorisv1 "github.com/apache/doris-operator/api/doris/v1"
 	"github.com/apache/doris-operator/pkg/common/utils/k8s"
 	"github.com/apache/doris-operator/pkg/common/utils/mysql"
 	"github.com/apache/doris-operator/pkg/common/utils/resource"
@@ -246,9 +246,13 @@ func (dcgs *DisaggregatedComputeGroupsController) reconcileStatefulset(ctx conte
 	}
 
 	if !volumeClaimTemplatesEqual(st.Spec.VolumeClaimTemplates, est.Spec.VolumeClaimTemplates) {
-		msg := fmt.Sprintf("compute group %s storage template is immutable after creation; modifying BE file_cache_path or persistent volume settings requires recreating the compute group", cg.UniqueId)
-		klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset immutable storage template changed, namespace=%s name=%s, err=%s", st.Namespace, st.Name, msg)
-		return &sc.Event{Type: sc.EventWarning, Reason: sc.CGStorageTemplateImmutable, Message: msg}, errors.New(msg)
+		if volumeClaimTemplatesResizeOnly(st.Spec.VolumeClaimTemplates, est.Spec.VolumeClaimTemplates) {
+			st.Spec.VolumeClaimTemplates = deepCopyVolumeClaimTemplates(est.Spec.VolumeClaimTemplates)
+		} else {
+			msg := fmt.Sprintf("compute group %s storage template is immutable after creation; modifying BE file_cache_path or persistent volume settings requires recreating the compute group", cg.UniqueId)
+			klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset immutable storage template changed, namespace=%s name=%s, err=%s", st.Namespace, st.Name, msg)
+			return &sc.Event{Type: sc.EventWarning, Reason: sc.CGStorageTemplateImmutable, Message: msg}, errors.New(msg)
+		}
 	}
 
 	// Direct-drop scale-down is owned by the graceful state machine when the image
@@ -352,6 +356,36 @@ func volumeClaimTemplatesEqual(new, old []corev1.PersistentVolumeClaim) bool {
 	return equality.Semantic.DeepEqual(normalizedNew, normalizedOld)
 }
 
+func volumeClaimTemplatesResizeOnly(new, old []corev1.PersistentVolumeClaim) bool {
+	if len(new) != len(old) {
+		return false
+	}
+
+	normalizedNew := make([]corev1.PersistentVolumeClaim, len(new))
+	normalizedOld := make([]corev1.PersistentVolumeClaim, len(old))
+	for i := range new {
+		normalizedNew[i] = normalizeVolumeClaimTemplate(new[i])
+		normalizedOld[i] = normalizeVolumeClaimTemplate(old[i])
+
+		newQuantity, newExists := normalizedNew[i].Spec.Resources.Requests[corev1.ResourceStorage]
+		oldQuantity, oldExists := normalizedOld[i].Spec.Resources.Requests[corev1.ResourceStorage]
+		if !newExists || !oldExists || newQuantity.Cmp(oldQuantity) < 0 {
+			return false
+		}
+		normalizedNew[i].Spec.Resources.Requests[corev1.ResourceStorage] = oldQuantity
+	}
+
+	return equality.Semantic.DeepEqual(normalizedNew, normalizedOld)
+}
+
+func deepCopyVolumeClaimTemplates(templates []corev1.PersistentVolumeClaim) []corev1.PersistentVolumeClaim {
+	copied := make([]corev1.PersistentVolumeClaim, len(templates))
+	for i := range templates {
+		copied[i] = *templates[i].DeepCopy()
+	}
+	return copied
+}
+
 func normalizeVolumeClaimTemplate(pvc corev1.PersistentVolumeClaim) corev1.PersistentVolumeClaim {
 	pvc.TypeMeta = metav1.TypeMeta{}
 	pvc.ObjectMeta = metav1.ObjectMeta{
@@ -360,6 +394,10 @@ func normalizeVolumeClaimTemplate(pvc corev1.PersistentVolumeClaim) corev1.Persi
 		Annotations: normalizeStringMap(pvc.Annotations),
 	}
 	pvc.Status = corev1.PersistentVolumeClaimStatus{}
+	delete(pvc.Annotations, dorisv1.ComponentResourceHash)
+	if len(pvc.Annotations) == 0 {
+		pvc.Annotations = nil
+	}
 	if pvc.Spec.VolumeMode == nil {
 		volumeMode := corev1.PersistentVolumeFilesystem
 		pvc.Spec.VolumeMode = &volumeMode
@@ -664,44 +702,7 @@ func (dcgs *DisaggregatedComputeGroupsController) ClearStatefulsetUnusedPVCs(ctx
 		return nil
 	}
 
-	var clearPVC []string
-	//we should use statefulset replicas for avoiding the phase=scaleDown, when phase `scaleDown` cg' replicas is less than statefuslet.
-	stsName := ddc.GetCGStatefulsetName(cg)
-	sts, err := k8s.GetStatefulSet(ctx, dcgs.K8sclient, ddc.Namespace, stsName)
-	if err != nil {
-		klog.Errorf("DisaggregatedComputeGroupsController ClearStatefulsetUnusedPVCs get statefulset namespace=%s, name=%s, failed, err=%s", ddc.Namespace, stsName, err.Error())
-		//waiting next reconciling.
-		return nil
-	}
-	replicas := *sts.Spec.Replicas
-	for _, pvc := range currentPVCs.Items {
-		pvcName := pvc.Name
-		sl := strings.Split(pvcName, stsName+"-")
-		if len(sl) != 2 {
-			klog.Errorf("DisaggregatedComputeGroupsController ClearStatefulsetUnusedPVCs namespace %s name %s not format pvc name format.", ddc.Namespace, pvcName)
-			continue
-		}
-		var index int64
-		var perr error
-		index, perr = strconv.ParseInt(sl[1], 10, 32)
-		if perr != nil {
-			klog.Errorf("DisaggregatedComputeGroupsController ClearStatefulsetUnusedPVCs namespace %s name %s index parse failed, err=%s", ddc.Namespace, pvcName, perr.Error())
-			continue
-		}
-		if int32(index) >= replicas {
-			clearPVC = append(clearPVC, pvcName)
-		}
-	}
-
-	var mergeError error
-	for _, pvcName := range clearPVC {
-		if err = k8s.DeletePVC(ctx, dcgs.K8sclient, ddc.Namespace, pvcName, pvcLabels); err != nil {
-			dcgs.K8srecorder.Event(ddc, string(sc.EventWarning), sc.PVCDeleteFailed, err.Error())
-			klog.Errorf("ClearStatefulsetUnusedPVCs deletePVCs failed: namespace %s, name %s delete pvc %s, err: %s .", ddc.Namespace, pvcName, pvcName, err.Error())
-			mergeError = utils.MergeError(mergeError, err)
-		}
-	}
-	return mergeError
+	return nil
 }
 
 func (dcgs *DisaggregatedComputeGroupsController) GetControllerName() string {

pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go

@@ -22,15 +22,55 @@ import (
 	"testing"
 
 	dv1 "github.com/apache/doris-operator/api/disaggregated/v1"
+	dorisv1 "github.com/apache/doris-operator/api/doris/v1"
 	sc "github.com/apache/doris-operator/pkg/controller/sub_controller"
 	appv1 "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"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
+func TestClearStatefulsetUnusedPVCsRetainsScaledDownClaims(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add core scheme failed: %v", err)
+	}
+	replicas := int32(2)
+	ddc := newTestDDC()
+	cg := newTestCG("cg1")
+	cg.Replicas = &replicas
+	ddc.Spec.ComputeGroups = []dv1.ComputeGroup{*cg}
+	labels := map[string]string{
+		dv1.DorisDisaggregatedClusterName:          ddc.Name,
+		dv1.DorisDisaggregatedComputeGroupUniqueId: cg.UniqueId,
+		dv1.DorisDisaggregatedPodType:              "compute",
+	}
+	pvc := &corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "data-doris-cg1-3",
+			Namespace: ddc.Namespace,
+			Labels:    labels,
+			UID:       types.UID("historical-uid"),
+		},
+	}
+	dcgs := &DisaggregatedComputeGroupsController{}
+	dcgs.K8sclient = fake.NewClientBuilder().WithScheme(scheme).WithObjects(pvc).Build()
+
+	if err := dcgs.ClearStatefulsetUnusedPVCs(context.Background(), ddc, dv1.ComputeGroupStatus{UniqueId: cg.UniqueId}); err != nil {
+		t.Fatalf("clear unused pvc failed: %v", err)
+	}
+	var retained corev1.PersistentVolumeClaim
+	if err := dcgs.K8sclient.Get(context.Background(), types.NamespacedName{Namespace: pvc.Namespace, Name: pvc.Name}, &retained); err != nil {
+		t.Fatalf("historical pvc was deleted: %v", err)
+	}
+	if retained.UID != pvc.UID {
+		t.Fatalf("historical pvc uid = %s, want %s", retained.UID, pvc.UID)
+	}
+}
+
 func TestReconcileStatefulsetRejectsStorageTemplateChange(t *testing.T) {
 	scheme := runtime.NewScheme()
 	if err := appv1.AddToScheme(scheme); err != nil {
@@ -40,7 +80,8 @@ func TestReconcileStatefulsetRejectsStorageTemplateChange(t *testing.T) {
 	ddc := newTestDDC()
 	cg := newTestCG("cg1")
 	existing := newTestStatefulSet(ddc.Namespace, ddc.GetCGStatefulsetName(cg), "100Gi")
-	desired := newTestStatefulSet(ddc.Namespace, ddc.GetCGStatefulsetName(cg), "200Gi")
+	desired := newTestStatefulSet(ddc.Namespace, ddc.GetCGStatefulsetName(cg), "100Gi")
+	desired.Spec.VolumeClaimTemplates[0].Spec.AccessModes = []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany}
 	dcgs := &DisaggregatedComputeGroupsController{}
 	dcgs.K8sclient = fake.NewClientBuilder().WithScheme(scheme).WithObjects(existing).Build()
 
@@ -56,6 +97,26 @@ func TestReconcileStatefulsetRejectsStorageTemplateChange(t *testing.T) {
 	}
 }
 
+func TestVolumeClaimTemplatesResizeOnlyAllowsExpansion(t *testing.T) {
+	existing := newTestStatefulSet("default", "doris-cg1", "100Gi").Spec.VolumeClaimTemplates
+	desired := newTestStatefulSet("default", "doris-cg1", "200Gi").Spec.VolumeClaimTemplates
+	existing[0].Annotations = map[string]string{dorisv1.ComponentResourceHash: "old-hash"}
+	desired[0].Annotations = map[string]string{dorisv1.ComponentResourceHash: "new-hash"}
+
+	if !volumeClaimTemplatesResizeOnly(desired, existing) {
+		t.Fatal("volumeClaimTemplatesResizeOnly should allow storage expansion")
+	}
+}
+
+func TestVolumeClaimTemplatesResizeOnlyRejectsShrink(t *testing.T) {
+	existing := newTestStatefulSet("default", "doris-cg1", "200Gi").Spec.VolumeClaimTemplates
+	desired := newTestStatefulSet("default", "doris-cg1", "100Gi").Spec.VolumeClaimTemplates
+
+	if volumeClaimTemplatesResizeOnly(desired, existing) {
+		t.Fatal("volumeClaimTemplatesResizeOnly should reject storage shrink")
+	}
+}
+
 func newTestDDC() *dv1.DorisDisaggregatedCluster {
 	return &dv1.DorisDisaggregatedCluster{
 		ObjectMeta: metav1.ObjectMeta{