package server

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"

	awssdk "github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/servicequotas"
	"github.com/pingcap/log"
	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/tidbcloud"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/selection"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	internalutil "github.com/tidbcloud/cluster-controller/server/internal/util"
	"github.com/tidbcloud/cluster-controller/server/pkg/cloudprovider"
	config2 "github.com/tidbcloud/cluster-controller/server/pkg/config"
	"github.com/tidbcloud/cluster-controller/server/server/cvt"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	apicommon "github.com/tidbcloud/infra-provider/api/tidbcloud/common"
	cloudaws "github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/cloud/azure"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/gitops"
	gitopsconsts "github.com/tidbcloud/infra-provider/pkg/gitops-sdk/consts"
	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/revision"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup/selfmanaged"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const (
	AuditLogManagedByPingcap = "audit_log_managed_by_pingcap"
)

type ClusterServer struct {
	apiv1.UnimplementedClusterServiceServer

	KubeClient           client.Client
	NodePoolClient       client.Client
	clusterReader        internalutil.ClusterReader
	auditLogViewer       AuditLogViewer
	azureCli             azure.Client
	awsCli               cloudaws.Client
	instanceTypeProvider gitops.InstanceTypeProvider
}

type ClusterServerOption func(*ClusterServer)

func WithLabelSelectorReader() ClusterServerOption {
	return func(s *ClusterServer) {
		s.clusterReader = &internalutil.LabelSelectorReader{}
	}
}

func WithAzureCli(cli azure.Client) ClusterServerOption {
	return func(s *ClusterServer) {
		s.azureCli = cli
	}
}

func WithAWSCli(cli cloudaws.Client) ClusterServerOption {
	return func(s *ClusterServer) {
		s.awsCli = cli
	}
}

func WithNodePoolClient(cli client.Client) ClusterServerOption {
	return func(s *ClusterServer) {
		s.NodePoolClient = cli
	}
}

func NewClusterServer(kubeClient client.Client, auditLogViewer AuditLogViewer, instanceTypeProvider gitops.InstanceTypeProvider, opts ...ClusterServerOption) *ClusterServer {
	cs := &ClusterServer{
		KubeClient:           kubeClient,
		auditLogViewer:       auditLogViewer,
		instanceTypeProvider: instanceTypeProvider,
	}

	for _, opt := range opts {
		opt(cs)
	}
	if cs.NodePoolClient == nil {
		cs.NodePoolClient = kubeClient
	}

	instanceTypeProvider, err := cloudprovider.NewInstanceTypeProvider(context.Background(), kubeClient)
	if err != nil {
		panic(fmt.Sprintf("failed to create instance type provider: %v", err))
	}
	cs.instanceTypeProvider = instanceTypeProvider
	return cs
}

