package metricsprovider

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"log/slog"
	"math"
	"net/http"
	"strconv"
	"time"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/tidbcloud/cluster-service-ng/pkg/dxf"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	appsv1 "k8s.io/api/apps/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	apimeta "k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/scale"
	"k8s.io/metrics/pkg/apis/custom_metrics"
	"k8s.io/metrics/pkg/apis/external_metrics"
	resourceclient "k8s.io/metrics/pkg/client/clientset/versioned/typed/metrics/v1beta1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider/defaults"
	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider/helpers"
)

const (
	tidbWorkerDSNTPL            = "cloud_admin:@tcp(%s:4000)/?tls=tidb"
	tidbWorkerStatusEndpointTPL = "https://%s.%s:10080"

	MetricNameDisableScaleIn            = "disable_scale_in" // dummy metric for testing
	MetricNameDXFHScaleTiDBWorkerHint   = "dxf_hscale_tidb_worker_hint"
	MetricNameDXFHScaleTiKVWorkerHint   = "dxf_hscale_tikv_worker_hint"    // this metrics works for deployment
	MetricNameDXFHScaleTiKVWorkerHintV2 = "dxf_hscale_tikv_worker_hint_v2" // this metrics works for tikvworkergroup cr
	// TODO(shenjun): come up a better name
	MetricNameExtCPUScaledRatio        = "ext_cpu_scaled_ratio"
	MetricNameExtTiDBCPUScaleInGate    = "ext_tidb_cpu_scale_in_gate"
	MetricNameExtTiProxyCPUScaleInGate = "ext_tiproxy_cpu_scale_in_gate"

	// TODO(shenjun):  not support yet
	MetricNameExtMemoryScaledRatio = "ext_memory_scaled_ratio"

	MetricNameExtAvgMemoryUtilization = "ext_avg_memory_utilization"

	MetricNameExtAvgBandwidthUtilization = "ext_avg_bandwidth_utilization"

	AnnotationTiDBCPUScaleInThreshold = "autoscaling.tidbcloud.com/tidb-scale-in-cpu-threshold"
	DefaultTiDBCPUScaleInThreshold    = int64(30)

	AnnotationTiProxyCPUScaleInThreshold = "autoscaling.tidbcloud.com/tiproxy-scale-in-cpu-threshold"
	DefaultTiProxyCPUScaleInThreshold    = int64(30)

	HintScaleIn   = 0
	HintDoNothing = 1
	HintScaleOut  = 2
)

var (
	// default value
	HorizontalPodAutoscalerCPUInitializationPeriod = metav1.Duration{Duration: 5 * time.Minute}
	HorizontalPodAutoscalerInitialReadinessDelay   = metav1.Duration{Duration: 30 * time.Second}
	HorizontalPodAutoscalerTolerance               = 0.1
)

type cmProvider struct {
	defaults.DefaultCustomMetricsProvider
	defaults.DefaultExternalMetricsProvider

	kubeClient            client.Client
	mapper                apimeta.RESTMapper
	resourceMetricsClient resourceclient.MetricsV1beta1Interface
	scaleClient           scale.ScalesGetter

	disableScalueIn int64

	bwCache                *bandwidthCache
	nodeExporterHTTPClient *http.Client
}

type CPUScaleInGateConfig struct {
	MetricName          string
	ThresholdAnnotation string
	DefaultThreshold    int64
}

var (
	TiDBCPUScaleInGateConfig = CPUScaleInGateConfig{
		MetricName:          MetricNameExtTiDBCPUScaleInGate,
		ThresholdAnnotation: AnnotationTiDBCPUScaleInThreshold,
		DefaultThreshold:    DefaultTiDBCPUScaleInThreshold,
	}
	TiProxyCPUScaleInGateConfig = CPUScaleInGateConfig{
		MetricName:          MetricNameExtTiProxyCPUScaleInGate,
		ThresholdAnnotation: AnnotationTiProxyCPUScaleInThreshold,
		DefaultThreshold:    DefaultTiProxyCPUScaleInThreshold,
	}
)

func getCPUScaleInThreshold(hpa *autoscalingv2.HorizontalPodAutoscaler, cfg CPUScaleInGateConfig) int64 {
	if hpa == nil || hpa.Annotations == nil {
		return cfg.DefaultThreshold
	}

	raw, ok := hpa.Annotations[cfg.ThresholdAnnotation]
	if !ok || raw == "" {
		return cfg.DefaultThreshold
	}

	threshold, err := strconv.ParseInt(raw, 10, 64)
	if err != nil {
		slog.Warn("invalid scale-in threshold annotation, fallback to default",
			"annotation", cfg.ThresholdAnnotation,
			"value", raw,
			"default", cfg.DefaultThreshold,
			"err", err)
		return cfg.DefaultThreshold
	}
	if threshold < 1 || threshold > 100 {
		slog.Warn("out-of-range scale-in threshold annotation, fallback to default",
			"annotation", cfg.ThresholdAnnotation,
			"value", threshold,
			"default", cfg.DefaultThreshold)
		return cfg.DefaultThreshold
	}
	return threshold
}

