package regionalservice

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/google/go-cmp/cmp"
	"github.com/shopspring/decimal"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/yaml"

	"k8s.io/apimachinery/pkg/util/intstr"

	"k8s.io/apimachinery/pkg/api/resource"

	prometheusv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"github.com/tidbcloud/observability/pkg/common/configs"
	corev1 "k8s.io/api/core/v1"

	"github.com/pingcap/log"
	v1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/common/utils"
	"github.com/tidbcloud/observability/pkg/dataservice"
	"github.com/tidbcloud/observability/pkg/resourcemanager"
	clusterctl "github.com/tidbcloud/observability/pkg/resourceworker/controllers/cluster"
	resourceworker "github.com/tidbcloud/observability/pkg/resourceworker/worker"
	"go.uber.org/zap"
)

type ClusterHandler struct {
}

var clusterHandler *ClusterHandler

var once sync.Once

func GetHandler() *ClusterHandler {
	once.Do(func() {
		if clusterHandler == nil {
			clusterHandler = new(ClusterHandler)
		}
	})

	return clusterHandler
}

const (
	applicationIDK8sInfra               = "k8s-infra"
	applicationIDDevTierInfraF          = "devtier-infra-f"
	BizeTypeDP                          = "datamigration"
	BizeTypeServerless                  = "devtier"
	BizeTypeTidbCloud                   = "tidbcloud"
	O11yEndpointNamePrefixDataplane     = "o11yd"  //o11y dataplane endpoint
	O11yEndpointNamePrefixTiDB          = "o11yc"  //o11y cluster endpoint
	O11yEndpointNamePrefixDataMigration = "o11ydm" //o11y dm endpoint
	TrustPolicyPrincipalKeyFederated    = "Federated"
	TrustPolicyPrincipalKeyAWS          = "AWS"
)

func getCompatibleRequest(req *message.CreateClusterReq) {
	if req.ApplicationID == applicationIDDevTierInfraF {
		req.MuteAlert = true
	}
}

func dealuseRDS(ctx context.Context, biztype string) context.Context {
	conf, ok := configs.GetServiceInfo().BizTypeConfigs[biztype]
	if ok && conf.UseRDS {
		ctx = context.WithValue(ctx, constants.UseRDSKey, true)
	}
	return ctx
}
func checkuseRDS(ctx context.Context) bool {
	if b, ok := ctx.Value(constants.UseRDSKey).(bool); ok && b {
		return true
	}
	return false
}

// CreateCluster
// @Description: create a new observability cluster
// @Parameter ctx
// @Parameter req message.CreateClusterReq
// @return message.CreateClusterResp
// @return errors.Error
func (p *ClusterHandler) CreateCluster(ctx context.Context, req message.CreateClusterReq) (resp message.CreateClusterResp, err error) {
	ctx = dealuseRDS(ctx, req.BizType)
	if len(req.ID) > 0 {
		if existed, e := resourcemanager.GetClusterReaderWriter(ctx).IsClusterIDExists(ctx, req.ID); e != nil {
			return resp, e
		} else if existed {
			return resp, errors.NewErrorf(errors.ParameterInvalid, "cluster id %s has been existed", req.ID)
		}
	} else {
		req.ID = utils.GenerateID()
	}

	// if existed, return both cluster info and errors.DuplicatedCluster
	if existed, e := resourcemanager.GetClusterReaderWriter(ctx).QueryClusters(ctx, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType, message.ContinueRequest{}); e != nil {
		return resp, e
	} else if len(existed.Items) > 0 {
		if existed.Items[0].Labels[constants.LabelShareMode] == string(constants.EXCLUSIVE) &&
			req.ShareMode == constants.ENHANCED {
			req.MonitorSpecID = existed.Items[0].Annotations[constants.AnnotationMonitorSpec]
			err = p.ModifyClusterMode(ctx, &(existed.Items[0]), req)
			if err != nil {
				log.Error("modify cluster mode failed", zap.Error(err))
				return
			}
		}

		if cluster, innerErr := parseClusterResource(ctx, &existed.Items[0]); innerErr != nil {
			return resp, errors.Wrap(errors.UnrecognizedError, "", innerErr)
		} else {
			resp.ObservabilityCluster = *cluster
			return resp, errors.NewErrorf(errors.DuplicatedCluster, "cluster has been existed, vendor = %s, region = %s, tenant id = %s, project id = %s, application id = %s, biz type = %s",
				req.Vendor, req.Region, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType)
		}
	}

	if req.ShareMode == constants.EXCLUSIVE || req.ShareMode == constants.ENHANCED {
		if len(req.MonitorSpecID) == 0 {
			specs := configs.GetServiceInfo().MonitorSpec
			for k := range specs {
				if specs[k].IsDefault {
					req.MonitorSpecID = k
				}
			}
		}
	}

	if req.ShareMode == constants.SHARED && len(req.ParentClusterID) > 0 && len(req.KeySpaceID) == 0 {
		return resp, errors.NewErrorf(errors.ParameterInvalid, "keySpaceID should not be empty in req with parentClusterID: %s", req.ParentClusterID)
	}

	if err = ensureBYOCVPCPeeringRoutesFromRequest(ctx, req); err != nil {
		return resp, err
	}

	var clusterResource, parentCluster *v1.Cluster
	var vmAccountID string

	err = errors.OfNullable(err).BreakIf(func() error {
		vmAccountID, err = prepareNamespace(ctx, req)
		return err
	}).BreakIf(func() error {
		parentCluster, err = prepareParentCluster(ctx, req)
		return err
	}).BreakIf(func() error {
		return prepareBucketResource(ctx, req)
	}).BreakIf(func() error {
		return prepareIAMResource(ctx, req)
	}).BreakIf(func() error {
		return prepareRoleBindingResource(ctx, req)
	}).BreakIf(func() error {
		clusterResource, err = BuildClusterResourceRequirements(ctx, req, vmAccountID, parentCluster)
		return err
	}).BreakIf(func() error {
		return resourcemanager.GetClusterReaderWriter(ctx).CreateCluster(ctx, clusterResource)
	}).BreakIf(func() error {
		cluster, inner := parseClusterResource(ctx, clusterResource)
		resp.ObservabilityCluster = *cluster
		return inner
	}).BreakIf(func() error {
		// create PrometheusRule after cluster is created, so as to set ownerReference on it.
		return preparePrometheusRuleResource(ctx, req, clusterResource)
	}).If(func(err error) {
		log.Error("create cluster failed", zap.Error(err))
	}).Present()

	return resp, err
}

func parseClusterResource(ctx context.Context, clusterResource *v1.Cluster) (*structs.ObservabilityCluster, error) {
	cluster := new(structs.ObservabilityCluster)
	var innerErr error
	err := errors.OfNullable(nil).BreakIf(func() error {
		cluster.ClusterBaseInfo, innerErr = parseClusterBaseInfo(ctx, clusterResource)
		return innerErr
	}).BreakIf(func() error {
		cluster.MonitorInfo, innerErr = parseMonitorInfo(ctx, clusterResource)
		return innerErr
	}).BreakIf(func() error {
		cluster.AgentInfo, innerErr = parseAgentInfo(ctx, &cluster.ClusterBaseInfo, clusterResource)
		return innerErr
	}).BreakIf(func() error {
		cluster.StorageInfo, innerErr = parseStorageInfo(ctx, &cluster.ClusterBaseInfo, clusterResource)
		return innerErr
	}).BreakIf(func() error {
		cluster.AlertInfo, innerErr = parseAlertInfo(ctx, clusterResource)
		return innerErr
	}).BreakIf(func() error {
		cluster.Status, innerErr = parseClusterStatus(ctx, clusterResource)
		return innerErr
	}).BreakIf(func() error {
		cluster.LastScaleTime, innerErr = parseClusterLastScaleTime(ctx, clusterResource)
		return innerErr
	}).If(func(err error) {
		log.Error("parse cluster resource failed",
			zap.Error(err),
			zap.Any("Cluster", cluster))
	}).Present()

	return cluster, err
}

// GetClusterDetail
// @Description: get cluster detail
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) GetClusterDetail(ctx context.Context, req message.DetailClusterReq) (resp message.DetailClusterResp, err error) {
	//set ctx if biztype use rds
	if req.BizType != "" {
		ctx = dealuseRDS(ctx, req.BizType)
	}
	clusterResource, err := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)

	if err != nil {
		log.Error("get cluster detail failed",
			zap.Error(err),
			zap.String("Cluster_ID", req.ID))
		return resp, err
	}

	cluster, err := parseClusterResource(ctx, clusterResource)

	if err != nil {
		return resp, err
	}
	resp.ObservabilityCluster = *cluster

	return resp, err
}

// UpdateCluster
// @Description: update cluster
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) UpdateCluster(ctx context.Context, req message.UpdateClusterReq) (resp message.UpdateClusterResp, err error) {
	cluster, getError := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if getError != nil {
		return resp, getError
	}

	// If status is empty, it may be pending in creating status.
	if cluster.Status.Phase != "" {
		accessResult := checkStatus(cluster.Status.Phase, constants.EndpointClusterUpdate)
		switch accessResult {
		case CONFLICT:
			return resp, errors.NewErrorf(errors.ClusterStatusConflict, "cluster %s is being %s", req.ID, cluster.Status.Phase)
		}
	}

	if req.MuteAlert != nil {
		cluster.Spec.MuteAlert = *req.MuteAlert
	}
	if req.RuleTemplateID != "" {
		if cluster.Labels == nil {
			cluster.Labels = make(map[string]string)
		}
		cluster.Labels[constants.LabelTemplateID] = req.RuleTemplateID
	}
	if req.MonitorSpecID != "" {
		if cluster.Annotations == nil {
			cluster.Annotations = make(map[string]string)
		}
		cluster.Annotations[constants.AnnotationMonitorSpec] = req.MonitorSpecID
	}

	err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
	if err != nil {
		log.Error("update cluster failed", zap.Error(err))
		return resp, err
	}

	resp.ClusterBaseInfo, err = parseClusterBaseInfo(ctx, cluster)
	if err != nil {
		log.Error("parse cluster base info failed", zap.Error(err))
		return resp, err
	}
	return resp, nil
}

//var maxTTL = 24 * 15

// DeleteCluster
// @Description: delete a cluster
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) DeleteCluster(ctx context.Context, req message.DeleteClusterReq) (resp message.DeleteClusterResp, err error) {
	//set ctx if biztype use rds
	if req.BizType != "" {
		ctx = dealuseRDS(ctx, req.BizType)
	}
	cluster, getError := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if getError != nil {
		return resp, getError
	}

	resp.ClusterBaseInfo, err = parseClusterBaseInfo(ctx, cluster)
	if err != nil {
		log.Error("parse cluster base info failed", zap.Error(err))
		return resp, err
	}

	// If status is empty, it may be pending in creating status. It is safe to delete the cluster.
	if cluster.Status.Phase != "" {
		accessResult := checkStatus(cluster.Status.Phase, constants.EndpointClusterDelete)
		switch accessResult {
		case IDEMPOTENT:
			return resp, nil
		case CONFLICT:
			return resp, errors.NewErrorf(errors.ClusterStatusConflict, "cluster %s is being %s", req.ID, cluster.Status.Phase)
		}
	}

	if err := deleteBYOCVPCPeeringRoutesFromCluster(ctx, cluster); err != nil {
		log.Error("delete BYOC VPC peering routes failed", zap.Error(err))
		return resp, err
	}

	//delete prometheusrule resource if exists
	isVMRule := isSharedVMRule(constants.ShareMode(cluster.Labels[constants.LabelShareMode]))
	err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, isVMRule).DeletePrometheusRules(ctx, req.ID)
	errs, ok := err.(errors.Error)
	if ok && errs.Code() != errors.PrometheusRuleNotFound {
		log.Error("DeletePrometheusRules failed", zap.Error(err))
		return
	}

	//delete cluster rolebinding if exists
	err = resourcemanager.GetClusterRoleBindingReaderWriter(ctx).Delete(ctx, req.ID)
	errs, ok = err.(errors.Error)
	if ok && errs.Code() != errors.RoleBindingNotFound {
		log.Error("delete cluster rolebinding failed", zap.Error(err))
		return
	}

	if archiveErr := archive(ctx, cluster, resp.ClusterBaseInfo); archiveErr != nil {
		log.Error("archive before deleting failed", zap.Error(archiveErr))
		return resp, errors.NewErrorf(errors.AWSClientError, "archive before deleting failed, err: %v", archiveErr)
	}

	if cluster.Labels[constants.LabelShareMode] == string(constants.SHARED) || cluster.Status.Phase == constants.PAUSED || cluster.Status.Phase == constants.CREATION_FAILED || cluster.Status.Phase == constants.CREATING {
		err = resourcemanager.GetClusterReaderWriter(ctx).DeleteCluster(ctx, req.ID)
	} else {
		err = resourcemanager.GetClusterReaderWriter(ctx).SoftDeleteCluster(ctx, req.ID)
	}

	if err != nil {
		log.Error("delete cluster failed", zap.Error(err))
		return resp, err
	}

	if err = removeAzureAccessARNFromIAMOnDelete(ctx, cluster, req.AgentAccessARN); err != nil {
		log.Error("remove azure access arn from iam failed", zap.Error(err))
		return resp, err
	}
	if err = removeGCPAccessARNFromIAMOnDelete(ctx, cluster, req.AgentAccessARN); err != nil {
		log.Error("remove gcp access arn from iam failed", zap.Error(err))
		return resp, err
	}

	if err = deleteDataplaneRoleIfNeeded(ctx, cluster); err != nil {
		log.Error("delete data plane role failed", zap.Error(err))
		return resp, err
	}

	if err = updateDataplaneIAMPoliciesOnDelete(ctx, cluster); err != nil {
		log.Error("update IAM policies failed", zap.Error(err))
		return resp, err
	}

	return resp, nil
}

func archive(ctx context.Context, cluster *v1.Cluster, baseInfo structs.ClusterBaseInfo) error {
	if cluster.Labels[constants.LabelShareMode] == string(constants.EXCLUSIVE) {
		return nil
	}
	archivecluster := structs.ArchivedClusterInfo{
		ClusterBaseInfo: baseInfo,
		VMAccountID:     strconv.Itoa(int(cluster.Spec.VictoriaAccount.AccountID)),
		VMProjectID:     strconv.Itoa(int(cluster.Spec.VictoriaAccount.ProjectID)),
	}
	if isNextGenCluster(cluster) {
		for _, category := range constants.AllCategory() {
			archivecluster.StorageInfo = append(archivecluster.StorageInfo, structs.StorageInfo{
				Category: category,
			})
		}
	}
	return resourcemanager.GetArchivedClusterReaderWriter().Archive(ctx,
		ObjectStorageResourceName(baseInfo),
		structs.ClusterUniqueKey{
			BizType:       baseInfo.BizType,
			TenantID:      baseInfo.TenantID,
			ProjectID:     baseInfo.ProjectID,
			ApplicationID: baseInfo.ApplicationID,
		},
		archivecluster)
}

