package cluster

import (
	"context"
	"fmt"

	"github.com/tidbcloud/observability/pkg/common/configs"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/utils/ptr"

	k8sapiv1 "k8s.io/api/core/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	vmv1beta1 "github.com/VictoriaMetrics/operator/api/operator/v1beta1"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/log"

	prometheusv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/utils"
	k8snetworkv1 "k8s.io/api/networking/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	vmInsertAPI       = "/insert/"
	vmSelectAPI       = "/select/"
	vmInsertPort      = 8480
	vmSelectPort      = 8481
	vmInsertSvcPrefix = "vminsert-"
	vmSelectSvcPrefix = "vmselect-"

	externalNameSuffix  = ".observability.svc.cluster.local"
	vmSelectUrlTemplate = "http://vmselect-%s.observability.svc.cluster.local:8481/select/%d:%d/prometheus"
	vmInsertUrlTemplate = "http://vminsert-%s.observability.svc.cluster.local:8480/insert/%d:%d/prometheus"

	vmAlertDefaultSpecID                 = "vmalert.default"
	vmAlertExtraArgLoggerMaxArgLen       = "10000"
	vmAlertRemoteWriteMaxBatchSize int32 = 5000

	alertManagerPort = 9093
	alertManagerUrl  = "http://%s.%s.%s.svc.cluster.local:%d"
)

func makeVMSelectIngress(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	ingressName := svcPrefix + clusterv1.Spec.Endpoint.Name + readSuffix
	ingressClass := cls.serviceInfo.AZConfig[clusterv1.Spec.VictoriaAccount.AvailabilityZone].ReadIngressClassName
	sourcePrefix := fmt.Sprintf(constants.ReadMetricsInternalPathPrefix, clusterv1.Name)
	targetServiceName := vmSelectSvcPrefix + clusterv1.Name

	ingress := makeVMIngress(ingressName, clusterv1.Namespace, ingressClass, vmSelectAPI,
		clusterv1.Spec.VictoriaAccount.AccountID,
		clusterv1.Spec.VictoriaAccount.ProjectID,
		sourcePrefix, targetServiceName, vmSelectPort, "", false)
	return &ingress, nil
}

func makeVMInsertIngress(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	ingressName := svcPrefix + clusterv1.Spec.Endpoint.Name
	ingressClass := cls.serviceInfo.AZConfig[clusterv1.Spec.VictoriaAccount.AvailabilityZone].IngressClassName
	sourcePrefix := fmt.Sprintf(constants.WriteMetricsPathPrefix, clusterv1.Name)
	targetServiceName := vmInsertSvcPrefix + clusterv1.Name

	ingress := makeVMIngress(ingressName, clusterv1.Namespace, ingressClass, vmInsertAPI,
		clusterv1.Spec.VictoriaAccount.AccountID,
		clusterv1.Spec.VictoriaAccount.ProjectID,
		sourcePrefix, targetServiceName, vmInsertPort, "", false)
	return &ingress, nil
}

func makeServerLessInsertIngress(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	ingressName := serverlessPrefix + clusterv1.Spec.Endpoint.Name
	ingressClass := cls.serviceInfo.AZConfig[clusterv1.Spec.VictoriaAccount.AvailabilityZone].IngressClassName
	sourcePrefix := fmt.Sprintf(constants.ServerlessWriteMetricsPathPrefix, clusterv1.Name)
	targetServiceName := vmInsertSvcPrefix + clusterv1.Name

	ingress := makeVMIngress(ingressName, clusterv1.Namespace, ingressClass, vmInsertAPI,
		clusterv1.Spec.VictoriaAccount.AccountID,
		clusterv1.Spec.VictoriaAccount.ProjectID,
		sourcePrefix, targetServiceName, vmInsertPort, "", true)
	return &ingress, nil

}

