package central_thanos

import (
	"context"
	"fmt"
	"strings"

	"github.com/pingcap/errors"
	"github.com/prometheus/common/model"
	"github.com/tidbcloud/metrics-server/central/metrics/prom"
	centralutil "github.com/tidbcloud/metrics-server/central/model/util"
	"github.com/tidbcloud/metrics-server/idl/pbgen/proto/core"
	"github.com/tidbcloud/metrics-server/internal/datasource"
	"github.com/tidbcloud/metrics-server/internal/datasource/client"
	"github.com/tidbcloud/metrics-server/pkg/tidb/component"
)

func QueryNodeMemoryUsedBytes(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	// user node level metrics for memory
	// promql := fmt.Sprintf("process_resident_memory_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"} or tiflash_proxy_process_resident_memory_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	promql := fmt.Sprintf("node_memory_MemTotal_bytes{cluster_id=\"tidb%d\"} - on(component, instance) node_memory_MemAvailable_bytes{cluster_id=\"tidb%d\"}", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryNodeMemoryTotalBytes(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("node_memory_MemTotal_bytes{cluster_id=\"tidb%d\"}", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryNodeCPUUsedSeconds(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	// user node level metrics for cpu
	// promql := fmt.Sprintf("process_cpu_seconds_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"} or tiflash_proxy_process_cpu_seconds_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	promql := fmt.Sprintf("round(sum(node_cpu_seconds_total{cluster_id=\"tidb%d\", mode!= \"idle\"}) by (component, instance))", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryNodeStorageUsedBytes(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	// promql := fmt.Sprintf("(label_replace(tikv_store_size_bytes{instance=~\"db-.*\", kubernetes_namespace=\"tidb%d\",type=\"capacity\"}, \"type\", \"\", \"\", \"\")-label_replace(tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\", type=\"available\"}, \"type\", \"\", \"\", \"\")) or (tiflash_system_current_metric_StoreSizeCapacity{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"} - tiflash_system_current_metric_StoreSizeAvailable{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"})",
	// 	clusterID, clusterID, clusterID, clusterID)
	promql := fmt.Sprintf("tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"used\"} or tiflash_system_current_metric_StoreSizeUsed{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryNodeStorageCapacityBytes(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	// promql := fmt.Sprintf("sum(pd_cluster_status{type=\"storage_capacity\", kubernetes_namespace=\"tidb%d\"})", clusterID)
	promql := fmt.Sprintf("tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"capacity\"} or tiflash_system_current_metric_StoreSizeCapacity{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryNodeStorageAvailableBytes(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"available\"} or tiflash_system_current_metric_StoreSizeAvailable{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBQueriesTotal(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("tidb_executor_statement_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBFailedQueriesTotal(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("tidb_server_execute_error_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBConnections(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("tidb_server_connections{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBQueryDurationSeconds(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("{__name__=~\"tidb_server_handle_query_duration_seconds.+\",instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBTotalConnection(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(tidb_server_connections{cluster_id=~\".*%d\"}) by (component, instance)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBActiveConnections(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(tidb_server_tokens{cluster_id=~\".*%d\"}) by (component, instance)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBDisconnections(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_server_disconnection_total{cluster_id=~\".*%d\"}[2m])) by (component, instance, result)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBDatabaseTime(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_server_handle_query_duration_seconds_sum{cluster_id=~\".*%d\", sql_type!=\"internal\"}[2m])) by (component, instance, sql_type)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBQueryPerSecond(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_executor_statement_total{cluster_id=~\".*%d\"}[2m])) by (component, instance, type)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBFailedQueries(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_server_execute_error_total{cluster_id=~\".*%d\"}[2m])) by (component, instance, type)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBCommandPerSecond(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_server_query_total{cluster_id=~\".*%d\"}[2m])) by (component, instance, type)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBQueriesUsingPlanCacheOps(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_server_plan_cache_total{cluster_id=~\".*%d\"}[2m])) by (component, instance, type)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBAverageQueryDuration(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_server_handle_query_duration_seconds_sum{cluster_id=~\".*%d\", sql_type!=\"internal\"}[2m])) by (component, instance, sql_type) / sum(rate(tidb_server_handle_query_duration_seconds_count{cluster_id=~\".*%d\", sql_type!=\"internal\"}[2m])) by (component, instance, sql_type) > 0", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryDBTransactionPerSecond(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_session_transaction_duration_seconds_count{cluster_id=~\".*%d\"}[2m])) by (component, instance, type, txn_mode)", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

// cdc related
// Changefeed checkpoint lag
func QueryTiCDCLatencySeconds(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_checkpoint_ts_lag{cluster_id=~\".*%d\", changefeed!=\"\"})", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryTiCDCCheckpointTS(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_checkpoint_ts{cluster_id=~\".*%d\", changefeed!=\"\"})", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

// Changefeed resolved ts lag
func QueryTiCDCResolvedTS(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_resolved_ts_lag{cluster_id=~\".*%d\", changefeed!=\"\"})", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

// Changefeed status
func QueryChangefeedStatus(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_status{cluster_id=~\".*%d\", changefeed!=\"\"})", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryTiCDCReplicaRowsPerSecond(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum by(changefeed) (rate(ticdc_sink_total_flushed_rows_count{cluster_id=~\".*%d\",changefeed!=\"\"}[1m])) or sum by(changefeed) (rate(ticdc_sinkv2_batch_row_count_sum{cluster_id=~\".*%d\", changefeed!=\"\"}[1m])) or sum by(changefeed) (rate(ticdc_sink_batch_row_count_sum{cluster_id=~\".*%d\",changefeed!=\"\"}[1m]))", clusterID, clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

// DM related
func QueryDMTaskState(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_status{cluster_id=~\".*%d\", changefeed!=\"\"})", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryKVRequestDuration(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("sum(rate(tidb_tikvclient_request_seconds_sum{store!=\"0\", cluster_id=~\".*%d\"}[2m])) by (type)/ sum(rate(tidb_tikvclient_request_seconds_count{store!=\"0\", cluster_id=~\".*%d\"}[2m])) by (type)", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func QueryUptime(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("(time() - process_start_time_seconds{cluster_id=~\".*%d\", component=~\"(tidb|tikv)\"}) or tiflash_system_asynchronous_metric_Uptime{cluster_id=~\".*%d\"}", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

// Control Plane
// metrics to add:
// node_disk_read_time_seconds_total
// node_disk_reads_completed_total
func QueryDiskReadLatency(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("rate(node_disk_read_time_seconds_total{cluster_id=~\".*%d\"}[2m]) / rate(node_disk_reads_completed_total{cluster_id=~\".*%d\"}[2m])", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

// metrics to add:
// node_disk_write_time_seconds_total
// node_disk_writes_completed_total
func QueryDiskWriteLatency(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("rate(node_disk_write_time_seconds_total{cluster_id=~\".*%d\"}[2m]) / rate(node_disk_writes_completed_total{cluster_id=~\".*%d\"}[2m])", clusterID, clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

// resource
func QueryRURead(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("rate(resource_manager_resource_unit_read_request_unit_sum{cluster_id=~\".*%d\", type=~\"|tp\"}[2m])", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}
func QueryRUWrite(ctx context.Context, clusterID uint64, metricName string) (*client.Response, error) {
	promql := fmt.Sprintf("rate(resource_manager_resource_unit_write_request_unit_sum{cluster_id=~\".*%d\", type=~\"|tp\"}[2m])", clusterID)
	return client.DoInstantQuery(ctx, promql, metricName)
}

func BuildNodeCPUSecondsTotalQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosNodeCPURespResult, err := QueryNodeCPUUsedSeconds(ctx, clusterID, vec.Name)
		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{
					{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeMemoryCapacityQuerier(cluster *core.Cluster, useClusterMetrics bool) 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{},
		}

		thanosNodeMemoryRespResult, err := QueryNodeMemoryTotalBytes(ctx, cluster.Id, vec.Name)
		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
			}

			if useClusterMetrics {
				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},
					},
					Value: model.SampleValue(value),
				})
			} else {
				vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
					Labels: []prom.MetricLabel{
						{Name: datasource.InstanceName, Value: datasource.UnifyDBName(m.Instance)},
						{Name: datasource.ComponentType, Value: datasource.UnifyDBName(m.Component)},
					},
					Value: model.SampleValue(value),
				})
			}
		}

		return &vec, nil
	}

}

func BuildNodeMemoryUsedQuerier(clusterID uint64, clusterName string, useClusterMetrics bool) 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{},
		}

		thanosNodeMemoryRespResult, err := QueryNodeMemoryUsedBytes(ctx, clusterID, vec.Name)
		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
			}

			if useClusterMetrics {
				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: clusterName},
					},
					Value: model.SampleValue(value),
				})
			} else {
				vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
					Labels: []prom.MetricLabel{
						{Name: datasource.InstanceName, Value: datasource.UnifyDBName(m.Instance)},
						{Name: datasource.ComponentType, Value: datasource.UnifyDBName(m.Component)},
					},
					Value: model.SampleValue(value),
				})
			}
		}

		return &vec, nil
	}
}

func BuildNodeStorageCapacityQuerier(cluster *core.Cluster) 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{},
		}
		// var componentCapacityMapper = map[string]uint64{}
		// for _, comp := range cluster.Components {
		// 	componentCapacityMapper[centralutil.CoreComponentType2componentName(comp.ComponentType).String()] = uint64(comp.StorageSizeGi) * 1024 * 1024 * 1024
		// }
		// // We can only get instance from thanos metrics.
		// thanosUsedNodeStorageRespResult, err := QueryNodeStorageUsedBytes(ctx, cluster.Id, vec.Name)
		// if err != nil || thanosUsedNodeStorageRespResult.Status != "success" {
		// 	return &vec, errors.Errorf("failed to query used node storage, err: %v, result: %+v", err, thanosUsedNodeStorageRespResult)
		// }

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

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

		// 	capacityValue := componentCapacityMapper[m.Component]
		// 	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},
		// 		},
		// 		Value: model.SampleValue(capacityValue),
		// 	})
		// }

		// return &vec, nil
		thanosNodeStorageRespResult, err := QueryNodeStorageCapacityBytes(ctx, cluster.Id, vec.Name)
		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},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeStorageUsedBytesQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosNodeStorageRespResult, err := QueryNodeStorageUsedBytes(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeStorageAvailableQuerier(cluster *core.Cluster) 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{},
		}

		thanosNodeStorageRespResult, err := QueryNodeStorageAvailableBytes(ctx, cluster.Id, vec.Name)
		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},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueriesTotalQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBQueriesRespResult, err := QueryDBQueriesTotal(ctx, clusterID, vec.Name)
		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: clusterName},
					{Name: datasource.LabelSQLType, Value: m.Type},
					{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBFailedQueriesTotalQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBFailedQueriesRespResult, err := QueryDBFailedQueriesTotal(ctx, clusterID, vec.Name)
		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: clusterName},
					{Name: datasource.LabelType, Value: m.Type},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueryDurationSecondsQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBQueryDurationRespResult, err := QueryDBQueryDurationSeconds(ctx, clusterID, vec.Name)
		if err != nil || thanosDBQueryDurationRespResult.Status != "success" {
			return nil, errors.Errorf("failed to query db query duration seconds, err: %v, result: %+v", err, thanosDBQueryDurationRespResult)
		}

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

			labels := []prom.MetricLabel{
				{Name: datasource.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
				{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
				{Name: datasource.LabelCluster, Value: clusterName},
				{Name: datasource.LabelSQLType, Value: m.SqlType},
				{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup},
			}

			var t string
			if strings.HasSuffix(m.MetricName, prom.VecMetricElementSum) {
				t = prom.VecMetricElementSum
			} else if strings.HasSuffix(m.MetricName, prom.VecMetricElementCount) {
				t = prom.VecMetricElementCount
			} else if strings.HasSuffix(m.MetricName, prom.VecMetricElementBucket) {
				t = 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:   t,
				Labels: labels,
				Value:  model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBConnectionsQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBConnectionsRespResult, err := QueryDBConnections(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBTotalConnectionQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBTotalConnectionRespResult, err := QueryDBTotalConnection(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBActiveConnectionsQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBActiveConnectionsRespResult, err := QueryDBActiveConnections(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBDisconnectionsQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBDisconnectionsRespResult, err := QueryDBDisconnections(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBDatabaseTimeQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBDatabaseTimeRespResult, err := QueryDBDatabaseTime(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueryPerSecondQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBQueryPerSecondRespResult, err := QueryDBQueryPerSecond(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBFailedQueriesQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBFailedQueriesRespResult, err := QueryDBFailedQueries(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBCommandPerSecondQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBCommandPerSecondRespResult, err := QueryDBCommandPerSecond(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBQueriesUsingPlanCacheOpsQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBQueriesUsingPlanCacheOpsRespResult, err := QueryDBQueriesUsingPlanCacheOps(ctx, clusterID, vec.Name)
		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.LabelInstance, Value: datasource.UnifyDBName(m.Instance)},
					{Name: datasource.LabelComponent, Value: datasource.UnifyDBName(m.Component)},
					{Name: datasource.LabelCluster, Value: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBAverageQueryDurationQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBAverageQueryDurationRespResult, err := QueryDBAverageQueryDuration(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildDBTransactionPerSecondQuerier(clusterID uint64, clusterName string) 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{},
		}

		thanosDBTransactionPerSecondRespResult, err := QueryDBTransactionPerSecond(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}

		return &vec, nil
	}
}

func BuildNodeCPUCapacityCoresQuerier(cluster *core.Cluster) 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{},
		}
		var componentCapacityMapper = map[string]float64{}
		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)
		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 := componentCapacityMapper[m.Component]
			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},
				},
				Value: model.SampleValue(capacityValue),
			})
		}

		return &vec, nil
	}
}

func BuildTiCDCLatencyQuerier(clusterID uint64, clusterName string) 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{},
		}

		ticdcLatencyRespResult, err := QueryTiCDCLatencySeconds(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildTiCDCCheckpointTSQuerier(clusterID uint64, clusterName string) 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{},
		}

		ticdcCheckpointTSRespResult, err := QueryTiCDCCheckpointTS(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildChangefeedStatusQuerier(clusterID uint64, clusterName string) 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{},
		}

		ticdcCheckpointTSRespResult, err := QueryChangefeedStatus(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildTiCDCReplicaRowsQuerier(clusterID uint64, clusterName string) 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{},
		}

		ticdcReplicaRowsPerSecondRespResult, err := QueryTiCDCReplicaRowsPerSecond(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildKVRequestDurationQuerier(cluster *core.Cluster, useClusterMetrics bool) 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:    "histogram",
			Metrics: []*prom.VecMetricElement{},
		}

		kvRequestRespResult, err := QueryKVRequestDuration(ctx, cluster.Id, vec.Name)
		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
			}

			if useClusterMetrics {
				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.LabelType, Value: m.Type},
					},
					Value: model.SampleValue(value),
				})
			} else {
				vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
					Labels: []prom.MetricLabel{
						{Name: datasource.InstanceName, Value: datasource.UnifyDBName(m.Instance)},
						{Name: datasource.ComponentType, Value: datasource.UnifyDBName(m.Component)},
					},
					Value: model.SampleValue(value),
				})
			}
		}

		return &vec, nil
	}

}

