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

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package computegroups

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/apache/doris-operator/pkg/common/utils/runtimeconfig"
	"regexp"
	"strconv"
	"strings"
	"sync"

	dv1 "github.com/apache/doris-operator/api/disaggregated/v1"
	"github.com/apache/doris-operator/pkg/common/utils"
	"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"
	"github.com/apache/doris-operator/pkg/common/utils/set"
	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/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/rest"
	"k8s.io/klog/v2"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var _ sc.DisaggregatedSubController = &DisaggregatedComputeGroupsController{}

var (
	disaggregatedComputeGroupsController = "disaggregatedComputeGroupsController"
)

type DisaggregatedComputeGroupsController struct {
	sc.DisaggregatedSubDefaultController
	RestConfig *rest.Config
}

func New(mgr ctrl.Manager) *DisaggregatedComputeGroupsController {
	return &DisaggregatedComputeGroupsController{
		DisaggregatedSubDefaultController: sc.DisaggregatedSubDefaultController{
			K8sclient:      mgr.GetClient(),
			K8srecorder:    mgr.GetEventRecorderFor(disaggregatedComputeGroupsController),
			ControllerName: disaggregatedComputeGroupsController,
		},
		RestConfig: mgr.GetConfig(),
	}
}

func (dcgs *DisaggregatedComputeGroupsController) Sync(ctx context.Context, obj client.Object) error {
	ddc := obj.(*dv1.DorisDisaggregatedCluster)
	if len(ddc.Spec.ComputeGroups) == 0 {
		klog.Errorf("disaggregatedComputeGroupsController sync disaggregatedDorisCluster namespace=%s,name=%s have not compute group spec.", ddc.Namespace, ddc.Name)
		dcgs.K8srecorder.Event(ddc, string(sc.EventWarning), string(sc.ComputeGroupsEmpty), "compute group empty, the cluster will not work normal.")
		return nil
	}

	if !dcgs.feAvailable(ddc) {
		dcgs.K8srecorder.Event(ddc, string(sc.EventNormal), string(sc.WaitFEAvailable), "fe have not ready.")
		return nil
	}

	// validating compute group information.
	if event, res := dcgs.validateComputeGroup(ddc.Spec.ComputeGroups); !res {
		klog.Errorf("disaggregatedComputeGroupsController namespace=%s name=%s validateComputeGroup have not match specifications %s.", ddc.Namespace, ddc.Name, sc.EventString(event))
		dcgs.K8srecorder.Eventf(ddc, string(event.Type), string(event.Reason), event.Message)
		return errors.New("validating compute group failed")
	}

	var errs []error
	cgs := ddc.Spec.ComputeGroups
	for i, _ := range cgs {

		if event, err := dcgs.computeGroupSync(ctx, ddc, &cgs[i]); err != nil {
			if event != nil {
				dcgs.K8srecorder.Event(ddc, string(event.Type), string(event.Reason), event.Message)
			}
			errs = append(errs, err)
			klog.Errorf("disaggregatedComputeGroupsController computeGroups sync failed, compute group Uniqueid %s  sync failed, err=%s", cgs[i].UniqueId, err.Error())
		}
	}

	if len(errs) != 0 {
		msg := fmt.Sprintf("disaggregatedComputeGroupsController sync namespace: %s ,ddc name: %s, compute group has the following error: ", ddc.Namespace, ddc.Name)
		for _, err := range errs {
			msg += err.Error()
		}
		return errors.New(msg)
	}

	return nil
}

// validate compute group config information.
func (dcgs *DisaggregatedComputeGroupsController) validateComputeGroup(cgs []dv1.ComputeGroup) (*sc.Event, bool) {
	dupl := dcgs.validateDuplicated(cgs)
	if dupl != "" {
		klog.Errorf("disaggregatedComputeGroupsController validateComputeGroup validate Duplicated have duplicate unique identifier %s.", dupl)
		return &sc.Event{Type: sc.EventWarning, Reason: sc.CGUniqueIdentifierDuplicate, Message: "unique identifier " + dupl + " duplicate in compute groups."}, false
	}

	if reg, res := dcgs.validateRegex(cgs); !res {
		klog.Errorf("disaggregatedComputeGroupsController validateComputeGroup validateRegex %s have not match regular expression", reg)
		return &sc.Event{Type: sc.EventWarning, Reason: sc.CGUniqueIdentifierNotMatchRegex, Message: reg}, false
	}

	return nil, true
}