func deleteDataplaneRoleIfNeeded(ctx context.Context, cluster *v1.Cluster) error {
	if cluster.Spec.Agent == nil {
		return nil
	}
	if configs.GetServiceInfo().Vendor == string(constants.Azure) {
		return deteleRoleAzure(ctx, cluster)
	}
	cs, err := resourcemanager.GetClusterReaderWriter(ctx).QueryClusters(ctx, cluster.Labels[constants.LabelTenantID], "", "", cluster.Labels[constants.LabelBizType], message.ContinueRequest{
		Limit: 1000,
	})
	if err != nil {
		return err
	}

	couldDelete := true
	for _, c := range cs.Items {
		// we consider role can't be deleted if there is cluster matches
		// 1. is 3-layers cluster(agent != nil)
		// 2. is not deleted or to_be_deleted
		// 3. is not the current cluster
		if c.Spec.Agent != nil &&
			!c.Status.Phase.ClusterHasBeenRequestedToDelete() &&
			c.Name != cluster.Name {
			couldDelete = false
		}
	}

	if couldDelete {
		iamList, err := resourcemanager.GetIAMReaderWriter(ctx).GetRoles(ctx, cluster.Namespace, map[string]string{
			constants.LabelBizType:   cluster.Labels[constants.LabelBizType],
			constants.LabelTenantID:  cluster.Labels[constants.LabelTenantID],
			constants.LabelProjectID: cluster.Labels[constants.LabelProjectID],
		})
		if err != nil {
			return fmt.Errorf("get dataplane roles failed, %w", err)
		}
		for _, iam := range iamList.Items {
			err := resourcemanager.GetIAMReaderWriter(ctx).DeleteRole(ctx, iam.Name, iam.Namespace)
			if err != nil {
				return fmt.Errorf("delete data plane role failed, %w", err)
			}
			// remove role in o11y iam role policy
			err = updateO11yIAMPoliciesOnDelete(ctx, cluster, iam.Labels[constants.LabelAlias])
			if err != nil {
				return fmt.Errorf("update o11y iam policies failed, %w", err)
			}
		}
	}
	return nil
}

func deteleRoleAzure(ctx context.Context, cluster *v1.Cluster) error {
	cs, err := resourcemanager.GetClusterReaderWriter(ctx).QueryClusters(ctx, cluster.Labels[constants.LabelTenantID], cluster.Labels[constants.LabelProjectID], "", cluster.Labels[constants.LabelBizType], message.ContinueRequest{
		Limit: 1000,
	})
	if err != nil {
		return err
	}
	couldDelete := true
	for _, c := range cs.Items {
		// we consider role can't be deleted if there is cluster matches
		// 1. is 3-layers cluster(agent != nil)
		// 2. is not deleted or to_be_deleted
		// 3. is not the current cluster
		if c.Spec.Agent != nil &&
			!c.Status.Phase.ClusterHasBeenRequestedToDelete() &&
			c.Name != cluster.Name {
			couldDelete = false
		}
	}
	ns := NamespaceName(cluster.Labels[constants.LabelBizType], cluster.Labels[constants.LabelTenantID])
	dataroleName := AgentDataPlaneRoleName(AgentBaseInfo{
		ClusterBaseInfo: structs.ClusterBaseInfo{
			BizType:   cluster.Labels[constants.LabelBizType],
			TenantID:  cluster.Labels[constants.LabelTenantID],
			ProjectID: cluster.Labels[constants.LabelProjectID],
		},
	})
	if couldDelete {
		return resourcemanager.GetIAMReaderWriter(ctx).DeleteRole(ctx, dataroleName, ns)
	} else {
		// remove the related workload sa in the iam.
		iam, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, dataroleName, ns)
		if err != nil {
			return err
		}
		iamCopy := iam.DeepCopy()
		workloadSAs := []string{}
		for _, sa := range iam.Spec.Azure.WorkloadServiceAccounts {
			if !strings.Contains(sa, cluster.Labels[constants.LabelApplicationID]) {
				workloadSAs = append(workloadSAs, sa)
			}
		}
		iamCopy.Spec.Azure.WorkloadServiceAccounts = workloadSAs
		return resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, iamCopy)
	}
}

func updateDataplaneIAMPoliciesOnDelete(ctx context.Context, cluster *v1.Cluster) error {
	if cluster.Spec.Agent == nil || configs.GetServiceInfo().Vendor != string(constants.AWS) {
		return nil
	}
	// Get all clusters in the namespace
	clusterList, err := resourcemanager.GetClusterReaderWriter(ctx).ListClusters(ctx, client.InNamespace(cluster.Namespace))
	if err != nil {
		return fmt.Errorf("failed to list clusters in namespace %s: %w", cluster.Namespace, err)
	}
	account := cluster.Spec.Agent.AWS.Account
	OIDCSet := map[string]struct{}{}
	for _, c := range clusterList.Items {
		// Skip self or clusters with different AWS accounts
		if c.Name == cluster.Name || c.Spec.Agent == nil || c.Spec.Agent.AWS.Account != account || c.Spec.Agent.AWS.OIDC == "" {
			continue
		}
		OIDCSet[c.Spec.Agent.AWS.OIDC] = struct{}{}
	}
	if len(OIDCSet) == 0 {
		return nil
	}

	dataroleName, err := AgentDataPlaneRoleNameWrap(ctx, cluster.Namespace, AgentBaseInfo{
		Account: account,
		ClusterBaseInfo: structs.ClusterBaseInfo{
			BizType:   cluster.Labels[constants.LabelBizType],
			TenantID:  cluster.Labels[constants.LabelTenantID],
			ProjectID: cluster.Labels[constants.LabelProjectID],
		},
	})
	if err != nil {
		return err
	}
	exists, err := resourcemanager.GetIAMReaderWriter(ctx).IsRoleExists(ctx, dataroleName, cluster.Namespace)
	if err != nil {
		return err
	}
	if !exists {
		return nil
	}
	iam, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, dataroleName, cluster.Namespace)
	if err != nil {
		return err
	}
	// check the oidc used in policy statement and remove the statements that are not needed.
	iamPolicy, err := GetIAMAssumePolicyAWS(ctx, iam)
	if err != nil {
		return err
	}
	validStmts := []Statement{}
	for _, s := range iamPolicy.Statement {
		oidcARN, ok := s.Principal[TrustPolicyPrincipalKeyFederated].(string)
		if !ok {
			continue
		}
		oidc := parseOIDC(oidcARN)
		// if oidc in the set, keep the statement
		if _, exists := OIDCSet[oidc]; exists {
			validStmts = append(validStmts, s)
		}
	}
	// update iam if there are changes and valid statements are not empty(statement should not be empty in aws)
	if len(validStmts) != len(iamPolicy.Statement) && len(validStmts) > 0 {
		log.Info("Updating IAM role 'Data_Role_Name' with trust policy diff 'Trust_Policy_Diff'",
			zap.String("Data_Role_Name", dataroleName),
			zap.String("Trust_Policy_Diff", cmp.Diff(validStmts, iamPolicy.Statement)),
		)
		iamPolicy.Statement = validStmts
		return updateIAMWithPolicy(ctx, iam, iamPolicy)
	}
	return nil
}

func updateO11yIAMPoliciesOnDelete(ctx context.Context, cluster *v1.Cluster, principal string) error {
	if configs.GetServiceInfo().Vendor != string(constants.AWS) {
		return nil
	}
	iamName := AgentRoleName(structs.ClusterBaseInfo{
		BizType:  cluster.Labels[constants.LabelBizType],
		TenantID: cluster.Labels[constants.LabelTenantID],
	})
	iam, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, iamName, cluster.Namespace)
	if err != nil {
		return err
	}
	// Perform diff and update IAM
	iamPolicy, err := GetIAMAssumePolicyAWS(ctx, iam)
	if err != nil {
		return err
	}

	validStmts := []Statement{}
	for _, s := range iamPolicy.Statement {
		roleArn, ok := s.Principal[TrustPolicyPrincipalKeyAWS].(string)
		// delete the statement if it contains the principal
		if ok && strings.Contains(roleArn, principal) {
			continue
		}
		validStmts = append(validStmts, s)
	}

	// update iam if there are changes and valid statements are not empty
	if len(validStmts) != len(iamPolicy.Statement) && len(validStmts) > 0 {
		iamPolicy.Statement = validStmts
		log.Info("Updating IAM role 'IAM_Role' by removing principal 'Principal_Removed'",
			zap.String("IAM_Role", iamName),
			zap.String("Principal_Removed", principal),
		)
		return updateIAMWithPolicy(ctx, iam, iamPolicy)
	}
	return nil
}

// parseOIDC return oidc.eks.us-west-2.amazonaws.com/id/EXAMPLED539D4633E53DE1B716D3041
// for arn like arn:aws:iam::123456789012:oidc-provider/oidc.eks.us-west-2.amazonaws.com/id/EXAMPLED539D4633E53DE1B716D3041
func parseOIDC(arn string) string {
	if arn == "" {
		return ""
	}
	parts := strings.Split(arn, "oidc-provider/")
	if len(parts) != 2 {
		return ""
	}

	return parts[1]
}

func updateIAMWithPolicy(ctx context.Context, iam *v1.IAM, policy *AssumeRolePolicy) error {
	iamCopy := iam.DeepCopy()
	policyJSON, err := json.Marshal(policy)
	if err != nil {
		return fmt.Errorf("failed to marshal IAM policy: %w", err)
	}

	iamCopy.Spec.Aws.AssumeRolePolicyDocument = base64.StdEncoding.EncodeToString(policyJSON)
	return resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, iamCopy)
}

// PauseCluster
// @Description: pause a cluster
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) PauseCluster(ctx context.Context, req message.PauseClusterReq) (resp message.PauseClusterResp, err error) {
	cluster, getError := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if getError != nil {
		return resp, getError
	}

	resp.ClusterBaseInfo, err = parseClusterBaseInfo(ctx, cluster)
	if err != nil {
		log.Error("parse cluster base info failed", zap.Error(err))
		return resp, err
	}

	accessResult := checkStatus(cluster.Status.Phase, constants.EndpointClusterPause)
	switch accessResult {
	case IDEMPOTENT:
		return resp, nil
	case CONFLICT:
		return resp, errors.NewErrorf(errors.ClusterStatusConflict, "cluster %s is being %s", req.ID, cluster.Status.Phase)
	}

	if clusterctl.IsSharedMode(cluster) {
		return resp, resourcemanager.GetClusterReaderWriter(ctx).PauseCluster(ctx, cluster)
	}

	err = resourcemanager.GetClusterReaderWriter(ctx).SoftPauseCluster(ctx, cluster)
	if err != nil {
		log.Error("pause cluster failed", zap.Error(err))
		return resp, err
	}
	return resp, nil
}

// ResumeCluster
// @Description: resume a cluster
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) ResumeCluster(ctx context.Context, req message.ResumeClusterReq) (resp message.ResumeClusterResp, err error) {
	cluster, getError := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if getError != nil {
		return resp, getError
	}

	resp.ClusterBaseInfo, err = parseClusterBaseInfo(ctx, cluster)
	if err != nil {
		log.Error("parse cluster base info failed", zap.Error(err))
		return resp, err
	}

	accessResult := checkStatus(cluster.Status.Phase, constants.EndpointClusterResume)
	switch accessResult {
	case IDEMPOTENT:
		return resp, nil
	case CONFLICT:
		return resp, errors.NewErrorf(errors.ClusterStatusConflict, "cluster %s is being %s", req.ID, cluster.Status.Phase)
	}

	if clusterctl.IsSharedMode(cluster) {
		return resp, resourcemanager.GetClusterReaderWriter(ctx).ResumeCluster(ctx, cluster)
	}

	bizType := cluster.Labels[constants.LabelBizType]
	replicas, err := configs.GetMonitorReplicasByBizType(bizType)
	if err != nil {
		return resp, err
	}
	cluster.Spec.Prometheus.Replicas = replicas

	err = resourcemanager.GetClusterReaderWriter(ctx).ResumeCluster(ctx, cluster)
	if err != nil {
		log.Error("resume cluster failed", zap.Error(err))
		return resp, err
	}
	return resp, nil
}

// MuteClusterAlert
// @Description: mute alert
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) MuteClusterAlert(ctx context.Context, req message.MuteClusterAlertReq) (resp message.MuteClusterAlertResp, err error) {
	cluster, getError := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if getError != nil {
		return resp, getError
	}

	cluster.Spec.MuteAlert = true

	err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
	if err != nil {
		log.Error("mute cluster alert failed", zap.Error(err))
		return resp, err
	}
	return resp, nil

}

// UnmuteClusterAlert
// @Description: unmute alert
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) UnmuteClusterAlert(ctx context.Context, req message.UnmuteClusterAlertReq) (resp message.UnmuteClusterAlertResp, err error) {
	cluster, getError := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if getError != nil {
		return resp, getError
	}

	cluster.Spec.MuteAlert = false

	err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
	if err != nil {
		log.Error("unmute cluster alert failed", zap.Error(err))
		return resp, err
	}
	return resp, nil
}

// QueryClusters
// @Description: query clusters
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) QueryClusters(ctx context.Context, req message.QueryClustersReq) (resp message.QueryClustersResp, err error) {
	//set ctx if biztype use rds
	if req.BizType != "" {
		ctx = dealuseRDS(ctx, req.BizType)
	}
	clusters, err := resourcemanager.GetClusterReaderWriter(ctx).QueryClusters(ctx, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType, req.ContinueRequest)
	if err != nil {
		log.Error("query clusters failed", zap.Error(err))
		return resp, err
	}

	resp.ContinueResponse = message.ContinueResponse{
		ContinueToken: clusters.GetContinue(),
	}

	resp.Clusters = make([]structs.ClusterBaseInfo, 0)
	for i := range clusters.Items {
		cluster, parseErr := parseClusterBaseInfo(ctx, &clusters.Items[i])
		if parseErr != nil {
			log.Error("parse cluster info failed",
				zap.Error(parseErr),
				zap.String("Cluster_ID", clusters.Items[i].Name))
			return resp, parseErr
		}
		resp.Clusters = append(resp.Clusters, cluster)
	}

	return
}

func (p *ClusterHandler) QueryClustersWithStorage(ctx context.Context, req message.QueryClustersReq) (resp message.QueryClustersWithStorageResp, err error) {
	if err := validateShowDeletedExactLookup(req); err != nil {
		return resp, err
	}

	if req.BizType != "" {
		ctx = dealuseRDS(ctx, req.BizType)
	}
	clusters, err := resourcemanager.GetClusterReaderWriter(ctx).QueryClusters(ctx, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType, req.ContinueRequest)
	if err != nil {
		log.Error("query clusters failed", zap.Error(err))
		return resp, err
	}

	resp.ContinueResponse = message.ContinueResponse{
		ContinueToken: clusters.GetContinue(),
	}

	resp.Clusters = make([]message.ClusterWithStorageInfo, 0, len(clusters.Items))
	for i := range clusters.Items {
		cluster, parseErr := buildClusterWithStorageInfo(ctx, &clusters.Items[i])
		if parseErr != nil {
			log.Error("build cluster with storage info failed",
				zap.Error(parseErr),
				zap.String("Cluster_ID", clusters.Items[i].Name))
			return resp, parseErr
		}
		resp.Clusters = append(resp.Clusters, cluster)
	}

	if len(resp.Clusters) > 0 || !req.ShowDeleted {
		return resp, nil
	}

	archived, archivedErr := dataservice.GetHandler().GetArchivedCluster(ctx, structs.ClusterUniqueKey{
		BizType:       req.BizType,
		TenantID:      req.TenantID,
		ProjectID:     req.ProjectID,
		ApplicationID: req.ApplicationID,
	})
	if archivedErr != nil {
		if errors.Is(archivedErr, errors.ObjectNotFoundError) {
			return resp, nil
		}
		return resp, archivedErr
	}

	resp.Clusters = append(resp.Clusters, buildArchivedClusterWithStorageInfo(ctx, archived))
	return
}