func BuildUptimeQuerier(cluster *core.Cluster, useClusterMetrics bool) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:    "tidbcloud_component_uptime",
			Help:    "The uptime of tidb components",
			Type:    "histogram",
			Metrics: []*prom.VecMetricElement{},
		}

		uptimeRespResult, err := QueryUptime(ctx, cluster.Id, vec.Name)
		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
			}

			if useClusterMetrics {
				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},
					},
					Value: model.SampleValue(value),
				})
			} else {
				vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
					Labels: []prom.MetricLabel{
						{Name: datasource.InstanceName, Value: datasource.UnifyDBName(m.Instance)},
						{Name: datasource.ComponentType, Value: datasource.UnifyDBName(m.Component)},
					},
					Value: model.SampleValue(value),
				})
			}
		}

		return &vec, nil
	}

}

func BuildDiskReadLatencyQuerier(cluster *core.Cluster, useClusterMetrics bool) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:    "tidbcloud_disk_read_latency",
			Help:    "The read latency",
			Type:    "histogram",
			Metrics: []*prom.VecMetricElement{},
		}

		latencyRespResult, err := QueryDiskReadLatency(ctx, cluster.Id, vec.Name)
		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
			}

			if useClusterMetrics {
				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.LabelDevice, Value: m.Device},
					},
					Value: model.SampleValue(value),
				})
			} else {
				vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
					Labels: []prom.MetricLabel{
						{Name: datasource.InstanceName, Value: datasource.UnifyDBName(m.Instance)},
						{Name: datasource.ComponentType, Value: datasource.UnifyDBName(m.Component)},
						{Name: datasource.LabelDevice, Value: m.Device},
					},
					Value: model.SampleValue(value),
				})
			}
		}

		return &vec, nil
	}

}

