package cluster

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/google/go-cmp/cmp"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler"

	k8snetworkv1 "k8s.io/api/networking/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	vmv1beta1 "github.com/VictoriaMetrics/operator/api/operator/v1beta1"
	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/configs"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/utils"
	k8sapiv1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	prometheusPort    = 9090
	sidecarCPURequest = "10m"
)

// peekPrometheusConditionStatus returns the prometheusConditionStatus and timeout Status
// if update timeout occurs, it will also return the current condition status of conditionType.
func peekPrometheusConditionStatus(conditions []prometheusv1.PrometheusCondition, conditionType prometheusv1.PrometheusConditionType, nowtime string) (prometheusv1.PrometheusConditionStatus, string, error) {
	for _, v := range conditions {
		if v.Type == conditionType {
			if nowtime != "" {
				// prometheus condition update time must newer than cluster
				clustertime, err := time.Parse(time.RFC3339, nowtime)
				if err != nil {
					return prometheusv1.PrometheusConditionUnknown, "", err
				}

				if v.LastTransitionTime.After(clustertime) {
					return v.Status, "", nil
				} else {
					if time.Now().Unix()-clustertime.Unix() > monitorTimeout {
						// if it exceeds the monitorTimeout, we consider it an update that not trigger recreation.
						// in this case, we just need to check if the conditionStatus is Ture.
						return v.Status, monitorUpdateTimeout, nil
					}
					return prometheusv1.PrometheusConditionUnknown, "", nil
				}
			}
			return v.Status, "", nil
		}
	}
	return prometheusv1.PrometheusConditionUnknown, "", nil
}

// getPodSecurityContext will set pod runs user id and use which gourp to mount pvc
func getPodSecurityContext() *k8sapiv1.PodSecurityContext {
	const (
		GROUP = 2000
		USER  = 1000
	)
	return &k8sapiv1.PodSecurityContext{
		FSGroup:      utils.Int64Ptr(GROUP),
		RunAsGroup:   utils.Int64Ptr(GROUP),
		RunAsNonRoot: utils.BoolPtr(true),
		RunAsUser:    utils.Int64Ptr(USER),
	}
}

// getPodSecurityContext will set pod runs user id and use which gourp to mount pvc
func getStorageSpec(storage *resource.Quantity) *prometheusv1.StorageSpec {
	const (
		AWS_SSD_NAME = "prometheus.ssd"
	)
	storagespec := &prometheusv1.StorageSpec{
		VolumeClaimTemplate: prometheusv1.EmbeddedPersistentVolumeClaim{
			Spec: k8sapiv1.PersistentVolumeClaimSpec{
				StorageClassName: utils.StringPtr(AWS_SSD_NAME),
			},
		},
	}
	storagespec.VolumeClaimTemplate.Spec.Resources.Requests = make(k8sapiv1.ResourceList)
	storagespec.VolumeClaimTemplate.Spec.Resources.Requests[k8sapiv1.ResourceStorage] = *storage
	return storagespec
}

func overSellFloat(quantity resource.Quantity, oversell float64) resource.Quantity {
	limit := quantity.DeepCopy()

	incValue := limit.AsApproximateFloat64() * oversell
	// use NewScaledQuantity to reduce precision loss
	inc := resource.NewScaledQuantity(int64(incValue*10000), -4)

	limit.Add(*inc)

	return limit
}

func overrideContainer(name string) (container k8sapiv1.Container) {
	container.Name = name
	container.Resources.Requests = make(k8sapiv1.ResourceList)
	container.Resources.Limits = make(k8sapiv1.ResourceList)
	container.Resources.Requests[k8sapiv1.ResourceCPU] = resource.MustParse(sidecarCPURequest)
	container.Resources.Limits[k8sapiv1.ResourceCPU] = resource.MustParse(sidecarCPURequest)
	return container
}

