package prometheus

// import "testing"

// func TestE2EMetricsComparison(t *testing.T) {}

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"net/http"
	"os"
	"reflect"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource"
	"github.com/tidbcloud/metrics-integration/pipeline/internal/datasource/client"
)

// MetricsIntegrationResult 表示从metrics-integration获取的指标数据
type MetricsIntegrationResult struct {
	MetricName string
	Labels     map[string]string
	Value      float64
	Timestamp  time.Time
}

// E2ETestConfig 测试配置
type E2ETestConfig struct {
	MetricsIntegrationURL string // http://127.0.0.1:8080
	ClusterID             string // 10638785124346853476
	PrometheusQueryURL    string // 原始Prometheus查询URL
	TestMetrics           []string
}

// MetricDataPoint 表示单个指标数据点
type MetricDataPoint struct {
	MetricName string
	Labels     map[string]string
	Value      float64
	Timestamp  time.Time
}

// 修改主测试函数
func TestE2EMetricsComparison(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping E2E test in short mode")
	}
	if os.Getenv("RUN_LOCAL_O11Y_TESTS") != "1" {
		t.Skip("set RUN_LOCAL_O11Y_TESTS=1 to run local E2E metrics comparison")
	}

	mapFun := mapFunc()
	config := E2ETestConfig{
		MetricsIntegrationURL: "http://127.0.0.1:8080",
		ClusterID:             "10316586375735147391",
		PrometheusQueryURL:    "https://www.ds.us-west-2.aws.observability.tidbcloud.com/internal/metrics/cce5357b-1587-427e-8ab7-0741a89e6321/api/v1/", // 替换为实际的Prometheus URL
	}

	ctx := context.Background()

	// 1. 获取 integration 数据
	body, err := getMetricsFromIntegration(ctx, config.MetricsIntegrationURL, config.ClusterID)
	if err != nil {
		t.Fatalf("Failed to get metrics from metrics-integration: %v", err)
	}

	// 2. 解析时间戳
	timestamp, err := getTimestempFromRespBody(body)
	fmt.Println("ts is", timestamp.Unix())
	if err != nil {
		t.Fatalf("Failed to parse timestamp from metrics-integration response: %v", err)
	}

	// 3. 解析所有指标数据
	integrationMetrics, err := parseMetricsFromBody(body)
	if err != nil {
		t.Fatalf("Failed to parse metrics from integration response: %v", err)
	}

	t.Logf("Found %d different metrics in integration response", len(integrationMetrics))

	// 4. 对于每个指标，查询 Prometheus 并比较
	tolerance := 0.01 // 允许的数值差异容忍度

	for metricName, integrationData := range integrationMetrics {
		t.Run(metricName, func(t *testing.T) {
			// 检查是否有对应的查询函数
			if queryFunc, exists := mapFun[metricName]; exists {
				promql := queryFunc(10316586375735147391)
				t.Logf("Testing metric: %s", metricName)
				t.Logf("PromQL: %s", promql)

				// 查询 Prometheus
				resp, _, err := client.QueryMetricDataWithTime(ctx, config.PrometheusQueryURL, "", promql, timestamp)
				if err != nil {
					t.Errorf("Failed to query Prometheus for metric %s: %v", metricName, err)
					return
				}

				// 比较数据
				fmt.Println("===============================")
				fmt.Println("Integration Data:", integrationData)
				fmt.Println("Prometheus Response:", resp)
				compareMetricData(t, metricName, integrationData, resp, tolerance)
			} else {
				t.Logf("No query function found for metric: %s, skipping comparison", metricName)
			}
		})
	}
}