func (dcgs *DisaggregatedComputeGroupsController) feAvailable(ddc *dv1.DorisDisaggregatedCluster) bool {
	//if fe deploy in k8s, should wait fe available
	//1. wait for fe ok.
	endpoints := corev1.Endpoints{}
	if err := dcgs.K8sclient.Get(context.Background(), types.NamespacedName{Namespace: ddc.Namespace, Name: ddc.GetFEServiceName()}, &endpoints); err != nil {
		klog.Infof("disaggregatedComputeGroupsController Sync wait fe service name %s available occur failed %s\n", ddc.GetFEServiceName(), err.Error())
		return false
	}

	for _, sub := range endpoints.Subsets {
		if len(sub.Addresses) > 0 {
			return true
		}
	}
	return false
}

func (dcgs *DisaggregatedComputeGroupsController) computeGroupSync(ctx context.Context, ddc *dv1.DorisDisaggregatedCluster, cg *dv1.ComputeGroup) (*sc.Event, error) {
	if cg.Replicas == nil {
		cg.Replicas = resource.GetInt32Pointer(1)
	}
	cvs := dcgs.GetConfigValuesFromConfigMaps(ddc.Namespace, resource.BE_RESOLVEKEY, cg.CommonSpec.ConfigMaps)
	st := dcgs.NewStatefulset(ddc, cg, cvs)
	if err := runtimeconfig.Prepare(ctx, dcgs.K8sclient, dcgs.K8srecorder, ddc, st, cvs); err != nil {
		return nil, err
	}
	internalSvc := dcgs.newInternalService(ddc, cg, cvs)
	externalSvc := dcgs.newExternalService(ddc, cg, cvs)
	dcgs.initialCGStatus(ddc, cg)

	dcgs.CheckSecretMountPath(ddc, cg.Secrets)
	dcgs.CheckSecretExist(ctx, ddc, cg.Secrets)

	// Reconcile internal headless service.
	// During upgrade from older versions, the existing service may not be headless (has a ClusterIP assigned).
	// Since K8s does not allow changing spec.clusterIP on an existing service, we must delete and recreate it.
	if err := dcgs.reconcileInternalService(ctx, internalSvc); err != nil {
		klog.Errorf("disaggregatedComputeGroupsController reconcile internal service namespace %s name %s failed, err=%s", internalSvc.Namespace, internalSvc.Name, err.Error())
		return &sc.Event{Type: sc.EventWarning, Reason: sc.ServiceApplyedFailed, Message: err.Error()}, err
	}

	// Reconcile external service for load-balanced access.
	event, err := dcgs.DefaultReconcileService(ctx, externalSvc)
	if err != nil {
		klog.Errorf("disaggregatedComputeGroupsController reconcile external service namespace %s name %s failed, err=%s", externalSvc.Namespace, externalSvc.Name, err.Error())
		return event, err
	}

	event, err = dcgs.reconcileStatefulset(ctx, st, ddc, cg)
	if err != nil {
		klog.Errorf("disaggregatedComputeGroupsController reconcile statefulset namespace %s name %s failed, err=%s", st.Namespace, st.Name, err.Error())
		return event, err
	}

	event, err = dcgs.ReconcilePVC(ctx, ddc, cvs, dv1.DisaggregatedBE, st, cg)
	if err != nil {
		klog.Errorf("computeGroupSync ReconcilePVC failed, namespace: %s, ddc name %s, cgName: %s, error=%s!", ddc.Namespace, ddc.Name, cg.UniqueId, err.Error())
	}

	return event, err
}

// reconcileInternalService reconciles the internal headless service for a compute group.
// If the existing service is not headless (upgrade scenario), it deletes and recreates it,
// because K8s does not allow mutating spec.clusterIP from a valid IP to "None".
func (dcgs *DisaggregatedComputeGroupsController) reconcileInternalService(ctx context.Context, svc *corev1.Service) error {
	var existingSvc corev1.Service
	err := dcgs.K8sclient.Get(ctx, types.NamespacedName{Namespace: svc.Namespace, Name: svc.Name}, &existingSvc)
	if err != nil {
		if apierrors.IsNotFound(err) {
			return k8s.CreateClientObject(ctx, dcgs.K8sclient, svc)
		}
		return err
	}

	// If the existing service is already headless, use normal reconcile path.
	if existingSvc.Spec.ClusterIP == "None" {
		_, err = dcgs.DefaultReconcileService(ctx, svc)
		return err
	}

	// Existing service is not headless — delete and recreate.
	klog.Infof("reconcileInternalService existing service %s/%s is not headless (clusterIP=%s), deleting and recreating as headless.",
		existingSvc.Namespace, existingSvc.Name, existingSvc.Spec.ClusterIP)
	if err = k8s.DeleteService(ctx, dcgs.K8sclient, svc.Namespace, svc.Name); err != nil {
		return err
	}
	return k8s.CreateClientObject(ctx, dcgs.K8sclient, svc)
}