func makeVMOpenSelectIngress(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	ingressName := svcPrefix + clusterv1.Spec.Endpoint.Name + openReadSuffix
	ingressClass := cls.serviceInfo.AZConfig[clusterv1.Spec.VictoriaAccount.AvailabilityZone].ReadIngressClassName
	sourcePrefix := fmt.Sprintf(constants.ReadMetricsExternalPathPrefixTemplate,
		clusterv1.Labels[constants.LabelBizType],
		clusterv1.Labels[constants.LabelTenantID],
		clusterv1.Labels[constants.LabelProjectID],
		clusterv1.Labels[constants.LabelApplicationID],
	)
	targetServiceName := vmSelectSvcPrefix + clusterv1.Name
	authUrl := fmt.Sprintf(cls.serviceInfo.BizTypeConfigs[clusterv1.Labels[constants.LabelBizType]].ExternalApiAuthUrlTemplate, clusterv1.Labels[constants.LabelTenantID])

	ingress := makeVMIngress(ingressName, clusterv1.Namespace, ingressClass, vmSelectAPI,
		clusterv1.Spec.VictoriaAccount.AccountID,
		clusterv1.Spec.VictoriaAccount.ProjectID,
		sourcePrefix, targetServiceName, vmSelectPort, authUrl, false)
	return &ingress, nil
}

const rewriteTargetTemplate = "%s%d:%d/prometheus/$2"
const serverlessTargetTemplate = "%smultitenant/prometheus/$2"

func makeVMIngress(name string, namespace string, ingressName string, targetPrefix string, accountID int32, projectID int32, sourcePath string, targetServiceName string, targetPort int32, authUrl string, serverless bool) k8snetworkv1.Ingress {
	var vmingress k8snetworkv1.Ingress
	vmingress.SetName(name)
	vmingress.SetNamespace(namespace)
	vmingress.Spec.IngressClassName = utils.StringPtr(ingressName)
	vmingress.Annotations = make(map[string]string)
	if len(authUrl) > 0 {
		vmingress.Annotations["nginx.ingress.kubernetes.io/auth-url"] = authUrl
		vmingress.Annotations["nginx.ingress.kubernetes.io/auth-cache-key"] = "$remote_user$http_authorization"

		vmingress.Annotations["nginx.ingress.kubernetes.io/enable-cors"] = "true"
		vmingress.Annotations["nginx.ingress.kubernetes.io/cors-allow-origin"] = "*"
		vmingress.Annotations["nginx.ingress.kubernetes.io/cors-allow-methods"] = "GET"
	}
	vmingress.Annotations["nginx.ingress.kubernetes.io/use-regex"] = "true"
	if serverless {
		vmingress.Annotations["nginx.ingress.kubernetes.io/rewrite-target"] = fmt.Sprintf(serverlessTargetTemplate, targetPrefix)
	} else {
		vmingress.Annotations["nginx.ingress.kubernetes.io/rewrite-target"] = fmt.Sprintf(rewriteTargetTemplate, targetPrefix, accountID, projectID)
	}
	var rule k8snetworkv1.IngressRule
	var httpRule k8snetworkv1.HTTPIngressRuleValue
	pathType := k8snetworkv1.PathTypeImplementationSpecific
	httpRule.Paths = append(httpRule.Paths, k8snetworkv1.HTTPIngressPath{
		Path:     sourcePath + "(/|$)(.*)",
		PathType: &pathType,
		Backend: k8snetworkv1.IngressBackend{
			Service: &k8snetworkv1.IngressServiceBackend{
				Name: targetServiceName,
				Port: k8snetworkv1.ServiceBackendPort{
					Number: targetPort,
				},
			},
		},
	})
	rule.HTTP = &httpRule
	vmingress.Spec.Rules = append(vmingress.Spec.Rules, rule)
	return vmingress
}

func makeVMSelectExternalService(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	return makeVMExternalService(clusterv1.Namespace, vmSelectSvcPrefix, clusterv1.Spec.VictoriaAccount.VMClusterName, clusterv1.Name)
}

func makeVMInsertExternalService(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	return makeVMExternalService(clusterv1.Namespace, vmInsertSvcPrefix, clusterv1.Spec.VictoriaAccount.VMClusterName, clusterv1.Name)
}