func NewCMProvider(
	kubeClient client.Client,
	mapper apimeta.RESTMapper,
	disableScalueIn int64,
	resourceMetricsClient resourceclient.MetricsV1beta1Interface,
	scaleClient scale.ScalesGetter,
) provider.MetricsProvider {
	return &cmProvider{
		kubeClient:            kubeClient,
		mapper:                mapper,
		disableScalueIn:       disableScalueIn,
		resourceMetricsClient: resourceMetricsClient,
		scaleClient:           scaleClient,
		bwCache:               newBandwidthCache(),
		nodeExporterHTTPClient: &http.Client{
			Timeout: nodeExporterScrapeTimeout,
		},
	}
}

func (p *cmProvider) buildTLSConfig(tlsSecret *corev1.Secret) (*tls.Config, error) {
	// get the ca cert key file from tidbgroup
	pem, ok := tlsSecret.Data["ca.crt"]
	if !ok {
		return nil, fmt.Errorf("cannot get ca.crt form secret %s", client.ObjectKeyFromObject(tlsSecret).String())
	}
	tlsCert, ok := tlsSecret.Data["tls.crt"]
	if !ok {
		return nil, fmt.Errorf("cannot get tls.crt form secret %s", client.ObjectKeyFromObject(tlsSecret).String())
	}
	tlsKey, ok := tlsSecret.Data["tls.key"]
	if !ok {
		return nil, fmt.Errorf("cannot get tls.key form secret %s", client.ObjectKeyFromObject(tlsSecret).String())
	}

	pool := x509.NewCertPool()
	if ok := pool.AppendCertsFromPEM(pem); !ok {
		return nil, errors.New("unable to append root cert to pool")
	}
	cert, err := tls.X509KeyPair(tlsCert, tlsKey)
	if err != nil {
		return nil, err
	}

	return &tls.Config{
		RootCAs:            pool,
		Certificates:       []tls.Certificate{cert},
		InsecureSkipVerify: true, //nolint:gosec // skip tls
	}, nil
}

// metricFor constructs a result for a single metric value.
func (p *cmProvider) metricFor(
	value int64,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
) (*custom_metrics.MetricValue, error) {
	// construct a reference referring to the described object
	objRef, err := helpers.ReferenceFor(p.mapper, name, info)
	if err != nil {
		return nil, err
	}

	return &custom_metrics.MetricValue{
		DescribedObject: objRef,
		Metric: custom_metrics.MetricIdentifier{
			Name: info.Metric,
		},
		// you'll want to use the actual timestamp in a real adapter
		Timestamp: metav1.Now(),
		Value:     *resource.NewQuantity(value, resource.DecimalSI),
	}, nil
}

func (p *cmProvider) findTiDBWokerBySharedPool(ctx context.Context, namespace, sharedPool string) (*v1alpha1.TiDBGroup, error) {
	tidbgroups := new(v1alpha1.TiDBGroupList)
	selector := labels.Set(map[string]string{
		schemav1.LabelSharedPoolKey: sharedPool,
		schemav1.LabelComponentKey:  "worker-tidb",
	}).AsSelector()
	if err := p.kubeClient.List(ctx,
		tidbgroups,
		&client.ListOptions{LabelSelector: selector, Namespace: namespace}); err != nil {
		slog.Error("list tidbgroups failed", "err", err)
		return nil, err
	}
	if len(tidbgroups.Items) == 0 {
		return nil, fmt.Errorf("tidbgroups for selector %s is not found", selector.String())
	}
	return &tidbgroups.Items[0], nil
}

func (p *cmProvider) findTiDBWokerSvc(ctx context.Context, tidbgroup types.NamespacedName) (*corev1.Service, error) {
	list := corev1.ServiceList{}
	opt := client.MatchingLabels{
		schemav1.LabelPingcapGroupKey: tidbgroup.Name,
	}
	if err := p.kubeClient.List(ctx, &list, opt); err != nil {
		slog.Error("list services failed", "err", err)
		return nil, err
	}
	if len(list.Items) == 0 {
		return nil, fmt.Errorf("service for tidbgroup %s is not found", tidbgroup.String())
	}
	return &list.Items[0], nil
}

func (p *cmProvider) findTLSConfig(ctx context.Context, secretKey types.NamespacedName) (*tls.Config, error) {
	tlsSecret := new(corev1.Secret)
	if err := p.kubeClient.Get(ctx, secretKey, tlsSecret); err != nil {
		slog.Error("get secret failed", "err", err)
		return nil, err
	}
	tlsConfig, err := p.buildTLSConfig(tlsSecret)
	if err != nil {
		slog.Error("build tls config failed", "err", err)
		return nil, err
	}
	return tlsConfig, nil
}

//nolint:unparam // timestamp is part of the resource metrics result; current callers only need metric values.
func (p *cmProvider) getResourceMetric(ctx context.Context, resourceName corev1.ResourceName, namespace string, selector labels.Selector, container string) (PodMetricsInfo, time.Time, error) {
	metrics, err := p.resourceMetricsClient.PodMetricses(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector.String()})
	if err != nil {
		return nil, time.Time{}, fmt.Errorf("unable to fetch metrics from resource metrics API: %w", err)
	}

	if len(metrics.Items) == 0 {
		return nil, time.Time{}, fmt.Errorf("no metrics returned from resource metrics API")
	}
	var res PodMetricsInfo
	if container != "" {
		res, err = getContainerMetrics(metrics.Items, resourceName, container)
		if err != nil {
			return nil, time.Time{}, fmt.Errorf("failed to get container metrics: %w", err)
		}
	} else {
		res = getPodMetrics(metrics.Items, resourceName)
	}
	timestamp := metrics.Items[0].Timestamp.Time
	return res, timestamp, nil
}