// reconcileStatefulset return bool means reconcile print error message.
func (dcgs *DisaggregatedComputeGroupsController) reconcileStatefulset(ctx context.Context, st *appv1.StatefulSet, cluster *dv1.DorisDisaggregatedCluster, cg *dv1.ComputeGroup) (*sc.Event, error) {
	//use new default value before apply new statefulset, when creating and apply spec change.
	ndf := func(st *appv1.StatefulSet, est *appv1.StatefulSet) {
		dcgs.useNewDefaultValuesInStatefulset(st)
	}

	var est appv1.StatefulSet
	if err := dcgs.K8sclient.Get(ctx, types.NamespacedName{Namespace: st.Namespace, Name: st.Name}, &est); apierrors.IsNotFound(err) {
		//if err = k8s.CreateClientObject(ctx, dcgs.K8sclient, st); err != nil {
		//	klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset create statefulset namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
		//	return &sc.Event{Type: sc.EventWarning, Reason: sc.CGCreateResourceFailed, Message: err.Error()}, err
		//}

		//use apply replace create, if use create the default image not replace with be image and annotation for equal not assign.
		if err = k8s.ApplyStatefulSet(ctx, dcgs.K8sclient, st, func(new, est *appv1.StatefulSet) bool {
			dcgs.RestrictConditionsEqual(new, est)
			//creating use the function to assign equal annotation.
			return resource.StatefulsetDeepEqualWithKey(new, est, dv1.DisaggregatedSpecHashValueAnnotation, false)
		}, ndf); err != nil {
			klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset create statefulset namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
			return &sc.Event{Type: sc.EventWarning, Reason: sc.CGCreateResourceFailed, Message: err.Error()}, err
		}

		return nil, nil
	} else if err != nil {
		klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset get statefulset failed, namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
		return nil, err
	}

	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)
	}

	// Finish an existing graceful action before starting SQL decommission. Keep
	// replicas unchanged while that action owns the pod lifecycle.
	if cluster.Spec.EnableDecommission && hasGracefulAction(&est) {
		replicas := *est.Spec.Replicas
		st.Spec.Replicas = &replicas
	}
	if pending, err := dcgs.reconcileDecommissionScaleDown(ctx, st, &est, cluster, cg); pending || err != nil {
		if err != nil {
			return &sc.Event{Type: sc.EventWarning, Reason: sc.CGSqlExecFailed, Message: err.Error()}, err
		}
		return nil, nil
	}

	// Direct-drop scale-down is owned by the graceful state machine when the image
	// supports it. SQL decommission is handled above before applying resources.
	gracefulInProgress := false
	if !cluster.Spec.EnableDecommission {
		gracefulInProgress = dcgs.reconcileGracefulStatefulSet(ctx, st, &est, cluster, cg)
	}

	if !gracefulInProgress {
		err := dcgs.preApplyStatefulSet(ctx, st, &est, cluster, cg)
		if err != nil {
			klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset preApplyStatefulSet namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
			return &sc.Event{Type: sc.EventWarning, Reason: sc.CGSqlExecFailed, Message: err.Error()}, err
		}

		// be decommission processing, skip apply statefulset.
		if skipApplyStatefulset(cluster, cg) {
			return nil, nil
		}
	}

	if cluster.Spec.EnableDecommission {
		dcgs.reconcileGracefulStatefulSet(ctx, st, &est, cluster, cg)
	}

	if st.Spec.UpdateStrategy.Type == appv1.OnDeleteStatefulSetStrategyType {
		dcgs.clearStatefulSetRollingUpdate(ctx, st.Namespace, st.Name)
	}
	if err := k8s.ApplyStatefulSet(ctx, dcgs.K8sclient, st, func(new, est *appv1.StatefulSet) bool {
		dcgs.RestrictConditionsEqual(new, est)
		//store annotations "doris.disaggregated.cluster/generation={generation}" on statefulset
		//store annotations "doris.disaggregated.cluster/update-{uniqueid}=true/false" on DorisDisaggregatedCluster
		businessEqual := dcgs.businessStatefulSetEqual(new, est)
		if !businessEqual {
			if len(new.Annotations) == 0 {
				new.Annotations = map[string]string{}
			}
			st_annos := (resource.Annotations)(new.Annotations)
			st_annos.Add(dv1.UpdateStatefulsetGeneration, strconv.FormatInt(cluster.Generation, 10))
			if len(cluster.Annotations) == 0 {
				cluster.Annotations = map[string]string{}
			}
			ddc_annos := (resource.Annotations)(cluster.Annotations)
			msUniqueIdKey := strings.ToLower(fmt.Sprintf(dv1.UpdateStatefulsetName, cluster.GetCGStatefulsetName(cg)))
			ddc_annos.Add(msUniqueIdKey, "true")
		}
		return businessEqual && gracefulStatefulSetControlEqual(new, est)

	}, ndf); err != nil {
		klog.Errorf("disaggregatedComputeGroupsController reconcileStatefulset apply statefulset namespace=%s name=%s failed, err=%s", st.Namespace, st.Name, err.Error())
		return &sc.Event{Type: sc.EventWarning, Reason: sc.CGApplyResourceFailed, Message: err.Error()}, err
	}
	return nil, nil
}