func BuildDiskWriteLatencyQuerier(cluster *core.Cluster, useClusterMetrics bool) datasource.MetricQuerierFunc {
	return func(ctx context.Context) (prom.Metric, error) {
		vec := prom.VecMetric{
			Name:    "tidbcloud_disk_write_latency",
			Help:    "The write latency",
			Type:    "histogram",
			Metrics: []*prom.VecMetricElement{},
		}

		latencyRespResult, err := QueryDiskWriteLatency(ctx, cluster.Id, vec.Name)
		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
			}

			if useClusterMetrics {
				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.LabelDevice, Value: m.Device},
					},
					Value: model.SampleValue(value),
				})
			} else {
				vec.Metrics = append(vec.Metrics, &prom.VecMetricElement{
					Labels: []prom.MetricLabel{
						{Name: datasource.InstanceName, Value: datasource.UnifyDBName(m.Instance)},
						{Name: datasource.ComponentType, Value: datasource.UnifyDBName(m.Component)},
						{Name: datasource.LabelDevice, Value: m.Device},
					},
					Value: model.SampleValue(value),
				})
			}
		}

		return &vec, nil
	}

}

func BuildTiCDCChangefeedResolvedTsLagQuerier(clusterID uint64, clusterName string) 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{},
		}

		ticdcOwnerResolvedTsRespResult, err := QueryTiCDCResolvedTS(ctx, clusterID, vec.Name)
		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: clusterName},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildRUReadQuerier(clusterID uint64, clusterName string) 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{},
		}

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

		for _, result := range ticdcProcessorResolvedTsRespResult.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: clusterName},
					{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}

func BuildRUWriteQuerier(clusterID uint64, clusterName string) 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{},
		}

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

		for _, result := range ticdcProcessorResolvedTsRespResult.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: clusterName},
					{Name: datasource.LabelResourceGroup, Value: m.ResourceGroup},
				},
				Value: model.SampleValue(value),
			})
		}
		return &vec, nil
	}
}