// dxf_hscale_tidb_worker_hint = 0 scale in
// dxf_hscale_tidb_worker_hint = 1 do nothing
// dxf_hscale_tidb_worker_hint = 2 scale out
func (p *cmProvider) GetMetricDXFHScaleTiDBWorkerHint(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		slog.Error("normailze failed", "err", err)
		return nil, err
	}
	if res != "tidbgroup" {
		return nil, fmt.Errorf("resource %s is not supported", res)
	}
	tidbworker := new(v1alpha1.TiDBGroup)
	if err = p.kubeClient.Get(ctx, name, tidbworker); err != nil {
		slog.Error("find tidbgroup failed", "err", err)
		return nil, err
	}
	currReplicas := int(*tidbworker.Spec.Replicas)
	// get schedule status from kernel via dxf schedule status api
	svc, err := p.findTiDBWokerSvc(ctx, name)
	if err != nil {
		return nil, err
	}
	// cluster tls
	secretKey := types.NamespacedName{
		Name:      fmt.Sprintf("%s-tidb-cluster-secret", name.Name),
		Namespace: name.Namespace,
	}
	tlsConfig, err := p.findTLSConfig(ctx, secretKey)
	if err != nil {
		return nil, err
	}
	endpoint := fmt.Sprintf(tidbWorkerStatusEndpointTPL, svc.Name, svc.Namespace)
	dxfClient := dxf.NewClient(endpoint, tlsConfig, secretKey.Name)
	schedStatus, err := dxfClient.GetDXFScheduleStatus(ctx)
	if err != nil {
		slog.Error("get dxf schedule status failed", "err", err)
		return nil, err
	}
	if schedStatus.TiDBWorker.RequiredCount == currReplicas {
		return p.metricFor(HintDoNothing, name, info)
	}
	// check if scale out is needed
	if schedStatus.TiDBWorker.RequiredCount > currReplicas {
		return p.metricFor(HintScaleOut, name, info)
	}
	// check if scale in is needed
	if schedStatus.TiDBWorker.RequiredCount < currReplicas {
		if schedStatus.Flags.PauseScaleIn.Enabled {
			return p.metricFor(HintDoNothing, name, info)
		}
		if schedStatus.TaskQueue.ScheduledCount == 0 {
			return p.metricFor(HintScaleIn, name, info)
		}
	}
	// do nothing
	return p.metricFor(HintDoNothing, name, info)
}

// dxf_hscale_tikv_worker_hint = 0 scale in
// dxf_hscale_tikv_worker_hint = 1 do nothing
// dxf_hscale_tikv_worker_hint = 2 scale out
func (p *cmProvider) GetMetricDXFHScaleTiKVWorkerHint(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		slog.Error("normailze failed", "err", err)
		return nil, err
	}
	if res != "deployment" {
		return nil, fmt.Errorf("resource %s is not supported", res)
	}
	tikvworker := new(appsv1.Deployment)
	if err = p.kubeClient.Get(ctx, name, tikvworker); err != nil {
		slog.Error("get deployment failed", "err", err)
		return nil, err
	}
	currReplicas := int(*tikvworker.Spec.Replicas)
	sharedPool := tikvworker.Labels[schemav1.LabelSharedPoolKey]
	if sharedPool == "" {
		slog.Error("data corrupted, cannot find sharedpool id", "label", tikvworker.Labels, "deployment", name.String())
		return nil, fmt.Errorf("data corrupted, cannot find sharedpool id")
	}

	tidbworker, err := p.findTiDBWokerBySharedPool(ctx, name.Namespace, sharedPool)
	if err != nil {
		return nil, err
	}
	tidbworkerKey := client.ObjectKeyFromObject(tidbworker)

	// get schedule status from kernel via dxf schedule status api
	svc, err := p.findTiDBWokerSvc(ctx, tidbworkerKey)
	if err != nil {
		return nil, err
	}
	// cluster tls
	secretKey := types.NamespacedName{
		Name:      fmt.Sprintf("%s-tidb-cluster-secret", tidbworkerKey.Name),
		Namespace: tidbworkerKey.Namespace,
	}
	tlsConfig, err := p.findTLSConfig(ctx, secretKey)
	if err != nil {
		return nil, err
	}
	endpoint := fmt.Sprintf(tidbWorkerStatusEndpointTPL, svc.Name, svc.Namespace)
	dxfClient := dxf.NewClient(endpoint, tlsConfig, secretKey.Name)
	schedStatus, err := dxfClient.GetDXFScheduleStatus(ctx)
	if err != nil {
		slog.Error("get dxf schedule status failed", "err", err)
		return nil, err
	}
	if schedStatus.TiKVWorker.RequiredCount == currReplicas {
		return p.metricFor(HintDoNothing, name, info)
	}
	// check if scale out is needed
	if schedStatus.TiKVWorker.RequiredCount > currReplicas {
		return p.metricFor(HintScaleOut, name, info)
	}
	// check if scale in is needed
	if schedStatus.TiKVWorker.RequiredCount < currReplicas {
		if schedStatus.Flags.PauseScaleIn.Enabled {
			return p.metricFor(HintDoNothing, name, info)
		}
		if schedStatus.TaskQueue.ScheduledCount == 0 {
			return p.metricFor(HintScaleIn, name, info)
		}
	}
	// do nothing
	return p.metricFor(HintDoNothing, name, info)
}