func (s *ClusterServer) CreateCluster(ctx context.Context, req *apiv1.CreateClusterRequest) (*apiv1.Cluster, error) {
	// Convert req to cloudv1.Cluster
	clusterReq := req.Cluster
	tidbGroupId := req.Cluster.DefaultTidbGroupId

	infraProviderType := getInfraProviderFromReq(clusterReq)
	if infraProviderType == "" {
		return nil, status.Errorf(codes.InvalidArgument, "invalid infra provider type")
	}

	cfg := config2.GetConfig()
	nameSpace := genNamespace(cfg.NamespacePrefix, clusterReq.ProjectId)

	if err := ensureNamespace(ctx, s.KubeClient, nameSpace, clusterReq.ProjectId, clusterReq.TenantId); err != nil {
		return nil, err
	}

	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      clusterReq.Id,
			Namespace: nameSpace,
			Annotations: map[string]string{
				consts.DefaultLoadBalancerIDAnnotationKey: tidbGroupId,
				gitopsconsts.OverrideHelperImageKey:       cfg.OverrideHelperImage,
				gitopsconsts.OverrideRegistryKey:          cfg.OverrideRegistry,
				gitopsconsts.OverrideRepoPrefixKey:        cfg.OverrideRepoPrefix,
			},
			Labels: buildClusterLabelsForCreating(infraProviderType, cfg.Region, clusterReq.TenantId, clusterReq.ProjectId, clusterReq.Id),
		},
		Spec: cloudv1.ClusterSpec{
			Kubernetes: func() cloudv1.Kubernetes {
				k8s := cloudv1.Kubernetes{
					Name:     clusterReq.Kubernetes.Name,
					Endpoint: clusterReq.Kubernetes.Endpoint,
					CAData:   []byte(clusterReq.Kubernetes.Ca),
					Account:  clusterReq.Kubernetes.AccountId,
					Network:  clusterReq.Kubernetes.Vpc,
					Provider: infraProviderType,
					Version:  clusterReq.Kubernetes.Version,
					Cloud: cloudv1.Cloud{
						AWS: func() *cloudv1.CloudAWS {
							awsReq := clusterReq.Kubernetes.GetAws()
							if awsReq == nil {
								return nil
							}

							subnetGroups := cloudv1.AWSSubnetGroups{
								{
									Name:    cloudv1.AWSSubnetGroupNodes,
									Subnets: getAwsSubnetFromReq(awsReq.GetNodeSubnets()),
								},
								{
									Name:    cloudv1.AWSSubnetGroupPublicNLB,
									Subnets: getAwsSubnetFromReq(awsReq.GetPublicLoadbalancerSubnets()),
								},
								{
									Name:    cloudv1.AWSSubnetGroupPrivateNLB,
									Subnets: getAwsSubnetFromReq(awsReq.GetPrivateLoadbalancerSubnets()),
								},
							}

							if len(awsReq.GetSecondaryNodeSubnets()) > 0 {
								subnetGroups = subnetGroups.AppendGroup(cloudv1.AWSSubnetGroup{
									Name:    cloudv1.AWSSubnetGroupSecondaryCIDR,
									Subnets: getAwsSubnetFromReq(awsReq.GetSecondaryNodeSubnets()),
								})
							}

							return &cloudv1.CloudAWS{
								SubnetGroups:     subnetGroups,
								EKSSecurityGroup: awsReq.SecurityGroup,
								AdminNodeRole:    awsReq.AdminNodeRole,
							}

						}(),
						GCP: func() *cloudv1.CloudGCP {
							gcpReq := clusterReq.Kubernetes.GetGcp()
							if gcpReq == nil {
								return nil
							}
							return &cloudv1.CloudGCP{
								NodeZones:   gcpReq.NodeZones,
								AdminNodeSa: gcpReq.AdminNodeSa,
							}
						}(),
						Azure: func() *cloudv1.CloudAzure {
							azureReq := clusterReq.Kubernetes.GetAzure()
							if azureReq == nil {
								return nil
							}
							// get node subnets
							nodeSubnets := []cloudv1.AzureNodeSubnet{}
							for _, nodeSubnet := range azureReq.NodeSubnets {
								nodeSubnets = append(nodeSubnets, cloudv1.AzureNodeSubnet{
									Name: nodeSubnet.Id,
								})
							}
							// get pod subnets
							podSubnets := []cloudv1.AzurePodSubnet{}
							for _, podSubnet := range azureReq.PodSubnets {
								podSubnets = append(podSubnets, cloudv1.AzurePodSubnet{
									Name: podSubnet.Id,
								})
							}
							return &cloudv1.CloudAzure{
								NodeZones:         azureReq.NodeZones,
								NodeSubnets:       nodeSubnets,
								PodSubnets:        podSubnets,
								ResourceGroup:     azureReq.ResourceGroup,
								NodeResourceGroup: azureReq.NodeResourceGroup,
							}
						}(),
					},
				}
				return k8s
			}(),
			SqlEndpoints: cloudv1.SqlEndpoints{
				DNS: cloudv1.DNS{Identity: clusterReq.GetClusterIdentifierHash()},
				// privatelink, loadbalancer will managed by other api, won't be enabled by default
				Public:  cloudv1.Service{Enabled: ptr.To(false)},
				Private: &cloudv1.Service{Enabled: ptr.To(false)},
			},
		},
	}
	initClusterFeatures(cluster)

	// set shared bucket if need
	if clusterReq.SharedBucket != nil {
		cluster.Spec.SharedBucket = &cloudv1.SharedBucket{
			BucketURL: clusterReq.SharedBucket.BucketUrl,
		}
	}

	// create kubeconfig secret when clusterReq.Kubernetes.Kubeconfig is set
	if clusterReq.Kubernetes.Kubeconfig != "" {
		if err := s.createKubeConfigSecret(ctx, cluster, clusterReq); err != nil {
			return nil, err
		}

	}

	// create ConfigRevision and set it into Cluster
	if err := s.createVersionAndConfigurations(ctx, cluster, clusterReq); err != nil {
		return nil, err
	}

	// setup security context, such as CMEK, JWKs
	if _, err := s.upsertSecurityContext(ctx, cluster, clusterReq.SecurityConfigurations); err != nil {
		return nil, err
	}

	if err := s.createInitSQLConfigMap(ctx, cluster, clusterReq.InitSqlStatements); err != nil {
		return nil, err
	}

	// generate clusterTopology
	// new cluster(after 2024/01/26) will not use local storage
	clusterTopology := cloudv1.ClusterTopology{}
	for _, component := range clusterReq.ComponentNodeSpecs {
		switch component.ComponentType {
		case apicommon.ComponentType_TIPROXY:
			clusterTopology.TiProxy.Replicas = int(component.Count)
			clusterTopology.TiProxy.InstanceType = component.InstanceType
			// Automatically enable `UseTiProxyAsLBTarget` if TiProxy is set when cluster creation.
			if component.Count > 0 {
				cluster.Spec.SqlEndpoints.UseTiProxyAsLBTarget = true
				cluster.Annotations[consts.AnnoKeyLBTargetOnTiProxy(tidbGroupId)] = consts.AnnoValTrue
			}
		case apicommon.ComponentType_TIDB:
			clusterTopology.TiDB.Replicas = int(component.Count)
			clusterTopology.TiDB.InstanceType = component.InstanceType

			cluster.Labels[cloudv1.ClusterLabelWantTokenBasedAuth] = "true"

			clusterTopology.TiDB.Security = &cloudv1.TiDBSecurity{
				SSLCASecret: nil,
				AuthTokenJwksSecret: &corev1.SecretReference{
					Name:      authPubKeySecretName(clusterReq.Id),
					Namespace: cluster.Namespace,
				},
			}

			clusterTopology.TiDB.Audit = &cloudv1.Audit{
				Enable: true,
			}

		case apicommon.ComponentType_TIKV:
			clusterTopology.TiKV.Replicas = int(component.Count)
			clusterTopology.TiKV.InstanceType = component.InstanceType
			clusterTopology.TiKV.StorageClassName = getStorageClassNameFromComponentReq(component)
			if component.Storage == nil {
				return nil, bizerr.New(bizerr.ErrCodeUnknown, "tikv storage is unknown")
			}
			clusterTopology.TiKV.StorageSize = storageSizeGi2quantity(component.Storage.SizeGi)
			clusterTopology.TiKV.StorageVolumes = getAdditionalStorageVolumesFromComponentReq(component)
		case apicommon.ComponentType_TIFLASH:
			clusterTopology.TiFlash.Replicas = int(component.Count)
			clusterTopology.TiFlash.InstanceType = component.InstanceType
			clusterTopology.TiFlash.StorageClassName = getStorageClassNameFromComponentReq(component)
			clusterTopology.TiFlash.StorageSize = storageSizeGi2quantity(component.Storage.SizeGi)
		case apicommon.ComponentType_PD:
			clusterTopology.PD.Replicas = int(component.Count)
			clusterTopology.PD.InstanceType = component.InstanceType
			clusterTopology.PD.StorageClassName = getStorageClassNameFromComponentReq(component)
			if component.Storage == nil {
				return nil, bizerr.New(bizerr.ErrCodeUnknown, "pd storage is unknown")
			}
			clusterTopology.PD.StorageSize = storageSizeGi2quantity(component.Storage.SizeGi)
		default:
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "unknown componentType %s", component.ComponentType)
		}
	}

	cluster.Spec.Cluster = clusterTopology

	if err := s.KubeClient.Create(ctx, cluster); err != nil {
		log.Error("Failed to create cluster", zap.Error(err))
		if apierrors.IsAlreadyExists(err) {
			return nil, status.Errorf(codes.AlreadyExists, "Cluster %s already exists", clusterReq.Id)
		}
		// setResponseStatusCodes(ctx, http.StatusInternalServerError)
		return nil, err
	}

	return cvt.Cluster2PB(cluster), nil
}