func makeVMExternalService(ns, vmComponentPrefix string, vmClusterName, clusterID string) (client.Object, error) {
	var externalService k8sapiv1.Service
	externalService.SetName(vmComponentPrefix + clusterID)
	externalService.SetNamespace(ns)
	externalService.Spec.Type = k8sapiv1.ServiceTypeExternalName
	externalService.Spec.ExternalName = vmComponentPrefix + vmClusterName + externalNameSuffix
	return &externalService, nil
}

func makeVMAlert(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	var vmAlert vmv1beta1.VMAlert
	vmAlert.SetNamespace(clusterv1.Namespace)
	vmAlert.SetName(clusterv1.Name)
	vmAlert.SetLabels(clusterv1.Labels)
	vmAlert.SetAnnotations(clusterv1.Annotations)

	//service account
	if account, ok := clusterv1.Annotations[constants.AnnotationServiceAccount]; ok {
		vmAlert.Spec.ServiceAccountName = account
	}

	if clusterv1.Spec.VictoriaAccount.Alerting != nil {
		notifiers, err := buildNotifiers(cls, clusterv1.Spec.VictoriaAccount.Alerting)
		if err != nil {
			return nil, fmt.Errorf("build notifiers failed: %w", err)
		}
		vmAlert.Spec.Notifiers = notifiers
	}

	// setting a bigger loggerMaxArgLen to avoid logs being truncated after vmalert v1.97
	vmAlert.Spec.ExtraArgs = map[string]string{
		"loggerMaxArgLen": vmAlertExtraArgLoggerMaxArgLen,
	}

	vmSelectUrl := fmt.Sprintf(vmSelectUrlTemplate,
		clusterv1.Spec.VictoriaAccount.VMClusterName,
		clusterv1.Spec.VictoriaAccount.AccountID,
		clusterv1.Spec.VictoriaAccount.ProjectID)
	vmAlert.Spec.Datasource = vmv1beta1.VMAlertDatasourceSpec{URL: vmSelectUrl}
	vmAlert.Spec.RemoteRead = &vmv1beta1.VMAlertRemoteReadSpec{URL: vmSelectUrl}

	vmInsertUrl := fmt.Sprintf(vmInsertUrlTemplate,
		clusterv1.Spec.VictoriaAccount.VMClusterName,
		clusterv1.Spec.VictoriaAccount.AccountID,
		clusterv1.Spec.VictoriaAccount.ProjectID)
	vmAlert.Spec.RemoteWrite = &vmv1beta1.VMAlertRemoteWriteSpec{
		URL:          vmInsertUrl,
		MaxBatchSize: ptr.To(vmAlertRemoteWriteMaxBatchSize),
	}

	specID := clusterv1.Annotations[constants.AnnotationMonitorSpec]
	if len(specID) == 0 {
		specID = vmAlertDefaultSpecID
	}
	spec := cls.serviceInfo.MonitorSpec[specID]

	// Set maxQueueSize for RemoteWrite
	if clusterv1.Spec.VictoriaAccount.MaxQueueSize != nil {
		vmAlert.Spec.RemoteWrite.MaxQueueSize = clusterv1.Spec.VictoriaAccount.MaxQueueSize
	} else if spec.MaxQueueSize != nil {
		// Fallback to monitor spec configuration for maxQueueSize
		vmAlert.Spec.RemoteWrite.MaxQueueSize = spec.MaxQueueSize
	}

	// ignore RuleNamespaceSelector, only select rules under current namespace with rule selector
	// this avoids expensive namespace scan in vm-operator
	vmAlert.Spec.RuleSelector = clusterv1.Spec.VictoriaAccount.RuleSelector

	//resource
	vmAlert.Spec.ReplicaCount = ptr.To(int32(1))
	vmAlert.Spec.Resources.Requests = make(k8sapiv1.ResourceList)
	vmAlert.Spec.Resources.Limits = make(k8sapiv1.ResourceList)
	resources := clusterv1.Spec.VictoriaAccount.AlertResources
	if resources.Requests != nil {
		vmAlert.Spec.Resources.Requests[k8sapiv1.ResourceMemory] = *resources.Requests.Memory()
		vmAlert.Spec.Resources.Requests[k8sapiv1.ResourceCPU] = *resources.Requests.Cpu()
	} else {
		vmAlert.Spec.Resources.Requests = configs.GetSpecContainer()[specID].Requests
	}
	if resources.Limits != nil {
		vmAlert.Spec.Resources.Limits[k8sapiv1.ResourceMemory] = *resources.Limits.Memory()
		vmAlert.Spec.Resources.Limits[k8sapiv1.ResourceCPU] = *resources.Limits.Cpu()
	} else {
		vmAlert.Spec.Resources.Limits = configs.GetSpecContainer()[specID].Limits
	}

	// external labels
	externalLabels := make(map[string]string)
	if clusterv1.Labels != nil {
		externalLabels[constants.ExternalLabelBizType] = clusterv1.Labels[constants.LabelBizType]
		externalLabels[constants.ExternalLabelRegion] = fmt.Sprintf("%s-%s", clusterv1.Labels[constants.LabelVendor], clusterv1.Labels[constants.LabelRegion])
		externalLabels[constants.ExternalLabelTenantID] = clusterv1.Labels[constants.LabelTenantID]
		externalLabels[constants.ExternalLabelApplicationID] = clusterv1.Labels[constants.LabelApplicationID]
		externalLabels[constants.ExternalLabelO11yID] = clusterv1.Name
	}
	if clusterv1.Labels[constants.LabelProjectID] != "" {
		externalLabels[constants.ExternalLabelProjectID] = clusterv1.Labels[constants.LabelProjectID]
	}
	vmAlert.Spec.ExternalLabels = externalLabels
	//sidecar resource
	vmAlert.Spec.Containers = append(vmAlert.Spec.Containers, overrideContainer("config-reloader"))

	//node
	vmAlert.Spec.NodeSelector = make(map[string]string)
	vmAlert.Spec.NodeSelector["node_type"] = "worker"
	vmAlert.Spec.NodeSelector["topology.kubernetes.io/zone"] = clusterv1.Spec.VictoriaAccount.AvailabilityZone

	vmAlert.Spec.EmbeddedProbes = &vmv1beta1.EmbeddedProbes{
		LivenessProbe: &k8sapiv1.Probe{
			InitialDelaySeconds: 30,
		},
		ReadinessProbe: &k8sapiv1.Probe{
			InitialDelaySeconds: 30,
		},
	}
	return &vmAlert, nil
}