// makePrometheus create prometheus crd
func makePrometheus(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	var newprome prometheusv1.Prometheus
	newprome.SetName(clusterv1.Spec.Prometheus.Name)
	newprome.SetNamespace(clusterv1.Namespace)
	//prometheus configs
	newprome.Spec.Replicas = &clusterv1.Spec.Prometheus.Replicas
	newprome.Spec.Version = clusterv1.Spec.Prometheus.Version
	if len(cls.serviceInfo.AccountInfo.Project) == 0 {
		newprome.Spec.Image = utils.StringPtr(fmt.Sprintf("%s.dkr.ecr.us-west-2.amazonaws.com/observability/prometheus:%s", cls.serviceInfo.AccountInfo.Account, clusterv1.Spec.Prometheus.Version))
	} else {
		newprome.Spec.Image = utils.StringPtr(fmt.Sprintf("us-west1-docker.pkg.dev/%s/observability/prometheus:%s", cls.serviceInfo.AccountInfo.Project, clusterv1.Spec.Prometheus.Version))
	}
	newprome.Spec.Retention = prometheusv1.Duration(clusterv1.Spec.Prometheus.Retention)
	newprome.Spec.EnableRemoteWriteReceiver = true
	newprome.Spec.SecurityContext = getPodSecurityContext()

	//alerting configs
	newprome.Spec.Alerting = clusterv1.Spec.Prometheus.Alerting
	//prometheusrule selector
	newprome.Spec.RuleNamespaceSelector = clusterv1.Spec.Prometheus.RuleNamespaceSelector
	newprome.Spec.RuleSelector = clusterv1.Spec.Prometheus.RuleSelector

	//resource
	newprome.Spec.Resources.Requests = make(k8sapiv1.ResourceList)
	newprome.Spec.Resources.Limits = make(k8sapiv1.ResourceList)
	newprome.Spec.Resources.Requests[k8sapiv1.ResourceMemory] = *clusterv1.Spec.Prometheus.Resources.Requests.Memory()
	newprome.Spec.Resources.Requests[k8sapiv1.ResourceCPU] = *clusterv1.Spec.Prometheus.Resources.Requests.Cpu()
	if clusterv1.Spec.Prometheus.Resources.Limits != nil {
		newprome.Spec.Resources.Limits[k8sapiv1.ResourceMemory] = *clusterv1.Spec.Prometheus.Resources.Limits.Memory()
		newprome.Spec.Resources.Limits[k8sapiv1.ResourceCPU] = *clusterv1.Spec.Prometheus.Resources.Limits.Cpu()
	} else {
		newprome.Spec.Resources.Limits[k8sapiv1.ResourceMemory] = overSellFloat(*clusterv1.Spec.Prometheus.Resources.Requests.Memory(), cls.serviceInfo.DefaultOversell)
		newprome.Spec.Resources.Limits[k8sapiv1.ResourceCPU] = overSellFloat(*clusterv1.Spec.Prometheus.Resources.Requests.Cpu(), cls.serviceInfo.DefaultOversell)
	}
	// query parameters
	newprome.Spec.Query = clusterv1.Spec.Prometheus.Query

	//sidecar resource
	// CAUTION:
	// The version for prometheus-config-reloader image in AWS and GCP, which are v0.57.0 and v0.56.3,
	// are different for now because of the Prometheus Operator deployment issue.
	// Please pay attention to this difference while upgrading Prometheus Operator.
	initContainer := overrideContainer("init-config-reloader")
	container := overrideContainer("config-reloader")
	if len(cls.serviceInfo.AccountInfo.Project) == 0 {
		initContainer.Image = fmt.Sprintf("%s.dkr.ecr.us-west-2.amazonaws.com/observability/prometheus-config-reloader:v0.57.0", cls.serviceInfo.AccountInfo.Account)
		container.Image = fmt.Sprintf("%s.dkr.ecr.us-west-2.amazonaws.com/observability/prometheus-config-reloader:v0.57.0", cls.serviceInfo.AccountInfo.Account)
	} else {
		initContainer.Image = fmt.Sprintf("us-west1-docker.pkg.dev/%s/observability/prometheus-config-reloader:v0.56.3", cls.serviceInfo.AccountInfo.Project)
		container.Image = fmt.Sprintf("us-west1-docker.pkg.dev/%s/observability/prometheus-config-reloader:v0.56.3", cls.serviceInfo.AccountInfo.Project)
	}
	newprome.Spec.InitContainers = append(newprome.Spec.InitContainers, initContainer)
	newprome.Spec.Containers = append(newprome.Spec.Containers, container)
	//storage
	newprome.Spec.Storage = getStorageSpec(clusterv1.Spec.Prometheus.Resources.Requests.Storage())
	//node
	newprome.Spec.NodeSelector = make(map[string]string)
	newprome.Spec.NodeSelector["node_type"] = "worker"
	newprome.Spec.NodeSelector["topology.kubernetes.io/zone"] = clusterv1.Spec.Prometheus.AvailabilityZone
	//labels
	newprome.SetLabels(clusterv1.Labels)
	newprome.Spec.PodMetadata = &prometheusv1.EmbeddedObjectMetadata{}
	newprome.Spec.PodMetadata.Labels = clusterv1.Labels
	if cls.serviceInfo.Vendor == string(constants.Azure) {
		newprome.Spec.PodMetadata.Labels[constants.LabelAzureUseWorkloadIdentity] = "true"
	}
	// 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.ExternalLabelProjectID] = clusterv1.Labels[constants.LabelProjectID]
	}
	newprome.Spec.ExternalLabels = externalLabels

	newprome.Spec.EnableAdminAPI = true
	//if all Selector is nil, operator will not create prometheus config,then external lables will not work
	//ref https://github.com/prometheus-operator/prometheus-operator/blob/main/pkg/prometheus/operator.go#L1926
	newprome.Spec.ServiceMonitorSelector = &metav1.LabelSelector{}

	//annotation
	newprome.SetAnnotations(clusterv1.Annotations)
	//service account
	if account, ok := clusterv1.Annotations[constants.AnnotationServiceAccount]; ok {
		newprome.Spec.ServiceAccountName = account
	}
	//thanos sidecar
	if clusterv1.Spec.Prometheus.MetricsToObjectStorage {
		//we must use version above v0.27 because we should use pure AWS-SDK auth for S3 storage and prefix buckets
		newprome.Spec.Thanos = &prometheusv1.ThanosSpec{
			Version: utils.StringPtr("v0.27.0"),
			ObjectStorageConfig: &k8sapiv1.SecretKeySelector{
				LocalObjectReference: k8sapiv1.LocalObjectReference{
					Name: clusterv1.Spec.Prometheus.Name,
				},
				Key: "thanos.yaml",
			},
		}
		if len(cls.serviceInfo.AccountInfo.Project) == 0 {
			newprome.Spec.Thanos.Image = utils.StringPtr(fmt.Sprintf("%s.dkr.ecr.us-west-2.amazonaws.com/observability/thanos:%s", cls.serviceInfo.AccountInfo.Account, *(newprome.Spec.Thanos).Version))
		} else {
			newprome.Spec.Thanos.Image = utils.StringPtr(fmt.Sprintf("us-west1-docker.pkg.dev/%s/observability/thanos:%s", cls.serviceInfo.AccountInfo.Project, *(newprome.Spec.Thanos).Version))
		}
	}
	return &newprome, nil
}