func (dcgs *DisaggregatedComputeGroupsController) reconcileGracefulStatefulSet(
	ctx context.Context,
	st, est *appv1.StatefulSet,
	cluster *dv1.DorisDisaggregatedCluster,
	cg *dv1.ComputeGroup,
) bool {
	if dcgs.RestConfig == nil {
		return false
	}
	var cgStatus *dv1.ComputeGroupStatus
	for i := range cluster.Status.ComputeGroupStatuses {
		if cluster.Status.ComputeGroupStatuses[i].UniqueId == cg.UniqueId {
			cgStatus = &cluster.Status.ComputeGroupStatuses[i]
			break
		}
	}
	if cgStatus == nil {
		return false
	}

	skipApply, err := dcgs.gracefulRolloutReconcile(ctx, dcgs.RestConfig, st, est, cluster, cg, cgStatus)
	if err != nil {
		klog.Errorf("reconcileStatefulset gracefulRolloutReconcile failed: %v", err)
	}
	if skipApply {
		// Do not fall through to the legacy DropBE workflow while the graceful
		// state machine owns this StatefulSet.
		ensureOnDeleteStrategy(st)
	}
	return skipApply
}

func volumeClaimTemplatesEqual(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])
	}

	return equality.Semantic.DeepEqual(normalizedNew, normalizedOld)
}

func normalizeVolumeClaimTemplate(pvc corev1.PersistentVolumeClaim) corev1.PersistentVolumeClaim {
	pvc.TypeMeta = metav1.TypeMeta{}
	pvc.ObjectMeta = metav1.ObjectMeta{
		Name:        pvc.Name,
		Labels:      normalizeStringMap(pvc.Labels),
		Annotations: normalizeStringMap(pvc.Annotations),
	}
	pvc.Status = corev1.PersistentVolumeClaimStatus{}
	if pvc.Spec.VolumeMode == nil {
		volumeMode := corev1.PersistentVolumeFilesystem
		pvc.Spec.VolumeMode = &volumeMode
	}
	return pvc
}

func normalizeStringMap(m map[string]string) map[string]string {
	if len(m) == 0 {
		return nil
	}
	normalized := make(map[string]string, len(m))
	for k, v := range m {
		normalized[k] = v
	}
	return normalized
}

// initial compute group status before sync resources. status changing with sync steps, and generate the last status by classify pods.
func (dcgs *DisaggregatedComputeGroupsController) initialCGStatus(ddc *dv1.DorisDisaggregatedCluster, cg *dv1.ComputeGroup) {
	cgss := ddc.Status.ComputeGroupStatuses
	//clusterId := ddc.GetCGId(cg)
	uniqueId := cg.UniqueId
	defaultStatus := dv1.ComputeGroupStatus{
		Phase:           dv1.Reconciling,
		UniqueId:        cg.UniqueId,
		StatefulsetName: ddc.GetCGStatefulsetName(cg),
		ServiceName:     ddc.GetCGServiceName(cg),
		//set for status updated.
		Replicas: *cg.Replicas,
	}

	for i := range cgss {
		if cgss[i].UniqueId == uniqueId {
			cgss[i].Replicas = *cg.Replicas

			return
		}
	}

	// Need to adjust by pointer
	(&ddc.Status).ComputeGroupStatuses = append((&ddc.Status).ComputeGroupStatuses, defaultStatus)
}

func (dcgs *DisaggregatedComputeGroupsController) businessStatefulSetEqual(new, est *appv1.StatefulSet) bool {
	nst := new.DeepCopy()
	eSt := est.DeepCopy()
	normalizeGracefulStatefulSetForCompare(nst)
	normalizeGracefulStatefulSetForCompare(eSt)
	equal := resource.StatefulsetDeepEqualWithKey(nst, eSt, dv1.DisaggregatedSpecHashValueAnnotation, false)
	if hashValue := nst.Annotations[dv1.DisaggregatedSpecHashValueAnnotation]; hashValue != "" {
		if new.Annotations == nil {
			new.Annotations = map[string]string{}
		}
		new.Annotations[dv1.DisaggregatedSpecHashValueAnnotation] = hashValue
	}
	return equal
}