func (s *ClusterServer) createKubeConfigSecret(ctx context.Context, cluster *cloudv1.Cluster, clusterReq *apiv1.Cluster) error {
	// delete the cluster kubeconfig by cluster-controller itself, because the cluster server doesn't know when cluster-controller will stop to use the cluster kubeconfig
	decodedKubeconfig, err := base64.StdEncoding.DecodeString(clusterReq.Kubernetes.Kubeconfig)
	if err != nil {
		return fmt.Errorf("failed to decode kubeconfig: %v", err)
	}
	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-kubeconfig", cluster.Name),
			Namespace: cluster.Namespace,
		},
		Data: map[string][]byte{
			"kubeconfig": decodedKubeconfig,
		},
	}
	if _, err = controllerutil.CreateOrUpdate(ctx, s.KubeClient, secret, func() error {
		// if the secret already exists, we need to update it
		secret.Data = map[string][]byte{
			"kubeconfig": decodedKubeconfig,
		}
		return nil
	}); err != nil {
		return fmt.Errorf("create kubeconfig secret: %w", err)
	}

	cluster.Spec.Kubernetes.KubeConfigSecretRef = &corev1.SecretReference{
		Name:      secret.Name,
		Namespace: secret.Namespace,
	}
	return nil
}

func (s *ClusterServer) createVersionAndConfigurations(ctx context.Context, cluster *cloudv1.Cluster, clusterReq *apiv1.Cluster) error {
	cr := &cloudv1.ConfigRevision{}
	if err := overlayConfigAndVersion(ctx, cr, clusterReq.ComponentConfigurations, clusterReq.ClusterVersion); err != nil {
		return fmt.Errorf("generate ConfigRevision: %w", err)
	}
	revision.RectifyConfigRevisionForCreate(cluster, cr, "")
	tilog.WithContextL(ctx).Info("create ConfigRevision", zap.String("cluster", cluster.Name), zap.String("name", cr.Name))
	if _, err := controllerutil.CreateOrUpdate(ctx, s.KubeClient, cr, func() error {
		return nil
	}); err != nil {
		return fmt.Errorf("create ConfigRevision: %w", err)
	}
	cluster.UpdateConfigRevision(cr, true)
	return nil
}

// When tidb-mgmt doesn't provide the init sql, temporarily fall back to use cluster-controller managed bootstrap sql
// TODO: Once config-center and tidb-mgmt fully support init-sql, we should return an error if no init-sql-statements is provided.
// cluster-controller will copy the initsql from this configmap to dataplane if it exist
func (s *ClusterServer) createInitSQLConfigMap(ctx context.Context, cluster *cloudv1.Cluster, sql string) error {
	if sql == "" {
		tilog.Warn(ctx, "no initSqlStatements provided, fallback to use cluster-controller hardcode value", zap.String("cluster", cluster.GetClusterID()))
		return nil
	}
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: cluster.Namespace,
			Name:      namer.BootstrapSQLConfigMapNameInMeta(cluster.GetClusterID()),
			Labels: map[string]string{
				cloudv1.LabelServiceType: cloudv1.ServiceTypeDedicated,
				cloudv1.TenantTagKey:     cluster.GetTenantID(),
				cloudv1.ProjectTagKey:    cluster.GetProjectID(),
				cloudv1.ClusterTagKey:    cluster.GetClusterID(),
			},
		},
		Data: map[string]string{
			consts.BootstrapSQLCMKey: sql,
		},
	}

	if _, err := controllerutil.CreateOrUpdate(ctx, s.KubeClient, cm, func() error {
		return nil
	}); err != nil {
		return err
	}
	return nil
}

func (s *ClusterServer) upsertSecurityContext(ctx context.Context, cluster *cloudv1.Cluster, conf *apiv1.Cluster_SecurityConfigurations) (clusterChanged bool, _ error) {
	if conf == nil {
		return false, nil
	}

	jwksSecret := buildJWKsSecret(cluster, conf.TidbAuthTokenJwks)
	// todo(chenfei): delete jwks secret when deleting cluster
	if _, err := controllerutil.CreateOrUpdate(ctx, s.KubeClient, jwksSecret, func() error {
		return nil
	}); err != nil {
		return false, err
	}

	// set CMEK info into cluster
	switch cluster.GetProviderType() {
	case cloudv1.ProviderTypeEKS:
		keyArn := conf.GetCmekAwsMasterKeyArn()
		if keyArn == "" {
			break // ignore if no key specified
		}

		eksClusterSpec := &cloudv1.EKSClusterSpec{
			TypeMeta: metav1.TypeMeta{
				Kind:       "EKSCluster",
				APIVersion: "eks.infra.pingcap.com/v1alpha1",
			},
			Security: &cloudv1.Security{
				MasterKeyARN: ptr.To(conf.GetCmekAwsMasterKeyArn()),
			},
		}
		if err := cloudv1.SetEKSClusterSpec(cluster, eksClusterSpec); err != nil {
			return false, bizerr.WrapError(err, "failed to set eks cluster spec")
		}
		cluster.Annotations[consts.CMEKEnabledKey] = "true"
		cluster.EnableFeature(consts.ClusterFeatureUseCMEK)
		clusterChanged = true
	case cloudv1.ProviderTypeGKE:
		keyName := conf.GetCmekGcpMasterKeyName()
		if keyName == "" {
			break // ignore if no key specified
		}
		return false, status.Errorf(codes.InvalidArgument, "GKE does not support CMEK")
	case cloudv1.ProviderTypeAKS:
		changed, err := s.buildCMEKForAzure(ctx, cluster, conf)
		if err != nil {
			return false, fmt.Errorf("failed to build CMEK for Azure: %w", err)
		}
		clusterChanged = changed
	}

	// set SQL client CA
	if conf.SqlClientCa != "" {
		return false, status.Errorf(codes.InvalidArgument, "SQL client CA is not supported for now")
	}
	return clusterChanged, nil
}