func (p *cmProvider) GetMetricDXFHScaleTiKVWorkerHintV2(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		slog.Error("normailze failed", "err", err)
		return nil, err
	}
	if res != "tikvworkergroup" {
		return nil, fmt.Errorf("resource %s is not supported", res)
	}
	tikvworker := new(v1alpha1.TiKVWorkerGroup)
	if err = p.kubeClient.Get(ctx, name, tikvworker); err != nil {
		slog.Error("get tikvworkergroup failed", "err", err)
		return nil, err
	}
	currReplicas := int(*tikvworker.Spec.Replicas)
	sharedPool := tikvworker.Labels[schemav1.LabelSharedPoolKey]
	if sharedPool == "" {
		slog.Error("data corrupted, cannot find sharedpool id", "label", tikvworker.Labels, "tikvworkergroup", name.String())
		return nil, fmt.Errorf("data corrupted, cannot find sharedpool id")
	}

	tidbworker, err := p.findTiDBWokerBySharedPool(ctx, name.Namespace, sharedPool)
	if err != nil {
		return nil, err
	}
	tidbworkerKey := client.ObjectKeyFromObject(tidbworker)

	// get schedule status from kernel via dxf schedule status api
	svc, err := p.findTiDBWokerSvc(ctx, tidbworkerKey)
	if err != nil {
		return nil, err
	}
	// cluster tls
	secretKey := types.NamespacedName{
		Name:      fmt.Sprintf("%s-tidb-cluster-secret", tidbworkerKey.Name),
		Namespace: tidbworkerKey.Namespace,
	}
	tlsConfig, err := p.findTLSConfig(ctx, secretKey)
	if err != nil {
		return nil, err
	}
	endpoint := fmt.Sprintf(tidbWorkerStatusEndpointTPL, svc.Name, svc.Namespace)
	dxfClient := dxf.NewClient(endpoint, tlsConfig, secretKey.Name)
	schedStatus, err := dxfClient.GetDXFScheduleStatus(ctx)
	if err != nil {
		slog.Error("get dxf schedule status failed", "err", err)
		return nil, err
	}
	if schedStatus.TiKVWorker.RequiredCount == currReplicas {
		return p.metricFor(HintDoNothing, name, info)
	}
	// check if scale out is needed
	if schedStatus.TiKVWorker.RequiredCount > currReplicas {
		return p.metricFor(HintScaleOut, name, info)
	}
	// check if scale in is needed
	if schedStatus.TiKVWorker.RequiredCount < currReplicas {
		if schedStatus.Flags.PauseScaleIn.Enabled {
			return p.metricFor(HintDoNothing, name, info)
		}
		if schedStatus.TaskQueue.ScheduledCount == 0 {
			return p.metricFor(HintScaleIn, name, info)
		}
	}
	// do nothing
	return p.metricFor(HintDoNothing, name, info)
}

