package slprober

import (
	"context"
	"crypto/rand"
	"fmt"
	"io"
	"math/big"
	"sync"
	"time"

	"github.com/pingcap/log"
	"go.uber.org/atomic"
	"go.uber.org/zap"
)

type ServiceStatus string
type TidbClusterStatus string

const (
	O11yServiceStatusDetectFailed          ServiceStatus = "detect_failed"
	O11yServiceStatusSkippedByNotActive    ServiceStatus = "skipped_by_not_active"
	O11yServiceStatusSkippedByNotDedicated ServiceStatus = "skipped_by_not_dedicated"
	O11yServiceStatusDetectedAsAlive       ServiceStatus = "alive"
)

const (
	ProbeTimeout     = time.Second * 5
	ProbeTimeoutLong = time.Minute * 1
)

type O11yServiceLevelProber struct {
}

func NewO11yServiceLevelProber() *O11yServiceLevelProber {
	return &O11yServiceLevelProber{}
}

type o11yProbeResult struct {
	cluster *ObservabilityCluster
	status  ServiceStatus
	err     error
}

func (m *O11yServiceLevelProber) ProbeObservabilityDedicatedPrometheus() {
	// query cluster
	timeBegin := time.Now()
	o11yClusters := queryObservabilityCluster()
	timeListDuration := time.Since(timeBegin)

	timeProbeBegin := time.Now()

	probeResultsCh := make(chan o11yProbeResult, len(o11yClusters))
	availableResults := atomic.NewUint32(0)
	var wg sync.WaitGroup

	for _, cluster := range o11yClusters {
		cls := cluster
		wg.Add(1)
		go m.probPrometheus(cls, &wg, probeResultsCh, availableResults)
	}
	waitTimeout(&wg, ProbeTimeout)
	resultsN := int(availableResults.Load()) // Not all results might be ready

	// Generate statistics summary for output
	statsProbeSuccess := 0
	statsProbeFailed := 0
	statsSkipNotDedicatedClusters := 0
	statsSkipNotActiveClusters := 0
	for i := 0; i < resultsN; i++ {
		r := <-probeResultsCh
		switch r.status {
		case O11yServiceStatusSkippedByNotDedicated:
			statsSkipNotDedicatedClusters++
		case O11yServiceStatusDetectedAsAlive:
			statsProbeSuccess++
		case O11yServiceStatusDetectFailed:
			statsProbeFailed++
		case O11yServiceStatusSkippedByNotActive:
			statsSkipNotActiveClusters++
		}
	}

	timeProbeDuration := time.Since(timeProbeBegin)
	timeDuration := time.Since(timeBegin)

	log.Info("O11yServiceLevelProber::ProbeDedicatedPrometheusFinished",
		zap.Int("all_clusters", len(o11yClusters)),
		zap.Int("skip_not_dedicated_clusters", statsSkipNotDedicatedClusters),
		zap.Int("probe_success_clusters", statsProbeSuccess),
		zap.Int("probe_failed_clusters", statsProbeFailed),
		zap.Int("skip_not_active_clusters", statsSkipNotActiveClusters),
		zap.Duration("list_elapsed", timeListDuration),
		zap.Duration("probe_elapsed", timeProbeDuration),
		zap.Duration("total_elapsed", timeDuration),
	)
}

func (m *O11yServiceLevelProber) probPrometheus(cluster ClusterBaseInfo,
	wg *sync.WaitGroup,
	resultCh chan o11yProbeResult,
	resultN *atomic.Uint32) {
	clusterDetail := GetClusterDetail(DetailClusterReq{
		CommonReqInfo: CommonReqInfo{
			Vendor: cluster.Vendor,
			Region: cluster.Region,
		},
		ID: cluster.ID,
	})
	status, err := m.probePrometheusImpl(clusterDetail)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbeDedicatedPrometheusError",
			zap.String("region", clusterDetail.Region),
			zap.String("o11y_cluster_id", clusterDetail.ID),
			zap.String("tenant_id", clusterDetail.TenantID),
			zap.String("project_id", clusterDetail.ProjectID),
			zap.String("application_id", clusterDetail.ApplicationID),
			zap.Error(err))
	}
	// if the cluster is not dedicated, metric should not be set and upload to sqs
	if status != O11yServiceStatusSkippedByNotDedicated {
		setO11yDedicatedMetricsAPIAvailableMetric(cluster, string(status))
	}
	resultCh <- o11yProbeResult{
		cluster: clusterDetail,
		status:  status,
		err:     err,
	}
	resultN.Add(1)
	wg.Done()
}