func buildClusterWithStorageInfo(ctx context.Context, clusterResource *v1.Cluster) (cluster message.ClusterWithStorageInfo, err error) {
	cluster.ClusterBaseInfo, err = parseClusterBaseInfo(ctx, clusterResource)
	if err != nil {
		return cluster, err
	}

	storageInfo, err := parseStorageInfo(ctx, &cluster.ClusterBaseInfo, clusterResource)
	if err != nil {
		return cluster, err
	}
	cluster.StorageInfo = storageInfo
	cluster.LogQueryUri = buildRegionalLogQueryURI(cluster.ClusterBaseInfo.BizType)
	cluster.MultitenantQueryUri = buildRegionalMultitenantQueryURI(
		cluster.ClusterBaseInfo.BizType,
		clusterResource.Spec.VictoriaAccount.AccountID,
	)
	cluster.DiagnosisQueryUri = buildRegionalDiagnosisQueryURI(cluster.ClusterBaseInfo.BizType)

	return cluster, nil
}

func buildArchivedClusterWithStorageInfo(ctx context.Context, archived structs.ArchivedClusterInfo) message.ClusterWithStorageInfo {
	resp := message.ClusterWithStorageInfo{
		ClusterBaseInfo: archived.ClusterBaseInfo,
		LogQueryUri:     buildRegionalLogQueryURI(archived.ClusterBaseInfo.BizType),
		DiagnosisQueryUri: buildRegionalDiagnosisQueryURI(
			archived.ClusterBaseInfo.BizType,
		),
	}
	if len(archived.VMAccountID) > 0 {
		vmAccountID, parseErr := parseVMAccountID(archived.VMAccountID)
		if parseErr != nil {
			log.Error("parse archived vm account id failed",
				zap.Error(parseErr),
				zap.String("applicationID", archived.ApplicationID),
				zap.String("vmAccountID", archived.VMAccountID),
			)
		} else {
			resp.MultitenantQueryUri = buildRegionalMultitenantQueryURI(
				archived.ClusterBaseInfo.BizType,
				vmAccountID,
			)
		}
	}
	resp.ClusterBaseInfo.InternalReadUri = buildHistoricalInternalReadURI(archived.ClusterBaseInfo)
	storageInfo, err := parseStorageInfo(ctx, &resp.ClusterBaseInfo, nil)
	if err != nil {
		log.Error("build archived cluster storage info failed", zap.Error(err), zap.String("applicationID", resp.ApplicationID))
		resp.StorageInfo = nil
		return resp
	}
	resp.StorageInfo = storageInfo
	return resp
}

func validateShowDeletedExactLookup(req message.QueryClustersReq) error {
	if !req.ShowDeleted {
		return nil
	}
	if req.BizType == "" || req.TenantID == "" || req.ApplicationID == "" {
		return errors.New(errors.ParameterInvalid, "showDeleted requires bizType, tenantID and applicationID")
	}
	return nil
}

func buildHistoricalInternalReadURI(baseInfo structs.ClusterBaseInfo) string {
	hostName := resolveDataServerHostName(baseInfo.BizType)
	if baseInfo.ProjectID == "" {
		return fmt.Sprintf(
			constants.HistoricalInternalReadUriTemplateWithNoProject,
			hostName,
			baseInfo.BizType,
			baseInfo.TenantID,
			baseInfo.ApplicationID,
		)
	}

	return fmt.Sprintf(
		constants.HistoricalInternalReadUriTemplate,
		hostName,
		baseInfo.BizType,
		baseInfo.TenantID,
		baseInfo.ProjectID,
		baseInfo.ApplicationID,
	)
}

// QueryClustersByRuleTemplateID
// @Description: query clusters by rule templateID
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) QueryClustersByRuleTemplateID(ctx context.Context, ruleTemplateID string, creq message.ContinueRequest) (resp message.QueryClustersResp, err error) {
	clusters, err := resourcemanager.GetClusterReaderWriter(ctx).QueryClustersByRuleTemplateID(ctx, ruleTemplateID, creq)
	if err != nil {
		log.Error("query clusters failed", zap.Error(err))
		return resp, err
	}

	resp.ContinueResponse = message.ContinueResponse{
		ContinueToken: clusters.GetContinue(),
	}

	resp.Clusters = make([]structs.ClusterBaseInfo, 0)
	for i := range clusters.Items {
		cluster, parseErr := parseClusterBaseInfo(ctx, &clusters.Items[i])
		if parseErr != nil {
			log.Error("parse cluster info failed",
				zap.Error(parseErr),
				zap.String("Cluster_ID", clusters.Items[i].Name))
			return resp, parseErr
		}
		//filter  to_be_delete and deleting  status clusters
		if clusters.Items[i].Status.Phase == constants.TO_BE_DELETED ||
			clusters.Items[i].Status.Phase == constants.DELETING {
			continue
		}
		resp.Clusters = append(resp.Clusters, cluster)
	}

	return
}

// MonitorClusterSpec
// @Description: modify spec of monitor instance
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) MonitorClusterSpec(ctx context.Context, req message.ModifyMonitorSpecReq) (resp message.ModifyMonitorSpecResp, err error) {
	cluster, err := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if err != nil {
		log.Error("get cluster failed", zap.Error(err))
		return
	}

	accessResult := checkStatus(cluster.Status.Phase, constants.EndpointClusterMonitorSpec)
	switch accessResult {
	case IDEMPOTENT:
		return resp, nil
	case CONFLICT:
		return resp, errors.NewErrorf(errors.ClusterStatusConflict, "cluster %s is being %s", req.ID, cluster.Status.Phase)
	}

	requirements, err := MonitorSpecRequirements(req.SpecID)

	if err != nil {
		log.Error("get monitor spec requirements failed", zap.Error(err))
		return
	} else {
		log.Info("update monitor spec of cluster",
			zap.String("Cluster_ID", req.ID),
			zap.String("Spec", req.SpecID))
	}

	cluster.Annotations[constants.AnnotationMonitorSpec] = req.SpecID

	if clusterctl.IsSharedMode(cluster) {
		cluster.Spec.VictoriaAccount.AlertResources = requirements
		// Update MaxQueueSize from monitor spec
		if s, ok := configs.GetServiceInfo().MonitorSpec[req.SpecID]; ok && s.MaxQueueSize != nil {
			cluster.Spec.VictoriaAccount.MaxQueueSize = s.MaxQueueSize
		}
	} else {
		cluster.Spec.Prometheus.Resources = requirements
		if s, ok := configs.GetServiceInfo().MonitorSpec[req.SpecID]; ok {
			cluster.Spec.Prometheus.Query = &prometheusv1.QuerySpec{
				MaxConcurrency: s.MaxQueryConcurrency,
				MaxSamples:     s.MaxQuerySamples,
			}
		}
	}

	cluster.Annotations[constants.AnnotationLastScaleTime] = strconv.FormatInt(time.Now().Unix(), 10)

	err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
	if err != nil {
		log.Error("update cluster failed", zap.Error(err))
	}
	return
}

// MonitorClusterMode
// @Description: modify share mode of monitor instance
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) MonitorClusterMode(ctx context.Context, req message.ModifyMonitorModeReq) (resp message.ModifyMonitorModeResp, err error) {
	// todo
	return
}

// ModifyMonitorStorage
// @Description: modify storage of monitor instance
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) ModifyMonitorStorage(ctx context.Context, req message.ModifyMonitorStorageReq) (resp message.ModifyMonitorStorageResp, err error) {
	cluster, err := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if err != nil {
		log.Error("get cluster failed", zap.Error(err))
		return
	}
	if constants.ShareMode(cluster.Annotations[constants.AnnotationShareMode]) == constants.SHARED {
		err = errors.NewErrorf(errors.MethodNotSupported, " this method is not supported by shared cluster")
		log.Error("update cluster failed", zap.Error(err))
		return
	}
	pvc, err := resourcemanager.GetPVCReaderWriter().Get(ctx, req.Namespace, req.PVCName)
	if err != nil {
		log.Error("get pvc failed", zap.Error(err))
		return
	}
	// update pvc resource
	pvc.Spec.Resources.Requests[corev1.ResourceStorage] = req.Request
	err = resourcemanager.GetPVCReaderWriter().Update(ctx, req.Namespace, pvc)
	if err != nil {
		log.Error("update pvc failed", zap.Error(err))
		return
	}
	// update cluster annotations and resource
	cluster.Annotations[constants.AnnotationLastScaleStorage] = strconv.FormatInt(time.Now().Unix(), 10)
	if cluster.Spec.Prometheus.Resources.Requests == nil {
		request := map[corev1.ResourceName]resource.Quantity{
			corev1.ResourceStorage: req.Request,
		}
		cluster.Spec.Prometheus.Resources.Requests = request
	} else {
		cluster.Spec.Prometheus.Resources.Requests[corev1.ResourceStorage] = req.Request
	}
	err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
	if err != nil {
		log.Error("update cluster failed", zap.Error(err))
		return
	}
	return
}

// ModifyMonitorResource
// @Description: modify monitor resource
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) ModifyMonitorResource(ctx context.Context, req message.ModifyMonitorResourceReq) (resp message.ModifyMonitorResourceResp, err error) {
	cluster, err := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if err != nil {
		log.Error("get cluster failed", zap.Error(err))
		return
	}

	switch req.ResourceType {
	case corev1.ResourceCPU:
		if req.ResourceMeta[req.ResourceType].SpecID == "" {
			err = errors.NewErrorf(errors.ParameterInvalid, "specID should not be empty")
			log.Error("failed to modify resource",
				zap.Error(err),
				zap.String("Resource_Type", string(req.ResourceType)),
			)
			return
		}
		newSpecID := req.ResourceMeta[req.ResourceType].SpecID

		_, err = p.MonitorClusterSpec(ctx, message.ModifyMonitorSpecReq{
			ID:     req.ID,
			SpecID: newSpecID,
		})

		if err != nil {
			log.Error("failed to modify resource",
				zap.Error(err),
				zap.String("Resource_Type", string(req.ResourceType)),
			)
			return resp, err
		}
	case corev1.ResourceMemory:
		if req.ResourceMeta[req.ResourceType].SpecID == "" {
			err = errors.NewErrorf(errors.ParameterInvalid, "specID should not be empty")
			log.Error("failed to modify resource",
				zap.Error(err),
				zap.String("Resource_Type", string(req.ResourceType)),
			)
			return resp, err
		}

		newSpecID := req.ResourceMeta[req.ResourceType].SpecID

		_, err = p.MonitorClusterSpec(ctx, message.ModifyMonitorSpecReq{
			ID:     req.ID,
			SpecID: newSpecID,
		})

		if err != nil {
			log.Error("failed to modify resource",
				zap.Error(err),
				zap.String("Resource_Type", string(req.ResourceType)),
			)
			return resp, err
		}

	case corev1.ResourceStorage:
		if constants.ShareMode(cluster.Annotations[constants.AnnotationShareMode]) == constants.SHARED {
			err = errors.NewErrorf(errors.MethodNotSupported, " this method is not supported by shared cluster")
			log.Error("update cluster failed", zap.Error(err))
			return
		}

		if req.ResourceMeta[req.ResourceType].Namespace == "" || req.ResourceMeta[req.ResourceType].PVCName == "" {
			err = errors.NewErrorf(errors.ParameterInvalid, "namespace or pvcname should not be empty")
			log.Error("failed to modify resource",
				zap.Error(err),
				zap.String("Resource_Type", string(req.ResourceType)),
			)
			return
		}
		pvc, err := resourcemanager.GetPVCReaderWriter().Get(ctx, req.ResourceMeta[req.ResourceType].Namespace, req.ResourceMeta[req.ResourceType].PVCName)
		if err != nil {
			return resp, err
		}
		storage := pvc.Spec.Resources.Requests.Storage()
		storage.Add(*storage)
		_, err = p.ModifyMonitorStorage(ctx, message.ModifyMonitorStorageReq{
			ID:        req.ID,
			Namespace: req.ResourceMeta[req.ResourceType].Namespace,
			PVCName:   req.ResourceMeta[req.ResourceType].PVCName,
			Request:   *storage,
		})
		if err != nil {
			log.Error("failed to modify resource",
				zap.Error(err),
				zap.String("Resource_Type", req.ResourceType.String()),
			)
			return resp, err
		}
	}

	return
}

func (p *ClusterHandler) ModifyClusterMode(ctx context.Context, cluster *v1.Cluster, req message.CreateClusterReq) error {
	req.ID = cluster.Name
	err := appendMonitor(ctx, cluster, req, cluster.Labels[constants.LabelAZ], req.ShareMode, "")
	if err != nil {
		return err
	}

	cluster.Labels[constants.LabelShareMode] = string(req.ShareMode)
	cluster.Annotations[constants.AnnotationShareMode] = string(req.ShareMode)

	return resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
}

// ModifyAgentObserved
// @Description: modify observed info for monitor agent
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) ModifyAgentObserved(ctx context.Context, req message.ModifyAgentObservedReq) (resp message.ModifyAgentObservedResp, err error) {
	cluster, err := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
	if err != nil {
		log.Error("get cluster failed", zap.Error(err))
		return
	}

	accessResult := checkStatus(cluster.Status.Phase, constants.EndpointClusterAgentObserved)
	switch accessResult {
	case IDEMPOTENT:
		return resp, nil
	case CONFLICT:
		return resp, errors.NewErrorf(errors.ClusterStatusConflict, "cluster %s is being %s", req.ID, cluster.Status.Phase)
	}

	if cluster.Spec.Agent == nil {
		return resp, errors.NewErrorf(errors.ClusterAgentNotSet, "cluster %s agent struce not set,", req.ID)
	}

	cluster.Spec.Agent.Observed = &v1.Observed{
		TiDBVersion: req.Observed.TiDBVersion,
		TiDBClusterTopology: &v1.ClusterTopologyType{
			TiDB: v1.TiDB{
				Replicas: req.Observed.TiDBClusterTopology.TiDB.Replicas,
			},
			PD: v1.PD{
				Replicas: req.Observed.TiDBClusterTopology.PD.Replicas,
			},
			TiKV: v1.TiKV{
				Replicas: req.Observed.TiDBClusterTopology.TiKV.Replicas,
			},
		},
	}
	if req.Observed.TiDBClusterTopology.TiFlash != nil {
		cluster.Spec.Agent.Observed.TiDBClusterTopology.TiFlash = &v1.TiFlash{
			Replicas: req.Observed.TiDBClusterTopology.TiFlash.Replicas,
		}
	}

	cluster.Annotations[constants.AnnotationLastObservedTime] = strconv.FormatInt(time.Now().Unix(), 10)

	err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
	if err != nil {
		log.Error("update cluster failed", zap.Error(err))
	}
	return
}

