package resources

import (
	"strings"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	dto "github.com/prometheus/client_model/go"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"sigs.k8s.io/yaml"

	gkestatus "github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous/status"
)

const testScaleWorkflowName = "scale-workflow"

func TestGKEHeterogeneousStatusCollectorProductionEntrypoints(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:                        "cluster-a",
		Namespace:                        "namespace-a",
		TenantID:                         "tenant-a",
		ProjectID:                        "project-a",
		LastPlacementProbeTime:           testTimePtr(time.Unix(123, 0)),
		LastSuccessfulPlacementProbeTime: testTimePtr(time.Unix(123, 0)),
		ProbeStatus:                      gkestatus.ProbeStatusSuccess,
		ClusterFullSynced:                true,
		Enabled:                          true,
		ValidIntent:                      true,
	}
	resource := gkeHeterogeneousStatus{}
	obj := statusConfigMapUnstructured(t, status)

	if resource.FilterOut(obj) {
		t.Fatalf("FilterOut() = true for status ConfigMap")
	}
	wrongPrefix := statusConfigMapUnstructured(t, status)
	wrongPrefix.SetName("ordinary-configmap")
	if !resource.FilterOut(wrongPrefix) {
		t.Fatalf("FilterOut() = false for non-status ConfigMap")
	}
	wrongDataKey := statusConfigMapUnstructured(t, status)
	wrongDataKey.Object["data"] = map[string]interface{}{"wrong-key": "invalid"}
	if metric := resource.Extract(wrongDataKey); metric != nil {
		t.Fatalf("Extract() = %v, want nil for missing status data key", metric)
	}

	metric := resource.Extract(obj)
	if metric == nil {
		t.Fatalf("Extract() = nil, want placement probe timestamp metric")
	}
	if got := metricGaugeValue(t, metric); got != 123 {
		t.Fatalf("placement probe timestamp = %v, want 123", got)
	}
	labels := metricLabels(t, metric)
	if labels["cluster_id"] != "cluster-a" || labels["probe_status"] != gkestatus.ProbeStatusSuccess {
		t.Fatalf("placement probe timestamp labels = %v", labels)
	}

	collector := newCollector(resource)
	descs := collectDescs(collector)
	if len(descs) != 1+len(resource.ExtraDescs()) {
		t.Fatalf("descriptors = %d, want %d", len(descs), 1+len(resource.ExtraDescs()))
	}
	assertDescContains(t, descs, "gke_heterogeneous_placement_probe_timestamp_seconds")
	assertDescContains(t, descs, "gke_heterogeneous_successful_placement_probe_timestamp_seconds")
	assertDescContains(t, descs, "gke_heterogeneous_cluster_full_synced")
	assertDescContains(t, descs, "gke_heterogeneous_nodes_total")
	assertDescContains(t, descs, "gke_heterogeneous_component_nodes_total")
	assertDescContains(t, descs, "gke_heterogeneous_scale_workflow_duration_seconds")
	assertDescContains(t, descs, "gke_heterogeneous_node_provision_duration_seconds")
	assertDescNotContains(t, descs, "gke_heterogeneous_pod_placement_info")

	collector.OnAdd(wrongPrefix)
	if metrics := collectCollectorMetrics(collector); len(metrics) != 0 {
		t.Fatalf("collector emitted %d metrics for filtered ConfigMap, want 0", len(metrics))
	}
	collector.OnAdd(wrongDataKey)
	if metrics := collectCollectorMetrics(collector); len(metrics) != 0 {
		t.Fatalf("collector emitted %d metrics for ConfigMap with missing status data, want 0", len(metrics))
	}
	collector.OnAdd(obj)
	metrics := collectCollectorMetrics(collector)
	if len(metrics) != 6 {
		t.Fatalf("collector emitted %d metrics, want 6", len(metrics))
	}
	if got := metricGaugeValue(t, findMetricByGaugeValue(t, metrics, 123)); got != 123 {
		t.Fatalf("collected placement probe timestamp = %v, want 123", got)
	}
}

