/*
Copyright 2022.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cluster

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/go-logr/logr"
	"github.com/prometheus/common/model"
	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/metrics"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler"
	"gopkg.in/yaml.v3"
	k8sapiv1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

// ClusterReconciler reconciles a Cluster object
type ClusterReconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

const (
	clusterfinalizerName      = "observability.pingcap.com/Cluster"
	errNotCluster             = "not Cluster resource"
	errAzNotSupport           = "this availabilityZone defined on cluster we not support now"
	monitorTimeout            = 300
	monitorUpdateTimeout      = "TimeOut"
	pollInterval              = 3 * time.Second
	pausedClusterPollInterval = 15 * time.Minute
	svcPrefix                 = "svc-"
	serverlessPrefix          = "serverless-"
	readSuffix                = "-read"
	openReadSuffix            = "-open"
)

func NewClusterReconciler(mgr ctrl.Manager, log logr.Logger, serviceInfo structs.ServiceInfo) *ClusterReconciler {
	return &ClusterReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: mgr.GetClient(),
			ExternalObj: clusterConnector{
				cli:         mgr.GetClient(),
				serviceInfo: serviceInfo,
				event:       mgr.GetEventRecorderFor("ClusterController"),
			},
			NewResource: func() client.Object {
				return &observabilitypingcapcomv1.Cluster{}
			},
		},
	}
}

type clusterConnector struct {
	cli         client.Client
	serviceInfo structs.ServiceInfo
	event       record.EventRecorder
}

func (csc clusterConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	return &Cluster{
		cli:         csc.cli,
		serviceInfo: csc.serviceInfo,
		event:       csc.event,
	}, nil
}

type Cluster struct {
	cli         client.Client
	serviceInfo structs.ServiceInfo
	event       record.EventRecorder
}

func (cls *Cluster) observeCompositeResource(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster) (bool, bool, client.Object, client.Object, error) {
	log = log.WithName("observeCompositeResource")

	monitorObservation, monitor, err := cls.ObserveMonitorResource(ctx, clusterv1)
	if err != nil {
		return false, false, nil, nil, err
	}
	log.V(1).Info("monitor crd info", "crd", fmt.Sprintf("%v", monitor))

	endpointObservation, endpoint, err := cls.ObserveEndpointServiceResource(ctx, clusterv1)
	if err != nil {
		return false, false, nil, nil, err
	}
	log.V(1).Info("endpoint service crd info", "crd", fmt.Sprintf("%v", endpoint))

	observation := reconciler.Combine(monitorObservation, endpointObservation)
	return observation.ResourceExists, observation.PartResourceExists, monitor, endpoint, err
}

func (cls *Cluster) observeClusterStatusCreating(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster, monitorCrd client.Object, endpointService client.Object) (reconciler.ExternalObservation, error) {
	log = log.WithName("observeClusterStatusCreating")
	monitorUpToDate, _, err := cls.ConvertMonitorStatus(ctx, clusterv1, monitorCrd, false)
	if err != nil {
		log.Error(err, "convert monitor status failed")
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false, PollInterval: pollInterval}, err
	}
	if !monitorUpToDate {
		log.Info("monitor is not ready")
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false, PollInterval: pollInterval}, nil
	}

	endpointUpToDate, _, err := cls.EndpointServiceStatus(ctx, clusterv1, endpointService, false)
	if err != nil {
		log.Error(err, "get endpoint service status failed")
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false, PollInterval: pollInterval}, err
	}

	if !endpointUpToDate {
		log.Info("endpoint service is not ready")
		cls.reconcilerRepoter(clusterv1.Name, pollInterval.String(), "")
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false, PollInterval: pollInterval}, nil
	}
	//Phase not get final state so we should update state
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func (cls *Cluster) observeClusterStatusUpdating(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (reconciler.ExternalObservation, error) {
	log = log.WithName("observeClusterStatusUpdating")

	resourceUpToDate, timeout, err := cls.ConvertMonitorStatus(ctx, clusterv1, obj, true /*checkTimeout*/)
	if err != nil {
		log.Error(err, "convert monitor status failed")
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return reconciler.ExternalObservation{}, err
	}
	if timeout && !resourceUpToDate {
		log.Error(err, "monitor updating timeout, cluster is not ready")
		cls.reconcilerRepoter(clusterv1.Name, "", monitorUpdateTimeout)
		return reconciler.ExternalObservation{ResourceExists: true, PollInterval: pollInterval}, nil
	}
	if !resourceUpToDate {
		log.Info("monitor is not ready")
		cls.reconcilerRepoter(clusterv1.Name, pollInterval.String(), "")
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false, PollInterval: pollInterval}, nil
	}
	//Phase not get final state so we should update state
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func (cls *Cluster) observeClusterStatusActive(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster, monitor client.Object) (reconciler.ExternalObservation, error) {
	log = log.WithName("observeClusterStatusActive")
	var equaled bool = false
	if cls.MonitorEqual(ctx, monitor, clusterv1) {
		// if subresource equaled we do noting
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		equaled = true
	} else {
		log.Info("monitor is not ready")
	}
	if equaled {
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, nil
	}
	//Phase not get final state so we should update state
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func (cls *Cluster) checkIfPausedClusterWithDataExpired(cluster *observabilitypingcapcomv1.Cluster) bool {
	if cluster.Status.Phase == constants.PAUSED {
		tstr := cluster.Labels[constants.LabelTimestampToPause]
		t, _ := strconv.ParseInt(tstr, 10, 64)
		bizType := cluster.Labels[constants.LabelBizType]
		monitorRetention := cls.serviceInfo.BizTypeConfigs[bizType].MonitorRetention
		monitorRetentionDuration, _ := model.ParseDuration(monitorRetention)
		monitorRetentionDuration = monitorRetentionDuration / model.Duration(time.Second)
		if (t + int64(monitorRetentionDuration)) < time.Now().Unix() {
			return true
		}
	}

	return false
}

func (cls *Cluster) observeClusterStatusPaused(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster) (reconciler.ExternalObservation, error) {
	log = log.WithName("observeClusterStatusPaused")
	if IsSharedMode(clusterv1) {
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, nil
	}
	if !cls.checkIfPausedClusterWithDataExpired(clusterv1) {
		log.Info("monitor data has not been expired")
		cls.reconcilerRepoter(clusterv1.Name, pausedClusterPollInterval.String(), "")
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false, PollInterval: pausedClusterPollInterval}, nil
	}

	var pvclist k8sapiv1.PersistentVolumeClaimList
	err := cls.cli.List(ctx, &pvclist, &client.ListOptions{
		Namespace: clusterv1.Namespace,
		LabelSelector: labels.SelectorFromSet(map[string]string{
			"app.kubernetes.io/name":      "prometheus",
			"operator.prometheus.io/name": clusterv1.Name}),
	})
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return reconciler.ExternalObservation{}, err
	}

	if len(pvclist.Items) == 0 {
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}

	//Phase not get final state so we should update state
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func (cls *Cluster) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	log := log.FromContext(ctx).WithName("Observe")
	clusterv1, ok := obj.(*observabilitypingcapcomv1.Cluster)
	if !ok {
		cls.reconcilerRepoter(obj.GetName(), "", errNotCluster)
		return reconciler.ExternalObservation{Droped: true}, errors.New(errNotCluster)
	}
	log.V(1).Info("cluster crd info", "crd", fmt.Sprintf("%v", clusterv1))

	if clusterv1.Status.Phase == constants.TO_BE_DELETED || clusterv1.Status.Phase == constants.TO_BE_PAUSED {
		if hasNewSnapshot(clusterv1) {
			err := cls.RestartWithSnapshot(ctx, clusterv1)
			if err != nil {
				cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
				return reconciler.ExternalObservation{}, err
			}
		}
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}

	if clusterv1.Status.Phase == constants.CREATION_FAILED {
		// final state, nothing to do
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil
	}
	//if cluste not set status phase, and it serverless cluster, we should update it status and do nothing
	if clusterv1.Status.Phase == "" {
		if RejectInsertIngress(clusterv1) && RejectReadIngress(cls, clusterv1) && !NextGenNeedEndpointService(clusterv1) {
			cls.reconcilerRepoter(clusterv1.Name, "", "")
			return reconciler.ExternalObservation{ResourceUpToDate: false, ResourceExists: true}, nil
		}
	}

	//check az if support
	if _, ok := cls.serviceInfo.AZConfig[cls.GetMonitorAZ(clusterv1)]; !ok {
		log.Info(errAzNotSupport)
		cls.reconcilerRepoter(clusterv1.Name, pollInterval.String(), "")
		cls.event.Eventf(clusterv1, "Warning", "Observer", "clusters %s/%s 's region %s not support", clusterv1.Namespace, clusterv1.Name, cls.GetMonitorAZ(clusterv1))
		return reconciler.ExternalObservation{PollInterval: pollInterval}, nil
	}
	//observe depended resource exists or not
	exists, partExists, monitor, endpointService, err := cls.observeCompositeResource(ctx, log, clusterv1)
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return reconciler.ExternalObservation{}, err
	}
	if !exists {
		return reconciler.ExternalObservation{ResourceExists: exists, PartResourceExists: partExists}, nil
	}

	// observe cluster and depended resource status
	switch clusterv1.Status.Phase {
	case constants.CREATING:
		return cls.observeClusterStatusCreating(ctx, log, clusterv1, monitor, endpointService)
	case constants.UPDATING:
		return cls.observeClusterStatusUpdating(ctx, log, clusterv1, monitor)
	case constants.ACTIVE:
		return cls.observeClusterStatusActive(ctx, log, clusterv1, monitor)
	case constants.PAUSED:
		return cls.observeClusterStatusPaused(ctx, log, clusterv1)
	}
	return reconciler.ExternalObservation{ResourceExists: true}, nil
}