func (m *O11yServiceLevelProber) probePrometheusImpl(clusterDetail *ObservabilityCluster) (status ServiceStatus, err error) {
	status = O11yServiceStatusDetectedAsAlive
	if clusterDetail.ShareMode != "exclusive" {
		status = O11yServiceStatusSkippedByNotDedicated
		return
	}
	if clusterDetail.Status != "ACTIVE" {
		status = O11yServiceStatusSkippedByNotActive
		return
	}

	ns := fmt.Sprintf("%s-%s", clusterDetail.BizType, clusterDetail.TenantID)
	urlWrite := fmt.Sprintf("http://svc-%s.%s.svc.cluster.local:%d%s/write", clusterDetail.ID, ns, 9090, ApiPrefix)

	client := NewClient(urlWrite)
	_, err = client.Write(context.Background(), &WriteRequest{
		TimeSeries: []TimeSeries{
			{
				Labels: []Label{
					{
						Name:  LABEL_NAME,
						Value: METRIC_NAME,
					},
				},
				Sample: Sample{
					Time:  time.Now(),
					Value: 123,
				},
			},
		},
	})
	if err != nil /*|| code == 200*/ {
		// set metric
		status = O11yServiceStatusDetectFailed
		return
	}
	// if 2xx continue
	urlQuery := fmt.Sprintf("http://svc-%s.%s.svc.cluster.local:%d%s/query?query=%s", clusterDetail.ID, ns, 9090, ApiPrefix, "my_metric_name")
	query, err := QueryPrometheus(urlQuery)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	// compare metric
	if len(query.Data.Result) == 0 {
		status = O11yServiceStatusDetectFailed
		return
	}
	if query.Data.Result[0].Metric[LABEL_NAME] != METRIC_NAME {
		status = O11yServiceStatusDetectFailed
		return
	}
	return
}

func setO11yDedicatedMetricsAPIAvailableMetric(cluster ClusterBaseInfo, status string) {
	O11yDedicatedPrometheusAPIAvailableCounter.
		WithLabelValues(cluster.Region, cluster.ID, cluster.TenantID, cluster.ProjectID, cluster.ApplicationID, status).
		Inc()
}

func (m *O11yServiceLevelProber) ProbeObservabilitySharedVMCluster() {
	c := GetConfig()
	timeProbeBegin := time.Now()
	status, err := m.probeVMImpl()
	O11ySharedVMAPIAvailableCounter.
		WithLabelValues(c.Region, "regional-vm-cluster", string(status)).
		Inc()
	timeProbeDuration := time.Since(timeProbeBegin)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbeSharedVMError",
			zap.String("region", c.Region),
			zap.Error(err))
	}

	log.Info("O11yServiceLevelProber::ProbeSharedVMFinished",
		zap.String("probe_shared_vm_result", string(status)),
		zap.Duration("probe_elapsed", timeProbeDuration),
	)
}

func (m *O11yServiceLevelProber) probeVMImpl() (status ServiceStatus, err error) {
	status = O11yServiceStatusDetectedAsAlive
	urlWrite := "http://vminsert-vmcluster.observability.svc.cluster.local:8480/insert/0:1/prometheus/api/v1/write"

	client := NewClient(urlWrite)
	_, err = client.Write(context.Background(), &WriteRequest{
		TimeSeries: []TimeSeries{
			{
				Labels: []Label{
					{
						Name:  LABEL_NAME,
						Value: METRIC_NAME,
					},
				},
				Sample: Sample{
					Time:  time.Now(),
					Value: 123,
				},
			},
		},
	})
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	//if 2xx continue
	urlQuery := fmt.Sprintf("http://vmselect-vmcluster.observability.svc.cluster.local:8481/select/0:1/prometheus/api/v1/query?query=%s", "my_metric_name")
	query, err := QueryVM(urlQuery)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}

	// compare metric
	if len(query.Data.Result) == 0 {
		status = O11yServiceStatusDetectFailed
		return
	}
	if query.Data.Result[0].Metric[LABEL_NAME] != METRIC_NAME {
		status = O11yServiceStatusDetectFailed
		return
	}

	return
}