func TestGKEHeterogeneousStatusCollectorRegisteredForGKE(t *testing.T) {
	collector := findGKEHeterogeneousStatusCollector(t, GetAllCollectors("gke"))
	status := gkestatus.Snapshot{
		ClusterID:                        "cluster-a",
		Namespace:                        "namespace-a",
		TenantID:                         "tenant-a",
		ProjectID:                        "project-a",
		LastPlacementProbeTime:           testTimePtr(time.Unix(123, 0)),
		LastSuccessfulPlacementProbeTime: testTimePtr(time.Unix(123, 0)),
		ProbeStatus:                      gkestatus.ProbeStatusSuccess,
		ClusterFullSynced:                true,
		Enabled:                          true,
		ValidIntent:                      true,
	}
	collector.OnAdd(statusConfigMapUnstructured(t, status))

	registry := prometheus.NewRegistry()
	registry.MustRegister(collector)
	families, err := registry.Gather()
	if err != nil {
		t.Fatalf("gather metrics: %v", err)
	}

	assertMetricFamily(t, families, "infra_exporter_gke_heterogeneous_placement_probe_timestamp_seconds")
	assertMetricFamily(t, families, "infra_exporter_gke_heterogeneous_successful_placement_probe_timestamp_seconds")
	assertMetricFamily(t, families, "infra_exporter_gke_heterogeneous_enabled")
	assertMetricFamily(t, families, "infra_exporter_gke_heterogeneous_valid_intent")
	assertMetricFamily(t, families, "infra_exporter_gke_heterogeneous_status_probe_success")
	assertMetricFamily(t, families, "infra_exporter_gke_heterogeneous_cluster_full_synced")
}

func TestGKEHeterogeneousStatusTimestampMetricsUsePlacementTimes(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:                        "cluster-a",
		Namespace:                        "namespace-a",
		TenantID:                         "tenant-a",
		ProjectID:                        "project-a",
		LastPlacementProbeTime:           testTimePtr(time.Unix(200, 0)),
		LastSuccessfulPlacementProbeTime: testTimePtr(time.Unix(100, 0)),
		ProbeStatus:                      gkestatus.ProbeStatusFailed,
		ClusterFullSynced:                true,
		Enabled:                          true,
		ValidIntent:                      true,
	}
	resource := gkeHeterogeneousStatus{}
	obj := statusConfigMapUnstructured(t, status)

	placementMetric := resource.Extract(obj)
	if placementMetric == nil {
		t.Fatalf("Extract() = nil, want attempted placement timestamp metric")
	}
	if got := metricGaugeValue(t, placementMetric); got != 200 {
		t.Fatalf("attempted placement timestamp = %v, want 200", got)
	}

	metrics := resource.ExtractMore(obj)
	successfulMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_successful_placement_probe_timestamp_seconds")
	if got := metricGaugeValue(t, successfulMetric); got != 100 {
		t.Fatalf("successful placement timestamp = %v, want 100", got)
	}

	disabled := gkestatus.Snapshot{
		ClusterID:   "cluster-a",
		Namespace:   "namespace-a",
		TenantID:    "tenant-a",
		ProjectID:   "project-a",
		ProbeStatus: gkestatus.ProbeStatusDisabled,
		Enabled:     false,
		ValidIntent: true,
	}
	if metric := resource.Extract(statusConfigMapUnstructured(t, disabled)); metric != nil {
		t.Fatalf("Extract() = %v, want nil without placement probe time", metric)
	}
}

func TestGKEHeterogeneousStatusMetricSkipsPlacementOnFailedProbe(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:         "cluster-a",
		Namespace:         "namespace-a",
		TenantID:          "tenant-a",
		ProjectID:         "project-a",
		ClusterGeneration: 12,
		ProbeStatus:       gkestatus.ProbeStatusFailed,
		ProbeError:        "list pods failed",
		Enabled:           true,
		ValidIntent:       true,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType: "tikv",
				Pods: []gkestatus.PodPlacement{
					{
						Pod:             "stale-pod",
						PlacementStatus: gkestatus.PlacementStatusHeterogeneous,
					},
				},
			},
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))

	if len(metrics) != 4 {
		t.Fatalf("failed probe metrics count = %d, want 4", len(metrics))
	}
	for _, metric := range metrics {
		labels := metricLabels(t, metric)
		if _, ok := labels["pod"]; ok {
			t.Fatalf("failed probe should not emit pod placement metrics; labels=%v", labels)
		}
		if labels["probe_status"] != gkestatus.ProbeStatusFailed {
			t.Fatalf("probe_status = %q, want failed; labels=%v", labels["probe_status"], labels)
		}
	}
	fullSyncedMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_cluster_full_synced")
	if got := metricGaugeValue(t, fullSyncedMetric); got != 0 {
		t.Fatalf("cluster full synced metric = %v, want 0", got)
	}
}