// ext_cpu_scaled_ratio is scaled based on cpu usage or utilization
func (p *cmProvider) GetMetricNameExtCPUScaledRatio(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		slog.Error("normailze failed", "err", err)
		return nil, err
	}
	if res != "horizontalpodautoscaler" {
		return nil, fmt.Errorf("resource %s is not supported", res)
	}
	hpa := new(autoscalingv2.HorizontalPodAutoscaler)
	if err = p.kubeClient.Get(ctx, name, hpa); err != nil {
		slog.Error("find hpa failed", "err", err)
		return nil, err
	}
	cpuMetricSource, ok := getResourceMetricSource(hpa.Spec.Metrics, corev1.ResourceCPU)
	if !ok {
		return nil, fmt.Errorf("resource metric cpu is not found in hpa %s", name)
	}
	scaleObj, err := getScale(ctx, p.scaleClient, p.mapper, name.Namespace, hpa.Spec.ScaleTargetRef)
	if err != nil {
		slog.Error("get scale for ref failed",
			"ref", hpa.Spec.ScaleTargetRef.String(),
			"err", err)
		return nil, err
	}
	selector, err := labels.Parse(scaleObj.Status.Selector)
	if err != nil {
		slog.Error("convert selector into a corresponding internal selector object failed",
			"selector", scaleObj.Status.Selector,
			"err", err)
		return nil, err
	}
	target := cpuMetricSource.Target
	if target.AverageUtilization == nil && target.AverageValue == nil {
		errMsg := "invalid resource metric source: neither an average utilization target nor an average value target was set"
		return nil, errors.New(errMsg)
	}

	if target.AverageValue != nil {
		errMsg := "average value is not support"
		return nil, errors.New(errMsg)
	}

	metrics, _, err := p.getResourceMetric(ctx, cpuMetricSource.Name, name.Namespace, selector, "")
	if err != nil {
		slog.Error("get resource metrics from pods failed",
			"resource name", cpuMetricSource.Name,
			"namespace", name.Namespace,
			"selector", selector.String(),
			"err", err)
		return nil, err
	}
	pods := new(corev1.PodList)
	if err = p.kubeClient.List(ctx, pods,
		&client.ListOptions{LabelSelector: selector, Namespace: name.Namespace}); err != nil {
		slog.Error("list pods failed", "err", err)
		return nil, err
	}
	if len(pods.Items) == 0 {
		return nil, fmt.Errorf("pods for selector %s is not found", selector.String())
	}

	_, unreadyPods, missingPods, ignoredPods := groupPods(
		pods.Items,
		metrics,
		cpuMetricSource.Name,
		HorizontalPodAutoscalerCPUInitializationPeriod.Duration,
		HorizontalPodAutoscalerInitialReadinessDelay.Duration,
	)
	removeMetricsForPods(metrics, ignoredPods)
	removeMetricsForPods(metrics, unreadyPods)
	requests, err := calculatePodRequests(pods.Items, "", cpuMetricSource.Name)
	if err != nil {
		return nil, err
	}

	if len(metrics) == 0 {
		return nil, fmt.Errorf("did not receive metrics for any ready pods")
	}
	targetUtilization := *target.AverageUtilization
	usageRatio, utilization, _, err := GetResourceUtilizationRatio(metrics, requests, targetUtilization)
	if err != nil {
		slog.Error("GetResourceUtilizationRatio failed",
			"err", err)
		return nil, err
	}

	rebalanceIgnored := len(unreadyPods) > 0 && usageRatio > 1.0

	if !rebalanceIgnored && len(missingPods) == 0 {
		if math.Abs(1.0-usageRatio) <= HorizontalPodAutoscalerTolerance {
			// do nothing if the change would be too small
			return p.metricFor(1, name, info)
		}
		// if we don't have any unready or missing pods, we can return the metric value based on utilization.
		if utilization > targetUtilization {
			return p.metricFor(100, name, info)
		} else {
			return p.metricFor(0, name, info)
		}
	}
	if len(missingPods) > 0 {
		if usageRatio < 1.0 {
			// on a scale-down, treat missing pods as using 100% of the resource request
			for podName := range missingPods {
				metrics[podName] = PodMetric{Value: requests[podName]}
			}
		} else if usageRatio > 1.0 {
			// on a scale-up, treat missing pods as using 0% of the resource request
			for podName := range missingPods {
				metrics[podName] = PodMetric{Value: 0}
			}
		}
	}
	if rebalanceIgnored {
		// on a scale-up, treat unready pods as using 0% of the resource request
		for podName := range unreadyPods {
			metrics[podName] = PodMetric{Value: 0}
		}
	}
	// re-run the utilization calculation with our new numbers
	newUsageRatio, _, _, err := GetResourceUtilizationRatio(metrics, requests, targetUtilization)
	if err != nil {
		slog.Error("GetResourceUtilizationRatio failed",
			"err", err)
		return nil, err
	}

	if math.Abs(1.0-newUsageRatio) <= HorizontalPodAutoscalerTolerance || (usageRatio < 1.0 && newUsageRatio > 1.0) || (usageRatio > 1.0 && newUsageRatio < 1.0) {
		// return the current replicas if the change would be too small,
		// or if the new usage ratio would cause a change in scale direction
		return p.metricFor(1, name, info)
	}
	currentReplicas := scaleObj.Spec.Replicas
	newReplicas := int32(math.Ceil(newUsageRatio * float64(len(metrics))))
	if (newUsageRatio < 1.0 && newReplicas > currentReplicas) || (newUsageRatio > 1.0 && newReplicas < currentReplicas) {
		// return the current replicas if the change of metrics length would cause a change in scale direction
		return p.metricFor(1, name, info)
	}

	// return the result, where the number of replicas considered is
	// however many replicas factored into our calculation
	if utilization > targetUtilization {
		return p.metricFor(100, name, info)
	} else {
		return p.metricFor(0, name, info)
	}
}

type cpuScaleInGateOptions struct {
	excludeDrainingPods bool
}

func (p *cmProvider) GetMetricExtTiDBCPUScaleInGate(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	return p.getCPUScaleInGateMetric(ctx, name, info, TiDBCPUScaleInGateConfig, MetricNameExtTiDBCPUScaleInGate, cpuScaleInGateOptions{})
}

func (p *cmProvider) GetMetricExtTiProxyCPUScaleInGate(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	return p.getCPUScaleInGateMetric(ctx, name, info, TiProxyCPUScaleInGateConfig, MetricNameExtTiProxyCPUScaleInGate, cpuScaleInGateOptions{
		excludeDrainingPods: true,
	})
}