func (m *O11yServiceLevelProber) ProbeObservabilitySharedLoki() {
	c := GetConfig()
	timeProbeBegin := time.Now()
	status, err := m.probeLokiImpl()

	O11ySharedLokiAPIAvailableCounter.
		WithLabelValues(c.Region, "regional-loki", string(status)).
		Inc()
	timeProbeDuration := time.Since(timeProbeBegin)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbeSharedLokiError",
			zap.String("region", c.Region),
			zap.Error(err))
	}
	log.Info("O11yServiceLevelProber::ProbeSharedLokiFinished",
		zap.String("probe_shared_loki_result", string(status)),
		zap.Duration("probe_elapsed", timeProbeDuration),
	)
}

func (m *O11yServiceLevelProber) probeLokiImpl() (status ServiceStatus, err error) {
	status = O11yServiceStatusDetectedAsAlive
	req := LokiLog{
		Streams: []Streams{
			{
				Stream: map[string]string{
					"foo": "bar2",
				},
				Values: [][]string{
					{"1570818238000000000", "fizzbuzz"},
				},
			},
		},
	}
	resp, err := WriteLoki(req)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		status = O11yServiceStatusDetectFailed
		return
	}
	// todo query loki
	return
}

var clusterID string

func (m *O11yServiceLevelProber) ProbeObservabilityCreateCluster() {
	c := GetConfig()
	timeProbeBegin := time.Now()
	status, err := m.probeCreateClusterImpl(c)
	O11yCreateClusterAPIAvailableCounter.
		WithLabelValues(c.Region, clusterID, string(status)).
		Inc()
	timeProbeDuration := time.Since(timeProbeBegin)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbeCreateClusterError",
			zap.String("region", c.Region),
			zap.String("o11y_cluster_id", clusterID),
			zap.Error(err))
	}
	log.Info("O11yServiceLevelProber::ProbeCreateClusterFinished",
		zap.String("probe_create_cluster_result", string(status)),
		zap.Duration("probe_elapsed", timeProbeDuration),
	)
}

func (m *O11yServiceLevelProber) probeCreateClusterImpl(c *RegionConfig) (status ServiceStatus, err error) {
	req := CreateClusterReq{
		ClusterBaseInfo: ClusterBaseInfo{
			CommonReqInfo: CommonReqInfo{
				Vendor: c.Vendor,
				Region: c.Region,
			},
			TenantID:           "tenantid",
			ProjectID:          "projectid",
			ApplicationID:      "applicationid",
			BizType:            "ft-biz",
			Version:            "v1",
			ShareMode:          "exclusive",
			Tags:               []string{"tag1", "default"},
			DataPropertyRights: 1,
		},
		AgentAccessARN: c.UserArn,
		MonitorSpecID:  "spec1",
	}
	status = O11yServiceStatusDetectedAsAlive
	resp, err := CreateCluster(req)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		status = O11yServiceStatusDetectFailed
		return
	}
	clustersResp := &CreateClusterResp{}
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	err = UnmarshalRespData(respContent, clustersResp)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	clusterID = clustersResp.ID
	return
}

func (m *O11yServiceLevelProber) ProbeObservabilityGetCluster() {
	if clusterID == "" {
		return
	}
	c := GetConfig()
	timeProbeBegin := time.Now()
	status, err := m.probeGetClusterImpl()
	O11yGetClusterAPIAvailableCounter.
		WithLabelValues(c.Region, clusterID, string(status)).
		Inc()
	timeProbeDuration := time.Since(timeProbeBegin)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbeGetClusterError",
			zap.String("region", c.Region),
			zap.String("o11y_cluster_id", clusterID),
			zap.Error(err))
	}
	log.Info("O11yServiceLevelProber::ProbeGetClusterFinished",
		zap.String("probe_get_cluster_result", string(status)),
		zap.Duration("probe_elapsed", timeProbeDuration),
	)
}

