package central_thanos

import (
	"context"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/juju/errors"
	"github.com/prometheus/common/model"
	"github.com/tidbcloud/metrics-integration/pipeline/central/config"
	"github.com/tidbcloud/metrics-integration/pipeline/central/entity"
	"github.com/tidbcloud/metrics-integration/pipeline/central/metrics/prom"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource/client"
	"github.com/tidbcloud/metrics-integration/pipeline/pkg/tidb/component"
)

func BuildNodeCPUSecondsTotalQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_cpu_seconds_total",
			Help:        "The CPU usage of TiDB/TiKV/TiFlash nodes",
			Type:        "counter",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosNodeCPURespResult, err := QueryNodeCPUUsedSeconds(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeCPURespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query used node cpu seconds, err: %v, result: %+v", err, thanosNodeCPURespResult)
		}

		for _, result := range thanosNodeCPURespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					// [cluster_name, cluster_id, instance, component, project_name]
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeCPUSecondsTotalRateQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_cpu_seconds_total_rate",
			Help:        "The CPU usage rate in cores of TiDB/TiKV/TiFlash nodes",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosNodeCPURespResult, err := QueryNodeCPUUsedSecondsRate(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeCPURespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query used node cpu seconds rate, err: %v, result: %+v", err, thanosNodeCPURespResult)
		}

		for _, result := range thanosNodeCPURespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeMemoryCapacityQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_memory_capacity_bytes",
			Help:        "The memory capacity bytes of TiDB/TiKV/TiFlash nodes",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		thanosNodeMemoryRespResult, err := QueryNodeMemoryTotalBytes(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeMemoryRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query total node memory, err: %v, result: %+v", err, thanosNodeMemoryRespResult)
		}

		for _, result := range thanosNodeMemoryRespResult.Data.Result {
			m := result.LabelSet

			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					// [cluster_name, cluster_id, instance, component, project_name]
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})

		}

		return &vec, nil
	}

}

func BuildNodeMemoryUsedQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_memory_used_bytes",
			Help:        "The used memory bytes of TiDB/TiKV/TiFlash nodes",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosNodeMemoryRespResult, err := QueryNodeMemoryUsedBytes(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeMemoryRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query used node memory, err: %v, result: %+v", err, thanosNodeMemoryRespResult)
		}

		for _, result := range thanosNodeMemoryRespResult.Data.Result {
			m := result.LabelSet

			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					// [cluster_name, cluster_id, instance, component, project_name]
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})

		}

		return &vec, nil
	}
}

func BuildNodeStorageCapacityQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_storage_capacity_bytes",
			Help:        "The disk capacity bytes of TiKV/TiFlash nodes",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosNodeStorageRespResult, err := QueryNodeStorageCapacityBytes(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeStorageRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query used node storage capacity, err: %v, result: %+v", err, thanosNodeStorageRespResult)
		}

		for _, result := range thanosNodeStorageRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetIntValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeStorageUsedBytesQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_storage_used_bytes",
			Help:        "The disk usage bytes of TiKV/TiFlash nodes",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosNodeStorageRespResult, err := QueryNodeStorageUsedBytes(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeStorageRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query used node storage, err: %v, result: %+v", err, thanosNodeStorageRespResult)
		}

		for _, result := range thanosNodeStorageRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetIntValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeStorageAvailableQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_storage_available_bytes",
			Help:        "The disk available bytes of TiKV/TiFlash nodes",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosNodeStorageRespResult, err := QueryNodeStorageAvailableBytes(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeStorageRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query used node storage available, err: %v, result: %+v", err, thanosNodeStorageRespResult)
		}

		for _, result := range thanosNodeStorageRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetIntValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueriesTotalQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_queries_total",
			Help:        "The total number of statements executed",
			Type:        "counter",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBQueriesRespResult, err := QueryDBQueriesTotal(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBQueriesRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db queries total, err: %v, result: %+v", err, thanosDBQueriesRespResult)
		}

		for _, result := range thanosDBQueriesRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelSQLType, Value: m.Type},
					{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBFailedQueriesTotalQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_failed_queries_total",
			Help:        "The total number of execution errors",
			Type:        "counter",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBFailedQueriesRespResult, err := QueryDBFailedQueriesTotal(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBFailedQueriesRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db failed queries total, err: %v, result: %+v", err, thanosDBFailedQueriesRespResult)
		}

		for _, result := range thanosDBFailedQueriesRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetIntValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelType, Value: m.Type},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueryDurationSecondsQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_query_duration_seconds",
			Help:        "The histogram of the duration of statements",
			Type:        "histogram",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		type queryResult struct {
			resp       *client.Response
			err        error
			name       string
			metricType string
			includeLE  bool
		}

		resultsCh := make(chan queryResult, 3)
		var wg sync.WaitGroup
		wg.Add(3)

		go func() {
			defer wg.Done()
			resp, err := QueryDBQueryDurationSecondsSum(ctx, cluster.Id, vec.Name+prom.VecMetricElementSum, currTime, currTime)
			resultsCh <- queryResult{
				resp:       resp,
				err:        err,
				name:       "sum",
				metricType: prom.VecMetricElementSum,
				includeLE:  false,
			}
		}()

		go func() {
			defer wg.Done()
			resp, err := QueryDBQueryDurationSecondsCount(ctx, cluster.Id, vec.Name+prom.VecMetricElementCount, currTime, currTime)
			resultsCh <- queryResult{
				resp:       resp,
				err:        err,
				name:       "count",
				metricType: prom.VecMetricElementCount,
				includeLE:  false,
			}
		}()

		go func() {
			defer wg.Done()
			resp, err := QueryDBQueryDurationSecondsBucket(ctx, cluster.Id, vec.Name+prom.VecMetricElementBucket, currTime, currTime)
			resultsCh <- queryResult{
				resp:       resp,
				err:        err,
				name:       "bucket",
				metricType: prom.VecMetricElementBucket,
				includeLE:  true,
			}
		}()

		go func() {
			wg.Wait()
			close(resultsCh)
		}()

		collected := make(map[string]queryResult, 3)
		for r := range resultsCh {
			if r.err != nil || r.resp == nil || r.resp.Status != "success" {
				return nil, errors.Errorf("failed to query db query duration seconds %s, err: %v, result: %+v", r.name, r.err, r.resp)
			}
			collected[r.name] = r
		}
		for _, key := range []string{"sum", "count", "bucket"} {
			if _, ok := collected[key]; !ok {
				return nil, errors.Errorf("missing db query duration query result %q (collected %d)", key, len(collected))
			}
		}

		appendDurationElements := func(resp *client.Response, metricType string, includeLE bool) error {
			for _, result := range resp.Data.Result {
				m := result.LabelSet
				value, err := datasource.GetFloatValue(result.Value)
				if err != nil {
					return err
				}

				labels := []prom.MetricLabel{
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelSQLType, Value: m.SqlType},
				}
				if includeLE {
					labels = append(labels, prom.MetricLabel{Name: datasource.LabelLE, Value: m.Le})
				}

				vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
					Type:   metricType,
					Labels: labels,
					Value:  model.SampleValue(value),
				})
			}
			return nil
		}

		if err := appendDurationElements(collected["sum"].resp, collected["sum"].metricType, collected["sum"].includeLE); err != nil {
			return nil, err
		}
		if err := appendDurationElements(collected["count"].resp, collected["count"].metricType, collected["count"].includeLE); err != nil {
			return nil, err
		}
		if err := appendDurationElements(collected["bucket"].resp, collected["bucket"].metricType, collected["bucket"].includeLE); err != nil {
			return nil, err
		}

		return &vec, nil
	}
}

func BuildDBConnectionsQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_connections",
			Help:        "Current number of connections in TiDB server",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBConnectionsRespResult, err := QueryDBConnections(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBConnectionsRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db connections, err: %v, result: %+v", err, thanosDBConnectionsRespResult)
		}

		for _, result := range thanosDBConnectionsRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetIntValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBTotalConnectionQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_total_connection",
			Help:        "The number of current connection",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBTotalConnectionRespResult, err := QueryDBTotalConnection(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBTotalConnectionRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db total connection, err: %v, result: %+v", err, thanosDBTotalConnectionRespResult)
		}

		for _, result := range thanosDBTotalConnectionRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBActiveConnectionsQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_active_connections",
			Help:        "The number of active connection",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBActiveConnectionsRespResult, err := QueryDBActiveConnections(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBActiveConnectionsRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db active connections, err: %v, result: %+v", err, thanosDBActiveConnectionsRespResult)
		}

		for _, result := range thanosDBActiveConnectionsRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBDisconnectionsQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_disconnections",
			Help:        "The number of clients disconnected",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBDisconnectionsRespResult, err := QueryDBDisconnections(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBDisconnectionsRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db disconnections, err: %v, result: %+v", err, thanosDBDisconnectionsRespResult)
		}

		for _, result := range thanosDBDisconnectionsRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelResult, Value: m.Result},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBDatabaseTimeQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_database_time",
			Help:        "A time model statistic that is the sum of all process' CPU consumption plus the sum of non-idle wait time.",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBDatabaseTimeRespResult, err := QueryDBDatabaseTime(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBDatabaseTimeRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db database time, err: %v, result: %+v", err, thanosDBDatabaseTimeRespResult)
		}

		for _, result := range thanosDBDatabaseTimeRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelSQLType, Value: m.SqlType},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueryPerSecondQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_query_per_second",
			Help:        "The number of SQL statements executed per second on all TiDB instances, which is counted according to SELECT, INSERT, UPDATE, and other types of statements",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBQueryPerSecondRespResult, err := QueryDBQueryPerSecond(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBQueryPerSecondRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db query per second, err: %v, result: %+v", err, thanosDBQueryPerSecondRespResult)
		}

		for _, result := range thanosDBQueryPerSecondRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelType, Value: m.Type},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBFailedQueriesQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_failed_queries",
			Help:        "The statistics of error types (such as syntax errors and primary key conflicts) according to the errors occurred when executing SQL statements per second on each TiDB instance.",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBFailedQueriesRespResult, err := QueryDBFailedQueries(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBFailedQueriesRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db failed queries, err: %v, result: %+v", err, thanosDBFailedQueriesRespResult)
		}

		for _, result := range thanosDBFailedQueriesRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelType, Value: m.Type},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBCommandPerSecondQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_command_per_second",
			Help:        "The number of commands processed by TiDB per second, which is classified according to the success or failure of command execution results",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBCommandPerSecondRespResult, err := QueryDBCommandPerSecond(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBCommandPerSecondRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db command per second, err: %v, result: %+v", err, thanosDBCommandPerSecondRespResult)
		}

		for _, result := range thanosDBCommandPerSecondRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelType, Value: m.Type},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueriesUsingPlanCacheOpsQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_queries_using_plan_cache_ops",
			Help:        "The statistics of queries using Plan Cache per second.\\nThe execution plan cache only supports the prepared statement command. If the value is 0, the execution plan cache has not been hit.",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBQueriesUsingPlanCacheOpsRespResult, err := QueryDBQueriesUsingPlanCacheOps(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBQueriesUsingPlanCacheOpsRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db queries using plan cache ops, err: %v, result: %+v", err, thanosDBQueriesUsingPlanCacheOpsRespResult)
		}

		for _, result := range thanosDBQueriesUsingPlanCacheOpsRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelType, Value: m.Type},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBAverageQueryDurationQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_average_query_duration",
			Help:        "The duration between the time that the client's network request is sent to TiDB and the time that the request is returned to the client after TiDB has executed it.",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()
		thanosDBAverageQueryDurationRespResult, err := QueryDBAverageQueryDuration(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBAverageQueryDurationRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db average query duration, err: %v, result: %+v", err, thanosDBAverageQueryDurationRespResult)
		}

		for _, result := range thanosDBAverageQueryDurationRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelSQLType, Value: m.SqlType},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBTransactionPerSecondQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_db_transaction_per_second",
			Help:        "The number of transactions executed per second",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		thanosDBTransactionPerSecondRespResult, err := QueryDBTransactionPerSecond(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosDBTransactionPerSecondRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query db transaction per second, err: %v, result: %+v", err, thanosDBTransactionPerSecondRespResult)
		}

		for _, result := range thanosDBTransactionPerSecondRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelTxnMode, Value: m.TxnMode},
					{Name: datasource.LabelType, Value: m.Type},
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeCPUCapacityCoresQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_node_cpu_capacity_cores",
			Help:        "The CPU limit cores of TiDB/TiKV/TiFlash nodes",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		componentCPUMap := cluster.NodeCPUCapacityComponentMap()
		vec.Timestamp = currTime.Unix()
		// todo: cpu cores.
		// cluster.ClusterInfo
		// for _, comp := range cluster.Components {
		// 	componentCapacityMapper[centralutil.CoreComponentType2componentName(comp.ComponentType).String()] = float64(comp.CpuNum)
		// }
		// We can only get instance from thanos metrics.
		thanosNodeCPUSecondsRespResult, err := QueryNodeCPUUsedSeconds(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || thanosNodeCPUSecondsRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query used cpu seconds, err: %v, result: %+v", err, thanosNodeCPUSecondsRespResult)
		}

		for _, result := range thanosNodeCPUSecondsRespResult.Data.Result {
			m := result.LabelSet
			valueTuple := result.Value
			if len(valueTuple) != 2 {
				return &vec, errors.Errorf("failed to query value(%+v) of used cpu seconds", valueTuple)
			}

			if m.Component == component.PD.String() {
				continue
			}

			capacityValue, ok := componentCPUMap[m.Component]
			if !ok {
				continue
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				// [cluster_name, cluster_id, instance, component, project_name]
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},

				Value: model.SampleValue(capacityValue),
			})
		}

		return &vec, nil
	}
}

func BuildTiCDCLatencyQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_changefeed_latency",
			Help:        "The progress (latency/lag) of data replication (the unit is second) between the upstream and the downstream",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ticdcLatencyRespResult, err := QueryTiCDCLatencySeconds(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ticdcLatencyRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query ticdc changefeed latency, err: %v, result: %+v", err, ticdcLatencyRespResult)
		}

		for _, result := range ticdcLatencyRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelChangefeedID, Value: m.Changefeed},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildTiCDCCheckpointTSQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {

		vec := prom.VecMetric{
			Name:        "tidbcloud_changefeed_checkpoint_ts",
			Help:        "The checkpoint timestamp of a changefeed",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ticdcCheckpointTSRespResult, err := QueryTiCDCCheckpointTS(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ticdcCheckpointTSRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query cdc changefeed checkpoint ts, err: %v, result: %+v", err, ticdcCheckpointTSRespResult)
		}

		for _, result := range ticdcCheckpointTSRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelChangefeedID, Value: m.Changefeed},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildChangefeedStatusQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_changefeed_status",
			Help:        "The status of a changefeed",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ticdcCheckpointTSRespResult, err := QueryChangefeedStatus(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ticdcCheckpointTSRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query cdc changefeed status, err: %v, result: %+v", err, ticdcCheckpointTSRespResult)
		}

		for _, result := range ticdcCheckpointTSRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelChangefeedID, Value: m.Changefeed},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildTiCDCReplicaRowsQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_changefeed_Replica_rows",
			Help:        "The number of replica rows that ChangeFeed writes to the downstream per second",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ticdcReplicaRowsPerSecondRespResult, err := QueryTiCDCReplicaRowsPerSecond(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ticdcReplicaRowsPerSecondRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query ticdc replication rows, err: %v, result: %+v", err, ticdcReplicaRowsPerSecondRespResult)
		}

		for _, result := range ticdcReplicaRowsPerSecondRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelChangefeedID, Value: m.Changefeed},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildKVRequestDurationQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_kv_request_duration",
			Help:        "The duration of kv request",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		kvRequestRespResult, err := QueryKVRequestDuration(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || kvRequestRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query kv request duration, err: %v, result: %+v", err, kvRequestRespResult)
		}

		for _, result := range kvRequestRespResult.Data.Result {
			m := result.LabelSet

			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelType, Value: m.Type},
				},
				Value: model.SampleValue(value),
			})

		}

		return &vec, nil
	}

}

func BuildUptimeQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_component_uptime",
			Help:        "The uptime of tidb components",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		uptimeRespResult, err := QueryUptime(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || uptimeRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query uptime, err: %v, result: %+v", err, uptimeRespResult)
		}

		for _, result := range uptimeRespResult.Data.Result {
			m := result.LabelSet

			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})

		}

		return &vec, nil
	}

}

func BuildDiskReadLatencyQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_disk_read_latency",
			Help:        "The read latency",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		latencyRespResult, err := QueryDiskReadLatency(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || latencyRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query disk read latency, err: %v, result: %+v", err, latencyRespResult)
		}

		for _, result := range latencyRespResult.Data.Result {
			m := result.LabelSet

			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelDevice, Value: m.Device},
				},
				Value: model.SampleValue(value),
			})

		}

		return &vec, nil
	}

}

func BuildDiskWriteLatencyQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_disk_write_latency",
			Help:        "The write latency",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		latencyRespResult, err := QueryDiskWriteLatency(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || latencyRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query disk write latency, err: %v, result: %+v", err, latencyRespResult)
		}

		for _, result := range latencyRespResult.Data.Result {
			m := result.LabelSet

			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}

			if m.Component == component.PD.String() {
				continue
			}

			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelDevice, Value: m.Device},
				},
				Value: model.SampleValue(value),
			})

		}

		return &vec, nil
	}

}