func TestGKEHeterogeneousStatusMetricIncludesSuccessfulPlacementWhenClusterNotFullSynced(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:                        "cluster-a",
		Namespace:                        "namespace-a",
		TenantID:                         "tenant-a",
		ProjectID:                        "project-a",
		LastSuccessfulPlacementProbeTime: testTimePtr(time.Unix(100, 0)),
		ProbeStatus:                      gkestatus.ProbeStatusSuccess,
		ClusterFullSynced:                false,
		Enabled:                          true,
		ValidIntent:                      true,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType: "tikv",
				PodGroups: []gkestatus.PodGroup{
					{
						PlacementStatus: gkestatus.PlacementStatusHeterogeneous,
						PodCount:        1,
						Pods:            []string{"tikv-0"},
					},
				},
				NodeUsages: []gkestatus.NodeUsage{
					{
						NodePool:      "pool-extra",
						MachineType:   "n2d-standard-8",
						MachineFamily: "n2d",
						UsedNodeCount: 1,
					},
				},
			},
		},
		ClusterNodeUsages: []gkestatus.NodeUsage{
			{
				NodePool:      "pool-extra",
				MachineType:   "n2d-standard-8",
				MachineFamily: "n2d",
				UsedNodeCount: 1,
			},
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))

	successMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_status_probe_success")
	if got := metricGaugeValue(t, successMetric); got != 1 {
		t.Fatalf("success metric = %v, want 1", got)
	}
	fullSyncedMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_cluster_full_synced")
	if got := metricGaugeValue(t, fullSyncedMetric); got != 0 {
		t.Fatalf("cluster full synced metric = %v, want 0", got)
	}
	successfulTimestampMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_successful_placement_probe_timestamp_seconds")
	if got := metricGaugeValue(t, successfulTimestampMetric); got != 100 {
		t.Fatalf("successful placement timestamp = %v, want 100", got)
	}
	podCountMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_component_pods_total")
	if got := metricGaugeValue(t, podCountMetric); got != 1 {
		t.Fatalf("component pod count metric = %v, want 1", got)
	}
	componentNodeMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_component_nodes_total")
	if got := metricGaugeValue(t, componentNodeMetric); got != 1 {
		t.Fatalf("component node metric = %v, want 1", got)
	}
	clusterNodeMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_nodes_total")
	if got := metricGaugeValue(t, clusterNodeMetric); got != 1 {
		t.Fatalf("cluster node metric = %v, want 1", got)
	}
}

