diff --git a/pkg/common/utils/k8s/client.go b/pkg/common/utils/k8s/client.go index ecd40b20..6209f9d3 100644 --- a/pkg/common/utils/k8s/client.go +++ b/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 } diff --git a/pkg/common/utils/k8s/client_test.go b/pkg/common/utils/k8s/client_test.go index 6b5c78b1..11b3329d 100644 --- a/pkg/common/utils/k8s/client_test.go +++ b/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) + } } diff --git a/pkg/common/utils/mysql/mysql.go b/pkg/common/utils/mysql/mysql.go index aabe3ce5..1b2944ca 100644 --- a/pkg/common/utils/mysql/mysql.go +++ b/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 { diff --git a/pkg/common/utils/mysql/mysql_test.go b/pkg/common/utils/mysql/mysql_test.go index cb7d8c20..35d05218 100644 --- a/pkg/common/utils/mysql/mysql_test.go +++ b/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 { diff --git a/pkg/common/utils/resource/persistent_volume_claim.go b/pkg/common/utils/resource/persistent_volume_claim.go index 8a9f122c..a1c0ebbb 100644 --- a/pkg/common/utils/resource/persistent_volume_claim.go +++ b/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{} diff --git a/pkg/common/utils/resource/persistent_volume_claim_test.go b/pkg/common/utils/resource/persistent_volume_claim_test.go index 1b47d2a0..d6655720 100644 --- a/pkg/common/utils/resource/persistent_volume_claim_test.go +++ b/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() { diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go index bb3bdd32..28064adc 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller.go +++ b/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 { diff --git a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go b/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go index 87249a4f..60d0b9f8 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/computegroups/controller_test.go +++ b/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{ diff --git a/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go b/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go index f9c60d44..236d2211 100644 --- a/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go +++ b/pkg/controller/sub_controller/disaggregated_cluster/disaggregated_fe/controller.go @@ -304,6 +304,11 @@ func (dfc *DisaggregatedFEController) reconcileStatefulset(ctx context.Context, // if fe scale, drop fe node by http if willRemovedAmount < 0 || cluster.Status.FEStatus.Phase == v1.ScaleDownFailed { if err := dfc.dropFEBySQLClient(ctx, dfc.K8sclient, cluster); err != nil { + if mysql.IsRetryableDropObserverError(err) { + cluster.Status.FEStatus.Phase = v1.Reconciling + klog.Infof("ScaleDownFE temporarily deferred by Doris, will retry: %s", err.Error()) + return nil, nil + } cluster.Status.FEStatus.Phase = v1.ScaleDownFailed klog.Errorf("ScaleDownFE failed, err:%s ", err.Error()) return &sc.Event{Type: sc.EventWarning, Reason: sc.FEHTTPFailed, Message: err.Error()}, diff --git a/pkg/controller/sub_controller/disaggregated_subcontroller.go b/pkg/controller/sub_controller/disaggregated_subcontroller.go index 7515888f..cfb5a4d7 100644 --- a/pkg/controller/sub_controller/disaggregated_subcontroller.go +++ b/pkg/controller/sub_controller/disaggregated_subcontroller.go @@ -37,7 +37,9 @@ import ( "github.com/spf13/viper" appv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" + k8sresource "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/record" @@ -412,6 +414,19 @@ func (d *DisaggregatedSubDefaultController) ReconcilePVC( if manager != string(v1.PVCProvisionerOperator) { continue } + + for j := range oldPvcList.Items { + oldPvc := &oldPvcList.Items[j] + if !pvcMatchesTemplate(oldPvc.Name, sts.Name, pvcTemplates[i].Name) { + continue + } + + newQuantity := pvcTemplates[i].Spec.Resources.Requests[corev1.ResourceStorage] + if event, err := d.reconcileExistingPVC(ctx, ddc, oldPvc, newQuantity); err != nil { + return event, err + } + } + for ordinal := range *commonSpec.Replicas { pvc := resource.BuildDisaggregatedPVC(pvcTemplates[i], selector, ddc.Namespace, sts.Name, strconv.FormatInt(int64(ordinal), 10)) oldPvc := getPvc(oldPvcList, pvc.Name) @@ -433,34 +448,56 @@ func (d *DisaggregatedSubDefaultController) ReconcilePVC( continue } - oldQuantity := oldPvc.Spec.Resources.Requests[corev1.ResourceStorage] - newQuantity := pvc.Spec.Resources.Requests[corev1.ResourceStorage] - //if !oldQuantity.Equal(newQuantity){ - if oldQuantity.Cmp(newQuantity) == -1 { - // pvc need update - oldPvc.Spec.Resources.Requests[corev1.ResourceStorage] = newQuantity - if err := d.K8sclient.Patch(ctx, oldPvc, client.Merge); err != nil { - message := fmt.Sprintf("ReconcilePVC patch pvc failed, namespace: %s, ddc name: %s, patch pvc %s, error: %s", ddc.Namespace, ddc.Name, pvc.Name, err.Error()) - klog.Errorf(message) - return &Event{Type: EventWarning, Reason: PVCUpdateFailed, Message: message}, err - } - message := fmt.Sprintf("ReconcilePVC patch pvc, namespace: %s, ddc name: %s update pvc %s .", ddc.Namespace, ddc.Name, pvc.Name) - klog.Infof(message) - d.K8srecorder.Event(ddc, string(EventNormal), PVCUpdate, message) - } + } + } - if oldQuantity.Cmp(newQuantity) == 1 { - message := fmt.Sprintf("ReconcilePVC pvc resize is rejected, PVC shrinking is not supported. namespace: %s, ddc name: %s, resize pvc %s", ddc.Namespace, ddc.Name, pvc.Name) - klog.Warningf(message) - d.K8srecorder.Event(ddc, string(EventWarning), PVCUpdateFailed, message) - } + return nil, nil +} +func (d *DisaggregatedSubDefaultController) reconcileExistingPVC( + ctx context.Context, + ddc *v1.DorisDisaggregatedCluster, + pvc *corev1.PersistentVolumeClaim, + newQuantity k8sresource.Quantity, +) (*Event, error) { + original := pvc.DeepCopy() + oldQuantity := pvc.Spec.Resources.Requests[corev1.ResourceStorage] + if oldQuantity.Cmp(newQuantity) < 0 { + pvc.Spec.Resources.Requests[corev1.ResourceStorage] = newQuantity + } + pvc.Finalizers = resource.RemoveOperatorPVCFinalizers(pvc.Finalizers) + + if !equality.Semantic.DeepEqual(original.Spec.Resources.Requests, pvc.Spec.Resources.Requests) || + !equality.Semantic.DeepEqual(original.Finalizers, pvc.Finalizers) { + if err := d.K8sclient.Patch(ctx, pvc, client.MergeFrom(original)); err != nil { + message := fmt.Sprintf("ReconcilePVC patch pvc failed, namespace: %s, ddc name: %s, patch pvc %s, error: %s", ddc.Namespace, ddc.Name, pvc.Name, err.Error()) + klog.Error(message) + return &Event{Type: EventWarning, Reason: PVCUpdateFailed, Message: message}, err } + message := fmt.Sprintf("ReconcilePVC patch pvc, namespace: %s, ddc name: %s update pvc %s .", ddc.Namespace, ddc.Name, pvc.Name) + klog.Info(message) + d.K8srecorder.Event(ddc, string(EventNormal), PVCUpdate, message) + } + + if oldQuantity.Cmp(newQuantity) > 0 { + message := fmt.Sprintf("ReconcilePVC pvc resize is rejected, PVC shrinking is not supported. namespace: %s, ddc name: %s, resize pvc %s", ddc.Namespace, ddc.Name, pvc.Name) + klog.Warning(message) + d.K8srecorder.Event(ddc, string(EventWarning), PVCUpdateFailed, message) } return nil, nil } +func pvcMatchesTemplate(pvcName, stsName, volumeName string) bool { + prefix := resource.BuildPVCName(stsName, "", volumeName) + if !strings.HasPrefix(pvcName, prefix) { + return false + } + ordinal := strings.TrimPrefix(pvcName, prefix) + _, err := strconv.ParseUint(ordinal, 10, 32) + return err == nil +} + func getPvc(pvcs corev1.PersistentVolumeClaimList, pvcName string) *corev1.PersistentVolumeClaim { for _, pvc := range pvcs.Items { if pvc.Name == pvcName { diff --git a/pkg/controller/sub_controller/disaggregated_subcontroller_test.go b/pkg/controller/sub_controller/disaggregated_subcontroller_test.go index 86e4457f..bbb98562 100644 --- a/pkg/controller/sub_controller/disaggregated_subcontroller_test.go +++ b/pkg/controller/sub_controller/disaggregated_subcontroller_test.go @@ -17,12 +17,101 @@ package sub_controller import ( + "context" + "fmt" + "testing" + v1 "github.com/apache/doris-operator/api/disaggregated/v1" + operatorresource "github.com/apache/doris-operator/pkg/common/utils/resource" + appv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" - "testing" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client/fake" ) +func TestReconcilePVCExpandsHistoricalClaims(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 := &v1.DorisDisaggregatedCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "doris", Namespace: "default"}, + } + cg := &v1.ComputeGroup{ + UniqueId: "cg1", + CommonSpec: v1.CommonSpec{ + Replicas: &replicas, + LogNotStore: true, + PersistentVolumes: []v1.PersistentVolume{{ + MountPaths: []string{"/data"}, + PVCProvisioner: v1.PVCProvisionerOperator, + PersistentVolumeClaimSpec: corev1.PersistentVolumeClaimSpec{ + Resources: corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{ + corev1.ResourceStorage: resource.MustParse("20Gi"), + }}, + }, + }}, + }, + } + selector := map[string]string{"app": "doris-cg1"} + sts := &appv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "doris-cg1", Namespace: ddc.Namespace}, + Spec: appv1.StatefulSetSpec{Selector: &metav1.LabelSelector{MatchLabels: selector}}, + } + + objects := make([]runtime.Object, 0, 4) + uids := make(map[string]types.UID, 4) + for ordinal := 0; ordinal < 4; ordinal++ { + name := operatorresource.BuildPVCName(sts.Name, fmt.Sprintf("%d", ordinal), "data") + uid := types.UID(fmt.Sprintf("uid-%d", ordinal)) + uids[name] = uid + objects = append(objects, &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: ddc.Namespace, + Labels: selector, + UID: uid, + Finalizers: []string{"apache.doris.org/pvc-finalizer", "kubernetes.io/pvc-protection"}, + }, + Spec: corev1.PersistentVolumeClaimSpec{ + Resources: corev1.VolumeResourceRequirements{Requests: corev1.ResourceList{ + corev1.ResourceStorage: resource.MustParse("10Gi"), + }}, + }, + }) + } + + controller := &DisaggregatedSubDefaultController{ + K8sclient: fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(objects...).Build(), + K8srecorder: record.NewFakeRecorder(20), + } + if _, err := controller.ReconcilePVC(context.Background(), ddc, map[string]interface{}{}, v1.DisaggregatedBE, sts, cg); err != nil { + t.Fatalf("reconcile pvc failed: %v", err) + } + + for name, uid := range uids { + var pvc corev1.PersistentVolumeClaim + if err := controller.K8sclient.Get(context.Background(), types.NamespacedName{Namespace: ddc.Namespace, Name: name}, &pvc); err != nil { + t.Fatalf("get pvc %s failed: %v", name, err) + } + if pvc.UID != uid { + t.Fatalf("pvc %s uid = %s, want %s", name, pvc.UID, uid) + } + if got := pvc.Spec.Resources.Requests[corev1.ResourceStorage]; got.Cmp(resource.MustParse("20Gi")) != 0 { + t.Fatalf("pvc %s storage = %s, want 20Gi", name, got.String()) + } + if len(pvc.Finalizers) != 1 || pvc.Finalizers[0] != "kubernetes.io/pvc-protection" { + t.Fatalf("pvc %s finalizers = %v", name, pvc.Finalizers) + } + } +} + func TestDisaggregatedSubDefaultController_BuildVolumesVolumeMountsAndPVCs_empty_persistentVolume(t *testing.T) { confMap := map[string]interface{}{} commonSpec := v1.CommonSpec{} diff --git a/pkg/controller/sub_controller/sub_controller.go b/pkg/controller/sub_controller/sub_controller.go index 2e2f01b7..2e7b5f2c 100644 --- a/pkg/controller/sub_controller/sub_controller.go +++ b/pkg/controller/sub_controller/sub_controller.go @@ -517,24 +517,32 @@ func (d *SubDefaultController) patchPVCs(ctx context.Context, dcr *dorisv1.Doris //patch already exist in k8s . prepared := true for _, pvc := range pvcs { + original := pvc.DeepCopy() oldCapacity := pvc.Spec.Resources.Requests[corev1.ResourceStorage] newCapacity := volume.PersistentVolumeClaimSpec.Resources.Requests[corev1.ResourceStorage] - if !oldCapacity.Equal(newCapacity) { + capacityChanged := !oldCapacity.Equal(newCapacity) + if capacityChanged { // if pvc need update, the resource have not prepared, return false. prepared = false - eventType := EventNormal - reason := PVCUpdate - message := pvc.Name + " update successfully!" pvc.Spec.Resources.Requests[corev1.ResourceStorage] = newCapacity - if err := d.K8sclient.Patch(ctx, &pvc, client.Merge); err != nil { - klog.Errorf("SubDefaultController namespace %s name %s patch pvc %s failed, %s", dcr.Namespace, dcr.Name, pvc.Name, err.Error()) - eventType = EventWarning - reason = PVCUpdateFailed - message = pvc.Name + " update failed, " + err.Error() - } + } + pvc.Finalizers = resource.RemoveOperatorPVCFinalizers(pvc.Finalizers) + finalizersChanged := len(original.Finalizers) != len(pvc.Finalizers) + if !capacityChanged && !finalizersChanged { + continue + } - d.K8srecorder.Event(dcr, string(eventType), reason, message) + eventType := EventNormal + reason := PVCUpdate + message := pvc.Name + " update successfully!" + if err := d.K8sclient.Patch(ctx, &pvc, client.MergeFrom(original)); err != nil { + klog.Errorf("SubDefaultController namespace %s name %s patch pvc %s failed, %s", dcr.Namespace, dcr.Name, pvc.Name, err.Error()) + eventType = EventWarning + reason = PVCUpdateFailed + message = pvc.Name + " update failed, " + err.Error() } + + d.K8srecorder.Event(dcr, string(eventType), reason, message) } // if need add new pvc, the resource prepared not finished, return false.