// makePrometheusService create service crd for prometheus instances
func makePrometheusService(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	var promesvc k8sapiv1.Service

	promesvc.SetName(svcPrefix + clusterv1.Spec.Endpoint.Name)
	promesvc.SetNamespace(clusterv1.Namespace)
	promesvc.Spec.Ports = append(promesvc.Spec.Ports, k8sapiv1.ServicePort{
		Port:     prometheusPort,
		Protocol: k8sapiv1.ProtocolTCP,
	})
	promesvc.Spec.Selector = make(map[string]string)
	promesvc.Spec.Selector["prometheus"] = clusterv1.Spec.Prometheus.Name
	if clusterv1.Spec.Endpoint.Shared {
		//Shared Endpoint Path
		promesvc.Spec.Type = k8sapiv1.ServiceTypeClusterIP
	} else {
		//Exclusive Endpoint Path
		promesvc.Spec.Type = k8sapiv1.ServiceTypeLoadBalancer
		//append aws-load-balancer-controller
		promesvc.Annotations = make(map[string]string)
		promesvc.Annotations["service.beta.kubernetes.io/aws-load-balancer-type"] = "nlb"
		promesvc.Annotations["service.beta.kubernetes.io/aws-load-balancer-nlb-target-type"] = "ip"
		promesvc.Annotations["service.beta.kubernetes.io/aws-load-balancer-internal"] = "true"
		promesvc.Annotations["service.beta.kubernetes.io/aws-load-balancer-cross-zone-load-balancing-enabled"] = "true"
	}

	return &promesvc, nil
}

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

	ingress := makePrometheusIngress(ingressName, clusterv1.Namespace,
		ingressClass, sourcePrefix, clusterv1.Spec.Endpoint.Name, "")
	return &ingress, nil
}

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

	ingress := makePrometheusIngress(ingressName, clusterv1.Namespace,
		ingressClass, sourcePrefix, clusterv1.Spec.Endpoint.Name, "")

	if IsEnhancedMode(clusterv1) {
		ingress.Annotations["nginx.ingress.kubernetes.io/mirror-target"] = fmt.Sprintf(vmInsertUrlTemplate, clusterv1.Spec.VictoriaAccount.VMClusterName, clusterv1.Spec.VictoriaAccount.AccountID, clusterv1.Spec.VictoriaAccount.ProjectID) + "/$2"
	}
	return &ingress, nil
}

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

	ingress := makePrometheusIngress(ingressName, clusterv1.Namespace,
		ingressClass, sourcePrefix, clusterv1.Spec.Endpoint.Name, authUrl)
	return &ingress, nil
}