func TestGKEHeterogeneousStatusMetricIncludesSuccessfulPlacement(t *testing.T) {
	placementRatio := 1.0
	effectiveRatio := 0.5
	status := gkestatus.Snapshot{
		ClusterID:         "cluster-a",
		Namespace:         "namespace-a",
		TenantID:          "tenant-a",
		ProjectID:         "project-a",
		ClusterGeneration: 12,
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		ClusterFullSynced: true,
		Enabled:           true,
		ValidIntent:       true,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType:     "tikv",
				ComputeClassName:  "tikv-priority",
				BaseInstanceType:  "n2-standard-8",
				DesiredReplicas:   2,
				TotalPods:         1,
				ScheduledPods:     1,
				HeterogeneousPods: 1,
				PlacementRatio:    &placementRatio,
				EffectiveRatio:    &effectiveRatio,
				PodGroups: []gkestatus.PodGroup{
					{
						PlacementStatus:      gkestatus.PlacementStatusHeterogeneous,
						ExpectedRole:         "extra",
						ExpectedInstanceType: "n2d-standard-8",
						ActualInstanceType:   "n2d-standard-8",
						ActualNodePool:       "pool-extra",
						PodCount:             1,
						Pods:                 []string{"tikv-0"},
					},
				},
				NodeUsages: []gkestatus.NodeUsage{
					{
						NodePool:        "pool-extra",
						MachineType:     "n2d-standard-8",
						MachineFamily:   "n2d",
						ExpectedRole:    "extra",
						PlacementStatus: gkestatus.PlacementStatusHeterogeneous,
						UsedNodeCount:   1,
						PodCount:        1,
					},
				},
			},
		},
		ClusterNodeUsages: []gkestatus.NodeUsage{
			{
				NodePool:      "pool-extra",
				MachineType:   "n2d-standard-8",
				MachineFamily: "n2d",
				UsedNodeCount: 1,
				PodCount:      1,
			},
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	for _, metric := range metrics {
		if labels := metricLabels(t, metric); labels["pod"] != "" {
			t.Fatalf("per-pod placement metric should not be emitted; labels=%v", labels)
		}
	}

	podCountMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_component_pods_total")
	if got := metricGaugeValue(t, podCountMetric); got != 1 {
		t.Fatalf("component pod count metric = %v, want 1", got)
	}
	podCountLabels := metricLabels(t, podCountMetric)
	if podCountLabels["placement_status"] != gkestatus.PlacementStatusHeterogeneous ||
		podCountLabels["expected_role"] != "extra" ||
		podCountLabels["expected_instance_type"] != "n2d-standard-8" ||
		podCountLabels["actual_instance_type"] != "n2d-standard-8" ||
		podCountLabels["actual_nodepool"] != "pool-extra" {
		t.Fatalf("component pod count labels are wrong: %v", podCountLabels)
	}

	clusterMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_nodes_total")
	if got := metricGaugeValue(t, clusterMetric); got != 1 {
		t.Fatalf("cluster node usage metric = %v, want 1", got)
	}
	clusterLabels := metricLabels(t, clusterMetric)
	if clusterLabels["nodepool"] != "pool-extra" || clusterLabels["machine_family"] != "n2d" {
		t.Fatalf("cluster node usage labels are wrong: %v", clusterLabels)
	}

	componentMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_component_nodes_total")
	if got := metricGaugeValue(t, componentMetric); got != 1 {
		t.Fatalf("component node usage metric = %v, want 1", got)
	}
	componentLabels := metricLabels(t, componentMetric)
	if componentLabels["component"] != "tikv" || componentLabels["placement_status"] != gkestatus.PlacementStatusHeterogeneous {
		t.Fatalf("component node usage labels are wrong: %v", componentLabels)
	}
}

func TestGKEHeterogeneousStatusMetricNormalizesUnknownPlacementLabels(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:         "cluster-a",
		Namespace:         "namespace-a",
		TenantID:          "tenant-a",
		ProjectID:         "project-a",
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		ClusterFullSynced: true,
		Enabled:           true,
		ValidIntent:       true,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType: "tikv",
				PodGroups: []gkestatus.PodGroup{
					{
						PlacementStatus: gkestatus.PlacementStatusUnknown,
						PodCount:        1,
						Pods:            []string{"tikv-0"},
					},
				},
			},
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	for _, metric := range metrics {
		if labels := metricLabels(t, metric); labels["pod"] != "" {
			t.Fatalf("per-pod placement metric should not be emitted; labels=%v", labels)
		}
	}

	podCountMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_component_pods_total")
	labels := metricLabels(t, podCountMetric)
	if labels["placement_status"] != gkestatus.PlacementStatusUnknown ||
		labels["expected_role"] != gkestatus.PlacementStatusUnknown ||
		labels["expected_instance_type"] != gkestatus.PlacementStatusUnknown ||
		labels["actual_instance_type"] != gkestatus.PlacementStatusUnknown ||
		labels["actual_nodepool"] != gkestatus.PlacementStatusUnknown {
		t.Fatalf("component pod count labels are wrong: %v", labels)
	}
}

func TestGKEHeterogeneousStatusMetricUsesPodGroupCountInsteadOfPodNames(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:         "cluster-a",
		Namespace:         "namespace-a",
		TenantID:          "tenant-a",
		ProjectID:         "project-a",
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		ClusterFullSynced: true,
		Enabled:           true,
		ValidIntent:       true,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType: "tikv",
				PodGroups: []gkestatus.PodGroup{
					{
						PlacementStatus:    gkestatus.PlacementStatusBase,
						ActualInstanceType: "n2-standard-8",
						ActualNodePool:     "pool-base",
						PodCount:           1000,
						Pods:               []string{"tikv-0", "tikv-1"},
					},
				},
			},
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	podCountMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_component_pods_total")

	if got := metricGaugeValue(t, podCountMetric); got != 1000 {
		t.Fatalf("component pod count metric = %v, want PodCount 1000", got)
	}
}