// ModifyAlertRulesTemplate
// @Description: modify alert rules template
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) ModifyAlertRulesTemplate(ctx context.Context, req message.ModifyAlertRulesTemplateReq) (resp message.ModifyAlertRulesTemplateResp, err error) {
	if req.MonitorPromRules.Groups == nil {
		return resp, errors.NewErrorf(errors.PrometheusRuleIsEmpty, "prometheus alert rule not allow empty")
	}
	ruleGroups := make([]prometheusv1.RuleGroup, 0)
	ruleGroupsNameMap := make(map[string]struct{})
	for _, rules := range req.MonitorPromRules.Groups {
		if _, ok := ruleGroupsNameMap[rules.Name]; ok {
			errMsg := fmt.Sprintf("rule template contains identical alert rule group name %s", rules.Name)
			log.Error(errMsg)
			err = errors.New(errors.ParameterInvalid, errMsg)
			return
		}
		ruleGroups = append(ruleGroups, rules)
		ruleGroupsNameMap[rules.Name] = struct{}{}
	}
	req.MonitorPromRules.Groups = ruleGroups

	promRules, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).GetPrometheusRule(ctx, req.TemplateID)
	if err != nil {
		errs, ok := err.(errors.Error)
		if ok && errs.Code() != errors.PrometheusRuleNotFound {
			log.Error("get prometheus rule failed", zap.Error(err))
			return
		}
		ruleresource, _ := BuildPrometheusRuleTemplateResourceRequirements(req)

		err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).Create(ctx, ruleresource)
		return resp, err
	}
	promRules.ObjectMeta.Labels[constants.ExternalLabelOverlay] = "true"
	promRules.Spec = req.MonitorPromRules
	err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).UpdatePrometheusRule(ctx, promRules)
	return resp, err
}

// DeleteAlertRulesTemplate
// @Description: delete alert rules template
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) DeleteAlertRulesTemplate(ctx context.Context, req message.DeleteAlertRulesTemplateReq) (resp message.DeleteAlertRulesTemplateResp, err error) {

	promRules, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).GetPrometheusRule(ctx, req.ID)
	if err != nil {
		return resp, err
	}

	err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).DeletePrometheusRules(ctx, req.ID)
	if err != nil {
		log.Error("delete prometheus rule failed", zap.Error(err))
	}
	resp.MonitorPromRules = promRules.Spec
	return resp, err
}

// QueryAlertRuleTemplate
// @Description: get prometheus rule template
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) QueryAlertRuleTemplate(ctx context.Context, req message.QueryAlertRulesTemplateReq) (resp message.QueryAlertRulesTemplateResp, err error) {
	promRules, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).GetPrometheusRule(ctx, req.ID)
	if err != nil {
		log.Error("GetPrometheusRule failed", zap.Error(err))
		return
	}
	resp.MonitorPromRules = promRules.Spec
	return
}

// ApplyLogAlertRules
// @Description: apply log alert rules template
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) ApplyLogAlertRules(ctx context.Context, req message.ApplyLogAlertRulesReq) (resp message.ModifyAlertRulesTemplateResp, err error) {
	if req.RulesConfig.Groups == nil {
		return resp, errors.NewErrorf(errors.PrometheusRuleIsEmpty, "log alert rule not allow empty")
	}

	// Process log alert rules and deduplicate by name
	ruleGroupsNameMap := make(map[string]struct{})
	for _, rules := range req.RulesConfig.Groups {
		if _, ok := ruleGroupsNameMap[rules.Name]; ok {
			return resp, errors.NewErrorf(errors.ParameterInvalid, "duplicate rule group name: %s", rules.Name)
		}
		ruleGroupsNameMap[rules.Name] = struct{}{}
	}
	// Convert Prometheus rules to YAML format for Loki
	rulesYAML, err := yaml.Marshal(req.RulesConfig)
	if err != nil {
		log.Error("failed to marshal log alert rules to YAML", zap.Error(err))
		return resp, errors.NewErrorf(errors.MarshalError, "failed to marshal log alert rules: %v", err)
	}

	// Create or update the loki-ha-alerting-rules ConfigMap
	configMap := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "loki-ha-alerting-rules",
			Namespace: constants.ObservabilityNamespace,
			Labels: map[string]string{
				"app.kubernetes.io/name":      "loki",
				"app.kubernetes.io/component": "alerting-rules",
				"loki_rule":                   "true",
			},
		},
		Data: map[string]string{
			"loki-ha-alerting-rules.yaml": string(rulesYAML),
		},
	}

	err = resourcemanager.GetConfigMapReaderWriter().CreateOrUpdate(ctx, configMap)
	if err != nil {
		log.Error("failed to create or update loki-ha-alerting-rules ConfigMap", zap.Error(err))
		return resp, errors.NewErrorf(errors.K8SClientError, "failed to create or update ConfigMap: %v", err)
	}

	log.Info("successfully applied log alert rules to ConfigMap")

	return resp, nil
}

func (p *ClusterHandler) QueryAlertRulesByCloudUser(ctx context.Context, req message.QueryAlertRulesReq) (resp message.QueryAlertRulesByCloudUserResp, err error) {
	//get from cache
	cacheEnabled := configs.GetServiceInfo().AlertRuleCacheConfig.Enabled
	cacheKey := fmt.Sprintf(constants.RULE_CACHEKEY_TEMPLATE, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType)
	if cacheEnabled {
		alertRuleGroups, ok := ruleQueryCache.get(cacheKey)
		if ok {
			resp.AlertRuleList = alertRuleGroups
			return
		}
	}

	externalRules, _, _, err := getOverLayRules(ctx, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType)
	if err != nil {
		log.Error("getExternalAlertRules failed", zap.Error(err))
		return
	}

	alertRuleGroups := make([]structs.AlertRuleGroup, 0)
	ruleLen := 0
	maxRuleLen := configs.GetServiceInfo().AlertRuleCacheConfig.MaxAlertRulesPerCluster

	for _, group := range externalRules.Spec.Groups {
		alertRuleGroup := structs.AlertRuleGroup{Name: group.Name}
		alertRuleGroup.Rules = make([]structs.AlertRuleDetail, 0)
		for _, rule := range group.Rules {
			defaultExpr := rule.Labels[constants.RuleLabelDefaultExpr]
			defaultFor := rule.Labels[constants.RuleLabelDefaultFor]

			ruleExpr := structs.GenerateOverLayAlertRuleEXPR(rule.Expr.String())

			duration, _ := time.ParseDuration(rule.For)
			defaultDuration, _ := time.ParseDuration(defaultFor)
			ruleDetail := structs.AlertRuleDetail{}
			ruleDetail.Threshold = &ruleExpr.Value
			ruleDetail.AlertRuleID = fmt.Sprintf(constants.RULEID_TEMPLATE, group.Name, rule.Labels[constants.RuleLabelID])

			ruleDetail.Status = getAlertRuleStatus(rule.Labels)
			ruleDetail.For = utils.Int64Ptr(int64(duration.Minutes()))
			ruleDetail.DefaultFor = defaultDuration.Minutes()
			if rule.Expr.String() != defaultExpr {
				defaultRuleExpr := structs.GenerateOverLayAlertRuleEXPR(defaultExpr)
				ruleDetail.DefaultThreshold = defaultRuleExpr.Value
			} else {
				ruleDetail.DefaultThreshold = ruleExpr.Value
			}
			ruleDetail.MonitorTarget = rule.Labels[constants.RuleLabelMonitortarget]
			ruleDetail.Operator = rule.Labels[constants.RuleLabelOperator]
			ruleDetail.CanEdit = checkCanEdit(rule.Labels)
			rule = replaceExternalRulePlaceHolder(rule)
			ruleDetail.Alert = rule.Annotations[constants.RuleAnnotationsDescription]
			ruleDetail.SortID, _ = strconv.Atoi(rule.Labels[constants.RuleLabelID])
			alertRuleGroup.Rules = append(alertRuleGroup.Rules, ruleDetail)
			ruleLen++
			if ruleLen >= maxRuleLen {
				break
			}
		}
		sort.Sort(alertRuleGroup)

		alertRuleGroups = append(alertRuleGroups, alertRuleGroup)
		if ruleLen >= maxRuleLen {
			break
		}
	}
	if cacheEnabled {
		ruleQueryCache.set(cacheKey, alertRuleGroups)
	}
	resp.AlertRuleList = alertRuleGroups
	return
}

func getAlertRuleStatus(labels map[string]string) constants.AlertRuleStatus {
	if labels[constants.RuleLabelStatus] == string(constants.RULE_DISABLED) {
		return constants.RULE_DISABLED
	}
	return constants.RULE_ACTIVE
}

func getOverLayRules(ctx context.Context, tenantID, projectID, applicationID, bizType string) (externalRule *prometheusv1.PrometheusRule, externalRulesDict map[string]prometheusv1.Rule, overlayRuleExists bool, err error) {
	clusterList, errs := resourcemanager.GetClusterReaderWriter(ctx).QueryClusters(ctx, tenantID, projectID, applicationID, bizType, message.ContinueRequest{})
	if errs != nil {
		log.Error("get cluster detail failed",
			zap.Error(errs),
			zap.String("Tenant_ID", tenantID),
			zap.String("Project_ID", projectID),
			zap.String("Application_ID", applicationID))
		return externalRule, externalRulesDict, overlayRuleExists, errs
	}

	if len(clusterList.Items) == 0 {
		err = errors.NewErrorf(errors.ClusterNotFound, "cluster not found, tenantID=%s,ProjectID=%s, ApplicationID=%s",
			tenantID, projectID, applicationID)
		return externalRule, externalRulesDict, overlayRuleExists, err
	}

	cluster := clusterList.Items[0]
	alertInfo, errs := parseAlertInfo(ctx, &cluster)
	if errs != nil {
		log.Error("parse alert info failed", zap.Error(errs))
	}
	if alertInfo.RuleTemplateID == "" {
		return externalRule, externalRulesDict, overlayRuleExists, errors.NewErrorf(errors.PrometheusRuleTemplateIDIsEmpty, "rule templateID is empty")
	}
	clusterBaseInfo, errs := parseClusterBaseInfo(ctx, &cluster)
	if errs != nil {
		log.Error("parseClusterBaseInfo failed",
			zap.Error(errs),
			zap.String("Cluster_ID", clusterBaseInfo.ID))
		return externalRule, externalRulesDict, overlayRuleExists, errs
	}

	overlayRuleExists = true
	//get overlay rules by name
	externalPromRules, externalRulesDict, err := getExternalAlertRules(ctx, fmt.Sprintf(constants.OVERLAY_NAME_TEMPATE, clusterBaseInfo.ID))
	if err != nil {
		errs, ok := err.(errors.Error)
		if ok && errs.Code() == errors.PrometheusRuleNotFound {
			overlayRuleExists = false
		} else {
			log.Error("getExternalAlertRules failed", zap.Error(err))
			return
		}
	}

	externalPromTemplateRules, externalTemplateRulesDict, errs := getExternalAlertRules(ctx, alertInfo.RuleTemplateID)

	if errs != nil {
		log.Error("getExternalAlertRules failed",
			zap.Error(errs),
			zap.String("Cluster_ID", clusterBaseInfo.ID))
		return externalRule, externalRulesDict, overlayRuleExists, errs
	}

	if overlayRuleExists {
		externalPromTemplateRules.ObjectMeta.ResourceVersion = externalPromRules.ObjectMeta.ResourceVersion
		for i, group := range externalPromTemplateRules.Spec.Groups {
			for k, rule := range group.Rules {
				ruleID := fmt.Sprintf(constants.RULEID_TEMPLATE, group.Name, rule.Labels[constants.RuleLabelID])
				overlayRule, exists := externalRulesDict[ruleID]
				rule.Labels[constants.RuleLabelDefaultFor] = rule.For
				rule.Labels[constants.RuleLabelDefaultExpr] = rule.Expr.String()
				if exists {
					modifiedTag, _ := strconv.Atoi(overlayRule.Labels[constants.RuleLabelUserModified])
					if modifiedTag > 0 {
						if modifiedTag&constants.RuleStatusModifiedMask > 0 {
							rule.Labels[constants.RuleLabelStatus] = overlayRule.Labels[constants.RuleLabelStatus]
						}
						if modifiedTag&constants.RuleThresholdModifiedMask > 0 {
							rule.Expr = overlayRule.Expr
						}
						if modifiedTag&constants.RuleForModifiedMask > 0 {
							rule.For = overlayRule.For
						}
					}
					rule.Labels[constants.RuleLabelUserModified] = overlayRule.Labels[constants.RuleLabelUserModified]
				}
				rule.Labels[constants.RuleLabelCanEdit] = checkCanEdit(rule.Labels)
				externalPromTemplateRules.Spec.Groups[i].Rules[k] = rule
				externalTemplateRulesDict[ruleID] = rule
			}
		}
	}

	externalPromTemplateRules.ObjectMeta.Labels[constants.LabelClusterID] = fmt.Sprintf(constants.OVERLAY_NAME_TEMPATE, clusterBaseInfo.ID)
	externalPromTemplateRules.ObjectMeta.Labels[constants.LabelBizType] = clusterBaseInfo.BizType
	externalPromTemplateRules.ObjectMeta.Labels[constants.LabelShareMode] = string(clusterBaseInfo.ShareMode)
	externalPromTemplateRules.ObjectMeta.Labels[constants.LabelTemplateID] = alertInfo.RuleTemplateID
	externalPromTemplateRules.ObjectMeta.Labels[constants.LabelTenantID] = clusterBaseInfo.TenantID
	externalPromTemplateRules.ObjectMeta.Labels[constants.LabelProjectID] = clusterBaseInfo.ProjectID
	externalPromTemplateRules.ObjectMeta.Labels[constants.LabelApplicationID] = clusterBaseInfo.ApplicationID
	externalPromTemplateRules.Name = fmt.Sprintf(constants.OVERLAY_NAME_TEMPATE, clusterBaseInfo.ID)
	externalPromTemplateRules.Namespace = cluster.GetNamespace()

	// lifecycle of overlay rule is managed by serverless_controller for serverless clusters.
	if strings.Contains(cluster.Labels[constants.LabelBizType], "devtier") && checkuseRDS(ctx) {
		return externalPromTemplateRules, externalTemplateRulesDict, overlayRuleExists, nil
	}
	err = controllerutil.SetOwnerReference(&cluster, externalPromTemplateRules, resourceworker.Scheme)
	if err != nil {
		log.Error("error setting owner reference for prometheus rule",
			zap.Error(err),
			zap.String("Prom_Template_Rule", externalPromTemplateRules.Name))
	}
	return externalPromTemplateRules, externalTemplateRulesDict, overlayRuleExists, nil
}