func makePrometheusIngress(name string, namespace string, ingressName string, pathPrefix string, endpointName string, authUrl string) k8snetworkv1.Ingress {
	var promeingress k8snetworkv1.Ingress
	promeingress.SetName(name)
	promeingress.SetNamespace(namespace)
	promeingress.Spec.IngressClassName = utils.StringPtr(ingressName)
	promeingress.Annotations = make(map[string]string)
	if len(authUrl) > 0 {
		promeingress.Annotations["nginx.ingress.kubernetes.io/auth-url"] = authUrl
		promeingress.Annotations["nginx.ingress.kubernetes.io/auth-cache-key"] = "$remote_user$http_authorization"

		promeingress.Annotations["nginx.ingress.kubernetes.io/enable-cors"] = "true"
		promeingress.Annotations["nginx.ingress.kubernetes.io/cors-allow-origin"] = "*"
		promeingress.Annotations["nginx.ingress.kubernetes.io/cors-allow-methods"] = "GET"
	}
	promeingress.Annotations["nginx.ingress.kubernetes.io/rewrite-target"] = "/$2"
	promeingress.Annotations["nginx.ingress.kubernetes.io/use-regex"] = "true"

	var rule k8snetworkv1.IngressRule
	var httprule k8snetworkv1.HTTPIngressRuleValue
	pathtype := k8snetworkv1.PathTypeImplementationSpecific
	httprule.Paths = append(httprule.Paths, k8snetworkv1.HTTPIngressPath{
		Path:     pathPrefix + "(/|$)(.*)",
		PathType: &pathtype,
		Backend: k8snetworkv1.IngressBackend{
			Service: &k8snetworkv1.IngressServiceBackend{
				Name: svcPrefix + endpointName,
				Port: k8snetworkv1.ServiceBackendPort{
					Number: prometheusPort,
				},
			},
		},
	})
	rule.HTTP = &httprule
	promeingress.Spec.Rules = append(promeingress.Spec.Rules, rule)
	return promeingress
}