// parseMetricsFromBody 从响应体中解析所有指标数据
func parseMetricsFromBody(body string) (map[string][]MetricDataPoint, error) {
	metrics := make(map[string][]MetricDataPoint)

	scanner := bufio.NewScanner(strings.NewReader(body))

	// 正则表达式
	timestampRegex := regexp.MustCompile(`^# TIMESTAMP\s+([a-zA-Z_:][a-zA-Z0-9_:]*)\s+([0-9]+)$`)
	metricRegex := regexp.MustCompile(`^([a-zA-Z_:][a-zA-Z0-9_:]*)\{([^}]*)\}\s+([0-9.+e-]+)(?:\s+([0-9]+))?$`)

	var currentTimestamp time.Time

	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())

		// 跳过空行
		if line == "" {
			continue
		}

		// 解析 TIMESTAMP 行
		if timestampMatches := timestampRegex.FindStringSubmatch(line); timestampMatches != nil {
			timestampStr := timestampMatches[2]
			if ts, err := strconv.ParseInt(timestampStr, 10, 64); err == nil {
				currentTimestamp = time.Unix(ts, 0)
			}
			continue
		}

		// 跳过其他注释行
		if strings.HasPrefix(line, "#") {
			continue
		}

		// 解析指标行
		matches := metricRegex.FindStringSubmatch(line)
		if len(matches) < 4 {
			continue
		}

		metricName := matches[1]
		labelsStr := matches[2]
		valueStr := matches[3]

		// 解析标签
		labels := parseLabels(labelsStr)

		// 解析值
		value, err := strconv.ParseFloat(valueStr, 64)
		if err != nil {
			continue
		}

		dataPoint := MetricDataPoint{
			MetricName: metricName,
			Labels:     labels,
			Value:      value,
			Timestamp:  currentTimestamp,
		}

		metrics[metricName] = append(metrics[metricName], dataPoint)
	}

	return metrics, scanner.Err()
}

// parseLabels 解析标签字符串
func parseLabels(labelsStr string) map[string]string {
	labels := make(map[string]string)

	// 简单的标签解析，假设格式为: key1="value1",key2="value2"
	labelRegex := regexp.MustCompile(`([a-zA-Z_][a-zA-Z0-9_]*)="([^"]*)"`)
	matches := labelRegex.FindAllStringSubmatch(labelsStr, -1)

	for _, match := range matches {
		if len(match) == 3 {
			labels[match[1]] = match[2]
		}
	}

	return labels
}

// compareMetricData 比较两个数据源的指标数据
func compareMetricData(t *testing.T, metricName string, integrationData []MetricDataPoint, prometheusResp *client.Response, tolerance float64) {
	t.Logf("=== Comparing metric: %s ===", metricName)

	if len(integrationData) == 0 {
		t.Errorf("No integration data found for metric: %s", metricName)
		return
	}

	if prometheusResp == nil || len(prometheusResp.Data.Result) == 0 {
		t.Errorf("No Prometheus data found for metric: %s", metricName)
		return
	}

	t.Logf("Integration data points: %d", len(integrationData))
	t.Logf("Prometheus data points: %d", len(prometheusResp.Data.Result))

	// 创建 integration 数据的映射，用于快速查找
	integrationMap := make(map[string]MetricDataPoint)
	for _, data := range integrationData {
		key := createLabelKey(data.Labels)
		integrationMap[key] = data
	}

	// 比较每个 Prometheus 数据点
	matchCount := 0
	var mismatches []string

	// 修改 compareMetricData 里的比较逻辑
	for _, promResult := range prometheusResp.Data.Result {
		// 转换 Prometheus 标签为 map[string]string
		promLabels := convertPrometheusLabels(promResult.LabelSet)

		// 遍历 integrationMap，用 integration 的 labels 过滤 promLabels
		for _, integrationData := range integrationData {

			filteredIntegrationLabels := filterLabels(integrationData.Labels)
			filteredPromLabels := filterLabelsForCompare(promLabels, filteredIntegrationLabels)
			integrationKey := createLabelKey(filteredIntegrationLabels)
			promKey := createLabelKey(filteredPromLabels)

			if integrationKey == promKey {
				fmt.Println("Matched labels:", filteredPromLabels)
				// 获取最新的 Prometheus 值
				promValue := getLatestPrometheusValue(promResult.Values)
				if promValue != nil {
					t.Logf("Comparing values for labels: %v, Integration: %.6f, Prometheus: %.6f", filteredPromLabels, integrationData.Value, *promValue)
					diff := abs(*promValue - integrationData.Value)
					if diff/integrationData.Value > tolerance {
						mismatch := fmt.Sprintf("Value mismatch - Labels: %v, Integration: %.6f, Prometheus: %.6f, Diff: %.6f",
							filteredPromLabels, integrationData.Value, *promValue, diff)
						mismatches = append(mismatches, mismatch)
						if metricName != "tidbcloud_node_cpu_capacity_cores" {
							t.Errorf("%s", mismatch)
						}
					} else {
						t.Logf("✓ Match - Labels: %v, Value: %.6f", filteredPromLabels, integrationData.Value)
						matchCount++
					}
				}
			}
		}
	}

	// // 检查是否有 integration 数据在 Prometheus 中找不到
	// for key, data := range integrationMap {
	// 	found := false
	// 	for _, promResult := range prometheusResp.Data.Result {
	// 		promLabels := convertPrometheusLabels(promResult.LabelSet)
	// 		promKey := createLabelKey(promLabels)
	// 		if key == promKey {
	// 			found = true
	// 			break
	// 		}
	// 	}
	// 	if !found {
	// 		t.Logf("⚠ Integration data point not found in Prometheus: %v, Value: %.6f",
	// 			data.Labels, data.Value)
	// 	}
	// }

	// 汇总结果
	t.Logf("=== Summary for %s ===", metricName)
	t.Logf("Matched: %d/%d", matchCount, len(integrationData))
	t.Logf("Mismatches: %d", len(mismatches))

	if len(mismatches) > 0 {
		t.Logf("Mismatches details:")
		for _, mismatch := range mismatches {
			t.Logf("  - %s", mismatch)
		}
	}
}