func (m *O11yServiceLevelProber) probeGetClusterImpl() (status ServiceStatus, err error) {
	resp, err := GetCluster(clusterID)
	status = O11yServiceStatusDetectedAsAlive
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		status = O11yServiceStatusDetectFailed
		return
	}

	clustersResp := &DetailClusterResp{}
	respContent, err := io.ReadAll(resp.Body)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}

	err = UnmarshalRespData(respContent, clustersResp)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	if clustersResp.ID != clusterID || clustersResp.TenantID != "tenantid" || clustersResp.ProjectID != "projectid" || clustersResp.ApplicationID != "applicationid" {
		status = O11yServiceStatusDetectFailed
		return
	}
	return
}

func (m *O11yServiceLevelProber) ProbeS3LogsUploadDelay() {
	// query cluster
	timeBegin := time.Now()
	o11yClusters := queryObservabilityCluster()
	timeListDuration := time.Since(timeBegin)
	timeProbeBegin := time.Now()

	probeResultsCh := make(chan o11yProbeResult, len(o11yClusters))
	availableResults := atomic.NewUint32(0)
	var wg sync.WaitGroup

	for _, cluster := range o11yClusters {
		cls := cluster
		wg.Add(1)
		go m.probeS3Logs(cls, &wg, probeResultsCh, availableResults)
	}
	waitTimeout(&wg, ProbeTimeoutLong)
	resultsN := int(availableResults.Load()) // Not all results might be ready
	// Generate statistics summary for output
	statsProbeSuccess := 0
	statsProbeFailed := 0
	statsSkipNotDedicatedClusters := 0
	statsSkipNotActiveClusters := 0
	for i := 0; i < resultsN; i++ {
		r := <-probeResultsCh
		switch r.status {
		case O11yServiceStatusSkippedByNotDedicated:
			statsSkipNotDedicatedClusters++
		case O11yServiceStatusDetectedAsAlive:
			statsProbeSuccess++
		case O11yServiceStatusDetectFailed:
			statsProbeFailed++
		case O11yServiceStatusSkippedByNotActive:
			statsSkipNotActiveClusters++
		}
	}

	timeProbeDuration := time.Since(timeProbeBegin)
	timeDuration := time.Since(timeBegin)
	log.Info("O11yServiceLevelProber::ProbeS3LogsUploadDelayFinished",
		zap.Int("all_clusters", len(o11yClusters)),
		zap.Int("skip_not_dedicated_clusters", statsSkipNotDedicatedClusters),
		zap.Int("probe_success_clusters", statsProbeSuccess),
		zap.Int("probe_failed_clusters", statsProbeFailed),
		zap.Int("skip_not_active_clusters", statsSkipNotActiveClusters),
		zap.Duration("list_elapsed", timeListDuration),
		zap.Duration("probe_elapsed", timeProbeDuration),
		zap.Duration("total_elapsed", timeDuration),
	)
}

func (m *O11yServiceLevelProber) probeS3Logs(cluster ClusterBaseInfo,
	wg *sync.WaitGroup,
	resultCh chan o11yProbeResult,
	resultN *atomic.Uint32) {
	clusterDetail := GetClusterDetail(DetailClusterReq{
		CommonReqInfo: CommonReqInfo{
			Vendor: cluster.Vendor,
			Region: cluster.Region,
		},
		ID: cluster.ID,
	})
	_, updateMinutes, status, err := m.probeS3LogsImpl(clusterDetail)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbeS3LogsUploadDelayError",
			zap.String("region", clusterDetail.Region),
			zap.String("o11y_cluster_id", clusterDetail.ID),
			zap.String("tenant_id", clusterDetail.TenantID),
			zap.String("project_id", clusterDetail.ProjectID),
			zap.String("application_id", clusterDetail.ApplicationID),
			zap.Error(err))
	}
	if status != O11yServiceStatusSkippedByNotDedicated {
		O11yS3LogsUploadDelayHistogram.
			WithLabelValues(clusterDetail.Region, clusterDetail.ID, clusterDetail.TenantID, clusterDetail.ProjectID, clusterDetail.ApplicationID, string(status)).Observe(float64(updateMinutes))
	}
	resultCh <- o11yProbeResult{
		cluster: clusterDetail,
		status:  status,
		err:     err,
	}
	resultN.Add(1)
	wg.Done()
}