func (s ClusterServer) buildCMEKForAzure(ctx context.Context, cluster *cloudv1.Cluster, conf *apiv1.Cluster_SecurityConfigurations) (bool, error) {
	logger := tilog.WithContextS(ctx)
	cmekReq := conf.GetAzureCmek()
	if cmekReq == nil {
		return false, nil
	}

	// Automatically append a key version if not specified.
	// A valid keyversion is required for creating disk encryption set, even if
	// it's not used actually. If the auto rotation is enabled, the latest key version will be used
	// even if one keyversion is specified.
	latestKeyVersion, err := AzureQueryLatestKeyvaultKeyVersion(ctx, s.azureCli, cmekReq)
	if err != nil {
		return false, status.Errorf(codes.InvalidArgument, "failed to query latest key version: %s", err)
	}
	logger.Infof("latest key version '%s' is auto appended", latestKeyVersion)

	azureCmek := &cloudv1.AzureCMEKSpec{
		ApplicationID:             cmekReq.ApplicationId,
		ManagedIdentityResourceID: cmekReq.ManagedIdentityResourceId,
		KeyVaultName:              cmekReq.KeyVaultName,
		KeyName:                   cmekReq.KeyName,
		KeyVersion:                latestKeyVersion,
	}
	if err := azureCmek.Validate(); err != nil {
		return false, status.Errorf(codes.InvalidArgument, "azure CMEK is not configured properly: %s", err)
	}

	aksClusterSpec := &cloudv1.AKSClusterSpec{
		TypeMeta: metav1.TypeMeta{
			Kind:       "AKSCluster",
			APIVersion: "aks.infra.pingcap.com/v1alpha1",
		},
		AzureCMEK: azureCmek,
	}

	if err := cloudv1.SetAKSClusterSpec(cluster, aksClusterSpec); err != nil {
		return false, bizerr.WrapError(err, "failed to set eks cluster spec")
	}
	cluster.EnableFeature(consts.ClusterFeatureUseCMEK)
	return true, nil
}

func (s *ClusterServer) GetCluster(ctx context.Context, req *apiv1.GetClusterRequest) (*apiv1.Cluster, error) {
	cluster, err := s.getClusterCR(ctx, req.ClusterId)
	if err != nil {
		return &apiv1.Cluster{}, err
	}

	return cvt.Cluster2PB(cluster), nil
}

func (s *ClusterServer) ListClusters(ctx context.Context, req *apiv1.ListClustersRequest) (*apiv1.ListClustersResponse, error) {
	selector := labels.NewSelector()
	if req.TenantId != "" {
		tenantReq, err := labels.NewRequirement(cloudv1.TenantTagKey, selection.Equals, []string{req.TenantId})
		if err != nil {
			return nil, status.Errorf(codes.InvalidArgument, "invalid tenant id(%s): %s", req.TenantId, err)
		}
		selector = selector.Add(*tenantReq)
	}
	if len(req.ProjectIds) != 0 {
		projectReq, err := labels.NewRequirement(cloudv1.ProjectTagKey, selection.In, req.ProjectIds)
		if err != nil {
			return nil, status.Errorf(codes.InvalidArgument, "invalid project ids(%s): %s", req.ProjectIds, err)
		}
		selector = selector.Add(*projectReq)
	}

	var clusterList cloudv1.ClusterList
	err := s.KubeClient.List(ctx, &clusterList, client.MatchingLabelsSelector{Selector: selector})
	if err != nil {
		tilog.Error(ctx, "Failed to get clusters", zap.Error(err), zap.Any("req", req))
		return nil, status.Errorf(codes.Internal, "failed to get clusters: %s", err)
	}

	out := &apiv1.ListClustersResponse{
		Clusters: lo.Map(clusterList.Items, func(cluster cloudv1.Cluster, _ int) *apiv1.Cluster {
			return cvt.Cluster2PB(&cluster)
		}),
	}
	return out, nil
}