func (dcgs *DisaggregatedComputeGroupsController) clearStatefulSetRollingUpdate(ctx context.Context, namespace, name string) {
	patch := []byte(`{"spec":{"updateStrategy":{"rollingUpdate":null}}}`)
	st := &appv1.StatefulSet{}
	st.Namespace = namespace
	st.Name = name
	if err := dcgs.K8sclient.Patch(ctx, st, client.RawPatch(types.MergePatchType, patch)); err != nil {
		klog.Errorf("clearStatefulSetRollingUpdate: failed to clear rollingUpdate for StatefulSet %s/%s: %v", namespace, name, err)
	}
}

// check compute groups unique identifier duplicated or not. return duplicated key.
func (dcgs *DisaggregatedComputeGroupsController) validateDuplicated(cgs []dv1.ComputeGroup) string {
	dupl := ""
	uniqueIds := set.NewSetString()
	for _, cg := range cgs {
		if uniqueIds.Find(cg.UniqueId) {
			dupl = dupl + cg.UniqueId + ";"
		}
		uniqueIds.Add(cg.UniqueId)
	}

	return dupl
}

// checking the cg name compliant with regular expression or not.
func (dcgs *DisaggregatedComputeGroupsController) validateRegex(cgs []dv1.ComputeGroup) (string, bool) {
	var regStr = ""
	for _, cg := range cgs {
		res, err := regexp.Match(compute_group_name_regex, []byte(cg.UniqueId))
		if !res {
			regStr = regStr + cg.UniqueId + " not match " + compute_group_name_regex
		}
		//for debugging, output the error in log
		if err != nil {
			klog.Errorf("disaggregatedComputeGroupsController validateRegex compute group name %s failed, err=%s", cg.UniqueId, err.Error())
		}
	}
	if regStr != "" {
		return regStr, false
	}

	return "", true
}

// clear not configed cg resources, delete not configed cg status from ddc.status .
func (dcgs *DisaggregatedComputeGroupsController) ClearResources(ctx context.Context, obj client.Object) (bool, error) {
	ddc := obj.(*dv1.DorisDisaggregatedCluster)

	var eCGs []dv1.ComputeGroupStatus
	var delComputeGroupIds []string
	for i, cgs := range ddc.Status.ComputeGroupStatuses {
		exist := false
		for _, cg := range ddc.Spec.ComputeGroups {
			if cgs.UniqueId == cg.UniqueId {
				eCGs = append(eCGs, ddc.Status.ComputeGroupStatuses[i])
				exist = true
				break
			}
		}

		if !exist {
			delComputeGroupIds = append(delComputeGroupIds, cgs.ComputeGroupId)
		}
	}

	//list the svcs and stss owner reference to dorisDisaggregatedCluster.
	cls := dcgs.GetCG2LayerCommonSchedulerLabels(ddc.Name)
	svcs, err := k8s.ListServicesInNamespace(ctx, dcgs.K8sclient, ddc.Namespace, cls)
	if err != nil {
		klog.Errorf("DisaggregatedComputeGroupsController ListServicesInNamespace failed, dorisdisaggregatedcluster name=%s", ddc.Name)
		return false, err
	}
	stss, err := k8s.ListStatefulsetInNamespace(ctx, dcgs.K8sclient, ddc.Namespace, cls)
	if err != nil {
		klog.Errorf("DisaggregatedComputeGroupsController ListStatefulsetInNamespace failed, dorisdisaggregatedcluster name=%s", ddc.Name)
		return false, err
	}

	//clear unused service and statefulset.
	delSvcNames := dcgs.findUnusedSvcs(svcs, ddc)
	delStsNames, delUniqueIds := dcgs.findUnusedStssAndUniqueIds(stss, ddc)

	if err = dcgs.clearCGInDorisMeta(ctx, delComputeGroupIds, ddc); err != nil {
		return false, err
	}
	if err = dcgs.clearSvcs(ctx, delSvcNames, ddc); err != nil {
		return false, err
	}
	if err = dcgs.clearStatefulsets(ctx, delStsNames, ddc); err != nil {
		return false, err
	}

	//clear unused pvc
	for i := range eCGs {
		err = dcgs.ClearStatefulsetUnusedPVCs(ctx, ddc, eCGs[i])
		if err != nil {
			klog.Errorf("disaggregatedComputeGroupsController ClearStatefulsetUnusedPVCs clear ComputeGroup reduced replicas PVC failed, namespace=%s, ddc name=%s, uniqueId=%s err=%s", ddc.Namespace, ddc.Name, eCGs[i].UniqueId, err.Error())
		}
	}

	for _, uniqueId := range delUniqueIds {
		//new fake computeGroup status for clear all pvcs owner reference to deleted compute group.
		fakeCgs := dv1.ComputeGroupStatus{
			UniqueId: uniqueId,
		}
		err = dcgs.ClearStatefulsetUnusedPVCs(ctx, ddc, fakeCgs)
		if err != nil {
			klog.Errorf("disaggregatedComputeGroupsController ClearStatefulsetUnusedPVCs clear deleted compute group failed, namespace=%s, ddc name=%s, uniqueId=%s err=%s", ddc.Namespace, ddc.Name, uniqueId, err.Error())
		}
	}

	ddc.Status.ComputeGroupStatuses = eCGs
	return true, nil
}