func TestGKEHeterogeneousStatusMetricDoesNotFallbackToOldPodsFormat(t *testing.T) {
	obj := statusConfigMapUnstructuredWithData(t, "cluster-a", "namespace-a", `
clusterId: cluster-a
namespace: namespace-a
tenantId: tenant-a
projectId: project-a
probeStatus: success
clusterFullSynced: true
enabled: true
validIntent: true
components:
- componentType: tikv
  pods:
  - pod: tikv-0
    placementStatus: base
`)

	metrics := gkeHeterogeneousStatus{}.ExtractMore(obj)

	for _, metric := range metrics {
		if strings.Contains(metric.Desc().String(), "gke_heterogeneous_component_pods_total") {
			t.Fatalf("old pods-only format emitted component pod metric: %v", metric.Desc())
		}
	}
}

func TestGKEHeterogeneousStatusMetricAggregatesPodGroupsByExportedLabels(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:         "cluster-a",
		Namespace:         "namespace-a",
		TenantID:          "tenant-a",
		ProjectID:         "project-a",
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		ClusterFullSynced: true,
		Enabled:           true,
		ValidIntent:       true,
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType: "tikv",
				PodGroups: []gkestatus.PodGroup{
					{
						PlacementStatus:      gkestatus.PlacementStatusHeterogeneous,
						ExpectedRole:         "extra",
						ExpectedInstanceType: "n2d-standard-8",
						ExpectedComputeClass: "compute-class-a",
						ActualInstanceType:   "n2d-standard-8",
						ActualComputeClass:   "compute-class-a",
						ActualNodePool:       "pool-extra",
						PodCount:             2,
					},
					{
						PlacementStatus:      gkestatus.PlacementStatusHeterogeneous,
						ExpectedRole:         "extra",
						ExpectedInstanceType: "n2d-standard-8",
						ExpectedComputeClass: "compute-class-b",
						ActualInstanceType:   "n2d-standard-8",
						ActualComputeClass:   "compute-class-b",
						ActualNodePool:       "pool-extra",
						PodCount:             3,
					},
				},
			},
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	var podCountMetrics []prometheus.Metric
	for _, metric := range metrics {
		if strings.Contains(metric.Desc().String(), "gke_heterogeneous_component_pods_total") {
			podCountMetrics = append(podCountMetrics, metric)
		}
	}

	if len(podCountMetrics) != 1 {
		t.Fatalf("component pod count metrics = %d, want one aggregated series", len(podCountMetrics))
	}
	if got := metricGaugeValue(t, podCountMetrics[0]); got != 5 {
		t.Fatalf("component pod count metric = %v, want aggregate count 5", got)
	}
}