func (s *ClusterServer) SetAuditLogConfiguration(ctx context.Context, req *apiv1.SetAuditLogConfigurationRequest) (*apiv1.AuditLogConfiguration, error) {
	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return &apiv1.AuditLogConfiguration{}, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if err := s.validateSetAuditLogConfiguration(req, cluster); err != nil {
		return nil, err
	}

	var audit *cloudv1.Audit
	if isAuditLogBucketManagedByCustomer(req) {
		delete(cluster.Annotations, AuditLogManagedByPingcap)
		audit = &cloudv1.Audit{
			Enable:    true,
			Region:    req.AuditLogConfiguration.GetBucketRegion(),
			BucketURL: req.AuditLogConfiguration.GetBucketUrl(),
		}

		if req.AuditLogConfiguration.GetS3() != nil {
			audit.RoleArn = req.AuditLogConfiguration.GetS3().GetRoleArn()

			if !util.IsAWSIAMARN(audit.RoleArn) {
				return nil, status.Errorf(codes.InvalidArgument, "the given role arn is not a valid AWS IAM ARN")
			}
			// The auditlog sidecar assume role ExternalID is set by cluster-controller automatically
			// So the global server does not need to pass it to cluster-server
		} else if req.AuditLogConfiguration.GetGcs() != nil { // nolint:staticcheck
			// GCP uses the node sa to access customer gcs directly
			// So not GCP specific info needs to set on Spec.Cluster.TiDB.Audit
			// See auditlog sidecar config in pkg audit.Config
		} else if req.AuditLogConfiguration.GetAzBlob() != nil {
			token := req.AuditLogConfiguration.GetAzBlob().GetSasToken()
			if token == "" {
				return nil, status.Errorf(codes.InvalidArgument, "no SAS token provided for Azure Blob Storage")
			}
			audit.Azure = &cloudv1.AzureAuditConfig{
				SASToken: token,
			}
		}
	} else { // using the bucket managed by PingCAP, in other words, using the shared bucket
		cluster.Annotations[AuditLogManagedByPingcap] = consts.AnnoValTrue
		audit = &cloudv1.Audit{
			Enable:    true,
			Region:    cluster.GetRegion(),
			BucketURL: allocateDBAuditLogBucketURLInSharedBucket(cluster),
		}
		switch pt := cluster.GetProviderType(); pt {
		case cloudv1.ProviderTypeEKS:
			audit.RoleArn = cluster.Status.SharedBucket.AWS.DBAuditLogWriterRoleARN
		}
	}

	cluster.Spec.Cluster.TiDB.Audit = audit
	if err := s.KubeClient.Update(ctx, cluster); err != nil {
		tilog.Error(ctx, "Failed to set audit log configuration", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to set audit log configuration: %s", err)
	}

	return req.AuditLogConfiguration, nil
}

func (s *ClusterServer) validateSetAuditLogConfiguration(req *apiv1.SetAuditLogConfigurationRequest, cluster *cloudv1.Cluster) error {
	if isAuditLogBucketManagedByCustomer(req) {
		if req.AuditLogConfiguration == nil {
			return status.Errorf(codes.InvalidArgument, "the AuditLogConfiguration is required when managed_by is Customer")
		}
	} else {
		if !cluster.SharedBucketEnabled() {
			return status.Errorf(codes.InvalidArgument, "shared bucket is not enabled on this cluster")
		}
		switch pt := cluster.GetProviderType(); pt {
		case cloudv1.ProviderTypeEKS:
			if cluster.Status.SharedBucket == nil ||
				cluster.Status.SharedBucket.AWS == nil ||
				cluster.Status.SharedBucket.AWS.DBAuditLogWriterRoleARN == "" {
				return status.Errorf(codes.InvalidArgument, "shared bucket is not ready, the DB audit log writer role ARN is empty")
			}
		}
	}
	return nil
}

func isAuditLogBucketManagedByCustomer(req *apiv1.SetAuditLogConfigurationRequest) bool {
	// For compatibility, we treat `UNSPECIFIED` as `CUSTOMER`
	return req.ManagedBy == apiv1.SetAuditLogConfigurationRequest_BUCKET_MANAGED_BY_CUSTOMER ||
		req.ManagedBy == apiv1.SetAuditLogConfigurationRequest_BUCKET_MANAGED_BY_UNSPECIFIED
}

func (s *ClusterServer) GetAuditLogConfiguration(ctx context.Context, req *apiv1.GetAuditLogConfigurationRequest) (*apiv1.AuditLogConfiguration, error) {
	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return &apiv1.AuditLogConfiguration{}, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if cluster.Spec.Cluster.TiDB.Audit == nil {
		return &apiv1.AuditLogConfiguration{}, nil
	}

	cfg := &apiv1.AuditLogConfiguration{
		BucketRegion:       cluster.Spec.Cluster.TiDB.Audit.Region,
		BucketUrl:          cluster.Spec.Cluster.TiDB.Audit.BucketURL,
		IsManagedByPingcap: cluster.Annotations[AuditLogManagedByPingcap] == consts.AnnoValTrue,
	}
	switch pt := cluster.GetProviderType(); pt {
	case consts.ProviderTypeEKS:
		role := cluster.Spec.Cluster.TiDB.Audit.RoleArn
		cfg.Storage = &apiv1.AuditLogConfiguration_S3_{
			S3: &apiv1.AuditLogConfiguration_S3{
				RoleArn: role,
			},
		}
	case consts.ProviderTypeGKE:
		cfg.Storage = &apiv1.AuditLogConfiguration_Gcs{
			Gcs: &apiv1.AuditLogConfiguration_GCS{},
		}
	case consts.ProviderTypeAKS:
		cfg.Storage = &apiv1.AuditLogConfiguration_AzBlob{
			AzBlob: &apiv1.AuditLogConfiguration_AZBlob{},
		}
	default:
		tilog.Error(ctx, "unsupported provider type", zap.String("provider", pt))
		return nil, status.Errorf(codes.Internal, "unsupported provider type: %s", pt)
	}
	return cfg, nil
}

func (s *ClusterServer) ValidateAuditLogConfiguration(ctx context.Context, req *apiv1.ValidateAuditLogConfigurationRequest) (*apiv1.ValidateAuditLogConfigurationResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))

	clusterId, err := strconv.ParseUint(req.ClusterId, 10, 64)
	if err != nil { // clusterproxy library requires uint64 clusterId
		logger.Error("cluster id is not uint64", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "invalid cluster id: %s", req.ClusterId)
	}

	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if cluster.Status.Proxy == nil {
		logger.Error("cluster.status.proxy is empty, can't get proxy URL")
		return nil, status.Errorf(codes.Internal, "Failed to get proxy URL for cluster %s, please try again later", req.ClusterId)
	}
	if cluster.Status.DB == nil {
		logger.Error("cluster.status.db is empty, can't get cluster version")
		return nil, status.Errorf(codes.Internal, "Failed to get cluster version from status %s, please try again later", req.ClusterId)
	}
	conf := config2.GetConfig()
	input := tidbcloud.ClusterProxyInput{
		ClusterID: clusterId,
		Version:   cluster.Status.DB.Version,
		BaseURL:   cluster.Status.Proxy.Endpoint,
		Cfg: tidbcloud.ProxyConfig{
			Auth0Domain:       conf.Auth0Domain,
			Auth0Audience:     conf.Auth0Audience,
			Auth0ClientID:     conf.Auth0ClientID,
			Auth0ClientSecret: conf.Auth0ClientSecret,
		},
		Timeout: httpTimeout,
	}
	proxy, err := tidbcloud.NewClusterProxyV2(ctx, input)
	if err != nil {
		logger.Error("failed to build clusterproxy client", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to build clusterproxy client %s: %v", req.ClusterId, err)
	}

	var args clusterproxy.AuditCheckModel
	args.Audit.Region = req.AuditLogConfiguration.BucketRegion
	args.Audit.BucketURL = req.AuditLogConfiguration.BucketUrl
	if p := req.AuditLogConfiguration.GetGcs(); p != nil {
		sa, err := s.getTidbWorkerNodeServiceAccount(ctx, cluster)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "failed to get worker node service account %s: %v", req.ClusterId, err)
		}
		args.ServiceAccount = sa
	} else if p := req.AuditLogConfiguration.GetS3(); p != nil {
		tidbRoleArn, externalID, err := s.getTidbWorkerNodeRole(ctx, cluster)
		if err != nil {
			return nil, status.Errorf(codes.Internal, "failed to get tidb worker node role arn and external id")
		}
		if cluster.Spec.Kubernetes.Cloud.AWS == nil {
			return nil, status.Errorf(codes.Internal, "can't find aws info from cluster CR")
		}
		adminRoleArn := cluster.Spec.Kubernetes.Cloud.AWS.AdminNodeRole
		if (adminRoleArn == "" && cluster.Spec.NetworkName != "") || adminRoleArn == tidbRoleArn {
			// Both admin node and tidb node are migrated from cloud-native arch
			// We don't need to pass tidb role for TestAuditConnection,
			// otherwise clusterproxy will assume the role itself which would always fail
		} else {
			args.Audit.TidbRoleARN = tidbRoleArn
		}

		args.Audit.RoleARN = p.GetRoleArn()
		args.ExternalID = externalID
	} else if p := req.AuditLogConfiguration.GetAzBlob(); p != nil {
		if p.SasToken != "" {
			args.Audit.SASToken = p.SasToken
		} else { // no sas token passed, use the persisted sas token for validation
			if cluster.Spec.Cluster.TiDB.Audit != nil &&
				cluster.Spec.Cluster.TiDB.Audit.Azure != nil {
				args.Audit.SASToken = cluster.Spec.Cluster.TiDB.Audit.Azure.SASToken
			}
		}
		if args.Audit.SASToken == "" {
			return nil, status.Errorf(codes.InvalidArgument, "no SAS token provided for Azure blob storage")
		}
	} else {
		logger.Error("unsupported storage for auditlog")
		return nil, status.Errorf(codes.InvalidArgument, "unsupported storage for auditlog")
	}
	auditResp, err := proxy.TestAuditConnection(ctx, args)
	if err != nil {
		logger.Error("failed to check connection for cluster %s: %v", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to check connection for cluster %s: %v", req.ClusterId, err)
	}
	if !auditResp.Pass {
		logger.Warn("check failed", zap.String("reason", auditResp.Message))
	}
	return &apiv1.ValidateAuditLogConfigurationResponse{
		IsValid: auditResp.Pass,
	}, nil
}