func (m *O11yServiceLevelProber) probeS3LogsImpl(clusterDetail *ObservabilityCluster) (lastModifiedTime time.Time, minutesToLastModified int, status ServiceStatus, err error) {
	status = O11yServiceStatusDetectedAsAlive
	if clusterDetail.ShareMode != "exclusive" {
		status = O11yServiceStatusSkippedByNotDedicated
		return
	}
	if clusterDetail.Status != "ACTIVE" {
		status = O11yServiceStatusSkippedByNotActive
		return
	}

	num, _ := rand.Int(rand.Reader, big.NewInt(10))
	time.Sleep(time.Duration(num.Int64()) * time.Second)
	tidbCluster, _ := GetTiDBCluster(*clusterDetail)
	tidbStatus, createTime := GetTidbClusterStatus(tidbCluster, clusterDetail.TenantID, clusterDetail.ApplicationID)
	if tidbStatus != "normal" {
		status = O11yServiceStatusSkippedByNotActive
		return
	}
	if clusterDetail.Vendor == "aws" {
		awsSess, _ := DefaultSession(clusterDetail.Region)
		lastModifiedTime, minutesToLastModified, _, err = GetMinutesSinceLastUpdateTimeS3(clusterDetail, awsSess, createTime)
		if err != nil {
			status = O11yServiceStatusDetectFailed
		}
	} else if clusterDetail.Vendor == "gcp" {
		lastModifiedTime, minutesToLastModified, _, _, err = GetMinutesSinceLastUpdateTimeGCS(clusterDetail, createTime)
		if err != nil {
			status = O11yServiceStatusDetectFailed
		}
	}

	return
}

func (m *O11yServiceLevelProber) ProbeS3MetricsUploadDelay() {
	// query cluster
	timeBegin := time.Now()
	o11yClusters := queryObservabilityCluster()
	timeListDuration := time.Since(timeBegin)
	timeProbeBegin := time.Now()

	probeResultsCh := make(chan o11yProbeResult, len(o11yClusters))
	availableResults := atomic.NewUint32(0)
	var wg sync.WaitGroup

	for _, cluster := range o11yClusters {
		cls := cluster
		wg.Add(1)
		m.probeS3Metrics(cls, &wg, probeResultsCh, availableResults)
	}
	waitTimeout(&wg, ProbeTimeoutLong)
	resultsN := int(availableResults.Load()) // Not all results might be ready
	// Generate statistics summary for output
	statsProbeSuccess := 0
	statsProbeFailed := 0
	statsSkipNotDedicatedClusters := 0
	statsSkipNotActiveClusters := 0
	for i := 0; i < resultsN; i++ {
		r := <-probeResultsCh
		switch r.status {
		case O11yServiceStatusSkippedByNotDedicated:
			statsSkipNotDedicatedClusters++
		case O11yServiceStatusDetectedAsAlive:
			statsProbeSuccess++
		case O11yServiceStatusDetectFailed:
			statsProbeFailed++
		case O11yServiceStatusSkippedByNotActive:
			statsSkipNotActiveClusters++
		}
	}

	timeProbeDuration := time.Since(timeProbeBegin)
	timeDuration := time.Since(timeBegin)
	log.Info("O11yServiceLevelProber::ProbeS3MetricsUploadDelayFinished",
		zap.Int("all_clusters", len(o11yClusters)),
		zap.Int("skip_not_dedicated_clusters", statsSkipNotDedicatedClusters),
		zap.Int("probe_success_clusters", statsProbeSuccess),
		zap.Int("probe_failed_clusters", statsProbeFailed),
		zap.Int("skip_not_active_clusters", statsSkipNotActiveClusters),
		zap.Duration("list_elapsed", timeListDuration),
		zap.Duration("probe_elapsed", timeProbeDuration),
		zap.Duration("total_elapsed", timeDuration),
	)
}