func getPrometheusCrd(ctx context.Context, cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (obj client.Object, err error) {
	var prom prometheusv1.Prometheus

	err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: clusterv1.Spec.Prometheus.Name}, &prom)
	return &prom, err
}

func getPrometheusAZ(clusterv1 *observabilitypingcapcomv1.Cluster) string {
	return clusterv1.Spec.Prometheus.AvailabilityZone
}

func prometheusStatus(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster, clusterTime string, obj client.Object) (bool, bool, error) {
	prometheus := obj.(*prometheusv1.Prometheus)
	status, timeoutStatus, err := peekPrometheusConditionStatus(prometheus.Status.Conditions, prometheusv1.PrometheusAvailable, clusterTime)
	if err != nil {
		return false, false, err
	}

	statusActive := status == prometheusv1.PrometheusConditionTrue
	timeout := timeoutStatus == monitorUpdateTimeout
	return statusActive, timeout, nil
}

func checkIfPrometheusPausedByReplicas(ctx context.Context, prometheus *prometheusv1.Prometheus) bool {
	return prometheus.Status.Replicas == 0
}

func checkIfPrometheusReadyByReplicas(ctx context.Context, prometheus *prometheusv1.Prometheus) (bool, error) {
	bizType := prometheus.Labels[constants.LabelBizType]
	replicas, err := configs.GetMonitorReplicasByBizType(bizType)
	if err != nil {
		return false, err
	}

	if prometheus.Status.Replicas == replicas && prometheus.Status.Replicas == prometheus.Status.AvailableReplicas {
		return true, nil
	}

	return false, nil
}

func (cls *Cluster) updatePrometheus(ctx context.Context, obj client.Object, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	log := log.FromContext(ctx).WithName("updatePrometheus")
	log.V(1).Info("update prometheus crd", "crd", fmt.Sprintf("%v", obj))
	clusterv1.Status.Phase = constants.UPDATING
	clusterv1.Status.UpdateAt = time.Now().UTC().Format(time.RFC3339)
	if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
		return err
	}

	_, err := controllerutil.CreateOrUpdate(ctx, cls.cli, obj, func() error {
		prome := obj.(*prometheusv1.Prometheus)
		expected, err := makePrometheus(cls, clusterv1)
		if err != nil {
			return err
		}
		expectedSpec := expected.(*prometheusv1.Prometheus).Spec
		prome.Spec = expectedSpec
		return nil
	})
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return fmt.Errorf("update prometheus crd failed: %w", err)
	}

	cls.event.Eventf(clusterv1, "Normal", "Updating", "Updating clusters %s/%s", clusterv1.Namespace, clusterv1.Name)
	return nil
}

func (cls *Cluster) updateVMOnPauseStatus(ctx context.Context, obj client.Object, clusterv1 *observabilitypingcapcomv1.Cluster, paused bool) error {
	vmrule := &vmv1beta1.VMRule{}
	// find the vmrule with the same namespaced name.
	err := cls.cli.Get(ctx, client.ObjectKeyFromObject(clusterv1), vmrule)
	if err != nil {
		return err
	}
	if VMAlertNeededByVMRule(clusterv1, vmrule) {
		vmalert, ok := obj.(*vmv1beta1.VMAlert)
		if !ok || vmalert == nil {
			return fmt.Errorf("failed converting obj %v into vmalert", obj)
		}
		expectedCount := pointer.Int32Ptr(1)
		if paused {
			expectedCount = pointer.Int32Ptr(0)
		}
		vmalert.Spec.ReplicaCount = expectedCount
		if err := cls.cli.Update(ctx, vmalert); err != nil {
			return fmt.Errorf("updating vmalert replicasCount to %d failed, %w", *expectedCount, err)
		}
	}

	// add an paused label on vmrule to help shared-vmalert not to process these alert anymore.
	if paused {
		vmrule.Labels[constants.LabelClusterPaused] = "true"
	} else {
		delete(vmrule.Labels, constants.LabelClusterPaused)
	}
	return cls.cli.Update(ctx, vmrule)
}