func BuildTiCDCChangefeedResolvedTsLagQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_ticdc_owner_resolved_ts_lag",
			Help:        "The resolved ts of a changefeed's owner",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ticdcOwnerResolvedTsRespResult, err := QueryTiCDCResolvedTS(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ticdcOwnerResolvedTsRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query cdc owner resolved ts lag, err: %v, result: %+v", err, ticdcOwnerResolvedTsRespResult)
		}

		for _, result := range ticdcOwnerResolvedTsRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelChangefeedID, Value: m.Changefeed},
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildRUReadQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_resource_manager_resource_unit_read_request_unit",
			Help:        "The resource manager read request",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ruRespResult, err := QueryRURead(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ruRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query read RU, err: %v, result: %+v", err, ruRespResult)
		}

		for _, result := range ruRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildRUWriteQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_resource_manager_resource_unit_write_request_unit",
			Help:        "The resource manager write request",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ruRespResult, err := QueryRUWrite(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ruRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query write RU, err: %v, result: %+v", err, ruRespResult)
		}

		for _, result := range ruRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildDMTaskStatusQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_dm_task_status",
			Help:        "The status of a DM task",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		dmTaskStatusRespResult, err := QueryDMTaskState(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || dmTaskStatusRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query dm tasks status, err: %v, result: %+v", err, dmTaskStatusRespResult)
		}
		for _, result := range dmTaskStatusRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelDMJobID, Value: m.JobID},
					{Name: datasource.LabelDMTask, Value: m.Task},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildDMSyncerReplicationLagBucketQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_dm_syncer_replication_lag",
			Help:        "The replication lag bucket of a DM syncer",
			Type:        "histogram",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		dmSyncerReplicationLagRespResult, err := QueryDMReplicationLagBucket(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || dmSyncerReplicationLagRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query dm syncer replication lag bucket, err: %v, result: %+v", err, dmSyncerReplicationLagRespResult)
		}
		for _, result := range dmSyncerReplicationLagRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			labels := []prom.MetricLabel{
				{Name: datasource.LabelCluster, Value: cluster.Name},
				{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
				{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
				{Name: datasource.LabelDMJobID, Value: m.JobID},
				{Name: datasource.LabelDMTask, Value: m.Task},
			}
			var metricType string
			if strings.HasSuffix(m.MetricName, prom.VecMetricElementSum) {
				metricType = prom.VecMetricElementSum
			} else if strings.HasSuffix(m.MetricName, prom.VecMetricElementCount) {
				metricType = prom.VecMetricElementCount
			} else if strings.HasSuffix(m.MetricName, prom.VecMetricElementBucket) {
				metricType = prom.VecMetricElementBucket
				labels = append(labels, prom.MetricLabel{Name: datasource.LabelLE, Value: m.Le})
			} else {
				return nil, errors.Errorf("unexpected metric name: %s", m.MetricName)
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Type:   metricType,
				Labels: labels,
				Value:  model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildDMSyncerReplicationLagGaugeQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_dm_syncer_replication_lag_gauge",
			Help:        "The replication lag gauge of a DM syncer",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		dmSyncerReplicationLagRespResult, err := QueryDMReplicationLagGauge(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || dmSyncerReplicationLagRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query dm syncer replication lag bucket, err: %v, result: %+v", err, dmSyncerReplicationLagRespResult)
		}
		for _, result := range dmSyncerReplicationLagRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelDMJobID, Value: m.JobID},
					{Name: datasource.LabelDMTask, Value: m.Task}},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildDMRelayReadErrorCount(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_dm_relay_read_error_count",
			Help:        "The error count of relay read",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		dmSyncerReplicationLagRespResult, err := QueryDMRelayReadErrorCount(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || dmSyncerReplicationLagRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query dm syncer replication lag bucket, err: %v, result: %+v", err, dmSyncerReplicationLagRespResult)
		}
		for _, result := range dmSyncerReplicationLagRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: []prom.MetricLabel{
					{Name: datasource.LabelCluster, Value: cluster.Name},
					{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
					{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
					{Name: datasource.LabelDMJobID, Value: m.JobID},
					{Name: datasource.LabelDMTask, Value: m.Task}},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildRUTotalQuerier(cluster *config.TargetCluster) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:        "tidbcloud_resource_manager_resource_request_unit_total",
			Help:        "The resource manager request unit total",
			Type:        "gauge",
			Metrics:     []*prom.VecMetricElement{},
			ServicePlan: cluster.ServicePlan,
		}
		currTime := time.Now()
		vec.Timestamp = currTime.Unix()

		ruRespResult, err := QueryRUTotal(ctx, cluster.Id, vec.Name, currTime, currTime)
		if err != nil || ruRespResult.Status != "success" {
			return &vec, errors.Errorf("failed to query total RU, err: %v, result: %+v", err, ruRespResult)
		}

		for _, result := range ruRespResult.Data.Result {
			m := result.LabelSet
			value, err := datasource.GetFloatValue(result.Value)
			if err != nil {
				return &vec, err
			}
			labels := []prom.MetricLabel{
				{Name: datasource.LabelCluster, Value: cluster.Name},
				{Name: datasource.LabelClusterID, Value: strconv.FormatUint(cluster.Id, 10)},
				{Name: datasource.LabelProjectID, Value: strconv.FormatUint(cluster.ProjectId, 10)},
			}
			if shouldIncludeResourceGroupLabelForRUTotal(cluster.ServicePlan) {
				labels = append(labels, prom.MetricLabel{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup})
			}
			vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
				Labels: labels,
				Value:  model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func shouldIncludeResourceGroupLabelForRUTotal(servicePlan string) bool {
	return !entity.ServicePlan(servicePlan).IsServerless()
}