func (m *O11yServiceLevelProber) probeS3Metrics(cluster ClusterBaseInfo,
	wg *sync.WaitGroup,
	resultCh chan o11yProbeResult,
	resultN *atomic.Uint32) {
	clusterDetail := GetClusterDetail(DetailClusterReq{
		CommonReqInfo: CommonReqInfo{
			Vendor: cluster.Vendor,
			Region: cluster.Region,
		},
		ID: cluster.ID,
	})

	_, updateMinutes, status, err := m.probeS3MetricsImpl(clusterDetail)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbeS3MetricsUploadDelayError",
			zap.String("region", clusterDetail.Region),
			zap.String("o11y_cluster_id", clusterDetail.ID),
			zap.String("tenant_id", clusterDetail.TenantID),
			zap.String("project_id", clusterDetail.ProjectID),
			zap.String("application_id", clusterDetail.ApplicationID),
			zap.Error(err))
	}
	if status != O11yServiceStatusSkippedByNotDedicated {
		O11yS3MetricsUploadDelayHistogram.
			WithLabelValues(clusterDetail.Region, clusterDetail.ID, clusterDetail.TenantID, clusterDetail.ProjectID, clusterDetail.ApplicationID, string(status)).Observe(float64(updateMinutes))
	}
	resultCh <- o11yProbeResult{
		cluster: clusterDetail,
		status:  status,
		//err:     err,
	}
	resultN.Add(1)
	wg.Done()
}

func (m *O11yServiceLevelProber) probeS3MetricsImpl(clusterDetail *ObservabilityCluster) (lastModifiedTime time.Time, minutesToLastModified int, status ServiceStatus, err error) {
	status = O11yServiceStatusDetectedAsAlive
	if clusterDetail.ShareMode != "exclusive" {
		status = O11yServiceStatusSkippedByNotDedicated
		return
	}
	if clusterDetail.Status != "ACTIVE" {
		status = O11yServiceStatusSkippedByNotActive
		return
	}

	num, _ := rand.Int(rand.Reader, big.NewInt(10))
	time.Sleep(time.Duration(num.Int64()) * time.Second)
	tidbCluster, _ := GetTiDBCluster(*clusterDetail)
	tidbStatus, createTime := GetTidbClusterStatus(tidbCluster, clusterDetail.TenantID, clusterDetail.ApplicationID)
	if tidbStatus != "normal" {
		status = O11yServiceStatusSkippedByNotActive
		return
	}
	if clusterDetail.Vendor == "aws" {
		awsSess, _ := DefaultSession(clusterDetail.Region)
		prefix := GenerateS3Prefix(clusterDetail)
		lastModifiedTime, minutesToLastModified, err = GetMinutesSinceLastS3Snapshot(clusterDetail, awsSess, prefix, createTime)
	} else if clusterDetail.Vendor == "gcp" {
		lastModifiedTime, minutesToLastModified, err = GetMinutesSinceLastGcsSnapshot(clusterDetail, createTime)
	}
	if err != nil {
		status = O11yServiceStatusDetectFailed
	}
	return
}

func (m *O11yServiceLevelProber) ProbeObservabilityPrometheusMetricsUploadDelay() {
	// query cluster
	timeBegin := time.Now()
	o11yClusters := queryObservabilityCluster()
	timeListDuration := time.Since(timeBegin)

	timeProbeBegin := time.Now()
	probeResultsCh := make(chan o11yProbeResult, len(o11yClusters))
	availableResults := atomic.NewUint32(0)
	var wg sync.WaitGroup

	for _, cluster := range o11yClusters {
		cls := cluster
		wg.Add(1)
		go m.probePrometheusMetrics(cls, &wg, probeResultsCh, availableResults)
	}
	waitTimeout(&wg, ProbeTimeout)
	resultsN := int(availableResults.Load()) // Not all results might be ready
	// Generate statistics summary for output
	statsProbeSuccess := 0
	statsProbeFailed := 0
	statsSkipNotDedicatedClusters := 0
	statsSkipNotActiveClusters := 0
	for i := 0; i < resultsN; i++ {
		r := <-probeResultsCh
		switch r.status {
		case O11yServiceStatusSkippedByNotDedicated:
			statsSkipNotDedicatedClusters++
		case O11yServiceStatusDetectedAsAlive:
			statsProbeSuccess++
		case O11yServiceStatusDetectFailed:
			statsProbeFailed++
		case O11yServiceStatusSkippedByNotActive:
			statsSkipNotActiveClusters++
		}
	}

	timeProbeDuration := time.Since(timeProbeBegin)
	timeDuration := time.Since(timeBegin)
	log.Info("O11yServiceLevelProber::ProbePrometheusMetricsUploadDelayFinished",
		zap.Int("all_clusters", len(o11yClusters)),
		zap.Int("skip_not_dedicated_clusters", statsSkipNotDedicatedClusters),
		zap.Int("probe_success_clusters", statsProbeSuccess),
		zap.Int("probe_failed_clusters", statsProbeFailed),
		zap.Int("skip_not_active_clusters", statsSkipNotActiveClusters),
		zap.Duration("list_elapsed", timeListDuration),
		zap.Duration("probe_elapsed", timeProbeDuration),
		zap.Duration("total_elapsed", timeDuration),
	)
}