// buildNotifiers returns the list of notifiers containing the alertmanager statefulset pod's fqdn.
// if the service alerting.Alertmanagers[0] is not found or the service selector not match any pods, a notifier with svc name will be returned.
func buildNotifiers(cls *Cluster, alerting *prometheusv1.AlertingSpec) ([]vmv1beta1.VMAlertNotifierSpec, error) {
	alertManager := alerting.Alertmanagers[0]
	notifiers := make([]vmv1beta1.VMAlertNotifierSpec, 0)
	defaultVMNotifier := func() vmv1beta1.VMAlertNotifierSpec {
		return vmv1beta1.VMAlertNotifierSpec{
			URL: fmt.Sprintf("http://%s.%s.svc.cluster.local:%d", alertManager.Name, alertManager.Namespace, alertManagerPort),
		}
	}
	// As per https://github.com/VictoriaMetrics/operator/issues/214
	// and that VMAlertNotifierSpec.DiscoverySelector only applies to vmalertmanager.
	// VMAlert do not support discovery for HA alertmanagers yet, we do a simple discovery here.
	// 1. Get pod name via alertmanager service selector
	svc := &k8sapiv1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      alertManager.Name,
			Namespace: alertManager.Namespace,
		},
	}
	err := cls.cli.Get(context.TODO(), client.ObjectKeyFromObject(svc), svc)
	if err != nil {
		if apierrors.IsNotFound(err) {
			notifiers = append(notifiers, defaultVMNotifier())
			return notifiers, nil
		}
		return nil, err
	}
	port := alertManagerPort
	for _, p := range svc.Spec.Ports {
		if p.Name == alertManager.Name {
			port = int(p.Port)
		}
	}

	podList := &k8sapiv1.PodList{}
	err = cls.cli.List(context.TODO(), podList, &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(svc.Spec.Selector),
		Namespace:     alertManager.Namespace,
	})
	if err != nil {
		return nil, err
	}
	// 2. Set all the alertmanagers in vmalert's notifiers.
	for _, pod := range podList.Items {
		notifiers = append(notifiers, vmv1beta1.VMAlertNotifierSpec{
			// build fqdn of alertmanager statefulset pod.
			URL: fmt.Sprintf(alertManagerUrl, pod.Name, alertManager.Name, alertManager.Namespace, port),
		})
	}
	return notifiers, nil
}