func updateParamValidate(canEdit string, ruleID string, updateReq structs.AlertRuleBase) error {
	if canEdit == constants.RULE_CANEDIT_NONE {
		return errors.NewErrorf(errors.PrometheusRuleCanNotBeEdit, "The alert rule %s is not allowed to be modified", ruleID)
	}
	if canEdit != constants.RULE_CANEDIT_ALL && (updateReq.For != nil || updateReq.Threshold != nil) {
		return errors.NewErrorf(errors.PrometheusRuleCanNotBeEdit, "The alert rule %s is not allowed to be modified", ruleID)
	}

	if updateReq.Status != "" && updateReq.Status != constants.RULE_ACTIVE && updateReq.Status != constants.RULE_DISABLED {
		return errors.NewErrorf(errors.PrometheusRuleStatusNotValid,
			"The rule %s status value can only be set to atcive or disabled", ruleID)
	}
	if updateReq.For != nil && (*updateReq.For < 0 || *updateReq.For > constants.MAX_TRIGGERINTERVAL) {
		return errors.NewErrorf(errors.ParameterInvalid, "for value must between [0, 600],ruleID:%s", ruleID)
	}
	return nil
}

func getAuditInfo(updateReq structs.AlertRuleBase, rule prometheusv1.Rule) structs.UpdateAlertRule {
	// avoid modifying the original rule in replaceExternalRulePlaceHolder.
	ruleModified := replaceExternalRulePlaceHolder(*rule.DeepCopy())
	overlayExpr := structs.GenerateOverLayAlertRuleEXPR(ruleModified.Expr.String())
	threshold := overlayExpr.Value
	newFor, newThreshold, newStatus := ruleModified.For, threshold, ruleModified.Labels[constants.RuleLabelStatus]
	if updateReq.For != nil {
		newFor = time.Duration(*updateReq.For * int64(time.Minute)).String()
	}
	if updateReq.Threshold != nil {
		newThreshold = *updateReq.Threshold
	}
	if updateReq.Status != "" {
		newStatus = string(updateReq.Status)
	}
	auditLog := structs.UpdateAlertRule{
		AlertName:    ruleModified.Annotations[constants.RuleAnnotationsDescription],
		OldThreshold: threshold,
		OldFor:       ruleModified.For,
		OldStatus:    ruleModified.Labels[constants.RuleLabelStatus],
		Status:       newStatus,
		For:          newFor,
		Threshold:    newThreshold,
	}
	return auditLog
}

func (p *ClusterHandler) UpdateAlertRulesByCloudUser(ctx context.Context, req message.UpdateAlertRulesByCloudUserReq) (resp message.UpdateAlertRulesByCloudUserResp, err error) {
	resp.Data = make(map[string]structs.UpdateAlertRule)
	externalPromTemplateRules, externalTemplateRulesDict, overlayRuleExists, errs := getOverLayRules(ctx, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType)
	if errs != nil {
		log.Error("getExternalAlertRules failed", zap.Error(errs))
		return resp, errs
	}

	for ruleID, updateReq := range req.AlertRules {
		var statusFlag, forFlag, thresholdFlag int
		rule, ok := externalTemplateRulesDict[ruleID]
		if !ok {
			return resp, errors.NewErrorf(errors.PrometheusRuleNotFound, "ruleID %s not exists", ruleID)
		}
		resp.Data[ruleID] = getAuditInfo(updateReq, rule)

		canEdit := checkCanEdit(rule.Labels)
		rule.Labels[constants.RuleLabelCanEdit] = canEdit
		if paramErr := updateParamValidate(canEdit, ruleID, updateReq); paramErr != nil {
			return resp, paramErr
		}

		if updateReq.Status != "" {
			rule.Labels[constants.RuleLabelStatus] = string(updateReq.Status)
			statusFlag = constants.RuleStatusModifiedMask
		}

		if updateReq.For != nil {
			duration := time.Duration(*updateReq.For * int64(time.Minute))
			rule.For = duration.String()
			forFlag = constants.RuleForModifiedMask
		}

		if updateReq.Threshold != nil {
			overlayExpr := structs.GenerateOverLayAlertRuleEXPR(rule.Expr.String())
			overlayExpr.Value = *updateReq.Threshold
			if overlayExpr.Expr == "" || overlayExpr.Operator == "" {
				return resp, errors.NewErrorf(errors.PrometheusRuleExprNotValid, "ruleID %s expr not valid", ruleID)
			}
			decimalVal := decimal.NewFromFloat(overlayExpr.Value)
			rule.Expr = intstr.FromString(overlayExpr.Expr + overlayExpr.Operator + decimalVal.String())
			thresholdFlag = constants.RuleThresholdModifiedMask
		}

		modifiedTag, _ := strconv.Atoi(rule.Labels[constants.RuleLabelUserModified])
		if modifiedTag > 0 {
			if modifiedTag&constants.RuleStatusModifiedMask > 0 {
				statusFlag = constants.RuleStatusModifiedMask
			}
			if modifiedTag&constants.RuleForModifiedMask > 0 {
				forFlag = constants.RuleForModifiedMask
			}
			if modifiedTag&constants.RuleThresholdModifiedMask > 0 {
				thresholdFlag = constants.RuleThresholdModifiedMask
			}
		}
		modifiedLabel := statusFlag + forFlag + thresholdFlag

		rule.Labels[constants.RuleLabelUserModified] = strconv.Itoa(modifiedLabel)
		externalTemplateRulesDict[ruleID] = rule
	}

	for i, g := range externalPromTemplateRules.Spec.Groups {
		for k, rule := range g.Rules {
			ruleID := fmt.Sprintf(constants.RULEID_TEMPLATE, g.Name, rule.Labels[constants.RuleLabelID])
			externalPromTemplateRules.Spec.Groups[i].Rules[k] = externalTemplateRulesDict[ruleID]
		}
	}

	promRule := &prometheusv1.PrometheusRule{
		ObjectMeta: metav1.ObjectMeta{
			Name:            externalPromTemplateRules.Name,
			Namespace:       externalPromTemplateRules.Namespace,
			Labels:          externalPromTemplateRules.ObjectMeta.Labels,
			Annotations:     map[string]string{},
			OwnerReferences: externalPromTemplateRules.OwnerReferences,
		},
		Spec: externalPromTemplateRules.Spec,
	}

	//save overlay
	if overlayRuleExists {
		promRule.ResourceVersion = externalPromTemplateRules.ObjectMeta.ResourceVersion
		err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).UpdatePrometheusRule(ctx, promRule)
	} else {
		err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).Create(ctx, promRule)
	}
	if configs.GetServiceInfo().AlertRuleCacheConfig.Enabled {
		cacheKey := fmt.Sprintf(constants.RULE_CACHEKEY_TEMPLATE, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType)
		ruleQueryCache.delete(cacheKey)
	}
	return resp, err
}

// UpdateAlertRules
// @Description: update prometheus rules for a cluster
// @Receiver p
// @Parameter ctx
// @Parameter req
// @return resp
// @return err
func (p *ClusterHandler) UpdateAlertRules(ctx context.Context, req message.UpdateAlertRulesReq) (resp message.UpdateAlertRulesResp, err error) {
	if req.MonitorPromRules.Groups == nil {
		return resp, errors.NewErrorf(errors.PrometheusRuleIsEmpty, "prometheus alert rule not allow empty")
	}
	isVMShared := isSharedVMRule(req.ShareMode)
	promRules, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, isVMShared).GetPrometheusRule(ctx, req.ID)
	if err != nil {
		errs, ok := err.(errors.Error)
		if ok && errs.Code() != errors.PrometheusRuleNotFound {
			log.Error("get prometheus rule failed", zap.Error(err))
			return
		}

		//create prometheus rules if not exists
		cluster, err := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ID)
		if err != nil {
			log.Error("get cluster failed", zap.Error(err))
			return resp, err
		}

		createClusterReq := message.CreateClusterReq{}
		createClusterReq.ApplicationID = cluster.Labels[constants.LabelApplicationID]
		createClusterReq.ProjectID = cluster.Labels[constants.LabelProjectID]
		createClusterReq.Region = cluster.Labels[constants.LabelRegion]
		createClusterReq.TenantID = cluster.Labels[constants.LabelTenantID]
		createClusterReq.BizType = cluster.Labels[constants.LabelBizType]
		createClusterReq.ID = cluster.Labels[constants.LabelClusterID]
		createClusterReq.MonitorPromRules = req.MonitorPromRules
		createClusterReq.ShareMode = constants.ShareMode(cluster.Labels[constants.LabelShareMode])
		shareMode := cluster.Annotations[constants.AnnotationShareMode]
		err = preparePrometheusRuleResource(ctx, createClusterReq, cluster)
		if err != nil {
			log.Error("create prometheusRule failed", zap.Error(err))
			return resp, err
		}

		if shareMode == string(constants.SHARED) {
			cluster.Spec.VictoriaAccount.Alerting = getAlertingConfig()
			cluster.Spec.VictoriaAccount.RuleSelector = &metav1.LabelSelector{
				MatchLabels: map[string]string{
					constants.LabelClusterID: req.ID,
				},
			}
			cluster.Spec.VictoriaAccount.RuleNamespaceSelector = &metav1.LabelSelector{
				MatchLabels: map[string]string{
					constants.LabelTenantID: cluster.Labels[constants.LabelTenantID],
				},
			}
		} else {
			cluster.Spec.Prometheus.Alerting = getAlertingConfig()

			cluster.Spec.Prometheus.RuleSelector = &metav1.LabelSelector{
				MatchLabels: map[string]string{
					constants.LabelClusterID: req.ID,
				},
			}
			cluster.Spec.Prometheus.RuleNamespaceSelector = &metav1.LabelSelector{
				MatchLabels: map[string]string{
					constants.LabelTenantID: cluster.Labels[constants.LabelTenantID],
				},
			}
		}
		err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, cluster)
		if err != nil {
			log.Error("update cluster failed", zap.Error(err))
		}
		return resp, err
	}
	if !req.Override {
		nameMaps := make(map[string]int)
		for k, rules := range promRules.Spec.Groups {
			nameMaps[rules.Name] = k
		}
		for _, rules := range req.MonitorPromRules.Groups {
			if k, ok := nameMaps[rules.Name]; ok {
				promRules.Spec.Groups[k] = rules
			} else {
				promRules.Spec.Groups = append(promRules.Spec.Groups, rules)
			}
		}
	} else {
		promRules.Spec = req.MonitorPromRules
	}

	err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, isVMShared).UpdatePrometheusRule(ctx, promRules)
	if err != nil {
		log.Error("update prometheus rule failed", zap.Error(err))
	}
	return resp, err
}

func prepareRoleBindingResource(ctx context.Context, req message.CreateClusterReq) error {
	if req.ShareMode == constants.SHARED {
		return nil
	}
	roleBindingSource, _ := BuildClusterRoleBindingResourceRequirements(req)
	return resourcemanager.GetClusterRoleBindingReaderWriter(ctx).Create(ctx, roleBindingSource)
}

func prepareParentCluster(ctx context.Context, req message.CreateClusterReq) (*v1.Cluster, error) {
	if len(req.ParentClusterID) == 0 {
		return nil, nil
	}
	parent, err := resourcemanager.GetClusterReaderWriter(ctx).GetCluster(ctx, req.ParentClusterID)
	if err != nil {
		return nil, err
	}

	if constants.SHARED == req.ShareMode && !parent.Spec.VictoriaAccount.SupportServerless {
		parent.Spec.VictoriaAccount.SupportServerless = true
		err = resourcemanager.GetClusterReaderWriter(ctx).UpdateCluster(ctx, parent)
		if err != nil {
			log.Error("update parentCluster.Spec.VictoriaAccount.SupportServerless failed", zap.Error(err))
			return nil, err
		}
	}
	return parent, nil
}

func prepareNamespace(ctx context.Context, req message.CreateClusterReq) (string, error) {
	if req.KeySpaceID != "" && req.ParentClusterID != "" && checkuseRDS(ctx) {
		//for serverless child cluster on rds mode, we don't need prepare namespace
		return "", nil
	}
	nsName := NamespaceName(req.BizType, req.TenantID)
	nsInfo, err := resourcemanager.GetNamespaceReaderWriter().Get(ctx, nsName)
	if err != nil {
		log.Error("get namespace failed", zap.Error(err))
		return "", err
	}
	var vmAccount string
	if nsInfo != nil {
		log.Info("namespace has been existed",
			zap.String("Namespace", nsName))
		if configs.GetServiceInfo().BizTypeConfigs[req.BizType].MultiTenancyRule &&
			isSharedVMRule(req.ShareMode) {
			multiTenancyRule, ok := nsInfo.Labels[constants.LabelMultiTenancyRule]
			if !ok || multiTenancyRule != "true" {
				nsInfo.Labels[constants.LabelMultiTenancyRule] = "true"
				return "", resourcemanager.GetNamespaceReaderWriter().Update(ctx, nsInfo)
			}
		}
		//if next gen agent is enabled, we need check vmaccount be created in ns
		var ok bool
		if req.AgentNextGen {
			vmAccount, ok = nsInfo.Labels[constants.LabelVMAccountID]
			if !ok {
				accountID, err := resourcemanager.GetDataAccountReaderWriter().UniqueInt32ForCluster(ctx)
				if err != nil {
					log.Error("get unique int32 for cluster failed", zap.Error(err))
					return "", err
				}
				vmAccount = strconv.Itoa(int(accountID))
				nsInfo.Labels[constants.LabelVMAccountID] = vmAccount
				return vmAccount, resourcemanager.GetNamespaceReaderWriter().Update(ctx, nsInfo)
			}
		}
		return vmAccount, nil
	}

	var namespace *corev1.Namespace

	return vmAccount, errors.OfNullable(nil).BreakIf(func() error {
		namespace, vmAccount, err = BuildNamespaceResourceRequirements(ctx, req.BizType, req.TenantID, req.ShareMode, req.AgentNextGen)
		return err
	}).BreakIf(func() error {
		return resourcemanager.GetNamespaceReaderWriter().Create(ctx, namespace)
	}).Map(func(err error) error {
		return errors.Ignore(err, errors.DuplicatedNamespace)
	}).If(func(err error) {
		log.Error("prepare namespace failed",
			zap.Error(err),
			zap.String("Namespace", nsName))
	}).Present()
}

func prepareBucketResource(ctx context.Context, req message.CreateClusterReq) error {
	bucketName := ObjectStorageResourceName(req.ClusterBaseInfo)
	bucketExisted, e := resourcemanager.GetObjectStorageReaderWriter(ctx).IsObjectStorageExisted(ctx, bucketName)

	if e != nil {
		log.Error("IsObjectStorageExisted failed", zap.Error(e))
		return e
	}
	if bucketExisted {
		log.Info("object storage bucket has been existed",
			zap.String("Bucket_Name", bucketName))
		return nil
	}

	bucketResource, err := BuildObjectStorageResourceRequirements(bucketName)
	if err != nil {
		log.Error("describe object storage resource failed", zap.Error(err))
		return err
	}

	err = resourcemanager.GetObjectStorageReaderWriter(ctx).CreateObjectStorage(ctx, bucketResource)
	err = errors.Ignore(err, errors.DuplicatedBucketName)
	if err != nil {
		log.Error("create bucket resource failed", zap.String("Bucket_Name", bucketName))
		return err
	}
	return nil
}