func (cls *Cluster) updatePrometheusReplicas(ctx context.Context, obj client.Object, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	log := log.FromContext(ctx).WithName("updatePrometheusReplicas")
	prome, ok := obj.(*prometheusv1.Prometheus)
	if !ok || prome == nil {
		return fmt.Errorf("failed converting obj %v into prometheus", obj)
	}

	//if resources config changed ,we should change prometheus crd
	if *prome.Spec.Replicas != clusterv1.Spec.Prometheus.Replicas {
		*prome.Spec.Replicas = clusterv1.Spec.Prometheus.Replicas
		log.V(1).Info("update prometheus crd", "crd", fmt.Sprintf("%v", prome))
		if err := cls.cli.Update(ctx, prome); err != nil {
			return err
		}
	}
	return nil
}

func equalPrometheus(ctx context.Context, cls *Cluster, obj client.Object, clusterv1 *observabilitypingcapcomv1.Cluster) bool {
	prom := obj.(*prometheusv1.Prometheus)
	logger := log.FromContext(ctx).WithName("equalPrometheus")
	expected, err := makePrometheus(cls, clusterv1)
	if err != nil {
		logger.Error(err, "make prometheus error")
		return false
	}
	expectedSpec := expected.(*prometheusv1.Prometheus).Spec

	if !equalPrometheusImages(&expectedSpec, &(prom.Spec)) {
		logger.Info("diff between current spec of prometheus and expected", "diff", cmp.Diff(prom.Spec, expectedSpec))
		return false
	}
	// TODO(just1900): we use DeepDerivative to replace DeepEquals,
	// as Prometheus is amended with default values while submitted to APIServer.
	// DeepDerivative works fine for our usecase now, but it may ignore some updates,
	// e.g. add a field to Prometheus that not present in makePrometheus().
	// The best way for this might be setting a last applied annotation like kubectl apply do,
	// and do a three way merge method when updating.
	equal := apiequality.Semantic.DeepDerivative(expectedSpec, prom.Spec)
	if !equal {
		logger.Info("diff between current spec of prometheus and expected", "diff", cmp.Diff(prom.Spec, expectedSpec))
	}
	return equal
}