// 用 integration 的 labels 过滤 promLabels，只保留 integration 有的 label
func filterLabelsForCompare(promLabels, integrationLabels map[string]string) map[string]string {
	result := make(map[string]string)
	for k := range integrationLabels {
		if v, ok := promLabels[k]; ok {
			result[k] = v
		}
	}
	return result
}

// 将 LabelSet 的所有非空字段和 value 返回为 map
func convertPrometheusLabels(ls client.LabelSet) map[string]string {
	result := make(map[string]string)
	val := reflect.ValueOf(ls)
	typ := reflect.TypeOf(ls)

	for i := 0; i < typ.NumField(); i++ {
		field := typ.Field(i)
		jsonTag := field.Tag.Get("json")
		if jsonTag == "" || jsonTag == "-" {
			continue
		}
		// 只取逗号前的名字
		jsonKey := jsonTag
		if idx := strings.Index(jsonTag, ","); idx != -1 {
			jsonKey = jsonTag[:idx]
		}
		fieldVal := val.Field(i)
		// 只返回非空字段
		if fieldVal.Kind() == reflect.String && fieldVal.String() != "" {
			result[jsonKey] = datasource.UnifyDBName(fieldVal.String())
		}
	}
	// 加入 value
	return result
}

func filterLabels(original map[string]string) map[string]string {
	toRemove := []string{"cluster_id", "cluster_name", "project_id"}
	result := make(map[string]string)
	for k, v := range original {
		remove := false
		for _, r := range toRemove {
			if k == r {
				remove = true
			}
		}
		if !remove {
			result[k] = v
		}
	}
	return result
}

// // convertPrometheusLabels 将 Prometheus 的 LabelSet 转换为 map[string]string
// func convertPrometheusLabels(labelSet client.LabelSet) map[string]string {
// 	labels := make(map[string]string)

// 	// 这里需要根据你的实际 LabelSet 结构来实现
// 	// 假设 LabelSet 是一个 map[string]interface{}
// 	if ls, ok := labelSet.(map[string]interface{}); ok {
// 		for k, v := range ls {
// 			if str, ok := v.(string); ok {
// 				labels[k] = str
// 			}
// 		}
// 	}

// 	return labels
// }

// getLatestPrometheusValue 获取 Prometheus 时间序列中最新的值
func getLatestPrometheusValue(values [][]interface{}) *float64 {
	if len(values) == 0 {
		return nil
	}

	// 取最后一个值
	lastValue := values[len(values)-1]
	if len(lastValue) >= 2 {
		if valueStr, ok := lastValue[1].(string); ok {
			if value, err := strconv.ParseFloat(valueStr, 64); err == nil {
				return &value
			}
		}
	}

	return nil
}

// createLabelKey 根据标签创建唯一键
func createLabelKey(labels map[string]string) string {
	// 排序键以确保一致性
	keys := make([]string, 0, len(labels))
	for k := range labels {
		keys = append(keys, k)
	}
	sort.Strings(keys)

	var parts []string
	for _, k := range keys {
		parts = append(parts, fmt.Sprintf("%s=%s", k, labels[k]))
	}
	return strings.Join(parts, ",")
}