// updataRole updates the o11y iam role, allowing accessARN to assume to o11y iam role.
func updateRole(ctx context.Context, roleName string, ns string, accessARN string) error {
	log.Info("iam role has been existed", zap.String("IAM_Role_Name", roleName))
	role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, ns)
	if err != nil {
		return err
	}
	iam, needUpdate, err := AppendAssumeEntity(ctx, role, accessARN)
	if err != nil {
		log.Error("append arn to role assume entities failed", zap.Error(err))
		return err
	}
	if needUpdate {
		err = resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, iam)
	}
	return err
}

func updateDataPlaneRole(ctx context.Context, roleName string, ns string, req message.CreateClusterReq) (*v1.IAM, error) {
	log.Info("iam role has been existed", zap.String("IAM_Role_Name", roleName))
	role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, ns)
	if err != nil {
		return role, err
	}
	iam, needUpdate, err := AppendDataPlaneAssumeEntity(ctx, role, req)
	if err != nil {
		log.Error("append arn to role assume entities failed", zap.Error(err))
		return iam, err
	}
	if needUpdate {
		err = resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, iam)
	}
	return iam, err
}

func ensureagentAccessRole(account string, agentaccessARN string) string {
	if len(account) > 0 {
		log.Info("agent access account is not empty, use it as agent access role", zap.String("Agent_Access_Account", account))
		return fmt.Sprintf("arn:aws:iam::%s:root", account)
	}
	return agentaccessARN
}

func prepareDataPlaneAgentRole(ctx context.Context, ns string, req message.CreateClusterReq) (string, error) {
	dataroleName, err := AgentDataPlaneRoleNameWrap(ctx, ns, buildAgentBaseInfoFromReq(req))
	if err != nil {
		return "", err
	}
	var iam *v1.IAM
	dataRoleExisted, err := resourcemanager.GetIAMReaderWriter(ctx).IsRoleExists(ctx, dataroleName, ns)
	if err != nil {
		log.Error("IsRoleExists failed", zap.Error(err))
		return "", err
	}
	if dataRoleExisted {
		//update role with new sa name
		// FIXME: there is quota limit for a single role to hold multiple k8s service accounts in azure & aws.
		iam, err = updateDataPlaneRole(ctx, dataroleName, ns, req)
		if err != nil {
			log.Error("update dataplane role failed", zap.Error(err))
			return "", err
		}
	} else {
		//create role in dataplane
		iam, err = BuildAgentDataPlaneRoleRequirements(req)

		if err != nil {
			log.Error("describe role resource failed", zap.Error(err))
			return "", err
		}

		err = resourcemanager.GetIAMReaderWriter(ctx).CreateRole(ctx, iam)

		if err != nil && errors.Ignore(err, errors.DuplicatedRoleName) == nil {
			iam, err = updateDataPlaneRole(ctx, dataroleName, ns, req)
			if err != nil {
				log.Error("update dataplane role failed", zap.Error(err))
				return "", err
			}
		}
		if err != nil {
			log.Error("create dataplane role failed", zap.Error(err))
			return "", err
		}
	}
	return BuildDataPlaneServiceAccountRoleARN(iam), nil
}

func prepareIAMResource(ctx context.Context, req message.CreateClusterReq) error {
	if iam := configs.GetServiceInfo().BizTypeConfigs[req.BizType].InheritedIAM; len(iam) > 0 {
		log.Info("skip preparing IAM resource because of InheritedIAM")
		return nil
	}
	ns := NamespaceName(req.BizType, req.TenantID)
	agentAccessRole := ensureagentAccessRole(req.AgentAccessAccount, req.AgentAccessARN)

	// 3-layers dp monitor don't need iam to write logs
	if req.AgentEnvironment != nil && strings.Contains(req.BizType, BizeTypeDP) {
		log.Info("skip preparing IAM resource because of dp in 3-layer")
		return nil
	}
	//deal with o11y dataplane agent role
	if req.AgentEnvironment != nil && !strings.Contains(req.BizType, BizeTypeDP) && !strings.Contains(req.BizType, BizeTypeServerless) {
		role, err := prepareDataPlaneAgentRole(ctx, ns, req)
		if err != nil {
			return err
		}
		agentAccessRole = role
	}
	// deal with o11y agent role
	if !AgentRoleNotNeeded(&req.ClusterBaseInfo) {
		if len(agentAccessRole) == 0 {
			log.Info("agent access role and account id all empty,we do nothing iam access role")
			return nil
		}
		roleName := AgentRoleName(req.ClusterBaseInfo)
		roleExisted, e := resourcemanager.GetIAMReaderWriter(ctx).IsRoleExists(ctx, roleName, ns)

		if e != nil {
			log.Error("IsRoleExists failed", zap.Error(e))
			return e
		}
		if roleExisted {
			return updateRole(ctx, roleName, ns, agentAccessRole)
		} else {
			req.AgentAccessARN = agentAccessRole
			iam, err := BuildAgentRoleRequirements(req)

			if err != nil {
				log.Error("describe role resource failed", zap.Error(err))
				return err
			}

			err = resourcemanager.GetIAMReaderWriter(ctx).CreateRole(ctx, iam)

			if err != nil && errors.Ignore(err, errors.DuplicatedRoleName) == nil {
				return updateRole(ctx, roleName, ns, agentAccessRole)
			}
			if err != nil {
				log.Error("create role failed", zap.Error(err))
				return err
			}
			return resourcemanager.GetNamespaceReaderWriter().
				CreateServiceAccount(ctx, ns, buildServiceAccountRoleARN(iam))
		}
	}

	switch configs.GetServiceInfo().Vendor {
	case string(constants.Azure):
		if err := ensureAzureAccessARNForAgentRoleNotNeeded(ctx, req); err != nil {
			log.Error("ensure azure access arn in iam failed", zap.Error(err))
			return err
		}
	case string(constants.GCP):
		if AgentRoleNotNeeded(&req.ClusterBaseInfo) {
			if err := ensureGCPAccessARNForAgentRoleNotNeeded(ctx, req); err != nil {
				log.Error("ensure gcp access arn in iam failed", zap.Error(err))
				return err
			}
		}
	}
	return nil
}

func ensureAzureAccessARNForAgentRoleNotNeeded(ctx context.Context, req message.CreateClusterReq) error {
	accessARN := strings.TrimSpace(req.AgentAccessARN)
	if accessARN == "" {
		return nil
	}

	ns := NamespaceName(req.BizType, req.TenantID)
	roleName := AgentRoleName(req.ClusterBaseInfo)
	role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, ns)
	if err != nil {
		if !errors.Is(err, errors.RoleNotFound) {
			return err
		}

		reqCopy := req
		reqCopy.AgentAccessARN = accessARN
		iam, buildErr := BuildAgentRoleRequirements(reqCopy)
		if buildErr != nil {
			return buildErr
		}
		createErr := resourcemanager.GetIAMReaderWriter(ctx).CreateRole(ctx, iam)
		if createErr == nil || errors.Ignore(createErr, errors.DuplicatedRoleName) == nil {
			return nil
		}
		return createErr
	}

	if role.Spec.Azure == nil || utils.IsIncludeString(role.Spec.Azure.AdditionalRoleMembers, accessARN) {
		return nil
	}

	roleCopy := role.DeepCopy()
	roleCopy.Spec.Azure.AdditionalRoleMembers = append(roleCopy.Spec.Azure.AdditionalRoleMembers, accessARN)
	return resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, roleCopy)
}

func ensureGCPAccessARNForAgentRoleNotNeeded(ctx context.Context, req message.CreateClusterReq) error {
	accessARN := strings.TrimSpace(req.AgentAccessARN)
	if accessARN == "" {
		return nil
	}

	normalizedAccessARN, err := checkGCPServiceAccount(accessARN)
	if err != nil {
		return err
	}

	ns := NamespaceName(req.BizType, req.TenantID)
	roleName := AgentRoleName(req.ClusterBaseInfo)
	role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, ns)
	if err != nil {
		if !errors.Is(err, errors.RoleNotFound) {
			return err
		}

		reqCopy := req
		reqCopy.AgentAccessARN = normalizedAccessARN
		iam, buildErr := BuildAgentRoleRequirements(reqCopy)
		if buildErr != nil {
			return buildErr
		}
		createErr := resourcemanager.GetIAMReaderWriter(ctx).CreateRole(ctx, iam)
		if createErr == nil || errors.Ignore(createErr, errors.DuplicatedRoleName) == nil {
			return nil
		}
		return createErr
	}

	if role.Spec.Gcp.UseExternalPrincipal {
		if utils.IsIncludeString(role.Spec.Gcp.AssumeRoleAccounts, normalizedAccessARN) {
			return nil
		}
		roleCopy := role.DeepCopy()
		roleCopy.Spec.Gcp.AssumeRoleAccounts = append(roleCopy.Spec.Gcp.AssumeRoleAccounts, normalizedAccessARN)
		return resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, roleCopy)
	}

	if role.Spec.Gcp.ProjectID == "" {
		if utils.IsIncludeString(role.Spec.Gcp.AssumeRoleAccounts, normalizedAccessARN) {
			return nil
		}
		roleCopy := role.DeepCopy()
		roleCopy.Spec.Gcp.AssumeRoleAccounts = append(roleCopy.Spec.Gcp.AssumeRoleAccounts, normalizedAccessARN)
		return resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, roleCopy)
	}

	// dataplane role management should still go through dataplane append path.
	return nil
}

func removeAzureAccessARNFromIAMOnDelete(ctx context.Context, cluster *v1.Cluster, accessARN string) error {
	if configs.GetServiceInfo().Vendor != string(constants.Azure) {
		return nil
	}
	if cluster == nil {
		return nil
	}

	trimmedAccessARN := strings.TrimSpace(accessARN)
	if trimmedAccessARN == "" {
		return nil
	}

	baseInfo := structs.ClusterBaseInfo{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(cluster.Labels[constants.LabelVendor]),
			Region: cluster.Labels[constants.LabelRegion],
		},
		BizType:       cluster.Labels[constants.LabelBizType],
		TenantID:      cluster.Labels[constants.LabelTenantID],
		ProjectID:     cluster.Labels[constants.LabelProjectID],
		ApplicationID: cluster.Labels[constants.LabelApplicationID],
		ShareMode:     constants.ShareMode(cluster.Labels[constants.LabelShareMode]),
	}

	ns := NamespaceName(baseInfo.BizType, baseInfo.TenantID)
	roleName := AgentRoleName(baseInfo)
	role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, ns)
	if err != nil {
		if errors.Is(err, errors.RoleNotFound) {
			return nil
		}
		return err
	}
	if role.Spec.Azure == nil {
		return nil
	}

	roleCopy := role.DeepCopy()
	oldAdditional := roleCopy.Spec.Azure.AdditionalRoleMembers
	newAdditional := make([]string, 0, len(oldAdditional))
	removed := false
	for _, member := range oldAdditional {
		if strings.TrimSpace(member) == trimmedAccessARN {
			removed = true
			continue
		}
		newAdditional = append(newAdditional, member)
	}
	if !removed {
		return nil
	}

	roleCopy.Spec.Azure.AdditionalRoleMembers = newAdditional
	if len(roleCopy.Spec.Azure.LastAdditionalRoleMembers) > 0 {
		newLast := make([]string, 0, len(roleCopy.Spec.Azure.LastAdditionalRoleMembers))
		for _, member := range roleCopy.Spec.Azure.LastAdditionalRoleMembers {
			if strings.TrimSpace(member) == trimmedAccessARN {
				continue
			}
			newLast = append(newLast, member)
		}
		roleCopy.Spec.Azure.LastAdditionalRoleMembers = newLast
	}

	if len(newAdditional) == 0 {
		return resourcemanager.GetIAMReaderWriter(ctx).DeleteRole(ctx, roleName, ns)
	}

	return resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, roleCopy)
}

func removeGCPAccessARNFromIAMOnDelete(ctx context.Context, cluster *v1.Cluster, accessARN string) error {
	if configs.GetServiceInfo().Vendor != string(constants.GCP) {
		return nil
	}
	if cluster == nil {
		return nil
	}

	trimmedAccessARN := strings.TrimSpace(accessARN)
	if trimmedAccessARN == "" {
		return nil
	}
	normalizedAccessARN, err := checkGCPServiceAccount(trimmedAccessARN)
	if err != nil {
		return err
	}

	baseInfo := structs.ClusterBaseInfo{
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(cluster.Labels[constants.LabelVendor]),
			Region: cluster.Labels[constants.LabelRegion],
		},
		BizType:       cluster.Labels[constants.LabelBizType],
		TenantID:      cluster.Labels[constants.LabelTenantID],
		ProjectID:     cluster.Labels[constants.LabelProjectID],
		ApplicationID: cluster.Labels[constants.LabelApplicationID],
		ShareMode:     constants.ShareMode(cluster.Labels[constants.LabelShareMode]),
	}

	ns := NamespaceName(baseInfo.BizType, baseInfo.TenantID)
	roleName := AgentRoleName(baseInfo)
	role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, ns)
	if err != nil {
		if errors.Is(err, errors.RoleNotFound) {
			return nil
		}
		return err
	}

	roleCopy := role.DeepCopy()
	oldAdditional := roleCopy.Spec.Gcp.AssumeRoleAccounts
	newAdditional := make([]string, 0, len(oldAdditional))
	removed := false
	for _, member := range oldAdditional {
		if strings.TrimSpace(member) == normalizedAccessARN {
			removed = true
			continue
		}
		newAdditional = append(newAdditional, member)
	}
	if !removed {
		return nil
	}

	roleCopy.Spec.Gcp.AssumeRoleAccounts = newAdditional
	if len(roleCopy.Spec.Gcp.LastAssumeRoleAccounts) > 0 {
		newLast := make([]string, 0, len(roleCopy.Spec.Gcp.LastAssumeRoleAccounts))
		for _, member := range roleCopy.Spec.Gcp.LastAssumeRoleAccounts {
			if strings.TrimSpace(member) == normalizedAccessARN {
				continue
			}
			newLast = append(newLast, member)
		}
		roleCopy.Spec.Gcp.LastAssumeRoleAccounts = newLast
	}

	if len(newAdditional) == 0 {
		return resourcemanager.GetIAMReaderWriter(ctx).DeleteRole(ctx, roleName, ns)
	}

	return resourcemanager.GetIAMReaderWriter(ctx).UpdateRole(ctx, roleCopy)
}

func isSharedVMRule(sharedMode constants.ShareMode) bool {
	return sharedMode == constants.SHARED
}