func (s *ClusterServer) PauseCluster(ctx context.Context, req *apiv1.PauseClusterRequest) (*apiv1.Cluster, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))

	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if cluster.Spec.Cluster.SuspendAction == nil {
		cluster.Spec.Cluster.SuspendAction = &cloudv1.SuspendAction{}
	}
	if !cluster.Spec.Cluster.SuspendAction.SuspendCompute {
		logger.Info("Pause the cluster")
		cluster.Spec.Cluster.SuspendAction.SuspendCompute = true
		if err := s.KubeClient.Update(ctx, cluster); err != nil {
			logger.Error("Failed to pause the cluster", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "Failed to pause the cluster %s: %s", req.ClusterId, err)
		}
	}

	pauseCondition := cluster.GetSuspendCondition()
	if pauseCondition == nil || pauseCondition.Status != cloudv1.ConditionTrue {
		logger.Info("cluster is pausing", zap.Any("condition", pauseCondition))
	} else {
		logger.Info("cluster is paused")
	}

	return cvt.Cluster2PB(cluster), nil
}

func (s *ClusterServer) ResumeCluster(ctx context.Context, req *apiv1.ResumeClusterRequest) (*apiv1.Cluster, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))

	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if cluster.Spec.Cluster.SuspendAction != nil && cluster.Spec.Cluster.SuspendAction.SuspendCompute {
		logger.Info("resume the cluster")
		cluster.Spec.Cluster.SuspendAction.SuspendCompute = false
		if err := s.KubeClient.Update(ctx, cluster); err != nil {
			logger.Error("Failed to resume the cluster", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "Failed to resume the cluster %s: %s", req.ClusterId, err)
		}
	}

	if !cluster.Suspending() && cluster.Status.Phase == cloudv1.ClusterPhaseRunning {
		logger.Info("cluster is resumed")
	} else {
		logger.Info("cluster is resuming")
	}

	return cvt.Cluster2PB(cluster), nil
}

func (s *ClusterServer) DeleteCluster(ctx context.Context, req *apiv1.DeleteClusterRequest) (*apiv1.Cluster, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))

	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if err := s.KubeClient.Delete(ctx, cluster); err != nil {
		logger.Error("Failed to delete cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to delete cluster %s: %s", req.ClusterId, err)
	}

	return cvt.Cluster2PB(cluster), nil
}

func (s *ClusterServer) GetComponentNodeInfos(ctx context.Context, req *apiv1.GetComponentNodeInfosRequest) (*apiv1.GetComponentNodeInfosResponse, error) {
	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err), zap.String("cluster", req.ClusterId))
		if apierrors.IsNotFound(err) {
			tilog.Info(ctx, "Cluster not found", zap.String("cluster", req.ClusterId))
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	return cvt.Cluster2ComponentNodeInfosPB(cluster), nil
}

func (s *ClusterServer) GetTiDBEKSUpgradeSelfManagedNodeGroups(ctx context.Context, req *apiv1.GetTiDBEKSUpgradeSelfManagedNodeGroupsRequest) (*apiv1.GetTiDBEKSUpgradeSelfManagedNodeGroupsResponse, error) {
	if req.GetClusterId() == "" || req.GetTargetK8SVersion() == "" {
		return nil, status.Errorf(codes.InvalidArgument, "cluster_id and target_k8s_version are required")
	}

	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err), zap.String("cluster", req.ClusterId))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if cluster.GetProviderType() != cloudv1.ProviderTypeEKS {
		return nil, status.Errorf(codes.FailedPrecondition, "cluster %s is not EKS (provider_type=%q)", req.ClusterId, cluster.GetProviderType())
	}

	plan, err := selfmanaged.ComputeTiDBEKSK8sUpgradeSelfManagedNGPlan(ctx, s.KubeClient, cluster, req.GetTargetK8SVersion())
	if err != nil {
		tilog.Error(ctx, "Failed to compute TiDB EKS upgrade self-managed NG plan", zap.Error(err), zap.String("cluster", req.ClusterId))
		return nil, status.Errorf(codes.Internal, "%v", err)
	}

	out := &apiv1.GetTiDBEKSUpgradeSelfManagedNodeGroupsResponse{}
	for _, b := range plan.OldSelfManaged {
		out.OldSelfManagedNodeGroups = append(out.OldSelfManagedNodeGroups, tidbEKSUpgradeNGBriefToPB(b))
	}
	for _, b := range plan.NewSelfManagedAtTargetK8sVersion {
		out.NewSelfManagedNodeGroupsAtTargetVersion = append(out.NewSelfManagedNodeGroupsAtTargetVersion, tidbEKSUpgradeNGBriefToPB(b))
	}
	for _, p := range plan.NewSelfManagedToCreate {
		out.NewSelfManagedNodeGroupsToCreate = append(out.NewSelfManagedNodeGroupsToCreate, &apiv1.TiDBEKSUpgradePlannedSelfManagedNodeGroup{
			OldNodeGroupName:   p.OldName,
			NewNodeGroupName:   p.NewName,
			ExtraTidbGroupName: p.ExtraGroupName,
		})
	}
	return out, nil
}

func tidbEKSUpgradeNGBriefToPB(b selfmanaged.TiDBEKSSelfManagedNGBrief) *apiv1.TiDBEKSUpgradeSelfManagedNodeGroup {
	msg := &apiv1.TiDBEKSUpgradeSelfManagedNodeGroup{
		Name:               b.Name,
		K8SVersion:         b.K8sVersion,
		ExtraTidbGroupName: b.ExtraGroupName,
		Component:          b.Component,
	}
	if b.DesiredCapacity != nil {
		v := *b.DesiredCapacity
		msg.DesiredCapacity = &v
	}
	return msg
}