func hasNewSnapshot(clusterv1 *observabilitypingcapcomv1.Cluster) bool {
	return clusterv1.Annotations != nil && len(clusterv1.Annotations[constants.AnnotationNewDataSnapshot]) > 0
}

type ResourceMaker func(*Cluster, *observabilitypingcapcomv1.Cluster) (obj client.Object, err error)

func (cls *Cluster) getAndCreate(ctx context.Context, namespace string, name string, obj client.Object, clusterv1 *observabilitypingcapcomv1.Cluster, makeResource ResourceMaker) error {
	key := types.NamespacedName{Namespace: namespace, Name: name}
	err := cls.cli.Get(ctx, key, obj)
	if err != nil {
		if client.IgnoreNotFound(err) != nil {
			return err
		}
		newres, err := makeResource(cls, clusterv1)
		if err != nil {
			return err
		}
		if err := cls.cli.Create(ctx, newres); err != nil {
			return err
		}
		// update obj
		return nil
	}
	return nil
}
func (cls *Cluster) Create(ctx context.Context, obj client.Object) (errResult error) {
	clusterv1, ok := obj.(*observabilitypingcapcomv1.Cluster)
	if !ok {
		cls.reconcilerRepoter(obj.GetName(), "", errNotCluster)
		return errors.New(errNotCluster)
	}
	// update state and then create monitor crd
	clusterv1.Status.Phase = constants.CREATING
	if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return err
	}

	//create monitor resource crd
	err := cls.MakeMonitorResource(ctx, clusterv1)
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return err
	}

	//create endpoint service crd
	// for serverless cluster, we don't need to create endpoint service
	// because it doesn't have insert ingress
	// for next-gen. only if userARN is set, we need to create endpoint service
	if !RejectInsertIngress(clusterv1) || NextGenNeedEndpointService(clusterv1) {
		err = cls.makeEndpointService(ctx, clusterv1)
		if err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}
	}
	cls.event.Eventf(clusterv1, "Normal", "Created", "Created clusters %s/%s", clusterv1.Namespace, clusterv1.Name)
	cls.reconcilerRepoter(clusterv1.Name, "", "")
	return nil
}