func (dcgs *DisaggregatedComputeGroupsController) clearStatefulsets(ctx context.Context, stsNames []string, ddc *dv1.DorisDisaggregatedCluster) error {
	for _, name := range stsNames {
		if err := k8s.DeleteStatefulset(ctx, dcgs.K8sclient, ddc.Namespace, name); err != nil {
			klog.Errorf("DisaggregatedComputeGroupsController clear statefulset failed, namespace=%s, name =%s, err=%s", ddc.Namespace, name, err.Error())
			return err
		}
	}
	return nil
}

func (dcgs *DisaggregatedComputeGroupsController) clearSvcs(ctx context.Context, svcNames []string, ddc *dv1.DorisDisaggregatedCluster) error {
	for _, name := range svcNames {
		if err := k8s.DeleteService(ctx, dcgs.K8sclient, ddc.Namespace, name); err != nil {
			klog.Errorf("DisaggregatedComputeGroupsController clear service failed, namespace=%s, name =%s, err=%s", ddc.Namespace, name, err.Error())
			return err
		}
	}
	return nil
}

func (dcgs *DisaggregatedComputeGroupsController) clearCGInDorisMeta(ctx context.Context, cgids []string, ddc *dv1.DorisDisaggregatedCluster) error {
	if len(cgids) == 0 {
		return nil
	}

	sqlClient, err := dcgs.getMasterSqlClient(ctx, ddc)
	if err != nil {
		klog.Errorf("DisaggregatedComputeGroupsController clearCGInDorisMeta dropCGBySQLClient getMasterSqlClient failed: %s", err.Error())
		dcgs.K8srecorder.Event(ddc, string(sc.EventWarning), string(sc.CGSqlExecFailed), "computeGroupSync dropCGBySQLClient failed: "+err.Error())
		return err
	}
	defer sqlClient.Close()

	for _, cgid := range cgids {
		//clear cg, the keepAmount = 0
		err = dcgs.scaledOutBENodesByDrop(sqlClient, cgid, 0)
		if err != nil {
			klog.Errorf("DisaggregatedComputeGroupsController clearCGInDorisMeta dropCGBySQLClient failed: %s", err.Error())
			dcgs.K8srecorder.Event(ddc, string(sc.EventWarning), string(sc.CGSqlExecFailed), "computeGroupSync dropCGBySQLClient failed: "+err.Error())
			return err
		}
	}

	return nil
}

func (dcgs *DisaggregatedComputeGroupsController) findUnusedSvcs(svcs []corev1.Service, ddc *dv1.DorisDisaggregatedCluster) []string {
	var unusedSvcNames []string
	for i, _ := range svcs {
		own := ownerReference2ddc(&svcs[i], ddc)
		if !own {
			//not owner reference to ddc, should skip the service.
			continue
		}

		svcUniqueId := getUniqueIdFromClientObject(&svcs[i])
		exist := false
		for j := 0; j < len(ddc.Spec.ComputeGroups); j++ {
			if ddc.Spec.ComputeGroups[j].UniqueId == svcUniqueId {
				exist = true
				break
			}
		}

		if !exist {
			unusedSvcNames = append(unusedSvcNames, svcs[i].Name)
		}
	}

	return unusedSvcNames
}

func (dcgs *DisaggregatedComputeGroupsController) findUnusedStssAndUniqueIds(stss []appv1.StatefulSet, ddc *dv1.DorisDisaggregatedCluster) ([]string /*sts*/, []string /*uniqueIds*/) {
	var unusedStsNames []string
	var unusedUniqueIds []string
	for i, _ := range stss {
		own := ownerReference2ddc(&stss[i], ddc)
		if !own {
			//not owner reference tto ddc should skip the statefulset.
			continue
		}

		stsUniqueId := getUniqueIdFromClientObject(&stss[i])
		exist := false
		for j := 0; j < len(ddc.Spec.ComputeGroups); j++ {
			if ddc.Spec.ComputeGroups[j].UniqueId == stsUniqueId {
				exist = true
				break
			}
		}
		if !exist {
			unusedStsNames = append(unusedStsNames, stss[i].Name)
			unusedUniqueIds = append(unusedUniqueIds, stsUniqueId)
		}
	}

	return unusedStsNames, unusedUniqueIds
}