func preparePrometheusRuleResource(ctx context.Context, req message.CreateClusterReq, clusterResource *v1.Cluster) error {
	if req.KeySpaceID != "" && checkuseRDS(ctx) {
		//for serverless cluster on rds mode, the lifecycle of rule is managed by serverless_controller.
		return nil
	}

	isVMRule := isSharedVMRule(req.ShareMode)
	isExist, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, isVMRule).IsPrometheusRuleIDExists(ctx, req.ID)
	if err != nil {
		log.Error("IsPrometheusRuleIDExists failed", zap.Error(err))
		return err
	}
	if req.RuleTemplateID != "" {
		templateRules, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).GetPrometheusRule(ctx, req.RuleTemplateID)
		if err != nil {
			log.Error("prometheusrule template dose not exist",
				zap.Error(err),
				zap.String("Rule_Template_ID", req.RuleTemplateID),
			)
			//return err
		}
		if templateRules != nil {
			req.MonitorPromRules = templateRules.Spec
		}
	}
	// when req.RuleTemplateID == "" or the requested prometheus rule template does not exist,
	// use the default prometheus rule template configured in the BizTypeConfig
	if req.MonitorPromRules.Groups == nil {
		ruleTemplateID := configs.GetServiceInfo().BizTypeConfigs[req.BizType].MonitorPromRuleTemplate
		templateRules, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).GetPrometheusRule(ctx, ruleTemplateID)
		if err != nil {
			log.Error("prometheusrule template dose not exist",
				zap.Error(err),
				zap.String("Rule_Template_ID", req.RuleTemplateID),
			)
		}
		if templateRules != nil {
			req.MonitorPromRules = templateRules.Spec
		}
	}

	ruleresource, err := BuildPrometheusRuleResourceRequirements(req, clusterResource)
	if err != nil {
		return err
	}
	if clusterResource.OwnerReferences != nil {
		if err = controllerutil.SetOwnerReference(clusterResource, ruleresource, resourceworker.Scheme); err != nil {
			return err
		}
	}

	if isExist {
		return resourcemanager.GetPrometheusRulesReaderWriter(ctx, isVMRule).UpdatePrometheusRule(ctx, ruleresource)
	}
	return resourcemanager.GetPrometheusRulesReaderWriter(ctx, isVMRule).Create(ctx, ruleresource)
}

func parseClusterBaseInfo(ctx context.Context, clusterResource *v1.Cluster) (structs.ClusterBaseInfo, error) {
	if clusterResource == nil || len(clusterResource.Labels) == 0 ||
		clusterResource.Labels[constants.LabelVendor] == "" ||
		clusterResource.Labels[constants.LabelRegion] == "" {

		return structs.ClusterBaseInfo{}, errors.New(errors.ParameterInvalid, "cluster resource is empty")
	}
	baseInfo := structs.ClusterBaseInfo{
		ID: clusterResource.Name,
		CommonReqInfo: structs.CommonReqInfo{
			Vendor: constants.ProvidedVendor(clusterResource.Labels[constants.LabelVendor]),
			Region: clusterResource.Labels[constants.LabelRegion],
		},
		TenantID:           clusterResource.Labels[constants.LabelTenantID],
		ProjectID:          clusterResource.Labels[constants.LabelProjectID],
		ApplicationID:      clusterResource.Labels[constants.LabelApplicationID],
		BizType:            clusterResource.Labels[constants.LabelBizType],
		Version:            constants.VersionCode(clusterResource.Annotations[constants.AnnotationVersion]),
		Note:               clusterResource.Annotations[constants.AnnotationNote],
		Tags:               utils.UnmarshalList(clusterResource.Annotations[constants.AnnotationTags]),
		ShareMode:          constants.ShareMode(clusterResource.Annotations[constants.AnnotationShareMode]),
		DataPropertyRights: getDataPropertyRights(clusterResource),
		InternalReadUri:    parseInternalReadUri(ctx, clusterResource),
		OpenReadUri:        parseO11yOpenReadUri(ctx, clusterResource),
		MuteAlert:          clusterResource.Spec.MuteAlert,
	}
	return baseInfo, nil
}

func parseInternalReadUri(ctx context.Context, clusterResource *v1.Cluster) string {
	readClusterID := clusterResource.Name
	if len(clusterResource.Labels[constants.LabelParentClusterID]) > 0 {
		readClusterID = clusterResource.Labels[constants.LabelParentClusterID]
	}

	bizType := clusterResource.Labels[constants.LabelBizType]
	if bizType == constants.DefaultByocBizType {
		hostName := resolveDataServerHostName(bizType)
		return fmt.Sprintf(constants.InternalUnifiedReadUriTemplate, hostName, readClusterID, constants.METRICS)
	}

	if isNextGenCluster(clusterResource) {
		return fmt.Sprintf(constants.InternalReadUriTemplateV2, configs.GetServiceInfo().LogicalRegionName,
			configs.GetServiceInfo().Vendor,
			constants.GetDomainPostfix(configs.GetServiceInfo().Env, configs.GetServiceInfo().DomainPostfix),
			readClusterID)
	}
	return fmt.Sprintf(constants.InternalReadUriTemplate,
		configs.GetServiceInfo().LogicalRegionName,
		configs.GetServiceInfo().Vendor,
		constants.GetDomainPostfix(configs.GetServiceInfo().Env, configs.GetServiceInfo().DomainPostfix),
		readClusterID)
}

func resolveDataServerHostName(bizType string) string {
	if bizType == constants.DefaultByocBizType {
		bizCfg := configs.GetServiceInfo().BizTypeConfigs[bizType]
		if bizCfg.O11yGlobalDSHostName != "" {
			return bizCfg.O11yGlobalDSHostName
		}
	}

	return fmt.Sprintf(
		"www.ds.%s.%s.%s",
		configs.GetServiceInfo().LogicalRegionName,
		configs.GetServiceInfo().Vendor,
		constants.GetDomainPostfix(configs.GetServiceInfo().Env, configs.GetServiceInfo().DomainPostfix),
	)
}

func parseO11yOpenReadUri(ctx context.Context, clusterResource *v1.Cluster) string {
	readClusterID := clusterResource.Name
	if len(clusterResource.Labels[constants.LabelParentClusterID]) > 0 {
		readClusterID = clusterResource.Labels[constants.LabelParentClusterID]
	}

	bizType := clusterResource.Labels[constants.LabelBizType]
	if bizType == constants.DefaultByocBizType {
		bizCfg := configs.GetServiceInfo().BizTypeConfigs[bizType]
		hostName := bizCfg.O11yGlobalDSHostName
		if hostName == "" {
			hostName = bizCfg.GlobalDataServerHostname
		}
		return fmt.Sprintf(constants.OpenUnifiedReadUriTemplate, hostName, readClusterID, constants.METRICS)
	}

	if isNextGenCluster(clusterResource) {
		return fmt.Sprintf(constants.OpenReadUriTemplateV2,
			configs.GetServiceInfo().Region,
			configs.GetServiceInfo().Vendor,
			constants.GetDomainPostfix(configs.GetServiceInfo().Env, configs.GetServiceInfo().DomainPostfix),
			"metrics",
			clusterResource.Labels[constants.LabelBizType],
			clusterResource.Labels[constants.LabelTenantID],
			clusterResource.Labels[constants.LabelApplicationID])
	}
	return fmt.Sprintf(constants.OpenReadUriTemplate,
		configs.GetServiceInfo().Region,
		configs.GetServiceInfo().Vendor,
		constants.GetDomainPostfix(configs.GetServiceInfo().Env, configs.GetServiceInfo().DomainPostfix),
		clusterResource.Labels[constants.LabelBizType],
		clusterResource.Labels[constants.LabelTenantID],
		clusterResource.Labels[constants.LabelProjectID],
		clusterResource.Labels[constants.LabelApplicationID])
}

func parseAlertInfo(ctx context.Context, clusterResource *v1.Cluster) (structs.AlertInfo, error) {
	return structs.AlertInfo{
		MuteAlert:      clusterResource.Spec.MuteAlert,
		RuleTemplateID: clusterResource.Labels[constants.LabelTemplateID],
	}, nil
}

func getDataPropertyRights(clusterResource *v1.Cluster) constants.DataPropertyRights {
	anno := clusterResource.Annotations[constants.AnnotationDataPropertyRights]
	if len(anno) > 0 && anno == "1" {
		return constants.ServiceProvider
	}
	return constants.CustomerTenant
}

func parseMonitorInfo(ctx context.Context, clusterResource *v1.Cluster) (structs.MonitorInfo, error) {
	if clusterResource == nil {
		return structs.MonitorInfo{}, errors.New(errors.ParameterInvalid, "cluster resource is empty")
	}

	var vmAccountID, vmProjectID int32
	if clusterResource.Labels[constants.LabelShareMode] == string(constants.SHARED) {
		vmAccountID = clusterResource.Spec.VictoriaAccount.AccountID
		vmProjectID = clusterResource.Spec.VictoriaAccount.ProjectID
	}

	monitorInfo := structs.MonitorInfo{
		SpecID:              clusterResource.Annotations[constants.AnnotationMonitorSpec],
		EndpointServiceName: clusterResource.Status.ServiceName,
		VMAccountID:         strconv.Itoa(int(vmAccountID)),
		VMProjectID:         strconv.Itoa(int(vmProjectID)),
	}

	azs := configs.GetServiceInfo().AZConfig
	if len(azs) == 0 {
		return monitorInfo, errors.New(errors.AbsenceOfConfig, "no available zone")
	}

	if dns := clusterResource.Status.DnsName; len(dns) > 0 {
		monitorInfo.DNSName = dns
	} else {
		az := clusterResource.Labels[constants.LabelAZ]
		if azConfig, ok := azs[az]; ok {
			monitorInfo.DNSName = azConfig.DnsName
		} else {
			return monitorInfo, errors.NewErrorf(errors.AbsenceOfConfig, "no az config named %s found", az)
		}
	}

	return monitorInfo, nil
}

const s3ArnTemplate = "arn:aws:s3:::%s/%s"
const gcpArnTemplate = "%s/%s"
const azureArnTemplate = "%s/%s"
const alicloudArnTemplate = "oss://%s/%s"

func parseStorageInfo(ctx context.Context, baseInfo *structs.ClusterBaseInfo, clusterResource *v1.Cluster) ([]structs.StorageInfo, error) {
	storageInfos := make([]structs.StorageInfo, 0)
	if baseInfo == nil {
		return storageInfos, errors.New(errors.ParameterInvalid, "cluster base info is empty")
	}
	bucketName := ObjectStorageResourceName(*baseInfo)
	arnTemplate, storageAccount := getStorageLocationMetadata()

	for _, category := range constants.AllCategory() {
		storageInfos = append(storageInfos, buildStorageInfo(
			*baseInfo,
			bucketName,
			storageAccount,
			arnTemplate,
			category,
		))
	}
	return storageInfos, nil
}

func getStorageLocationMetadata() (string, string) {
	switch configs.GetServiceInfo().Vendor {
	case string(constants.AWS):
		return s3ArnTemplate, configs.GetAccountID()
	case string(constants.GCP):
		return gcpArnTemplate, ""
	case string(constants.Azure):
		return azureArnTemplate, configs.GetAzureStorageAccount()
	case string(constants.AliCloud):
		return alicloudArnTemplate, configs.GetAccountID()
	default:
		return "", ""
	}
}

func buildStorageInfo(
	baseInfo structs.ClusterBaseInfo,
	bucketName string,
	storageAccount string,
	arnTemplate string,
	category constants.ObservabilityCategory,
) structs.StorageInfo {
	objectPath := buildStorageObjectPath(baseInfo, bucketName, category)
	return structs.StorageInfo{
		Category:       category,
		BucketName:     bucketName,
		StorageAccount: storageAccount,
		DataARN:        fmt.Sprintf(arnTemplate, bucketName, objectPath),
		ObjectPath:     objectPath,
	}
}

func buildStorageObjectPath(baseInfo structs.ClusterBaseInfo, bucketName string, category constants.ObservabilityCategory) string {
	if category != constants.SLOWQUERYS && category != constants.SQLSTATEMENTS {
		return ObjectStorageDataPrefix(baseInfo, category)
	}

	// Vector keeps dedicated/next-gen statement files under logs/statementlog/,
	// while serverless uses logs/statement/. Slow query files stay under logs/slowlog/.
	logCategoryPath := buildStorageLogCategoryPath(baseInfo, category)

	if strings.Contains(baseInfo.BizType, BizeTypeServerless) {
		serverlessPath := fmt.Sprintf("0/%s/serverless/%s/%s/%s/logs/%s/",
			baseInfo.BizType,
			baseInfo.TenantID,
			baseInfo.ProjectID,
			baseInfo.ApplicationID,
			logCategoryPath,
		)
		if (baseInfo.Vendor == constants.AliCloud ||
			configs.GetServiceInfo().Vendor == string(constants.AliCloud)) && bucketName != "" {
			return fmt.Sprintf("%s/%s", bucketName, serverlessPath)
		}
		return serverlessPath
	}

	if baseInfo.BizType == constants.DefaultByocBizType || strings.Contains(baseInfo.BizType, "nextgen") {
		return fmt.Sprintf("0/%s/%s/logs/%s/",
			"next-gen",
			baseInfo.ApplicationID,
			logCategoryPath,
		)
	}

	return fmt.Sprintf("0/%s/%s/%s/logs/%s/",
		baseInfo.TenantID,
		baseInfo.ProjectID,
		baseInfo.ApplicationID,
		logCategoryPath,
	)
}

func buildStorageLogCategoryPath(baseInfo structs.ClusterBaseInfo, category constants.ObservabilityCategory) string {
	switch category {
	case constants.SLOWQUERYS:
		return "slowlog"
	case constants.SQLSTATEMENTS:
		if strings.Contains(baseInfo.BizType, BizeTypeServerless) {
			return "statement"
		}
		return "statementlog"
	default:
		return string(category)
	}
}

func buildRegionalLogQueryURI(bizType string) string {
	path := fmt.Sprintf("/loki/%s", constants.ReleaseNameKubePrometheusStack)
	if bizType == constants.DefaultByocBizType {
		path = fmt.Sprintf("/internal/loki/%s", constants.ReleaseNameKubePrometheusStack)
	}

	return fmt.Sprintf("https://%s%s", resolveDataServerHostName(bizType), path)
}

func buildRegionalMultitenantQueryURI(bizType string, vmAccountID int32) string {
	if vmAccountID == 0 {
		return ""
	}

	query := url.Values{}
	query.Add("extra_filters[]", fmt.Sprintf(`{vm_account_id="%d"}`, vmAccountID))

	if bizType == constants.DefaultByocBizType {
		return fmt.Sprintf(
			"%s?%s",
			fmt.Sprintf(
				constants.InternalUnifiedReadUriTemplate,
				resolveDataServerHostName(bizType),
				"multitenant",
				constants.METRICS,
			),
			query.Encode(),
		)
	}

	return fmt.Sprintf(
		"https://%s/internal/v2/metrics/multitenant/api/v1/?%s",
		resolveDataServerHostName(bizType),
		query.Encode(),
	)
}