func (cls *Cluster) updateClusterUpdating(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (errResult error) {
	log = log.WithName("updateClusterUpdating")
	//updating status
	resourceUpToDate, _, err := cls.ConvertMonitorStatus(ctx, clusterv1, obj, true)
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return err
	}
	// only update cluster status when resource is up to date(i.e. PrometheusAvailable is true) no matter timeout or not.
	if resourceUpToDate {
		clusterv1.Status.Phase = constants.ACTIVE
		if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}
	}
	cls.reconcilerRepoter(clusterv1.Name, "", "")
	return nil
}

func (cls *Cluster) updateClusterCreating(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster) (errResult error) {
	log = log.WithName("updateClusterCreating")
	//for serverless cluster, we don't need to create endpoint service
	// because it doesn't have insert ingress
	// for next-gen. only if userARN is set, we need to create endpoint service
	if RejectInsertIngress(clusterv1) && !NextGenNeedEndpointService(clusterv1) {
		log.Info("serverless or no need userARN next-gen cluster, no need to create endpoint service!")
		clusterv1.Status.Phase = constants.ACTIVE
		clusterv1.Status.DnsName = cls.serviceInfo.AZConfig[cls.GetMonitorAZ(clusterv1)].DnsName
		if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		return nil
	}
	// creating status
	//get endpoint service status and update cluster status
	endpointService, err := cls.getEndpointService(ctx, log, clusterv1.Namespace, svcPrefix+clusterv1.Spec.Endpoint.Name)
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return err
	}
	if endpointService.Status.Phase == constants.ACTIVE {
		clusterv1.Status.Phase = constants.ACTIVE
		clusterv1.Status.ServiceName = endpointService.Status.ServiceName
		clusterv1.Status.DnsName = cls.serviceInfo.AZConfig[cls.GetMonitorAZ(clusterv1)].DnsName
		if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}
	}
	cls.reconcilerRepoter(clusterv1.Name, "", "")
	return nil
}