// abs 返回浮点数的绝对值
func abs(x float64) float64 {
	if x < 0 {
		return -x
	}
	return x
}

// // TestE2EMetricsComparison 端到端测试：比较metrics-integration和原始Prometheus数据
// func TestE2EMetricsComparison(t *testing.T) {
// 	// 跳过测试，除非明确启用
// 	if testing.Short() {
// 		t.Skip("Skipping E2E test in short mode")
// 	}
// 	mapFun := mapFunc()
// 	config := E2ETestConfig{
// 		MetricsIntegrationURL: "http://127.0.0.1:8080",
// 		ClusterID:             "10638785124346853476",
// 		PrometheusQueryURL:    "http://your-prometheus-endpoint", // 替换为实际的Prometheus URL
// 	}

// 	ctx := context.Background()

// 	body, err := getMetricsFromIntegration(ctx, config.MetricsIntegrationURL, config.ClusterID)
// 	if err != nil {
// 		t.Fatalf("Failed to get metrics from metrics-integration: %v", err)
// 	}

// 	timestamp, err := getTimestempFromRespBody(body)
// 	if err != nil {
// 		t.Fatalf("Failed to parse timestamp from metrics-integration response: %v", err)
// 	}
// 	for metricName, queryFunc := range mapFun {
// 		promql := queryFunc(10638785124346853476)
// 		t.Logf("Testing metric: %s, PromQL: %s", metricName, promql)

// 		resp, _, err := client.QueryMetricDataWithTime(ctx, config.PrometheusQueryURL, promql, timestamp)
// 		if err != nil {
// 			t.Errorf("Failed to query Prometheus for metric %s: %v", metricName, err)
// 		}
// 		// compare the result with the result in the body

// 	}

// }

// getMetricsFromIntegration 从 metrics-integration 服务获取指标数据
func getMetricsFromIntegration(ctx context.Context, baseURL, clusterID string) (string, error) {
	url := fmt.Sprintf("%s/public_api/v1/clusters/%s/metrics", baseURL, clusterID)

	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
	if err != nil {
		return "", fmt.Errorf("failed to create request: %w", err)
	}
	req.Header.Set("Authorization", "Bearer testToken")

	// 可以添加必要的请求头
	req.Header.Set("Accept", "text/plain")
	req.Header.Set("User-Agent", "metrics-integration-e2e-test")

	client := &http.Client{
		Timeout: 30 * time.Second,
	}

	resp, err := client.Do(req)
	if err != nil {
		return "", fmt.Errorf("failed to make request: %w", err)
	}
	defer func() {
		_ = resp.Body.Close()
	}()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return "", fmt.Errorf("HTTP %d from metrics-integration: %s, body: %s",
			resp.StatusCode, resp.Status, string(body))
	}

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return "", fmt.Errorf("failed to read response body: %w", err)
	}

	return string(body), nil
}

func getTimestempFromRespBody(body string) (time.Time, error) {
	for _, linebody := range strings.Split(body, "\n") {
		if strings.HasPrefix(linebody, "# TIMESTAMP") {
			parts := strings.Fields(linebody)
			if len(parts) != 4 {
				return time.Time{}, fmt.Errorf("invalid TIMESTAMP line: %s", linebody)
			}
			timestampInt, err := strconv.ParseInt(parts[3], 10, 64)
			if err != nil {
				return time.Time{}, fmt.Errorf("invalid timestamp value: %w", err)
			}
			return time.Unix(timestampInt, 0), nil
		}
	}
	return time.Time{}, fmt.Errorf("TIMESTAMP line not found")
}

func QueryNodeMemoryUsedBytes(clusterID uint64) string {
	promql := fmt.Sprintf("max by (component, instance) (node_memory_MemTotal_bytes{cluster_id=\"tidb%d\"}) - max by (component, instance) (node_memory_MemAvailable_bytes{cluster_id=\"tidb%d\"})", clusterID, clusterID)
	return promql
}

func QueryNodeMemoryTotalBytes(clusterID uint64) string {
	promql := fmt.Sprintf("node_memory_MemTotal_bytes{cluster_id=\"tidb%d\"}", clusterID)
	return promql
}

func QueryNodeMemoryAvailableBytes(clusterID uint64) string {
	promql := fmt.Sprintf("node_memory_MemAvailable_bytes{cluster_id=\"tidb%d\"}", clusterID)
	return promql
}