// This function will do the following two things:
//
//  1. Compare Images
//     There are four containers in Prometheus Pod, and three images are referred:
//     (1). InitContainer: init-config-reloader, which uses quay.io/prometheus-operator/prometheus-config-reloader:<tag> by default
//     (2). Containers:
//     i. prometheus, which uses quay.io/prometheus/prometheus:<tag> by default
//     ii. config-reloader: which uses quay.io/prometheus-operator/prometheus-config-reloader:<tag> by default
//     iii. thanos-sidecar: which uses quay.io/thanos/thanos:<tag> by default
//
//     Since we switch to use self-hosted images, the baseImage part changes but the tag part doesn't, and we would like to
//     consider images with same tag are same, which will avoid affecting the deployed Prometheus restart:
//     i.e.
//     aaaaaaaa:v1 == bbbbbbb:v1 is true as the tag v1 are same
//     aaaaaaaa:v1 == bbbbbbb:v2 is false as the tag changes from v1 to v2
//
//  2. If image comparison passes, i.e., at least the tags are same, it will set the original Prometheus's images to the expected ones,
//     which is used to make sure that the later DeepDerivative will eliminate the images issue during comparison
//
//gocyclo:ignore
func equalPrometheusImages(expectedSpec *prometheusv1.PrometheusSpec, actualSpec *prometheusv1.PrometheusSpec) bool {
	// same implementation as DeepDerivative
	if expectedSpec == nil {
		return true
	}

	// compare images for init-config-reloader
	if len(expectedSpec.InitContainers) != 0 {
		for idx, container := range expectedSpec.InitContainers {
			if container.Name == "init-config-reloader" {
				for i, c := range actualSpec.InitContainers {
					if c.Name == "init-config-reloader" {
						// for original deployed containers, there are no image specified
						// in AWS, the default tag is v0.57.0, and in GCP, the default tag is v0.56.3
						if (c.Image == "" &&
							!equalImageTag(container.Image, "quay.io/prometheus-operator/prometheus-config-reloader:v0.57.0") &&
							!equalImageTag(container.Image, "quay.io/prometheus-operator/prometheus-config-reloader:v0.56.3")) ||
							!equalImageTag(container.Image, c.Image) {
							return false
						} else {
							actualSpec.InitContainers[i].Image = expectedSpec.InitContainers[idx].Image
						}
					}
				}
			}
		}
	}

	// compare images for prometheus
	if expectedSpec.Version != "" {
		if expectedSpec.Version != actualSpec.Version {
			return false
		} else {
			actualSpec.Version = expectedSpec.Version
			actualSpec.Image = utils.StringPtr(*expectedSpec.Image)
		}
	}

	// compare images for config-reloader
	if len(expectedSpec.Containers) != 0 {
		for idx, container := range expectedSpec.Containers {
			if container.Name == "config-reloader" {
				for i, c := range actualSpec.Containers {
					if c.Name == "config-reloader" {
						if (c.Image == "" &&
							!equalImageTag(container.Image, "quay.io/prometheus-operator/prometheus-config-reloader:v0.57.0") &&
							!equalImageTag(container.Image, "quay.io/prometheus-operator/prometheus-config-reloader:v0.56.3")) ||
							!equalImageTag(container.Image, c.Image) {
							return false
						} else {
							actualSpec.Containers[i].Image = expectedSpec.Containers[idx].Image
						}
					}
				}
			}
		}
	}

	// compare images for thanos-sidecar
	if expectedSpec.Thanos != nil && expectedSpec.Thanos.Image != nil {
		if !equalImageTag(*(expectedSpec.Thanos.Image), *(actualSpec.Thanos.Image)) {
			return false
		} else {
			actualSpec.Thanos.Image = utils.StringPtr(*(expectedSpec.Thanos.Image))
		}
	}

	return true
}

// Since we switch to use self-hosted images, the baseImage part changes but the tag part doesn't, and we would like to
// consider images with same tag are same, which will avoid affecting the deployed Prometheus restart:
// i.e.
// aaaaaaaa:v1 == bbbbbbb:v1 is true as the tag v1 are same
// aaaaaaaa:v1 == bbbbbbb:v2 is false as the tag changes from v1 to v2
func equalImageTag(image1 string, image2 string) bool {
	if image1 == "" || image2 == "" {
		return true
	}

	imageSplit1 := strings.Split(image1, ":")
	imageSplit2 := strings.Split(image2, ":")

	tag1 := "latest"
	if len(imageSplit1) == 2 {
		tag1 = imageSplit1[1]
	}
	tag2 := "latest"
	if len(imageSplit2) == 2 {
		tag2 = imageSplit2[1]
	}
	return tag1 == tag2
}