func (cls *Cluster) updateClusterPausing(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (errResult error) {
	log = log.WithName("updateClusterPausing")
	//updating status
	paused, err := cls.CheckIfMonitorPaused(ctx, clusterv1, obj)
	if err != nil {
		return err
	}
	if !paused {
		err := cls.UpdateMonitorOnPauseStatus(ctx, clusterv1, obj, true /* paused */)
		if err != nil {
			log.Error(err, "update monitor failed when pausing o11y cluster")
			return err
		}
	}
	if paused {
		clusterv1.Status.Phase = constants.PAUSED
		if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}

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

	cls.reconcilerRepoter(clusterv1.Name, "", "")
	return nil
}

func (cls *Cluster) updateClusterPaused(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster) (errResult error) {
	log = log.WithName("updateClusterPaused")
	//updating status

	if IsSharedMode(clusterv1) {
		return nil
	}

	if !cls.checkIfPausedClusterWithDataExpired(clusterv1) {
		log.Info("monitor data has not been expired")
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		return nil
	}

	var pvclist k8sapiv1.PersistentVolumeClaimList
	err := cls.cli.List(ctx, &pvclist, &client.ListOptions{
		Namespace: clusterv1.Namespace,
		LabelSelector: labels.SelectorFromSet(map[string]string{
			"app.kubernetes.io/name":      "prometheus",
			"operator.prometheus.io/name": clusterv1.Name}),
	})
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return err
	}

	for _, item := range pvclist.Items {
		if item.Namespace != clusterv1.Namespace {
			continue
		}

		var pvc k8sapiv1.PersistentVolumeClaim
		if err := cls.getAndDelete(ctx, log, item.Namespace, item.Name, &pvc); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}

		log.Info("delete one pvc with expired data", "cluster_name", clusterv1.Name, "pvc_name", item.Name, "namespace", item.Namespace)
	}

	cls.reconcilerRepoter(clusterv1.Name, "", "")
	return nil
}