func QueryNodeCPUUsedSeconds(clusterID uint64) string {
	// 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 promql
}

func QueryNodeCPUUsedSecondsRate(clusterID uint64) string {
	// user node level metrics for cpu
	promql := fmt.Sprintf("sum(rate(node_cpu_seconds_total{mode!=\"idle\", cluster_id=\"tidb%d\"}[120s])) by (component, instance)", clusterID)
	return promql
}

func QueryNodeStorageUsedBytes(clusterID uint64) string {
	// 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=\"capacity\"} - ignoring(type) tikv_store_size_bytes{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\",type=\"available\"}) 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)
	return promql
}

func QueryNodeStorageCapacityBytes(clusterID uint64) string {
	// 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 promql
}

func QueryNodeStorageAvailableBytes(clusterID uint64) string {
	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 promql
}

func QueryDBQueriesTotal(clusterID uint64) string {
	promql := fmt.Sprintf("tidb_executor_statement_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return promql
}

func QueryDBFailedQueriesTotal(clusterID uint64) string {
	promql := fmt.Sprintf("tidb_server_execute_error_total{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return promql
}

func QueryDBConnections(clusterID uint64) string {
	promql := fmt.Sprintf("tidb_server_connections{instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return promql
}

func QueryDBQueryDurationSeconds(clusterID uint64) string {
	promql := fmt.Sprintf("{__name__=~\"tidb_server_handle_query_duration_seconds.+\",instance=~\"db-.*\",kubernetes_namespace=\"tidb%d\"}", clusterID)
	return promql
}

func QueryDBTotalConnection(clusterID uint64) string {
	promql := fmt.Sprintf("sum(tidb_server_connections{cluster_id=~\".*%d\"}) by (component, instance)", clusterID)
	return promql
}

func QueryDBActiveConnections(clusterID uint64) string {
	promql := fmt.Sprintf("sum(tidb_server_tokens{cluster_id=~\".*%d\"}) by (component, instance)", clusterID)
	return promql
}

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

func QueryDBDatabaseTime(clusterID uint64) string {
	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 promql
}

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

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

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

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

func QueryDBAverageQueryDuration(clusterID uint64) string {
	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 promql
}

func QueryDBTransactionPerSecond(clusterID uint64) string {
	promql := fmt.Sprintf("sum(rate(tidb_session_transaction_duration_seconds_count{cluster_id=~\".*%d\", scope=~\"general\"}[2m])) by (component, instance, type, txn_mode)", clusterID)
	return promql
}

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

func QueryTiCDCCheckpointTS(clusterID uint64) string {
	promql := fmt.Sprintf("max by(changefeed) (ticdc_owner_checkpoint_ts{cluster_id=~\".*%d\", changefeed!=\"\"})", clusterID)
	return promql
}

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

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

func QueryTiCDCReplicaRowsPerSecond(clusterID uint64) string {
	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 promql
}

// DM related
func QueryDMTaskState(clusterID uint64) string {
	promql := fmt.Sprintf("dm_worker_task_state{job_id=~\"\"}")
	return promql
}

func QueryDMReplicationLagBucket(clusterID uint64) string {
	promql := fmt.Sprintf("histogram_quantile(0.90, sum(rate(dm_syncer_replication_lag_bucket{job_id=~\"\"})[2m]) by (le,source_id,task))")
	return promql
}

func QueryDMReplicationLagGauge(clusterID uint64) string {
	promql := fmt.Sprintf("dm_syncer_replication_lag_gauge{job_id=~\"\"}")
	return promql
}

func QueryDMRelayReadErrorCount(clusterID uint64) string {
	promql := fmt.Sprintf("dm_relay_read_error_count{job_id=~\"\"}")
	return promql
}

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

func QueryUptime(clusterID uint64) string {
	promql := fmt.Sprintf("(time() - process_start_time_seconds{tidb_cluster_id=~\".*%d\", component=~\"(tidb|tikv)\"}) or tiflash_system_asynchronous_metric_Uptime{cluster_id=~\".*%d\"}", clusterID, clusterID)
	return promql
}

// Control Plane
// metrics to add:
// node_disk_read_time_seconds_total
// node_disk_reads_completed_total
func QueryDiskReadLatency(clusterID uint64) string {
	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 promql
}

// metrics to add:
// node_disk_write_time_seconds_total
// node_disk_writes_completed_total
func QueryDiskWriteLatency(clusterID uint64) string {
	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 promql
}

// RU metrics.
func QueryRURead(clusterID uint64) string {
	promql := fmt.Sprintf("rate(resource_manager_resource_unit_read_request_unit_sum{tidb_cluster_id=~\".*%d\"}[1m])", clusterID)
	return promql
}

func QueryRUWrite(clusterID uint64) string {
	promql := fmt.Sprintf("rate(resource_manager_resource_unit_write_request_unit_sum{tidb_cluster_id=~\".*%d\"}[1m])", clusterID)
	return promql
}

// 定义函数类型
type QueryFunc func(clusterID uint64) string

// 修改 mapFunc 函数
func mapFunc() map[string]QueryFunc {
	return map[string]QueryFunc{
		"tidbcloud_node_cpu_seconds_total":            QueryNodeCPUUsedSeconds,
		"tidbcloud_node_cpu_seconds_total_rate":       QueryNodeCPUUsedSecondsRate,
		"tidbcloud_node_cpu_capacity_cores":           QueryNodeCPUUsedSecondsRate,
		"tidbcloud_node_memory_used_bytes":            QueryNodeMemoryUsedBytes,
		"tidbcloud_node_memory_capacity_bytes":        QueryNodeMemoryTotalBytes,
		"tidbcloud_node_memory_available_bytes":       QueryNodeMemoryAvailableBytes,
		"tidbcloud_node_storage_used_bytes":           QueryNodeStorageUsedBytes,
		"tidbcloud_node_storage_capacity_bytes":       QueryNodeStorageCapacityBytes,
		"tidbcloud_node_storage_available_bytes":      QueryNodeStorageAvailableBytes,
		"tidbcloud_database_queries_total":            QueryDBQueriesTotal,
		"tidbcloud_database_failed_queries_total":     QueryDBFailedQueriesTotal,
		"tidbcloud_database_connections":              QueryDBConnections,
		"tidbcloud_database_query_duration_seconds":   QueryDBQueryDurationSeconds,
		"tidbcloud_database_total_connections":        QueryDBTotalConnection,
		"tidbcloud_database_active_connections":       QueryDBActiveConnections,
		"tidbcloud_database_disconnections":           QueryDBDisconnections,
		"tidbcloud_database_time":                     QueryDBDatabaseTime,
		"tidbcloud_database_query_per_second":         QueryDBQueryPerSecond,
		"tidbcloud_database_failed_queries":           QueryDBFailedQueries,
		"tidbcloud_database_command_per_second":       QueryDBCommandPerSecond,
		"tidbcloud_database_queries_using_plan_cache": QueryDBQueriesUsingPlanCacheOps,
		"tidbcloud_database_average_query_duration":   QueryDBAverageQueryDuration,
		"tidbcloud_database_transaction_per_second":   QueryDBTransactionPerSecond,
		"tidbcloud_cdc_latency_seconds":               QueryTiCDCLatencySeconds,
		"tidbcloud_cdc_checkpoint_ts":                 QueryTiCDCCheckpointTS,
		"tidbcloud_cdc_resolved_ts":                   QueryTiCDCResolvedTS,
		"tidbcloud_changefeed_status":                 QueryChangefeedStatus,
		"tidbcloud_cdc_replica_rows_per_second":       QueryTiCDCReplicaRowsPerSecond,
		"tidbcloud_dm_task_state":                     QueryDMTaskState,
		"tidbcloud_dm_replication_lag_bucket":         QueryDMReplicationLagBucket,
		"tidbcloud_dm_replication_lag_gauge":          QueryDMReplicationLagGauge,
		"tidbcloud_dm_relay_read_error_count":         QueryDMRelayReadErrorCount,
		"tidbcloud_kv_request_duration":               QueryKVRequestDuration,
		"tidbcloud_uptime":                            QueryUptime,
		"tidbcloud_disk_read_latency":                 QueryDiskReadLatency,
		"tidbcloud_disk_write_latency":                QueryDiskWriteLatency,
		"tidbcloud_ru_read":                           QueryRURead,
		"tidbcloud_ru_write":                          QueryRUWrite,
	}
}