// getCPUScaleInGateMetric implements a direction-aware scale-in gate:
// returns 0 (allow scale-in) when average CPU utilization is below threshold,
// otherwise returns 1 (block scale-in). The Resource CPU target stays independent
// and continues to drive scale-out (typically 70%).
func (p *cmProvider) getCPUScaleInGateMetric(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	cfg CPUScaleInGateConfig,
	metricName string,
	opts cpuScaleInGateOptions,
) (*custom_metrics.MetricValue, error) {
	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		slog.Error("normalize failed", "err", err)
		return nil, err
	}
	if res != "horizontalpodautoscaler" {
		return nil, fmt.Errorf("resource %s is not supported", res)
	}

	hpa := new(autoscalingv2.HorizontalPodAutoscaler)
	if err = p.kubeClient.Get(ctx, name, hpa); err != nil {
		slog.Error("find hpa failed", "err", err)
		return nil, err
	}

	threshold := getCPUScaleInThreshold(hpa, cfg)

	scaleObj, err := getScale(ctx, p.scaleClient, p.mapper, name.Namespace, hpa.Spec.ScaleTargetRef)
	if err != nil {
		slog.Error("get scale for ref failed",
			"ref", hpa.Spec.ScaleTargetRef.String(),
			"err", err)
		return nil, err
	}
	selector, err := labels.Parse(scaleObj.Status.Selector)
	if err != nil {
		slog.Error("convert selector into a corresponding internal selector object failed",
			"selector", scaleObj.Status.Selector,
			"err", err)
		return nil, err
	}

	metrics, _, err := p.getResourceMetric(ctx, corev1.ResourceCPU, name.Namespace, selector, "")
	if err != nil {
		slog.Error("get resource metrics from pods failed",
			"namespace", name.Namespace,
			"selector", selector.String(),
			"err", err)
		return nil, err
	}
	pods := new(corev1.PodList)
	if err = p.kubeClient.List(ctx, pods,
		&client.ListOptions{LabelSelector: selector, Namespace: name.Namespace}); err != nil {
		slog.Error("list pods failed", "err", err)
		return nil, err
	}
	if len(pods.Items) == 0 {
		return nil, fmt.Errorf("pods for selector %s is not found", selector.String())
	}

	_, unreadyPods, missingPods, ignoredPods := groupPods(
		pods.Items,
		metrics,
		corev1.ResourceCPU,
		HorizontalPodAutoscalerCPUInitializationPeriod.Duration,
		HorizontalPodAutoscalerInitialReadinessDelay.Duration,
	)
	removeMetricsForPods(metrics, ignoredPods)
	if opts.excludeDrainingPods {
		drainingPods := drainingPodNames(pods.Items)
		// Draining pods may not report metrics, so remove them from missing pods.
		missingPods = missingPods.Difference(drainingPods)
		removeMetricsForPods(metrics, drainingPods)
	}
	if len(unreadyPods) > 0 || len(missingPods) > 0 {
		// Be conservative: if metrics are incomplete, block scale-in.
		return p.metricFor(1, name, info)
	}
	if len(metrics) == 0 {
		return nil, fmt.Errorf("did not receive cpu metrics for any eligible pods")
	}

	requests, err := calculatePodRequests(pods.Items, "", corev1.ResourceCPU)
	if err != nil {
		return nil, err
	}
	_, utilization, _, err := GetResourceUtilizationRatio(metrics, requests, 100)
	if err != nil {
		slog.Error("GetResourceUtilizationRatio failed", "err", err)
		return nil, err
	}

	slog.Debug(metricName,
		"hpa", name,
		"selector", selector.String(),
		"utilization", utilization,
		"threshold", threshold)

	if int64(utilization) < threshold {
		return p.metricFor(0, name, info)
	}
	return p.metricFor(1, name, info)
}

func (p *cmProvider) GetMetricNameExtMemoryScaledRatio(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	return nil, fmt.Errorf("%s is not implemented yet", MetricNameExtMemoryScaledRatio)
}

func (p *cmProvider) GetMetricExtAvgMemoryUtilization(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	_ labels.Selector,
) (*custom_metrics.MetricValue, error) {
	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		return nil, fmt.Errorf("normalize metric info: %w", err)
	}
	if res != "horizontalpodautoscaler" {
		return nil, fmt.Errorf("ext_avg_memory_utilization: resource %q not supported (want horizontalpodautoscaler)", res)
	}

	hpa := new(autoscalingv2.HorizontalPodAutoscaler)
	if err = p.kubeClient.Get(ctx, name, hpa); err != nil {
		return nil, fmt.Errorf("find hpa %s: %w", name, err)
	}
	scaleObj, err := getScale(ctx, p.scaleClient, p.mapper, name.Namespace, hpa.Spec.ScaleTargetRef)
	if err != nil {
		return nil, fmt.Errorf("get scale for ref %s: %w", hpa.Spec.ScaleTargetRef.String(), err)
	}
	selector, err := labels.Parse(scaleObj.Status.Selector)
	if err != nil {
		return nil, fmt.Errorf("parse pod selector %q: %w", scaleObj.Status.Selector, err)
	}

	metrics, _, err := p.getResourceMetric(ctx, corev1.ResourceMemory, name.Namespace, selector, "")
	if err != nil {
		return nil, fmt.Errorf("get memory metrics for pods: %w", err)
	}
	pods := new(corev1.PodList)
	if err = p.kubeClient.List(ctx, pods,
		&client.ListOptions{LabelSelector: selector, Namespace: name.Namespace}); err != nil {
		return nil, fmt.Errorf("list pods: %w", err)
	}
	if len(pods.Items) == 0 {
		return nil, fmt.Errorf("pods for selector %s is not found", selector.String())
	}

	_, unreadyPods, missingPods, ignoredPods := groupPods(
		pods.Items,
		metrics,
		corev1.ResourceMemory,
		HorizontalPodAutoscalerCPUInitializationPeriod.Duration,
		HorizontalPodAutoscalerInitialReadinessDelay.Duration,
	)
	removeMetricsForPods(metrics, ignoredPods)
	removeMetricsForPods(metrics, unreadyPods)
	removeMetricsForPods(metrics, missingPods)
	removeMetricsForPods(metrics, drainingPodNames(pods.Items))
	if len(metrics) == 0 {
		return nil, fmt.Errorf("did not receive memory metrics for any eligible pods")
	}

	limits, err := calculatePodLimits(pods.Items, "", corev1.ResourceMemory)
	if err != nil {
		return nil, err
	}
	_, utilization, _, err := GetResourceUtilizationRatio(metrics, limits, 100)
	if err != nil {
		return nil, fmt.Errorf("calculate memory utilization with limits: %w", err)
	}

	slog.Info("ext_avg_memory_utilization",
		"hpa", name, "selector", selector.String(), "pct", utilization)
	return p.metricFor(int64(utilization), name, info)
}