func (cls *Cluster) updateClusterResuming(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster, obj client.Object) (errResult error) {
	log = log.WithName("updateClusterResuming")
	//updating status

	isMonitorReady, err := cls.CheckIfMonitorReadyByReplicas(ctx, clusterv1, obj)
	if err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return err
	}
	if !isMonitorReady {
		err = cls.UpdateMonitorOnPauseStatus(ctx, clusterv1, obj, false /* paused */)
		if err != nil {
			log.Error(err, "update monitor replicas for resuming o11y cluster failed")
			return err
		}
	}
	if isMonitorReady {
		if !IsSharedMode(clusterv1) {
			clusterv1.Annotations[constants.AnnotationNewDataSnapshot] = ""
			err := cls.RestartWithSnapshot(ctx, clusterv1)
			if err != nil {
				cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
				return err
			}
		}
		clusterv1.Status.Phase = constants.ACTIVE
		if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}

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

	return nil
}

func (cls *Cluster) updateClusterEmpty(ctx context.Context, log logr.Logger, clusterv1 *observabilitypingcapcomv1.Cluster) (errResult error) {
	log = log.WithName("updateClusterEmpty")
	//updating status
	if RejectInsertIngress(clusterv1) && RejectReadIngress(cls, clusterv1) {
		log.Info("serverless cluster, no need to create endpoint service!")
		clusterv1.Status.Phase = constants.ACTIVE
		clusterv1.Status.DnsName = cls.serviceInfo.AZConfig[cls.GetMonitorAZ(clusterv1)].DnsName
		if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}
		cls.reconcilerRepoter(clusterv1.Name, "", "")
		return nil
	}
	log.Error(errors.New("error status"), "not serverless clluster, there may error status")
	return nil
}

func (cls *Cluster) Update(ctx context.Context, obj client.Object) (errResult error) {
	log := log.FromContext(ctx).WithName("Update")
	clusterv1, ok := obj.(*observabilitypingcapcomv1.Cluster)
	if !ok {
		cls.reconcilerRepoter(obj.GetName(), "", errNotCluster)
		return errors.New(errNotCluster)
	}

	//we should observer monitor crd
	monitorCrd, err := cls.GetMonitorCrd(ctx, clusterv1)
	if err != nil {
		log.Error(err, "get monitor crd failed")
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return err
	}

	//update status depende cluster status
	switch clusterv1.Status.Phase {
	case constants.CREATING:
		return cls.updateClusterCreating(ctx, log, clusterv1)
	case constants.UPDATING:
		//updating status
		return cls.updateClusterUpdating(ctx, log, clusterv1, monitorCrd)
	case constants.PAUSING:
		return cls.updateClusterPausing(ctx, log, clusterv1, monitorCrd)
	case constants.PAUSED:
		return cls.updateClusterPaused(ctx, log, clusterv1)
	case constants.RESUMING:
		return cls.updateClusterResuming(ctx, log, clusterv1, monitorCrd)
	case "":
		// if cluster not set status phase, and it serverless cluster, we should update it status and do nothing
		return cls.updateClusterEmpty(ctx, log, clusterv1)
	}
	//all active evnet will be update here
	//deal with monitor update
	if cls.MonitorEqual(ctx, monitorCrd, clusterv1) {
		// if subresource equaled we do noting
		cls.reconcilerRepoter(clusterv1.Name, "", "")
	} else {
		log.Info("monitor need update")
		if err = cls.UpdateMonitorCRD(ctx, clusterv1, monitorCrd); err != nil {
			log.Error(err, "update monitor crd failed")
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return err
		}
	}
	cls.reconcilerRepoter(clusterv1.Name, "", "")
	return nil
}