// ClearStatefulsetUnusedPVCs
// 1.delete unused pvc skip cluster is Suspend
// 2.delete unused pvc for statefulset
// 3.delete pvc if not used by any statefulset
func (dcgs *DisaggregatedComputeGroupsController) ClearStatefulsetUnusedPVCs(ctx context.Context, ddc *dv1.DorisDisaggregatedCluster, cgs dv1.ComputeGroupStatus) error {
	var cg *dv1.ComputeGroup
	for i := range ddc.Spec.ComputeGroups {
		/*	uniqueId := ddc.GetCGId(&ddc.Spec.ComputeGroups[i])
			if clusterId == cgs.ClusterId {
				cg = &ddc.Spec.ComputeGroups[i]
			}*/
		if ddc.Spec.ComputeGroups[i].UniqueId == cgs.UniqueId {
			cg = &ddc.Spec.ComputeGroups[i]
		}
	}

	currentPVCs := corev1.PersistentVolumeClaimList{}
	//pvcMap := make(map[string]*corev1.PersistentVolumeClaim)

	pvcLabels := dcgs.newCGPodsSelector(ddc.Name, cgs.UniqueId)

	if err := dcgs.K8sclient.List(ctx, &currentPVCs, client.InNamespace(ddc.Namespace), client.MatchingLabels(pvcLabels)); err != nil {
		dcgs.K8srecorder.Event(ddc, string(sc.EventWarning), sc.PVCListFailed, fmt.Sprintf("DisaggregatedComputeGroupsController ClearStatefulsetUnusedPVCs list pvc failed:%s!", err.Error()))
		return err
	}

	// now only clear scale down pod.
	if cg == nil {
		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 {
			retain, err := runtimeconfig.RetainPVCOnScaleDown(sts, pvcName)
			if err != nil {
				return err
			}
			if retain {
				continue
			}
			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
}

func (dcgs *DisaggregatedComputeGroupsController) GetControllerName() string {
	return dcgs.ControllerName
}

func (dcgs *DisaggregatedComputeGroupsController) UpdateComponentStatus(obj client.Object) error {
	ddc := obj.(*dv1.DorisDisaggregatedCluster)
	cgss := ddc.Status.ComputeGroupStatuses
	if len(cgss) == 0 {
		klog.Errorf("disaggregatedComputeGroupsController updateComponentStatus compute group status is empty!")
		return nil
	}

	errChan := make(chan error, len(cgss))
	wg := sync.WaitGroup{}
	wg.Add(len(cgss))
	for i, _ := range cgss {
		go func(idx int) {
			defer wg.Done()
			errChan <- dcgs.updateCGStatus(ddc, &cgss[idx])
		}(i)
	}

	wg.Wait()
	close(errChan)
	errMs := ""
	for err := range errChan {
		if err != nil {
			errMs += err.Error()
		}
	}

	for _, cgs := range ddc.Status.ComputeGroupStatuses {
		if cgs.ComputeGroupId == "" {
			dcgs.recordComputeGroupIds(ddc)
			break
		}
	}

	var fullAvailableCount int32
	var availableCount int32
	for _, cgs := range ddc.Status.ComputeGroupStatuses {
		if cgs.Phase == dv1.Ready {
			fullAvailableCount++
		}
		if cgs.AvailableReplicas > 0 {
			availableCount++
		}
	}
	ddc.Status.ClusterHealth.CGCount = int32(len(ddc.Status.ComputeGroupStatuses))
	ddc.Status.ClusterHealth.CGFullAvailableCount = fullAvailableCount
	ddc.Status.ClusterHealth.CGAvailableCount = availableCount
	if errMs == "" {
		return nil
	}
	return errors.New(errMs)
}

func (dcgs *DisaggregatedComputeGroupsController) recordComputeGroupIds(ddc *dv1.DorisDisaggregatedCluster) error {
	// get user and password
	adminUserName, password := dcgs.GetManagementAdminUserAndPWD(context.Background(), ddc)

	// get host and port
	// When the operator and dcr are deployed in different namespace, it will be inaccessible, so need to add the dcr svc namespace
	host := ddc.GetFEVIPAddresss()
	confMap := dcgs.GetConfigValuesFromConfigMaps(ddc.Namespace, resource.FE_RESOLVEKEY, ddc.Spec.FeSpec.ConfigMaps)
	queryPort := resource.GetPort(confMap, resource.QUERY_PORT)
	cfg := mysql.NewDBConfig()
	cfg.User = adminUserName
	cfg.Password = password
	cfg.Host = host
	cfg.Port = strconv.FormatInt(int64(queryPort), 10)

	tlsConfig, secretName := dcgs.DisaggregatedSubDefaultController.FindSecretTLSConfig(confMap, ddc)
	secret, _ := k8s.GetSecret(context.Background(), dcgs.K8sclient, ddc.Namespace, secretName)

	db, err := mysql.NewDorisSqlDB(cfg, tlsConfig, secret)
	if err != nil {
		klog.Errorf("DisaggregatedComputeGroupsController recordComputeGroupIds new doris client failed,err=%s", err.Error())
		return err
	}
	defer db.Close()

	backends, err := db.ShowBackends()
	if err != nil {
		klog.Errorf("DisaggregatedComputeGroupsController recordComputeGroupIds show backends failed, err=%s", err.Error())
		return err
	}

	m := map[string]string{} //statefulsetname:computegroupid
	for _, backend := range backends {
		tags := map[string]string{}
		err = json.Unmarshal([]byte(backend.Tag), &tags)
		if err != nil {
			klog.Errorf("DisaggregatedComputeGroupsController recordComputeGroupIds backend tag stirng to map failed, tag: %s, err: %s", backend.Tag, err.Error())
			return err
		}
		if _, ok := tags[mysql.COMPUTE_GROUP_ID]; !ok {
			errMsg := fmt.Sprintf("DisaggregatedComputeGroupsController recordComputeGroupIds backend tag get compute_group_name failed, tag: %s, err: no compute_group_id field found ", backend.Tag)
			klog.Errorf(errMsg)
			return errors.New(errMsg)
		}

		podName := strings.Split(backend.Host, ".")[0]
		re, _ := regexp.Compile("(.*)-[0-9]+$")
		matchs := re.FindStringSubmatch(podName)
		stsName := matchs[len(matchs)-1]
		m[stsName] = tags[mysql.COMPUTE_GROUP_ID]
	}

	for i, cgs := range ddc.Status.ComputeGroupStatuses {
		ddc.Status.ComputeGroupStatuses[i].ComputeGroupId = m[cgs.StatefulsetName]
	}
	return nil
}

func (dcgs *DisaggregatedComputeGroupsController) updateCGStatus(ddc *dv1.DorisDisaggregatedCluster, cgs *dv1.ComputeGroupStatus) error {
	stfName := cgs.StatefulsetName
	sts, err := k8s.GetStatefulSet(context.Background(), dcgs.K8sclient, ddc.Namespace, stfName)
	if err != nil {
		klog.Errorf("DisaggregatedComputeGroupsController updateCGStatus get statefulset %s failed, err=%s", stfName, err.Error())
		return err
	}

	//check statefulset updated or not, if this reconcile update the sts, should exclude the circumstance that get old sts and the pods not updated.
	updateStatefulsetKey := strings.ToLower(fmt.Sprintf(dv1.UpdateStatefulsetName, stfName))
	if _, updated := ddc.Annotations[updateStatefulsetKey]; updated {
		generation := dcgs.DisaggregatedSubDefaultController.ReturnStatefulsetUpdatedGeneration(sts, updateStatefulsetKey)
		//if this reconcile not update statefulset will not check the generation equals or not.
		if ddc.Generation != generation {
			return errors.New("waiting statefulset updated")
		}
	}

	selector := dcgs.newCGPodsSelector(ddc.Name, cgs.UniqueId)
	var podList corev1.PodList
	if err := dcgs.K8sclient.List(context.Background(), &podList, client.InNamespace(ddc.Namespace), client.MatchingLabels(selector)); err != nil {
		return err
	}

	updateRevision := sts.Status.UpdateRevision
	//check all pods controlled by new statefulset.
	allUpdated := dcgs.DisaggregatedSubDefaultController.StatefulsetControlledPodsAllUseNewUpdateRevision(updateRevision, podList.Items)

	var availableReplicas int32
	var creatingReplicas int32
	var failedReplicas int32
	//get all pod status that controlled by st.
	for _, pod := range podList.Items {
		if ready := k8s.PodIsReady(&pod.Status); ready {
			availableReplicas++
		} else if pod.Status.Phase == corev1.PodRunning || pod.Status.Phase == corev1.PodPending {
			creatingReplicas++
		} else {
			failedReplicas++
		}
	}

	cgs.AvailableReplicas = availableReplicas
	if hasGracefulAction(sts) {
		switch cgs.Phase {
		case dv1.Ready:
			cgs.Phase = dv1.GracefulRolling
		case dv1.Reconciling, dv1.GracefulRolling, dv1.GracefulScaling, dv1.GracefulDeleting:
			// Keep the phase in a reconciling/graceful state while the StatefulSet
			// still carries graceful-action annotation.
		}
		return nil
	}
	if allUpdated && availableReplicas == cgs.Replicas {
		cgs.Phase = dv1.Ready
	}
	return nil
}

func hasGracefulAction(sts *appv1.StatefulSet) bool {
	if sts == nil || sts.Annotations == nil {
		return false
	}
	return sts.Annotations[gracefulActionAnnotation] != ""
}