func (s *ClusterServer) GetClusterRole(ctx context.Context, req *apiv1.GetClusterRoleRequest) (*apiv1.ClusterRole, error) {
	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err), zap.String("cluster", req.ClusterId))
		if apierrors.IsNotFound(err) {
			tilog.Info(ctx, "Cluster not found", zap.String("cluster", req.ClusterId))
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	switch cluster.GetProviderType() {
	case cloudv1.ProviderTypeEKS: // ref: `Context.GetAWSNodeRoleARN`
		roleArn, externalID, err := s.getTidbWorkerNodeRole(ctx, cluster)
		if err != nil {
			return nil, status.Error(codes.Internal, err.Error())
		}
		return &apiv1.ClusterRole{CloudIamIdentity: roleArn, ExternalId: externalID}, nil
	case cloudv1.ProviderTypeGKE: // ref: `Context.GetGCPNodePoolServiceAccountEmail`
		saEmail, err := s.getTidbWorkerNodeServiceAccount(ctx, cluster)
		if err != nil {
			return nil, status.Error(codes.Internal, err.Error())
		}
		return &apiv1.ClusterRole{CloudIamIdentity: saEmail}, nil
	case cloudv1.ProviderTypeAKS:
		// AKS nodepool identity is k8s level, it's stored in KubernetesCluster CR
		// cluster-server is designed to not access KubernetesCluster CR directly
		// And this identity is not used by end users at the moment
		// So we just return an empty string here
		return &apiv1.ClusterRole{CloudIamIdentity: ""}, nil
	}

	return nil, status.Errorf(codes.InvalidArgument, "unsupported provider type %s", cluster.GetProviderType())
}

func (s *ClusterServer) CheckClusterPublicLBInboundRulesStatus(ctx context.Context, req *apiv1.CheckClusterPublicLBInboundRulesStatusRequest) (*apiv1.CheckClusterPublicLBInboundRulesStatusResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId))

	if req.ClusterId == "" {
		return nil, status.Errorf(codes.InvalidArgument, "ClusterId is required")
	}

	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if cluster.GetProviderType() != cloudv1.ProviderTypeEKS {
		return nil, status.Errorf(codes.InvalidArgument, "only support aws for now")
	}

	region := cluster.GetRegion()

	quota, err := s.checkInboundRulesQuota(ctx, region)
	if err != nil {
		logger.Error("Failed to get inbound rules quota, using default value", zap.Error(err), zap.Int32("quota", quota))
	}

	logger.Info("Retrieved inbound rules quota limit", zap.Int32("quota", quota))

	return &apiv1.CheckClusterPublicLBInboundRulesStatusResponse{
		Full:  int32(len(req.InboundRules)) > quota,
		Quota: int64(quota),
	}, nil
}

// TODO(tennix): once all the Cluster CRs has persisted node identity in status
// We should use the Cluster.Status.NodeIdentity as the single source of truth
func (s *ClusterServer) getTidbWorkerNodeRole(ctx context.Context, cluster *cloudv1.Cluster) (arn string, externalID string, err error) {
	if cluster.Status.NodeIdentity != nil && cluster.Status.NodeIdentity.AWS != nil {
		arn := cluster.Status.NodeIdentity.AWS.RoleARN
		externalID := cluster.Status.NodeIdentity.AWS.ExternalID
		return arn, externalID, nil
	}

	externalID = cluster.GetProjectID()
	roleName := namer.NewName(cluster.ClusterScopeName(), "ng-role", "", -1)

	// TODO: get external id
	if cluster.Spec.NetworkName != "" {
		networkName := cluster.Spec.NetworkName
		networkNamespace := cluster.Namespace
		externalID = namer.GenExternalID(networkNamespace, networkName)
		roleName = cluster.Spec.Kubernetes.Name
	}

	arn = util.GetRoleARN(cluster.Spec.Kubernetes.Account, roleName)
	return
}

// TODO(tennix): once all the Cluster CRs has persisted node identity in status
// We should use the Cluster.Status.NodeIdentity as the single source of truth
func (s *ClusterServer) getTidbWorkerNodeServiceAccount(_ context.Context, cluster *cloudv1.Cluster) (string, error) {
	if cluster.Status.NodeIdentity != nil && cluster.Status.NodeIdentity.GCP != nil {
		return cluster.Status.NodeIdentity.GCP.ServiceAccount, nil
	}

	var saEmail string
	if cluster.Spec.NetworkName != "" {
		// .Spec.Kubernetes.Account is project on gcp
		gcpProject := cluster.Spec.Kubernetes.Account
		saName := namer.UniqueIdentifier(metav1.ObjectMeta{
			Namespace: cluster.Namespace,
			Name:      cluster.GetProviderType() + "-" + config2.GetConfig().Region,
		}, namer.GCPNodePoolServiceAccountNameOptions()...)
		saEmail = fmt.Sprintf("%s@%s.iam.gserviceaccount.com", saName, gcpProject)
	} else {
		saName := namer.NewName(cluster.Spec.Kubernetes.Name, "np-sa", "", namer.GKEServiceAccountNameLimit)
		saEmail = fmt.Sprintf("%s@%s.iam.gserviceaccount.com", saName, cluster.Spec.Kubernetes.Account)
	}
	return saEmail, nil
}

func (s *ClusterServer) getClusterCR(ctx context.Context, clusterId string) (*cloudv1.Cluster, error) {
	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, clusterId)
	if err != nil {
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err), zap.String("cluster", clusterId))
		if apierrors.IsNotFound(err) {
			tilog.Info(ctx, "Cluster not found", zap.String("cluster", clusterId))
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", clusterId)
		}
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterId, err)
	}
	return cluster, nil
}