func (cls *Cluster) getAndDelete(ctx context.Context, log logr.Logger, namespace string, name string, obj client.Object) error {
	log = log.WithName("getAndDelete")
	err := cls.cli.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, obj)
	if err == nil {
		err = cls.cli.Delete(ctx, obj)
		if err != nil {
			return err
		}
	} else if client.IgnoreNotFound(err) != nil {
		log.Error(err, "get crd and delete crd failed")
		return err
	}
	return nil
}

func (cls *Cluster) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	clusterv1, ok := obj.(*observabilitypingcapcomv1.Cluster)
	if !ok {
		cls.reconcilerRepoter(obj.GetName(), "", errNotCluster)
		return reconciler.ExternalDeletion{}, errors.New(errNotCluster)
	}
	//update status
	if clusterv1.Status.Phase != constants.DELETING {
		clusterv1.Status.Phase = constants.DELETING
		if err := cls.cli.Status().Update(ctx, clusterv1); err != nil {
			cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
			return reconciler.ExternalDeletion{}, err
		}
	}

	// delete prometheus endpoint service crd
	if err := cls.deleteEndpointService(ctx, clusterv1); err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return reconciler.ExternalDeletion{}, err
	}

	//delete monitor resource crd
	if err := cls.DeleteMonitorResource(ctx, clusterv1); err != nil {
		cls.reconcilerRepoter(clusterv1.Name, "", err.Error())
		return reconciler.ExternalDeletion{}, err
	}

	cls.event.Eventf(clusterv1, "Normal", "Deleted", "Deleted clusters %s/%s", clusterv1.Namespace, clusterv1.Name)
	cls.reconcilerRepoter(clusterv1.Name, "", "")
	return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
}

func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	// cluster key will with value name and it namespace
	logger := r.Log.WithValues("Cluster", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)
	return r.reconciler.Reconcile(ctx, req, clusterfinalizerName)
}

// SetupWithManager sets up the controller with the Manager.
func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&observabilitypingcapcomv1.Cluster{}).
		Complete(r)
}

func makeThanosObjectConfigSecret(cls *Cluster, clusterv1 *observabilitypingcapcomv1.Cluster) (client.Object, error) {
	var secretfile structs.ThanosObjectStorageSecret
	switch cls.serviceInfo.Vendor {
	case string(constants.AWS):
		{
			secretfile.Type = "s3"
			secretfile.Config.Bucket = clusterv1.Annotations[constants.AnnotationObjectStorageBucket]
			secretfile.Config.Endpoint = clusterv1.Annotations[constants.AnnotationObjectStorageEndpoint]
		}
	case string(constants.GCP):
		{
			secretfile.Type = "gcs"
			secretfile.Config.Bucket = clusterv1.Annotations[constants.AnnotationObjectStorageBucket]
		}
	case string(constants.Azure):
		secretfile.Type = "azure"
		secretfile.Config.Container = clusterv1.Annotations[constants.AnnotationObjectStorageBucket]
		secretfile.Config.StorageAccount = clusterv1.Annotations[constants.AnnotationBlobStorageAccount]
	}
	secretfile.Prefix = clusterv1.Annotations[constants.AnnotationObjectStorageMetricsPrefix]
	//Marshal struct to yaml string
	out, err := yaml.Marshal(secretfile)
	if err != nil {
		return nil, err
	}
	var secret k8sapiv1.Secret
	secret.Name = clusterv1.Spec.Prometheus.Name
	secret.Namespace = clusterv1.Namespace
	secret.Data = make(map[string][]byte)
	secret.Data["thanos.yaml"] = out
	return &secret, nil
}

func (cls *Cluster) reconcilerRepoter(name string, interval string, err string) {
	metrics.HandleRegionalControllerRunsMetrics(cls.serviceInfo.Vendor, cls.serviceInfo.Region, "ClusterController", name, interval, err)
}