func TestGKEHeterogeneousStatusMetricIncludesTerminalWorkflowOnFailedProbe(t *testing.T) {
	duration := 600.0
	status := gkestatus.Snapshot{
		ClusterID:   "cluster-a",
		Namespace:   "namespace-a",
		TenantID:    "tenant-a",
		ProjectID:   "project-a",
		ProbeStatus: gkestatus.ProbeStatusFailed,
		Enabled:     true,
		ValidIntent: true,
		ScaleWorkflow: &gkestatus.ScaleWorkflowStatus{
			WorkflowNamespace: "namespace-a",
			WorkflowName:      testScaleWorkflowName,
			WorkflowUID:       "workflow-uid",
			WorkflowPhase:     "Completed",
			Operation:         gkestatus.ScaleWorkflowOperationScaleOut,
			ComponentType:     "tikv",
			BeginAt:           testTimePtr(time.Unix(1000, 0)),
			EndAt:             testTimePtr(time.Unix(1600, 0)),
			DurationSeconds:   &duration,
		},
		Components: []gkestatus.ComponentStatus{
			{
				ComponentType: "tikv",
				Pods: []gkestatus.PodPlacement{
					{
						Pod:             "stale-pod",
						PlacementStatus: gkestatus.PlacementStatusHeterogeneous,
					},
				},
			},
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	durationMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_scale_workflow_duration_seconds")
	labels := metricLabels(t, durationMetric)
	if labels["workflow_name"] != testScaleWorkflowName || labels["component"] != "tikv" {
		t.Fatalf("workflow duration labels are wrong: %v", labels)
	}
	if _, ok := labels["workflow_uid"]; ok {
		t.Fatalf("workflow_uid should not be exported as a metric label: %v", labels)
	}
	if got := metricGaugeValue(t, durationMetric); got != duration {
		t.Fatalf("workflow duration = %v, want %v", got, duration)
	}
	provisionMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_node_provision_duration_seconds")
	if got := metricGaugeValue(t, provisionMetric); got != duration {
		t.Fatalf("node provision duration = %v, want %v", got, duration)
	}
	for _, metric := range metrics {
		labels := metricLabels(t, metric)
		if _, ok := labels["pod"]; ok {
			t.Fatalf("failed probe should not emit pod placement metrics; labels=%v", labels)
		}
	}
}

func TestGKEHeterogeneousStatusMetricIncludesTerminalWorkflowWhenDisabled(t *testing.T) {
	duration := 600.0
	status := gkestatus.Snapshot{
		ClusterID:   "cluster-a",
		Namespace:   "namespace-a",
		TenantID:    "tenant-a",
		ProjectID:   "project-a",
		ProbeStatus: gkestatus.ProbeStatusDisabled,
		Enabled:     false,
		ValidIntent: true,
		ScaleWorkflow: &gkestatus.ScaleWorkflowStatus{
			WorkflowNamespace: "namespace-a",
			WorkflowName:      testScaleWorkflowName,
			WorkflowUID:       "workflow-uid",
			WorkflowPhase:     "Completed",
			Operation:         gkestatus.ScaleWorkflowOperationDisable,
			ComponentType:     "tikv",
			BeginAt:           testTimePtr(time.Unix(1000, 0)),
			EndAt:             testTimePtr(time.Unix(1600, 0)),
			DurationSeconds:   &duration,
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	durationMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_scale_workflow_duration_seconds")
	labels := metricLabels(t, durationMetric)
	if labels["workflow_name"] != testScaleWorkflowName || labels["workflow_phase"] != "Completed" {
		t.Fatalf("disabled terminal workflow labels are wrong: %v", labels)
	}
	if got := metricGaugeValue(t, durationMetric); got != duration {
		t.Fatalf("disabled terminal workflow duration = %v, want %v", got, duration)
	}
	for _, metric := range metrics {
		if strings.Contains(metric.Desc().String(), "gke_heterogeneous_node_provision_duration_seconds") {
			t.Fatalf("disabled workflow should not emit node provision duration: %v", metric.Desc())
		}
	}
}

func TestGKEHeterogeneousStatusMetricSkipsNodeProvisionForOldDisabledWorkflowWithoutOperation(t *testing.T) {
	duration := 600.0
	status := gkestatus.Snapshot{
		ClusterID:   "cluster-a",
		Namespace:   "namespace-a",
		TenantID:    "tenant-a",
		ProjectID:   "project-a",
		ProbeStatus: gkestatus.ProbeStatusDisabled,
		Enabled:     false,
		ValidIntent: true,
		ScaleWorkflow: &gkestatus.ScaleWorkflowStatus{
			WorkflowNamespace: "namespace-a",
			WorkflowName:      testScaleWorkflowName,
			WorkflowUID:       "workflow-uid",
			WorkflowPhase:     "Completed",
			ComponentType:     "tikv",
			BeginAt:           testTimePtr(time.Unix(1000, 0)),
			EndAt:             testTimePtr(time.Unix(1600, 0)),
			DurationSeconds:   &duration,
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	durationMetric := findMetricByDescContains(t, metrics, "gke_heterogeneous_scale_workflow_duration_seconds")
	if got := metricGaugeValue(t, durationMetric); got != duration {
		t.Fatalf("old disabled workflow duration = %v, want %v", got, duration)
	}
	for _, metric := range metrics {
		if strings.Contains(metric.Desc().String(), "gke_heterogeneous_node_provision_duration_seconds") {
			t.Fatalf("old disabled workflow without operation should not emit node provision duration: %v", metric.Desc())
		}
	}
}

func TestGKEHeterogeneousStatusMetricSkipsStaleRunningWorkflow(t *testing.T) {
	status := gkestatus.Snapshot{
		ClusterID:         "cluster-a",
		Namespace:         "namespace-a",
		TenantID:          "tenant-a",
		ProjectID:         "project-a",
		ProbeStatus:       gkestatus.ProbeStatusSuccess,
		ClusterFullSynced: true,
		Enabled:           true,
		ValidIntent:       true,
		ScaleWorkflow: &gkestatus.ScaleWorkflowStatus{
			WorkflowNamespace: "namespace-a",
			WorkflowName:      testScaleWorkflowName,
			WorkflowUID:       "workflow-uid",
			WorkflowPhase:     "Stale",
			ComponentType:     "tikv",
			BeginAt:           testTimePtr(time.Unix(1000, 0)),
			InProgress:        true,
			Stale:             true,
		},
	}

	metrics := gkeHeterogeneousStatus{}.ExtractMore(statusConfigMapUnstructured(t, status))
	for _, metric := range metrics {
		labels := metricLabels(t, metric)
		if labels["workflow_name"] == testScaleWorkflowName {
			t.Fatalf("stale running workflow should not be emitted; labels=%v", labels)
		}
	}
}

func statusConfigMapUnstructured(t *testing.T, status gkestatus.Snapshot) *unstructured.Unstructured {
	t.Helper()

	data, err := yaml.Marshal(status)
	if err != nil {
		t.Fatalf("marshal status: %v", err)
	}
	return statusConfigMapUnstructuredWithData(t, status.ClusterID, status.Namespace, string(data))
}

func statusConfigMapUnstructuredWithData(t *testing.T, clusterID, namespace, data string) *unstructured.Unstructured {
	t.Helper()

	cm := corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      gkestatus.ConfigMapNamePrefix + clusterID,
			Namespace: namespace,
		},
		Data: map[string]string{
			gkestatus.ConfigMapDataKey: data,
		},
	}
	return toUnstructured(t, &cm)
}

func metricGaugeValue(t *testing.T, metric prometheus.Metric) float64 {
	t.Helper()

	var pb dto.Metric
	if err := metric.Write(&pb); err != nil {
		t.Fatalf("write metric: %v", err)
	}
	return pb.Gauge.GetValue()
}

func findMetricByGaugeValue(t *testing.T, metrics []prometheus.Metric, value float64) prometheus.Metric {
	t.Helper()

	for _, metric := range metrics {
		if metricGaugeValue(t, metric) == value {
			return metric
		}
	}
	t.Fatalf("metric with gauge value %v not found", value)
	return nil
}

func findMetricByDescContains(t *testing.T, metrics []prometheus.Metric, text string) prometheus.Metric {
	t.Helper()

	for _, metric := range metrics {
		if strings.Contains(metric.Desc().String(), text) {
			return metric
		}
	}
	t.Fatalf("metric descriptor containing %q not found", text)
	return nil
}

func assertMetricDescNotContains(t *testing.T, metrics []prometheus.Metric, text string) {
	t.Helper()

	for _, metric := range metrics {
		if strings.Contains(metric.Desc().String(), text) {
			t.Fatalf("metric descriptor containing %q found: %v", text, metric.Desc())
		}
	}
}

func collectDescs(collector *Collector) []*prometheus.Desc {
	ch := make(chan *prometheus.Desc, 32)
	collector.Describe(ch)
	close(ch)

	var out []*prometheus.Desc
	for desc := range ch {
		out = append(out, desc)
	}
	return out
}

func collectCollectorMetrics(collector *Collector) []prometheus.Metric {
	ch := make(chan prometheus.Metric, 32)
	collector.Collect(ch)
	close(ch)

	var out []prometheus.Metric
	for metric := range ch {
		out = append(out, metric)
	}
	return out
}

func assertDescContains(t *testing.T, descs []*prometheus.Desc, metricName string) {
	t.Helper()

	for _, desc := range descs {
		if strings.Contains(desc.String(), metricName) {
			return
		}
	}
	t.Fatalf("descriptor for %s not found in %v", metricName, descs)
}

func assertDescNotContains(t *testing.T, descs []*prometheus.Desc, metricName string) {
	t.Helper()

	for _, desc := range descs {
		if strings.Contains(desc.String(), metricName) {
			t.Fatalf("descriptor for %s should not be registered: %v", metricName, desc)
		}
	}
}

func findGKEHeterogeneousStatusCollector(t *testing.T, collectors []*Collector) *Collector {
	t.Helper()

	for _, collector := range collectors {
		if _, ok := collector.Res.(gkeHeterogeneousStatus); ok {
			return collector
		}
	}
	t.Fatalf("gkeHeterogeneousStatus collector not found")
	return nil
}

func assertMetricFamily(t *testing.T, families []*dto.MetricFamily, name string) {
	t.Helper()

	for _, family := range families {
		if family.GetName() == name {
			if len(family.Metric) == 0 {
				t.Fatalf("metric family %s has no metrics", name)
			}
			return
		}
	}
	t.Fatalf("metric family %s not found", name)
}

func testTimePtr(t time.Time) *time.Time {
	return &t
}