// will execute in status TO_BE_DELETED, TO_BE_PAUSED
func (cls *Cluster) updatePrometheusTSDB(ctx context.Context, clusterv1 *observabilitypingcapcomv1.Cluster) error {
	snapshotName := clusterv1.Annotations[constants.AnnotationNewDataSnapshot]
	tsdbPath := "/prometheus"
	if len(snapshotName) > 0 {
		tsdbPath = fmt.Sprintf("/prometheus/snapshots/%s", snapshotName)
	}

	var prom prometheusv1.Prometheus
	err := cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: clusterv1.Spec.Prometheus.Name}, &prom)
	if err != nil {
		return err
	}

	if clusterv1.Status.Phase == constants.TO_BE_DELETED || clusterv1.Status.Phase == constants.TO_BE_PAUSED {
		prom.Spec.Containers = append(prom.Spec.Containers, k8sapiv1.Container{
			Name: "prometheus",
			Args: []string{
				"--web.console.templates=/etc/prometheus/consoles",
				"--web.console.libraries=/etc/prometheus/console_libraries",
				"--storage.tsdb.retention.time=15d",
				"--config.file=/etc/prometheus/config_out/prometheus.env.yaml",
				fmt.Sprintf("--storage.tsdb.path=%s", tsdbPath),
				"--web.enable-lifecycle",
				"--web.enable-admin-api",
				"--web.enable-remote-write-receiver",
				"--web.route-prefix=/",
				"--web.config.file=/etc/prometheus/web_config/web-config.yaml",
				"--storage.tsdb.max-block-duration=2h",
				"--storage.tsdb.min-block-duration=2h",
			},
		})
		prom.Spec.Containers = append(prom.Spec.Containers, k8sapiv1.Container{
			Name: "thanos-sidecar",
			Args: []string{
				"sidecar",
				"--prometheus.url=http://127.0.0.1:9090/",
				"--grpc-address=:10901",
				"--http-address=:10902",
				"--objstore.config=$(OBJSTORE_CONFIG)",
				fmt.Sprintf("--tsdb.path=%s", tsdbPath),
			},
		})

		prom.Spec.RuleNamespaceSelector = clusterv1.Spec.Prometheus.RuleNamespaceSelector
		// mute when TO_BE_PAUSED
		prom.Spec.RuleSelector = &metav1.LabelSelector{
			MatchLabels: map[string]string{
				"muteall": "true",
			},
		}

		if err := cls.cli.Update(ctx, &prom); err != nil {
			return err
		}

		clusterv1.Annotations[constants.AnnotationNewDataSnapshot] = ""
		timestamp := time.Now().Add(time.Minute * 10).Unix()
		if clusterv1.Status.Phase == constants.TO_BE_DELETED {
			clusterv1.Labels[constants.LabelTimestampToDelete] = strconv.FormatInt(timestamp, 10)
		} else if clusterv1.Status.Phase == constants.TO_BE_PAUSED {
			clusterv1.Labels[constants.LabelTimestampToPause] = strconv.FormatInt(timestamp, 10)
		}
		if err := cls.cli.Update(ctx, clusterv1); err != nil {
			return err
		}
	}

	return nil
}

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

	allExist := true
	partExist := false
	var prom prometheusv1.Prometheus
	err := cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: clusterv1.Spec.Prometheus.Name}, &prom)
	if client.IgnoreNotFound(err) != nil {
		return reconciler.SubResourceObservation{}, nil, err
	}
	allExist = allExist && (err == nil)
	partExist = partExist || (err == nil)

	log.V(1).Info("prometheus crd info", "crd", fmt.Sprintf("%v", prom))
	var promService k8sapiv1.Service
	err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: svcPrefix + clusterv1.Spec.Endpoint.Name}, &promService)
	if client.IgnoreNotFound(err) != nil {
		return reconciler.SubResourceObservation{}, nil, err
	}
	allExist = allExist && (err == nil)
	partExist = partExist || (err == nil)

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

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

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

	if clusterv1.Spec.Prometheus.MetricsToObjectStorage {
		var thanosSecret k8sapiv1.Secret
		err = cls.cli.Get(ctx, types.NamespacedName{Namespace: clusterv1.Namespace, Name: clusterv1.Spec.Prometheus.Name}, &thanosSecret)
		if client.IgnoreNotFound(err) != nil {
			return reconciler.SubResourceObservation{}, nil, err
		}
		allExist = allExist && (err == nil)
		partExist = partExist || (err == nil)

		log.V(1).Info("thanos secret config  info", "secret", fmt.Sprintf("%v", thanosSecret))
	}

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