func getVMAZ(clusterv1 *observabilitypingcapcomv1.Cluster) string {
	return clusterv1.Spec.VictoriaAccount.AvailabilityZone
}

func getVMCrd(ctx context.Context, cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (obj client.Object, err error) {
	var vmalert vmv1beta1.VMAlert

	err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: clusterv1.Name}, &vmalert)
	return &vmalert, client.IgnoreNotFound(err)
}

func equalVM(ctx context.Context, cls *Cluster, obj client.Object, clusterv1 *observabilitypingcapcomv1.Cluster) bool {
	log := log.FromContext(ctx).WithName("equalVM")
	needVM, err := VMAlertNeeded(ctx, cls, clusterv1)
	if err != nil {
		// simply requeue if error occurs.
		log.Error(err, "failed to check if vmalert is needed")
		return false
	}
	if !needVM {
		return obj == nil || obj.GetName() == ""
	}
	// if muteAlert is false, we suppose obj to be not nil.
	if obj == nil {
		return false
	}
	vmAlert := obj.(*vmv1beta1.VMAlert)
	expected, err := makeVMAlert(cls, clusterv1)
	if err != nil {
		// simply requeue if error occurs.
		log.Error(err, "failed to make vmalert and requeue")
		return false
	}
	expectedSpec := expected.(*vmv1beta1.VMAlert).Spec
	return apiequality.Semantic.DeepEqual(vmAlert.Spec, expectedSpec)
}