// the params in path is metricLabelSelector
// example
// path: /apis/custom.metrics.k8s.io/v1beta1/namespaces/tidb1263855058001718/tidbgroups.core.pingcap.com/db-242afe13-worker/import_into_job_count?metricLabelSelector=status+in+%28pending%2Crunning%29
// kubectl get --raw "/apis/custom.metrics.k8s.io/v1beta1/namespaces/tidb1263855058001718/tidbgroups.core.pingcap.com/db-242afe13-worker/import_into_job_count?metricLabelSelector=status%20in%20(running,pending)"
func (p *cmProvider) GetMetricByName(
	ctx context.Context,
	name types.NamespacedName,
	info provider.CustomMetricInfo,
	metricSelector labels.Selector,
) (*custom_metrics.MetricValue, error) {
	slog.Info(
		"GetMetircByName is called",
		"info", info.String(),
		"name", name.String(),
		"metricSelector", metricSelector.String(),
	)
	switch info.Metric {
	case MetricNameDXFHScaleTiDBWorkerHint:
		return p.GetMetricDXFHScaleTiDBWorkerHint(ctx, name, info, metricSelector)
	case MetricNameDXFHScaleTiKVWorkerHint:
		return p.GetMetricDXFHScaleTiKVWorkerHint(ctx, name, info, metricSelector)
	case MetricNameDXFHScaleTiKVWorkerHintV2:
		return p.GetMetricDXFHScaleTiKVWorkerHintV2(ctx, name, info, metricSelector)
	case MetricNameExtCPUScaledRatio:
		return p.GetMetricNameExtCPUScaledRatio(ctx, name, info, metricSelector)
	case MetricNameExtTiDBCPUScaleInGate:
		return p.GetMetricExtTiDBCPUScaleInGate(ctx, name, info, metricSelector)
	case MetricNameExtTiProxyCPUScaleInGate:
		return p.GetMetricExtTiProxyCPUScaleInGate(ctx, name, info, metricSelector)
	case MetricNameExtMemoryScaledRatio:
		return p.GetMetricNameExtMemoryScaledRatio(ctx, name, info, metricSelector)
	case MetricNameExtAvgMemoryUtilization:
		return p.GetMetricExtAvgMemoryUtilization(ctx, name, info, metricSelector)
	case MetricNameExtAvgBandwidthUtilization:
		return p.GetMetricExtAvgBandwidthUtilization(ctx, name, info, metricSelector)
	default:
		return nil, fmt.Errorf("custom metric %s is not supported", info.Metric)
	}
}