func (s *ClusterServer) GetControlplaneK8SResource(ctx context.Context, req *apiv1.GetK8SResourceRequest) (*apiv1.GetK8SResourceResponse, error) {
	if req.ClusterId == "" || req.ResourceKind == "" || req.ResourceName == "" || req.Namespace == "" {
		return nil, status.Errorf(codes.InvalidArgument, "ClusterId, ResourceKind, ResourceName and Namespace are required")
	}

	mapper := s.KubeClient.RESTMapper()
	if mapper == nil {
		return nil, status.Errorf(codes.Internal, "Failed to get RESTMapper from kube client")
	}

	gvkList, err := mapper.KindsFor(schema.GroupVersionResource{Resource: strings.ToLower(req.ResourceKind) + "s"})
	if err != nil {
		tilog.Error(ctx, "Failed to get GVK for resource kind", zap.String("resourceKind", req.ResourceKind), zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get GVK for resource kind %s: %v", req.ResourceKind, err)
	}

	for _, gvk := range gvkList {
		if gvk.Kind != req.ResourceKind {
			continue
		}

		obj := &unstructured.Unstructured{}
		obj.SetGroupVersionKind(gvk)
		if err := s.KubeClient.Get(ctx, client.ObjectKey{Namespace: req.Namespace, Name: req.ResourceName}, obj); err != nil {
			continue
		}

		clusterId := obj.GetLabels()[cloudv1.ClusterTagKey]
		if clusterId != req.ClusterId {
			continue
		}

		data, err := json.MarshalIndent(obj.Object, "", "  ")
		if err != nil {
			tilog.Error(ctx, "Failed to marshal object", zap.String("resourceKind", req.ResourceKind), zap.Error(err))
			return nil, status.Errorf(codes.Internal, "Failed to marshal object %s/%s: %v", req.ResourceKind, req.ResourceName, err)
		}

		return &apiv1.GetK8SResourceResponse{Matadata: string(data)}, nil
	}

	return nil, status.Errorf(codes.NotFound, "Resource %s/%s not found in cluster %s", req.ResourceKind, req.ResourceName, req.ClusterId)
}

// UpdateClusterAWSCloud idempotently append new subnet group to cloud.AWS.SubnetGroups
func (s *ClusterServer) UpdateClusterAWSCloud(ctx context.Context, req *apiv1.UpdateClusterAWSCloudRequest) (*apiv1.Cluster, error) {
	cluster, err := s.clusterReader.Get(ctx, s.KubeClient, req.ClusterId)
	if err != nil {
		tilog.Error(ctx, "Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", req.ClusterId)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", req.ClusterId, err)
	}

	if len(req.GetSecondaryNodeSubnets()) <= 0 {
		return nil, status.Errorf(codes.InvalidArgument, "Secondary node subnets is required, got empty")
	}

	newSubnets := getAwsSubnetFromReq(req.GetSecondaryNodeSubnets())

	// Create a copy of the original subnet groups
	originGroups := cluster.Spec.Kubernetes.Cloud.AWS.SubnetGroups.DeepCopy()

	// Update the cluster's subnet groups by adding or merging the secondary CIDR subnet group
	originGroups = originGroups.AppendGroup(cloudv1.AWSSubnetGroup{
		Name:    cloudv1.AWSSubnetGroupSecondaryCIDR,
		Subnets: newSubnets,
	})
	cluster.Spec.Kubernetes.Cloud.AWS.SubnetGroups = originGroups

	if err := s.KubeClient.Update(ctx, cluster); err != nil {
		tilog.Error(ctx, "Failed to update aws cloud subnets", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to update aws cloud subnets: %s", err)
	}

	return cvt.Cluster2PB(cluster), nil
}

func (s *ClusterServer) ValidateSecurityConfiguration(ctx context.Context, req *apiv1.ValidateSecurityConfigurationRequest) (*apiv1.ValidateSecurityConfigurationResponse, error) {
	logger := tilog.WithContextS(ctx)

	switch provider := req.GetProvider(); provider {
	case apicommon.Provider_AWS, apicommon.Provider_GCP:
		return &apiv1.ValidateSecurityConfigurationResponse{IsValid: true, Message: "skipped"}, nil
	case apicommon.Provider_AZURE:
		valid, message, err := s.validateAzureCMEK(ctx, req.SecurityConfigurations.GetAzureCmek())
		if err != nil {
			return nil, err
		}

		return &apiv1.ValidateSecurityConfigurationResponse{
			IsValid: valid,
			Message: message,
		}, nil
	default:
		logger.Errorf("Unsupported provider type: %s", provider)
		return nil, status.Errorf(codes.InvalidArgument, "Unsupported provider type: %s", provider)
	}
}

func (s *ClusterServer) validateAzureCMEK(ctx context.Context, cmekReq *apiv1.Cluster_SecurityConfigurations_AzureCmek) (valid bool, message string, _ error) {
	logger := tilog.WithContextS(ctx)

	keyVersion, err := AzureQueryLatestKeyvaultKeyVersion(ctx, s.azureCli, cmekReq)
	if err != nil {
		logger.Infof("get latest key version failed: %s", err)
		return false, "Has no permissions to query key versions", nil
	}
	logger.Infof("latest key version '%s' is used", keyVersion)

	cmek := &cloudv1.AzureCMEKSpec{
		ApplicationID:             cmekReq.ApplicationId,
		ManagedIdentityResourceID: cmekReq.ManagedIdentityResourceId,
		KeyVaultName:              cmekReq.KeyVaultName,
		KeyName:                   cmekReq.KeyName,
		KeyVersion:                keyVersion,
	}
	if err := cmek.Validate(); err != nil {
		logger.Errorf("invalid Azure CMEK configuration: %s", err)
		return false, "", status.Errorf(codes.InvalidArgument, "validate azureCMEKSpec: %s", err)
	}

	return azureValidateKeyvaultKeyPermission(ctx, s.azureCli, cmek)
}

// checkInboundRulesQuota get inbound rules quota
func (s *ClusterServer) checkInboundRulesQuota(ctx context.Context, region string) (int32, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("region", region))

	if s.awsCli == nil {
		return 60, fmt.Errorf("AWS client not initialized")
	}

	quota, err := s.getVPCInboundRulesQuotaUsage(ctx, region)
	if err != nil {
		logger.Error("Failed to get VPC inbound rules quota", zap.Error(err))
		quota = 60
	}

	logger.Info("Retrieved inbound rules quota and usage",
		zap.Int32("quota", quota))

	return quota, nil
}

// getVPCInboundRulesQuotaUsage get VPC inbound rules quota limit
func (s *ClusterServer) getVPCInboundRulesQuotaUsage(ctx context.Context, region string) (int32, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("region", region))

	quotaClient := s.awsCli.ServiceQuotas(nil, cloudaws.WithRegion(region))
	targetQuotaName := "Inbound or outbound rules per security group"
	serviceCode := "vpc"

	// paginate to get all VPC service quotas
	paginator := servicequotas.NewListServiceQuotasPaginator(
		quotaClient,
		&servicequotas.ListServiceQuotasInput{
			ServiceCode: awssdk.String(serviceCode),
		},
	)

	for paginator.HasMorePages() {
		output, err := paginator.NextPage(ctx)
		if err != nil {
			return 0, err
		}

		// iterate over the current page's quotas
		for _, quota := range output.Quotas {
			if quota.QuotaName != nil && *quota.QuotaName == targetQuotaName && quota.Value != nil {
				quotaValue := int32(*quota.Value)
				logger.Info("Found and retrieved VPC inbound rules quota",
					zap.String("quotaName", targetQuotaName),
					zap.Int32("quota", quotaValue))
				return quotaValue, nil
			}
		}
	}

	return 0, fmt.Errorf("target quota not found")
}