func parseVMAccountID(vmAccountID string) (int32, error) {
	id, err := strconv.ParseInt(vmAccountID, 10, 32)
	if err != nil {
		return 0, err
	}

	return int32(id), nil
}

func buildRegionalDiagnosisQueryURI(bizType string) string {
	return fmt.Sprintf("https://%s/internal/diagnosis/api/v1/", resolveDataServerHostName(bizType))
}

func isNextGenCluster(clusterResource *v1.Cluster) bool {
	label, ok := clusterResource.Labels[constants.LabelNextGen]
	if !ok {
		return false
	}
	if label == "true" {
		return true
	}
	return false
}

const roleArnTemplate = "arn:aws:iam::%s:role/%s"
const aliRoleArnTemplate = "acs:ram::%s:role/%s"
const gcpRoleArnTemplate = "%s@%s"

func parseAgentInfo(ctx context.Context, baseInfo *structs.ClusterBaseInfo, clusterResource *v1.Cluster) (agentInfo structs.AgentInfo, err error) {
	if baseInfo == nil {
		return agentInfo, errors.New(errors.ParameterInvalid, "cluster base info is empty")
	}

	if iam := configs.GetServiceInfo().BizTypeConfigs[baseInfo.BizType].InheritedIAM; len(iam) > 0 {
		agentInfo.RoleARN = iam
	} else {
		roleName := AgentRoleName(*baseInfo)

		role, err := resourcemanager.GetIAMReaderWriter(ctx).GetRole(ctx, roleName, NamespaceName(baseInfo.BizType, baseInfo.TenantID))
		if err != nil {
			if (errors.Is(err, errors.RoleNotFound) && AgentRoleNotNeeded(baseInfo)) || (errors.Is(err, errors.RoleNotFound) && isNextGenCluster(clusterResource)) || (errors.Is(err, errors.RoleNotFound) && baseInfo.Vendor == constants.GCP) {
				return agentInfo, nil
			}
			log.Error("get role failed", zap.Error(err))
			return agentInfo, err
		}
		agentInfo.RoleARN = buildServiceAccountRoleARN(role)
	}

	return
}

func AgentRoleNotNeeded(baseInfo *structs.ClusterBaseInfo) bool {
	// 1. dp agent will not need o11y role, but exclusive-mode dp clusters need o11y role.
	// 2. azure 3-layers agent will not need o11y agent role, but exclusive-mode & seed clusters still need it.
	return (strings.Contains(baseInfo.BizType, BizeTypeDP) && baseInfo.ShareMode == constants.SHARED) ||
		(baseInfo.Vendor == constants.Azure && baseInfo.ShareMode == constants.SHARED && baseInfo.ApplicationID != constants.ApplicationIDSeed)
}

func parseClusterStatus(ctx context.Context, clusterResource *v1.Cluster) (constants.ResourceStatus, error) {
	if clusterResource == nil {
		return "", errors.New(errors.ParameterInvalid, "cluster resource is empty")
	}
	return clusterResource.Status.Phase, nil
}

func parseClusterLastScaleTime(ctx context.Context, clusterResource *v1.Cluster) (map[string]string, error) {
	if clusterResource == nil {
		err := errors.New(errors.ParameterInvalid, "cluster resource is empty")
		return nil, err
	}
	t := make(map[string]string)
	if clusterResource.Annotations[constants.AnnotationLastScaleTime] == "" {
		t["spec"] = "0"
	} else {
		t["spec"] = clusterResource.Annotations[constants.AnnotationLastScaleTime]
	}
	if clusterResource.Annotations[constants.AnnotationLastScaleStorage] == "" {
		t["storage"] = "0"
	} else {
		t["storage"] = clusterResource.Annotations[constants.AnnotationLastScaleStorage]
	}

	return t, nil
}

func buildServiceAccountRoleARN(iam *v1.IAM) string {
	switch configs.GetServiceInfo().Vendor {
	case string(constants.AWS):
		return fmt.Sprintf(roleArnTemplate, configs.GetAccountID(), iam.Labels[constants.LabelAlias])
	case string(constants.GCP):
		return fmt.Sprintf(gcpRoleArnTemplate, iam.Labels[constants.LabelAlias], configs.GetAccountID())
	case string(constants.Azure):
		// in azure, identity's clientId is used as the value of serviceAccount annotation,
		// but the clientId is generated after identity creation, so it might be empty during create api processing.
		// it will be updated by iam controller until IAM identity is created.
		return iam.Status.ClientId
	case string(constants.AliCloud):
		return fmt.Sprintf(aliRoleArnTemplate, configs.GetAccountID(), iam.Labels[constants.LabelAlias])
	}
	return ""
}

func getExternalAlertRules(ctx context.Context, name string) (externalRule *prometheusv1.PrometheusRule, externalRulesDict map[string]prometheusv1.Rule, err error) {
	externalRulesDict = make(map[string]prometheusv1.Rule, 0)
	externalRules := make([]prometheusv1.RuleGroup, 0)

	overlayAlertRules, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, false).GetPrometheusRule(ctx, name)
	if err != nil {
		return
	}

	if overlayAlertRules != nil && overlayAlertRules.Spec.Groups != nil {
		for _, ruelGroup := range overlayAlertRules.Spec.Groups {
			if ruelGroup.Rules == nil {
				continue
			}
			ruleList := make([]prometheusv1.Rule, 0)
			for _, rule := range ruelGroup.Rules {
				//filter record alert
				if rule.Alert == "" {
					continue
				}
				if visibility, ok := rule.Labels[constants.RuleLabelVisibility]; !ok ||
					visibility != "external" {
					continue
				}

				//filter alert rule if ID is empty
				LabelRuleID, exists := rule.Labels[constants.RuleLabelID]

				if !exists || len(LabelRuleID) <= 0 {
					continue
				}
				ruleID := fmt.Sprintf(constants.RULEID_TEMPLATE, ruelGroup.Name,
					rule.Labels[constants.RuleLabelID])

				externalRulesDict[ruleID] = rule
				ruleList = append(ruleList, rule)
			}
			if len(ruleList) == 0 {
				continue
			}
			group := prometheusv1.RuleGroup{
				Name:     ruelGroup.Name,
				Interval: ruelGroup.Interval,
				Rules:    ruleList,
			}
			externalRules = append(externalRules, group)
		}
	}
	overlayAlertRules.Spec.Groups = externalRules
	return overlayAlertRules, externalRulesDict, nil

}

func checkCanEdit(labels map[string]string) string {
	if labels[constants.RuleLabelCanEdit] == constants.RULE_CANEDIT_NONE {
		return labels[constants.RuleLabelCanEdit]
	}

	if labels[constants.RuleLabelMonitortarget] == "" {
		return constants.RULE_CANEDIT_STATUS
	}

	if labels[constants.RuleLabelOperator] == "" {
		return constants.RULE_CANEDIT_STATUS
	}

	if labels[constants.RuleLabelCanEdit] == constants.RULE_CANEDIT_ALL {
		return labels[constants.RuleLabelCanEdit]
	}

	return constants.RULE_CANEDIT_STATUS

}

func replaceExternalRulePlaceHolder(externalrule prometheusv1.Rule) prometheusv1.Rule {
	overlayExpr := structs.GenerateOverLayAlertRuleEXPR(externalrule.Expr.String())
	minu := decimal.NewFromFloat(overlayExpr.Value)
	mul := decimal.NewFromFloat(100)
	thresholdPercentage := fmt.Sprintf("%v", minu.Mul(mul))
	duration, _ := time.ParseDuration(externalrule.For)
	triggerInteval := fmt.Sprintf("%.0f", duration.Minutes())
	placeHolderVariables := map[string]string{
		constants.PlaceHolderThreshold:           fmt.Sprintf("%v", overlayExpr.Value),
		constants.PlaceHolderSeverity:            externalrule.Labels[constants.RuleLabelSeverity],
		constants.PlaceHolderTriggerInteval:      triggerInteval,
		constants.PlaceHolderThresholdPercentage: thresholdPercentage,
	}
	externalrule.Annotations[constants.RuleAnnotationsMessage] = replaceByPlaceHold(externalrule.Annotations[constants.RuleAnnotationsMessage],
		placeHolderVariables)
	externalrule.Annotations[constants.RuleAnnotationsDescription] = replaceByPlaceHold(externalrule.Annotations[constants.RuleAnnotationsDescription],
		placeHolderVariables)
	externalrule.Annotations[constants.RuleAnnotationsSummary] = replaceByPlaceHold(externalrule.Annotations[constants.RuleAnnotationsSummary],
		placeHolderVariables)
	externalrule.Labels[constants.RuleAnnotationsMessage] = replaceByPlaceHold(externalrule.Labels[constants.RuleAnnotationsMessage],
		placeHolderVariables)
	externalrule.Labels[constants.RuleAnnotationsDescription] = replaceByPlaceHold(externalrule.Labels[constants.RuleAnnotationsDescription],
		placeHolderVariables)

	externalrule.Alert = replaceByPlaceHold(externalrule.Alert,
		placeHolderVariables)
	return externalrule
}

func (p *ClusterHandler) GetFullPrometheusRule(ctx context.Context, id string) (resp message.OpenAlertRule, err error) {
	rule, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, true).GetPrometheusRule(ctx, id)
	if err != nil {
		return resp, err
	}
	resp.PrometheusRuleSpec = rule.Spec
	return
}

func (p *ClusterHandler) UpdateFullPrometheusRule(ctx context.Context, id string, rules message.OpenAlertRule) (resp message.OpenAlertRule, err error) {
	rule, err := resourcemanager.GetPrometheusRulesReaderWriter(ctx, true).GetPrometheusRule(ctx, id)
	if err != nil {
		return resp, err
	}

	rule.Spec.Groups = rules.Groups
	err = resourcemanager.GetPrometheusRulesReaderWriter(ctx, true).UpdatePrometheusRule(ctx, rule)
	if err != nil {
		return resp, err
	}
	return
}

func (p *ClusterHandler) GetClusterExternal(ctx context.Context, req message.GetClusterExternalReq) (resp message.GetClusterExternalResp, err error) {
	clusters, err := resourcemanager.GetClusterReaderWriter(ctx).QueryClusters(ctx, req.TenantID, req.ProjectID, req.ApplicationID, req.BizType, message.ContinueRequest{})
	if err != nil {
		log.Error("query clusters failed", zap.Error(err))
		return resp, err
	}
	if len(clusters.Items) == 0 {
		log.Info("no clusters found for the given cluster info",
			zap.String("Tenant_ID", req.TenantID),
			zap.String("Project_ID", req.ProjectID),
			zap.String("Application_ID", req.ApplicationID),
			zap.String("Biz_Type", req.BizType),
		)
		return resp, errors.NewErrorf(errors.ClusterNotFound, "no clusters found for the given cluster info")
	}
	if len(clusters.Items) > 1 {
		log.Warn("multiple clusters found for the given cluster info",
			zap.String("Tenant_ID", req.TenantID),
			zap.String("Project_ID", req.ProjectID),
			zap.String("Application_ID", req.ApplicationID),
			zap.String("Biz_Type", req.BizType),
		)
		return resp, errors.NewErrorf(errors.ParameterInvalid, "multiple clusters found for the given cluster info")
	}
	cluster := &clusters.Items[0]
	resp, err = parseClusterExternalInfo(ctx, cluster)
	if err != nil {
		log.Error("parse cluster external info failed", zap.Error(err))
		return resp, errors.NewErrorf(errors.ParameterInvalid, "parse cluster external info failed: %v", err)
	}
	return resp, nil
}

func parseClusterExternalInfo(ctx context.Context, clusterResource *v1.Cluster) (message.GetClusterExternalResp, error) {
	if clusterResource == nil || len(clusterResource.Labels) == 0 ||
		clusterResource.Labels[constants.LabelVendor] == "" ||
		clusterResource.Labels[constants.LabelRegion] == "" {

		return message.GetClusterExternalResp{}, errors.New(errors.ParameterInvalid, "cluster resource is empty")
	}
	externalInfo := message.GetClusterExternalResp{
		ID:            clusterResource.Name,
		Vendor:        constants.ProvidedVendor(clusterResource.Labels[constants.LabelVendor]),
		Region:        clusterResource.Labels[constants.LabelRegion],
		TenantID:      clusterResource.Labels[constants.LabelTenantID],
		ProjectID:     clusterResource.Labels[constants.LabelProjectID],
		ApplicationID: clusterResource.Labels[constants.LabelApplicationID],
		BizType:       clusterResource.Labels[constants.LabelBizType],
		Metrics: &message.ExternalMetricsConfig{
			OpenReadUri: parseOpenReadReadUriExternal(ctx, string(constants.METRICS), clusterResource),
		},
		SlowQuery: &message.ExternalSlowQueryConfig{
			OpenReadUri: parseOpenReadReadUriExternal(ctx, string(constants.SLOWQUERYS), clusterResource),
		},
		KeyVisual: &message.ExternalKeyVisualConfig{
			OpenReadUri: parseOpenReadReadUriExternal(ctx, string(constants.KEYVISUALIZER), clusterResource),
		},
		SqlStatements: &message.ExternalSqlStatementsConfig{
			OpenReadUri: parseOpenReadReadUriExternal(ctx, string(constants.SQLSTATEMENTS), clusterResource),
		},
	}
	return externalInfo, nil
}

func parseOpenReadReadUriExternal(ctx context.Context, category string, clusterResource *v1.Cluster) string {
	bizType := clusterResource.Labels[constants.LabelBizType]
	bizCfg := configs.GetServiceInfo().BizTypeConfigs[bizType]
	if bizCfg.GlobalDataServerHostname != "" {
		if category != string(constants.METRICS) {
			category = string(constants.DIAGNOSIS)
		}
		return fmt.Sprintf(constants.OpenUnifiedReadUriTemplate, bizCfg.GlobalDataServerHostname, clusterResource.Name, category)
	}

	if isNextGenCluster(clusterResource) {
		return fmt.Sprintf(constants.OpenReadUriTemplateV2,
			pickRegion(configs.GetServiceInfo().Region, configs.GetServiceInfo().LogicalRegionName),
			configs.GetServiceInfo().Vendor,
			constants.GetDomainPostfix(configs.GetServiceInfo().Env, configs.GetServiceInfo().DomainPostfix),
			category,
			clusterResource.Labels[constants.LabelBizType],
			clusterResource.Labels[constants.LabelTenantID],
			clusterResource.Labels[constants.LabelApplicationID])
	}

	return fmt.Sprintf(constants.OpenReadUriTemplateWithCategories,
		pickRegion(configs.GetServiceInfo().Region, configs.GetServiceInfo().LogicalRegionName),
		configs.GetServiceInfo().Vendor,
		constants.GetDomainPostfix(configs.GetServiceInfo().Env, configs.GetServiceInfo().DomainPostfix),
		category,
		clusterResource.Labels[constants.LabelBizType],
		clusterResource.Labels[constants.LabelTenantID],
		clusterResource.Labels[constants.LabelProjectID],
		clusterResource.Labels[constants.LabelApplicationID])
}

func pickRegion(region string, logic_region string) string {
	if len(logic_region) != 0 {
		return logic_region
	}
	return region
}