func (p *cmProvider) GetMetricBySelector(
	ctx context.Context,
	namespace string,
	selector labels.Selector,
	info provider.CustomMetricInfo,
	metricSelector labels.Selector,
) (*custom_metrics.MetricValueList, error) {
	slog.Info("GetMetricBySelector is called",
		"info", info.String(),
		"selector", selector.String(),
		"namespace", namespace,
		"metricSelector", metricSelector.String())

	_, res, err := info.Normalized(p.mapper)
	if err != nil {
		slog.Error("normailze failed", "err", err)
		return nil, err
	}
	out := &custom_metrics.MetricValueList{
		Items: make([]custom_metrics.MetricValue, 0),
	}

	switch info.Metric {
	case MetricNameDXFHScaleTiDBWorkerHint:
		if res != "tidbgroup" {
			return nil, fmt.Errorf("resource %s is not supported", res)
		}
		tidbworkers := v1alpha1.TiDBGroupList{}
		if err := p.kubeClient.List(ctx, &tidbworkers, &client.ListOptions{LabelSelector: selector, Namespace: namespace}); err != nil {
			slog.Error("list tidbgroups failed", "err", err)
			return nil, err
		}
		if len(tidbworkers.Items) == 0 {
			return nil, fmt.Errorf("tidbgroups for selector %s is not found", selector.String())
		}
		for i := range tidbworkers.Items {
			tidbworker := tidbworkers.Items[i]
			objkey := client.ObjectKeyFromObject(&tidbworker)
			mv, err := p.GetMetricDXFHScaleTiDBWorkerHint(ctx, objkey, info, metricSelector)
			if err != nil {
				slog.Error("get value for metric dxf_hscale_tidb_worker_hint failed", "err", err)
				return nil, err
			}
			out.Items = append(out.Items, *mv)
		}
	case MetricNameDXFHScaleTiKVWorkerHint:
		if res != "deployment" {
			return nil, fmt.Errorf("resource %s is not supported", res)
		}
		tikvworkers := new(appsv1.DeploymentList)
		if err := p.kubeClient.List(ctx, tikvworkers, &client.ListOptions{LabelSelector: selector, Namespace: namespace}); err != nil {
			slog.Error("list deployment failed", "err", err)
			return nil, err
		}
		if len(tikvworkers.Items) == 0 {
			return nil, fmt.Errorf("deployments for selector %s is not found", selector.String())
		}
		for i := range tikvworkers.Items {
			tikvworker := tikvworkers.Items[i]
			objkey := client.ObjectKeyFromObject(&tikvworker)
			mv, err := p.GetMetricDXFHScaleTiKVWorkerHint(ctx, objkey, info, metricSelector)
			if err != nil {
				slog.Error("get value for metric dxf_hscale_tikv_worker_hint failed", "err", err)
				return nil, err
			}
			out.Items = append(out.Items, *mv)
		}

	case MetricNameDXFHScaleTiKVWorkerHintV2:
		if res != "tikvworkergroup" {
			return nil, fmt.Errorf("resource %s is not supported", res)
		}
		tikvworkers := new(v1alpha1.TiKVWorkerGroupList)
		if err := p.kubeClient.List(ctx, tikvworkers, &client.ListOptions{LabelSelector: selector, Namespace: namespace}); err != nil {
			slog.Error("list tikvworkergroup failed", "err", err)
			return nil, err
		}
		if len(tikvworkers.Items) == 0 {
			return nil, fmt.Errorf("tikvworkergroups for selector %s is not found", selector.String())
		}
		for i := range tikvworkers.Items {
			tikvworker := tikvworkers.Items[i]
			objkey := client.ObjectKeyFromObject(&tikvworker)
			mv, err := p.GetMetricDXFHScaleTiKVWorkerHintV2(ctx, objkey, info, metricSelector)
			if err != nil {
				slog.Error("get value for metric dxf_hscale_tikv_worker_hint_v2 failed", "err", err)
				return nil, err
			}
			out.Items = append(out.Items, *mv)
		}

	default:
		return nil, fmt.Errorf("custom metric %s with selector %s is not supported", info.Metric, metricSelector.String())
	}

	return out, nil
}

func (p *cmProvider) ListAllMetrics() []provider.CustomMetricInfo {
	return []provider.CustomMetricInfo{
		{
			GroupResource: schema.GroupResource{
				Group:    "core.pingcap.com",
				Resource: "tidbgroups",
			},
			Metric:     MetricNameDXFHScaleTiDBWorkerHint,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "apps",
				Resource: "deployments",
			},
			Metric:     MetricNameDXFHScaleTiKVWorkerHint,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "core.pingcap.com",
				Resource: "tikvworkergroups",
			},
			Metric:     MetricNameDXFHScaleTiKVWorkerHintV2,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "autoscaling",
				Resource: "horizontalpodautoscalers",
			},
			Metric:     MetricNameExtCPUScaledRatio,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "autoscaling",
				Resource: "horizontalpodautoscalers",
			},
			Metric:     MetricNameExtTiDBCPUScaleInGate,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "autoscaling",
				Resource: "horizontalpodautoscalers",
			},
			Metric:     MetricNameExtTiProxyCPUScaleInGate,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "autoscaling",
				Resource: "horizontalpodautoscalers",
			},
			Metric:     MetricNameExtMemoryScaledRatio,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "autoscaling",
				Resource: "horizontalpodautoscalers",
			},
			Metric:     MetricNameExtAvgMemoryUtilization,
			Namespaced: true,
		},
		{
			GroupResource: schema.GroupResource{
				Group:    "core.pingcap.com",
				Resource: "tiproxygroups",
			},
			Metric:     MetricNameExtAvgBandwidthUtilization,
			Namespaced: true,
		},
	}
}

func (p *cmProvider) GetExternalMetric(
	ctx context.Context,
	namespace string,
	metricSelector labels.Selector,
	info provider.ExternalMetricInfo,
) (*external_metrics.ExternalMetricValueList, error) {
	items := make([]external_metrics.ExternalMetricValue, 0)
	items = append(items, external_metrics.ExternalMetricValue{
		MetricName:   MetricNameDisableScaleIn,
		MetricLabels: map[string]string{},
		Value:        *resource.NewQuantity(p.disableScalueIn, resource.DecimalSI),
		Timestamp:    metav1.Now(),
	})
	return &external_metrics.ExternalMetricValueList{
		Items: items,
	}, nil
}

func (p *cmProvider) ListAllExternalMetrics() []provider.ExternalMetricInfo {
	return []provider.ExternalMetricInfo{
		{
			Metric: MetricNameDisableScaleIn,
		},
	}
}