func (m *O11yServiceLevelProber) probePrometheusMetrics(cluster ClusterBaseInfo,
	wg *sync.WaitGroup,
	resultCh chan o11yProbeResult,
	resultN *atomic.Uint32) {
	clusterDetail := GetClusterDetail(DetailClusterReq{
		CommonReqInfo: CommonReqInfo{
			Vendor: cluster.Vendor,
			Region: cluster.Region,
		},
		ID: cluster.ID,
	})
	status, delay, err := m.probePrometheusMetricsImpl(clusterDetail)
	if err != nil {
		log.Error("O11yServiceLevelProber::ProbePrometheusMetricsUploadDelayError",
			zap.String("region", clusterDetail.Region),
			zap.String("o11y_cluster_id", clusterDetail.ID),
			zap.String("tenant_id", clusterDetail.TenantID),
			zap.String("project_id", clusterDetail.ProjectID),
			zap.String("application_id", clusterDetail.ApplicationID),
			zap.Error(err))
	}
	if status != O11yServiceStatusSkippedByNotDedicated {
		O11yPrometheusMetricsUploadDelayHistogram.
			WithLabelValues(clusterDetail.Region, clusterDetail.ID, clusterDetail.TenantID, clusterDetail.ProjectID, clusterDetail.ApplicationID, string(status)).Observe(delay)
	}
	resultCh <- o11yProbeResult{
		cluster: clusterDetail,
		status:  status,
		err:     err,
	}
	resultN.Add(1)
	wg.Done()
}

func (m *O11yServiceLevelProber) probePrometheusMetricsImpl(clusterDetail *ObservabilityCluster) (status ServiceStatus, delay float64, err error) {
	status = O11yServiceStatusDetectedAsAlive
	if clusterDetail.ShareMode != "exclusive" {
		status = O11yServiceStatusSkippedByNotDedicated
		return
	}
	if clusterDetail.Status != "ACTIVE" {
		status = O11yServiceStatusSkippedByNotActive
		return
	}
	ns := fmt.Sprintf("%s-%s", clusterDetail.BizType, clusterDetail.TenantID)
	url := fmt.Sprintf("http://svc-%s.%s.svc.cluster.local:%d%s/query?query=up{component=\"tidb\"}", clusterDetail.ID, ns, 9090, ApiPrefix)
	res, err := RequestPrometheus(url)
	if err != nil {
		status = O11yServiceStatusDetectFailed
		return
	}
	if len(res.Data.Result) == 0 {
		status = O11yServiceStatusDetectFailed
		return
	}
	ts, ok := res.Data.Result[0].Value[0].(float64)
	if ok {
		tm := time.Unix(int64(ts), 0)
		delay = time.Since(tm).Minutes()
	}

	return
}

func queryObservabilityCluster() []ClusterBaseInfo {
	c := GetConfig()
	var o11yClusters = make([]ClusterBaseInfo, 0)
	for _, biz := range c.Config.BizType {
		queryReq := QueryClustersReq{
			CommonReqInfo: CommonReqInfo{
				Vendor: c.Vendor,
				Region: c.Region,
			},
			BizType: biz,
		}
		o11yClusters = append(o11yClusters, QueryCluster(queryReq).Clusters...)
	}
	return o11yClusters
}

func (m *O11yServiceLevelProber) ObservabilityClusterControlAPI() {
	m.ProbeObservabilityCreateCluster()
	if clusterID != "" {
		m.ProbeObservabilityGetCluster()
		clusterID = ""
	}
	ClearFtNamespaces()
}