//gocyclo:ignore
func observeVMResource(ctx context.Context, cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (reconciler.SubResourceObservation, client.Object, error) {
	var obj client.Object
	log := log.FromContext(ctx).WithName("observeVMResource")

	allExist := true
	partExist := false
	var err error
	if !RejectInsertIngress(clusterv1) {
		var externalWriteService k8sapiv1.Service
		err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: vmInsertSvcPrefix + clusterv1.Name}, &externalWriteService)
		if client.IgnoreNotFound(err) != nil {
			return reconciler.SubResourceObservation{}, nil, err
		}
		allExist = allExist && (err == nil)
		partExist = partExist || (err == nil)

		log.V(1).Info("vminsert external service crd info", "crd", fmt.Sprintf("%v", externalWriteService))
	}

	if !RejectReadIngress(cls, clusterv1) {
		var externalReadService k8sapiv1.Service
		err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: vmSelectSvcPrefix + clusterv1.Name}, &externalReadService)

		if client.IgnoreNotFound(err) != nil {
			return reconciler.SubResourceObservation{}, nil, err
		}
		allExist = allExist && (err == nil)
		partExist = partExist || (err == nil)

		log.V(1).Info("vmselect external service crd info", "crd", fmt.Sprintf("%v", externalReadService))

	}

	if !RejectInsertIngress(clusterv1) {
		var ingressWrite k8snetworkv1.Ingress
		err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: svcPrefix + clusterv1.Spec.Endpoint.Name}, &ingressWrite)
		if err != nil {
			if client.IgnoreNotFound(err) != nil {
				return reconciler.SubResourceObservation{}, nil, err
			}
			allExist = allExist && false
		} else {
			partExist = partExist || true
		}
		log.V(1).Info("write ingress crd info", "crd", fmt.Sprintf("%v", ingressWrite))
	}

	if SupportServerless(clusterv1) {
		var serverlessIngress k8snetworkv1.Ingress
		err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: serverlessPrefix + clusterv1.Spec.Endpoint.Name}, &serverlessIngress)
		if err != nil {
			if client.IgnoreNotFound(err) != nil {
				return reconciler.SubResourceObservation{}, nil, err
			}
			allExist = allExist && false
		} else {
			partExist = partExist || true
		}
		log.V(1).Info("serverless write ingress crd info", "crd", fmt.Sprintf("%v", serverlessIngress))
	}

	if !RejectReadIngress(cls, clusterv1) {
		var ingressRead k8snetworkv1.Ingress
		err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: svcPrefix + clusterv1.Spec.Endpoint.Name + readSuffix}, &ingressRead)
		if err != nil {
			if client.IgnoreNotFound(err) != nil {
				return reconciler.SubResourceObservation{}, nil, err
			}
			allExist = allExist && (err == nil)
		} else {
			partExist = partExist || (err == nil)
		}

		log.V(1).Info("read ingress crd info", "crd", fmt.Sprintf("%v", ingressRead))
	}

	if !RejectReadIngress(cls, clusterv1) {
		var ingressOpenRead k8snetworkv1.Ingress
		err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: svcPrefix + clusterv1.Spec.Endpoint.Name + openReadSuffix}, &ingressOpenRead)
		if err != nil {
			if client.IgnoreNotFound(err) != nil {
				return reconciler.SubResourceObservation{}, nil, err
			}
			allExist = allExist && (err == nil)
		} else {
			partExist = partExist || (err == nil)
		}
		log.V(1).Info("open read ingress crd info", "crd", fmt.Sprintf("%v", ingressOpenRead))
	}

	var vmAlert vmv1beta1.VMAlert
	err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: clusterv1.Name}, &vmAlert)
	if client.IgnoreNotFound(err) != nil {
		return reconciler.SubResourceObservation{}, nil, err
	} else {
		if err == nil {
			// exist
			obj = &vmAlert
			log.V(1).Info("vmalert", "crd", fmt.Sprintf("%v", vmAlert))
		}
		needVM, err1 := VMAlertNeeded(ctx, cls, clusterv1)
		if err1 != nil {
			return reconciler.SubResourceObservation{}, nil, err1
		}
		allExist = allExist && (err == nil || !needVM)
		partExist = partExist || (err == nil || !needVM)
	}

	return reconciler.SubResourceObservation{
		ResourceExists:     allExist,
		PartResourceExists: partExist,
	}, obj, nil

}

// skip create vmalert if cluster is muted or multi-tenancy rule is enabled
func VMAlertNeeded(ctx context.Context, cls *Cluster, cluster *observabilitypingcapcomv1.Cluster) (bool, error) {
	if cluster.Spec.MuteAlert {
		return false, nil
	}
	biztype := cluster.Labels[constants.LabelBizType]
	if !configs.MultiTenancyRuleEnabled(biztype, cluster.Labels[constants.LabelApplicationID]) {
		return true, nil
	}
	vmRule := &vmv1beta1.VMRule{}
	err := cls.cli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, vmRule)
	if err != nil {
		return false, client.IgnoreNotFound(err)
	}
	if vmRule.Labels[constants.LabelMultiTenancyRule] != "true" {
		return true, nil
	}
	return false, nil
}

// the result is same as VMAlertNeeded, but reduce one `cli.Get` operator
func VMAlertNeededByVMRule(cluster *observabilitypingcapcomv1.Cluster, vmRule *vmv1beta1.VMRule) bool {
	return !cluster.Spec.MuteAlert &&
		(!configs.MultiTenancyRuleEnabled(cluster.Labels[constants.LabelBizType], cluster.Labels[constants.LabelApplicationID]) ||
			vmRule.Labels[constants.LabelMultiTenancyRule] != "true")
}