package core

import (
	"context"
	"errors"
	"fmt"
	"github.com/guo-shaoge/tiflash-autoscale/pkg/config"
	"github.com/guo-shaoge/tiflash-autoscale/pkg/meta"
	"github.com/guo-shaoge/tiflash-autoscale/pkg/metric"
	"github.com/guo-shaoge/tiflash-autoscale/pkg/util/log/logger"
	"go.uber.org/zap"
	"sort"
	"sync"
	"time"
)

const (
	AutoPausePolicyStr string = "auto pause"
	EvictPolicyStr            = "evict tenant"
	MergePolicyStr            = "merge tenant"
	ScalePolicyStr            = "scale in/out"
)

const SyncResumeFailBecauseAlreadyAllocating = "someone is already resuming"

type sharePodInfo struct {
	tenantDesc *meta.TenantDesc
	podDesc    *meta.PodDesc
}

type mergeCandidateInfo struct {
	podName   string
	tenantCnt uint
	avgCPU    float64
	podDesc   *meta.PodDesc
}

type TenantMgr struct {
	ctx context.Context

	mu struct {
		sync.RWMutex
		tenants map[string]*meta.TenantDesc
	}

	podMgr    *PodMgr
	configMgr *ConfigMgr
	// need mutext because two threads may get config:
	// 1. http request may call GetOrCreateTenantDesc, which will need config for tenant.
	// 2. analyzeTaskLoop will refresh config periodically.
	configMu struct {
		sync.RWMutex
		config *config.Config
	}

	logger *zap.Logger

	lastMergeTs time.Time
}

// NewTenantMgr creates new TenantMgr.
func NewTenantMgr(ctx context.Context, podMgr *PodMgr,
	configMgr *ConfigMgr, parentLogger *zap.Logger) *TenantMgr {
	res := &TenantMgr{
		ctx:       ctx,
		podMgr:    podMgr,
		configMgr: configMgr,
		logger:    parentLogger.With(zap.String(logger.LogIDMod, "tenantMgr")),
	}
	res.mu.tenants = make(map[string]*meta.TenantDesc)
	res.configMu.config = configMgr.GetConfig()
	res.lastMergeTs = time.Now()
	return res
}

func (t *TenantMgr) startBgWorker(wg *sync.WaitGroup) {
	wg.Add(1)
	go func() {
		t.analyzeTaskLoop()
	}()
}

func (t *TenantMgr) getConfig() *config.Config {
	t.configMu.RLock()
	defer t.configMu.RUnlock()
	return t.configMu.config
}

func (t *TenantMgr) tryRefreshConfig(oldConfig *config.Config) bool {
	if t.configMgr.HasNewerConfig(oldConfig) {
		t.configMu.Lock()
		defer t.configMu.Unlock()
		t.configMu.config = t.configMgr.GetConfig()
		return true
	}
	return false
}

// GetOrCreateTenantDesc return tenantDesc by name or create if necessary.
func (t *TenantMgr) GetOrCreateTenantDesc(name string, rgName string) (*meta.TenantDesc, bool) {
	res := t.GetTenantDescByName(name)

	if res != nil {
		return res, false
	}

	enableShare := *t.getConfig().GetTenantConfig(name).EnableShare

	t.mu.Lock()
	defer t.mu.Unlock()
	if res, ok := t.mu.tenants[name]; ok {
		return res, false
	}

	res = meta.NewTenantDesc(name, metric.ResourceGroupNameType(rgName), enableShare)
	t.mu.tenants[name] = res
	return res, true
}

// GetAllTenantNames return all tenant names.
func (t *TenantMgr) GetAllTenantNames() []string {
	t.mu.RLock()
	defer t.mu.RUnlock()
	res := make([]string, 0, len(t.mu.tenants))
	for name, _ := range t.mu.tenants {
		res = append(res, name)
	}
	return res
}

// GetTenantDescByName return tenantDesc by name.
func (t *TenantMgr) GetTenantDescByName(name string) *meta.TenantDesc {
	t.mu.RLock()
	defer t.mu.RUnlock()
	res, ok := t.mu.tenants[name]
	if !ok {
		return nil
	}
	return res
}

// SyncResume will resume tenant, it's blocking.
// TODO ctx need to pass down to rpc
func (t *TenantMgr) SyncResume(tenantName string, rgName string, minPod int, sourceMod string, traceId string) (meta.PodsMap, error) {
	tenantDesc, created := t.GetOrCreateTenantDesc(tenantName, rgName)
	if created {
		logger := t.logger.With(zap.Any(logger.LogIDMod, sourceMod), zap.Any(logger.LogIDTrace, traceId))

		curConfig := t.configMgr.GetConfig()
		initCoreNum := *curConfig.GetTenantConfig(tenantName).TenantInitCoreNum
		coreNumPerPod := *curConfig.GetClusterConfig().CoreNumPerPod

		initPod := initCoreNum / coreNumPerPod
		oriMinPod := minPod
		if minPod < int(initPod) {
			minPod = int(initPod)
		}
		tenantDesc.UpdateMemLimitRecoveryTs()

		logger.Info("new tenant created", zap.String("tenant", tenantDesc.ToString()), zap.Any("rgName", rgName),
			zap.Any("oriMinPod", oriMinPod), zap.Any("newMinPod", minPod),
			zap.Any("initCoreNum", initCoreNum), zap.Any("coreNumPerPod", coreNumPerPod),
			zap.Any("initPod", initPod))
	}
	if string(tenantDesc.GetRGName()) != rgName {
		return nil, fmt.Errorf("unexpected resource group name, expect: %v, got: %v", tenantDesc.GetRGName(), rgName)
	}
	return t.syncResumeByTenantDesc(tenantDesc, minPod, sourceMod, traceId)
}

func getPodsStr(podsMap meta.PodsMap) string {
	var resStr string
	for _, podDesc := range podsMap {
		resStr += podDesc.ToString()
	}
	return resStr
}

func (t *TenantMgr) syncResumeByTenantDesc(tenantDesc *meta.TenantDesc,
	minPod int, sourceMod string, traceId string) (res meta.PodsMap, err error) {
	if minPod <= 0 {
		return nil, errors.New("min pod should be greater than zero when SyncResume")
	}

	removed, usablePodsMap := tenantDesc.RemoveUnusablePods()
	curPodsStr := getPodsStr(usablePodsMap)

	logger := t.logger.With(zap.Any(logger.LogIDMod, sourceMod), zap.Any(logger.LogIDTrace, traceId))
	logger.Info("SyncTakePods", zap.Any("minPod", minPod),
		zap.Any("curPods", curPodsStr),
		zap.Any("removed usable", removed))
	if len(usablePodsMap) >= minPod {
		return usablePodsMap, nil
	}

	tenantDesc.ToResumingIfNecessary()

	if tenantDesc.IsAllocatingPods() || !tenantDesc.SetIsAllocatingPods(true) {
		// someone is already allocating, wait.
		return nil, errors.New(SyncResumeFailBecauseAlreadyAllocating)
	}

	defer func() {
		// change to resumed if got any pods.
		if !tenantDesc.ToResumedOrPaused() {
			logger.Info("SyncTakePods failed, pods still zero", zap.Any("tenant", tenantDesc.GetName()))
		}
		if !tenantDesc.SetIsAllocatingPods(false) {
			logger.Error("set allocating pod flag failed")
		}
		tenantDesc.BroadcastAllocPodsFinish()
		logger.Info("SyncTakePods done, BroadcastAllocPodsFinish", zap.Any("len(usablePodsMap)", len(usablePodsMap)),
			zap.Any("pods", getPodsStr(usablePodsMap)))
	}()
	removed, usablePodsMap = tenantDesc.RemoveUnusablePods()
	if len(usablePodsMap) >= minPod {
		return usablePodsMap, nil
	}

	enableShare := tenantDesc.GetEnableShare()

	newPods := t.podMgr.SyncTakePods(uint(minPod-len(usablePodsMap)), tenantDesc.GetName(), enableShare, sourceMod, traceId)
	tenantDesc.AddPodsByDesc(newPods)
	_, usablePodsMap = tenantDesc.RemoveUnusablePods()
	return usablePodsMap, nil
}

// analyzeTaskLoop is the control loop which will create goroutines for each tenant to analyze and scale-out/in.
func (t *TenantMgr) analyzeTaskLoop() {
	for {
		curConfig := t.getConfig()
		var configUpdated bool
		if t.tryRefreshConfig(curConfig) {
			newConfig := t.getConfig()
			t.logger.Info("update config",
				zap.Any("old ts", curConfig.GetUpdateTime()),
				zap.Any("new ts", newConfig.GetUpdateTime()))
			curConfig = newConfig
			configUpdated = true
		}
		intervalSec := *curConfig.GetClusterConfig().MetricIntervalSec + 2
		t.logger.Info("start to analyze", zap.Any("intervalSec", intervalSec))

		tenantNames := t.GetAllTenantNames()
		wg := &sync.WaitGroup{}

		start := time.Now()
		for _, tenantName := range tenantNames {
			tenantDesc := t.GetTenantDescByName(tenantName)
			if tenantDesc == nil {
				t.logger.Info("cannot find tenant, skip analyze", zap.Any("name", tenantName))
				continue
			}

			if configUpdated {
				tenantConfig := curConfig.GetTenantConfig(tenantName)
				needDerefSharingPods := tenantDesc.UpdateEnableShare(*tenantConfig.EnableShare)
				if needDerefSharingPods {
					errMsg := tenantDesc.DerefAllSharingPods("tenant config disable sharing")
					t.logger.Info("deref all sharing pods for tenant because sharing disabled",
						zap.Any("name", tenantName), zap.Any("err", errMsg))
				}
			}

			wg.Add(1)
			go func() {
				// main analyze func.
				defer wg.Done()
				t.doAnalyzeMain(curConfig, tenantDesc)
			}()
		}

		doneCh := make(chan struct{})
		go func() {
			wg.Wait()
			close(doneCh)
		}()

		select {
		case <-doneCh:
			t.logger.Info("all main analyze task done successfully, start merge analyze",
				zap.Any("elapsed", time.Since(start)))
		case <-t.ctx.Done():
			t.logger.Info("got context cancel, finish analyzeTaskLoop")
			return
		}

		// merge analyze to avoid too many pods in share pool.
		start = time.Now()
		t.doAnalyzeMergePods(start, intervalSec, curConfig, tenantNames)
		t.logger.Info("merge analyze done", zap.Any("elapsed", time.Since(start)))

		select {
		case <-time.After(time.Duration(intervalSec) * time.Second):
		case <-t.ctx.Done():
			t.logger.Info("got ctx.Done, exit")
			return
		}
	}
}

// why: in `tryEvict()`, will evict tenant from pods that is not suitable to share,
//
//	which will cause too many new pods allocated. so need to merge these pods in SharePool.
//
// how:
//  1. get all sharing pods for all tenants.
//  2. get all merge candidate(only consider cpu metric for now) from all sharing pods.
//  3. sort all candidate pods by avg cpu and tenant cnt.
//  4. get replace info(src tenant -> target pod)
//  5. start replace.
func (t *TenantMgr) doAnalyzeMergePods(start time.Time, analyzeInterval uint, config *config.Config, tenantNames []string) {
	logger := t.logger.With(zap.Any("policy", MergePolicyStr))
	clusterConfig := config.GetClusterConfig()
	sharePoolConfig := clusterConfig.SharePoolConfig
	if !*sharePoolConfig.EnableMerge {
		logger.Info("skip merge share pool because merge not enable")
		return
	}

	coreNumPerPod := *config.GetClusterConfig().CoreNumPerPod

	mergeIntervalSec := *clusterConfig.AutoPauseWindowSec + analyzeInterval*2

	okTs := t.lastMergeTs.Add(time.Duration(mergeIntervalSec) * time.Second)
	if start.Before(okTs) {
		logger.Info("skip merge share pool because merge interval not reach",
			zap.Any("merge interval", mergeIntervalSec), zap.Any("start", start),
			zap.Any("lastMergeTs", t.lastMergeTs), zap.Any("ok", okTs))
		return
	}

	allSharePods := t.getAllSharePods(tenantNames, logger)
	if len(allSharePods)*int(coreNumPerPod) <= int(*sharePoolConfig.SoftHighCoreNum) {
		logger.Info("skip merge share pool", zap.Any("sharing pods num", len(allSharePods)),
			zap.Any("coreNumPerPod", coreNumPerPod), zap.Any("soft limit", *sharePoolConfig.SoftHighCoreNum))
		return
	}
	t.lastMergeTs = time.Now()

	mergeCandidates := t.getAllMergeCandidates(sharePoolConfig, clusterConfig, allSharePods, logger)
	if len(mergeCandidates) <= 1 {
		return
	}

	replaceInfo, msg := t.getReplaceInfo(sharePoolConfig, mergeCandidates, coreNumPerPod, logger)
	if len(msg) != 0 {
		logger.Info("get replace info fail", zap.String("msg", msg))
		return
	}

	var logMsg string
	for fromPodName, toPod := range replaceInfo {
		logMsg += fmt.Sprintf("%s->%s(%s); ", fromPodName, toPod.GetName(), toPod.GetStatusStr())
	}
	logger.Info("replace info: ", zap.String("info", logMsg))

	t.doReplace(replaceInfo, tenantNames, logger)
}

// getAllSharePods return all running/sharing pods for all sharable tenants.
func (t *TenantMgr) getAllSharePods(tenantNames []string, logger *zap.Logger) map[string]*sharePodInfo {
	allSharePods := make(map[string]*sharePodInfo)

	filter := func(pod *meta.PodDesc) bool {
		return pod.IsSharing() || pod.IsRunning()
	}

	for _, tenantName := range tenantNames {
		tenantDesc := t.GetTenantDescByName(tenantName)
		if tenantDesc == nil {
			logger.Info("cannot find tenant", zap.Any("tenant", tenantName))
			continue
		}
		if !tenantDesc.IsResumed() {
			continue
		}
		if !tenantDesc.GetEnableShare() {
			logger.Info("tenant not enable share", zap.Any("tenant", tenantName))
			continue
		}
		sharePods := tenantDesc.GetPods(filter)
		for _, pod := range sharePods {
			allSharePods[pod.GetName()] = &sharePodInfo{
				tenantDesc: tenantDesc,
				podDesc:    pod,
			}
		}
	}
	return allSharePods
}

// getAllMergeCandidates returns all pods that can be merge candidate(core logic: PodDesc.OkToMerge)
// 1. find all sharing pods, check if ok to merge(cpu/cooldown)
func (t *TenantMgr) getAllMergeCandidates(
	sharePoolConfig *config.SharePoolConfig,
	clusterConfig *config.ClusterConfig,
	allSharePods map[string]*sharePodInfo,
	logger *zap.Logger) []*mergeCandidateInfo {
	var mergeCandidates []*mergeCandidateInfo

	for podName, info := range allSharePods {
		msg, avgCPU, ok := info.podDesc.OkToMerge(
			*sharePoolConfig.MergeLowCPULimit, *sharePoolConfig.MergeCoolDownSec,
			*clusterConfig.MetricWindowSec, *clusterConfig.MetricIntervalSec,
			*clusterConfig.CoreNumPerPod)
		if !ok {
			logger.Info("okToMerge info", zap.String("msg", msg))
			continue
		}

		mergeCandidates = append(mergeCandidates, &mergeCandidateInfo{
			podName:   podName,
			tenantCnt: info.podDesc.GetTenantCnt(),
			avgCPU:    avgCPU,
			podDesc:   info.podDesc,
		})
	}

	sort.Slice(mergeCandidates, func(i, j int) bool {
		c1 := mergeCandidates[i]
		c2 := mergeCandidates[j]
		if c1.avgCPU == c2.avgCPU {
			return c1.tenantCnt < c2.tenantCnt
		}
		return c1.avgCPU < c2.avgCPU
	})
	return mergeCandidates
}

// getReplaceInfo returns <fromPod, toPod> map, need to migrate all tenants of fromPod to toPod.
func (t *TenantMgr) getReplaceInfo(
	sharePoolConfig *config.SharePoolConfig,
	candidates []*mergeCandidateInfo,
	coreNumPerPod uint,
	logger *zap.Logger) (map[string]*meta.PodDesc, string) {

	var msg string
	if len(candidates) < 2 {
		msg = "merge candidates less than 2, skip merge"
		return nil, msg
	}

	replaceInfo := make(map[string]*meta.PodDesc)
	mergeHighCPU := *sharePoolConfig.MergeHighCPULimit * float64(coreNumPerPod)

	prevCPUSum := candidates[0].avgCPU
	prevTenantSum := candidates[0].tenantCnt
	prevCandidate := candidates[0]
	for i := 1; i < len(candidates); i++ {
		curCandidate := candidates[i]
		if prevCPUSum+curCandidate.avgCPU >= mergeHighCPU {
			break
		}

		// replace `from` pod to `to` pod
		var from *mergeCandidateInfo
		var to *mergeCandidateInfo
		if prevTenantSum < curCandidate.tenantCnt {
			from = prevCandidate
			to = curCandidate
		} else {
			from = curCandidate
			to = prevCandidate
		}

		prevCPUSum += curCandidate.avgCPU
		prevTenantSum += curCandidate.tenantCnt
		prevCandidate = to

		replaceInfo[from.podName] = to.podDesc
	}
	return replaceInfo, ""
}

// doReplace do the replace according to replaceInfo.
func (t TenantMgr) doReplace(replaceInfo map[string]*meta.PodDesc, tenantNames []string, logger *zap.Logger) {
	for _, tenantName := range tenantNames {
		tenantDesc := t.GetTenantDescByName(tenantName)
		if tenantDesc == nil {
			logger.Info("cannot find tenant, skip merge", zap.Any("tenant", tenantName))
			continue
		}
		if !tenantDesc.IsResumed() {
			logger.Info("tenant not resumed, skip merge", zap.Any("tenant", tenantName))
			continue
		}
		if !tenantDesc.GetEnableShare() {
			logger.Info("tenant not enable share, skip merge", zap.Any("tenant", tenantName))
			continue
		}
		replaceMsg, err := tenantDesc.ReplaceMergePods(replaceInfo)
		if err != nil {
			logger.Error("tenant replace fail", zap.Error(err))
			continue
		}
		logger.Info("tenant replace pods successfully", zap.String("msg", replaceMsg), zap.String("tenant", tenantName))
	}
}

func (t *TenantMgr) doAnalyzeMain(config *config.Config, tenantDesc *meta.TenantDesc) {
	tenantName := tenantDesc.GetName()
	rgName := tenantDesc.GetRGName()
	logger := t.logger.With(zap.Any("tenant", tenantName), zap.Any("rgName", rgName))
	if !tenantDesc.IsResumed() {
		logger.Info("tenant is not resumed, skip analyze")
		return
	}

	start := time.Now()
	logger.Info("analyze for tenant begins")
	defer func() {
		logger.Info("analyze for tenant ends", zap.Any("elapsed", time.Since(start)))
	}()

	// 1. AutoPause
	// 2. Evict: if shared pod is not rich, move self if workload is not highest.
	// 4. ScaleIn/ScaleOut
	//     1. CPU dim
	//     2. Mem dim
	//     3. Static dim
	//     4. call ResizePods()

	tenantConfig := config.GetTenantConfig(tenantName)

	if *tenantConfig.EnableAutoPause {
		t.tryAutoPause(config, tenantDesc)
	}
	if !tenantDesc.IsResumed() {
		logger.Info("tenant auto pause succeed, skip analyze", zap.String("state", tenantDesc.GetStatusStr()))
		return
	}

	if tenantDesc.HasSharingPods() {
		t.tryEvict(config, tenantDesc)
	}

	t.tryScale(config, tenantDesc)
}

// if the mpp task is zero for all pods of the tenant for N seconds, we can pause this tenant.
// return true if trigger auto pause.
func (t *TenantMgr) tryAutoPause(config *config.Config, tenantDesc *meta.TenantDesc) bool {
	tenantName := tenantDesc.GetName()
	rgName := tenantDesc.GetRGName()
	logger := t.logger.With(zap.Any("tenant", tenantName), zap.Any("rgName", rgName), zap.Any("policy", AutoPausePolicyStr))
	podsTaskCntMetricWindow := tenantDesc.GetTaskCntMetricWindow()
	clusterConfig := config.GetClusterConfig()

	lastResumedTs := tenantDesc.GetLastResumedTs()
	okTs := lastResumedTs.Add(time.Duration(*clusterConfig.AutoPauseWindowSec) * time.Second)
	now := time.Now()
	if now.Before(okTs) {
		logger.Info("skip auto pause because resume not cooldown",
			zap.Any("lastResumedTs", lastResumedTs), zap.Any("okTs", okTs))
		return false
	}
	lastMergeTs := tenantDesc.GetLastMergeTs()
	okTs = lastMergeTs.Add(time.Duration(*clusterConfig.AutoPauseWindowSec) * time.Second)
	if now.Before(okTs) {
		logger.Info("skip auto pause because merge not cooldown",
			zap.Any("lastMergeTs", lastMergeTs), zap.Any("okTs", okTs))
		return false
	}
	lastEvictTs := tenantDesc.GetLastEvictTs()
	okTs = lastEvictTs.Add(time.Duration(*clusterConfig.AutoPauseWindowSec) * time.Second)
	if now.Before(okTs) {
		logger.Info("skip auto pause because evict not cooldown",
			zap.Any("lastEvictTs", lastEvictTs), zap.Any("okTs", okTs))
		return false
	}

	for podName, window := range podsTaskCntMetricWindow {
		validNum, err := metric.IsTaskCntWindowValid(*clusterConfig.AutoPauseWindowSec,
			*clusterConfig.MetricIntervalSec, tenantDesc.GetRGName(), window)
		if err != nil {
			logger.Info("failed to auto pause", zap.String("pod", podName), zap.Error(err))
			return false
		}
		if validNum == 0 {
			logger.Info("no valid task cnt metric point for this pod, consider it no task running", zap.String("pod", podName))
			continue
		}

		var gotFirstTaskCnt bool
		var firstTaskCnt uint
		for _, point := range window {
			taskCnt, ok := point.GetTaskCnt(rgName)
			if !ok {
				logger.Info("cannot get task cnt metric point by tenant, skip this point")
				continue
			}
			if !gotFirstTaskCnt {
				firstTaskCnt = taskCnt
				gotFirstTaskCnt = true
			} else if firstTaskCnt != taskCnt {
				logger.Info("skip auto pause because cnt is not zero",
					zap.String("pod", podName),
					zap.Any("metric ts", point.GetTs()),
					zap.Any("first taskcnt", firstTaskCnt),
					zap.Any("taskCnt", taskCnt))
				return false
			}
		}
	}

	trigger, msg := tenantDesc.AutoPause("tenantMgr tryAutoPause")
	if len(msg) != 0 {
		logger.Error("AutoPause failed", zap.String("msg", msg))
		return false
	}
	logger.Info("auto pause done", zap.Any("trigger", trigger))
	return trigger
}

// 1. find all pods of this tenant that is not suitable to share
// 2. for each evict pod, find highest ru usage tenant. evict self tenant if not the highest tenant.
func (t *TenantMgr) tryEvict(config *config.Config, tenantDesc *meta.TenantDesc) (trigger bool) {
	clusterConfig := config.GetClusterConfig()
	sharePoolConfig := clusterConfig.SharePoolConfig
	tenantName := tenantDesc.GetName()
	rgName := tenantDesc.GetRGName()
	logger := t.logger.With(zap.Any("tenant", tenantName), zap.Any("rgName", rgName), zap.Any("policy", EvictPolicyStr))

	if !*sharePoolConfig.EnableEvict {
		logger.Info("skip evict pods because evict not enable")
		return
	}

	podsCPUMetricWindow := tenantDesc.GetCPUMetricWindow()
	// Pods that not suitable for sharing, try to migrate self tenant to other pods.
	var evictPods []string
	highCPULimit := *sharePoolConfig.EvictHighCPULimit * float64(*clusterConfig.CoreNumPerPod)

	for podName, window := range podsCPUMetricWindow {
		err := metric.IsWindowValid(*clusterConfig.MetricWindowSec, *clusterConfig.MetricIntervalSec, len(window))
		if err != nil {
			logger.Info("failed to evict sharing tenant", zap.String("pod", podName), zap.Error(err))
			return
		}
		if window.GetAvgCPU() >= highCPULimit {
			evictPods = append(evictPods, podName)
		}
	}

	if len(evictPods) == 0 {
		logger.Info("no evict pods", zap.Any("evict high cpu", highCPULimit))
		return
	}

	podsRUMetricWindow := tenantDesc.GetRUMetricWindow()
	realEvictPods := make([]string, 0, len(evictPods))
	for _, evictPodName := range evictPods {
		podDesc := tenantDesc.GetPodByName(evictPodName)
		if podDesc == nil {
			logger.Error("cannot find pod from tenantDesc to evict", zap.String("pod", evictPodName))
			continue
		}
		if !podDesc.IsSharing() {
			logger.Info("pod is not share status to evict", zap.String("status", podDesc.GetStatusStr()))
			continue
		}

		window, ok := podsRUMetricWindow[evictPodName]
		if !ok {
			logger.Info("cannot find ru metric for pod", zap.Any("pod", evictPodName))
			continue
		}

		if err := metric.IsWindowValid(*clusterConfig.MetricWindowSec, *clusterConfig.MetricIntervalSec, len(window)); err != nil {
			logger.Info("ru metric window not valid, skip evict", zap.Any("pod", evictPodName), zap.Error(err))
			continue
		}

		highestRG := metric.GetHighestRUUsageTenant(window, logger)

		if len(highestRG) == 0 {
			logger.Info("get highest ru usage tenant failed")
			return false
		}
		if highestRG != rgName {
			// Evict all tenants except the highest ru usage tenant.
			realEvictPods = append(realEvictPods, evictPodName)
		} else {
			logger.Info("pod resource is not suitable to share, self is highest user. others will migrate",
				zap.Any("pod", evictPodName))
		}
	}

	if len(realEvictPods) == 0 {
		logger.Info("no pod to evict for tenant")
		return false
	}

	traceId := fmt.Sprintf("tryEvict-%v", rgName)
	if err := tenantDesc.RemovePodsByName(realEvictPods, traceId); err != nil {
		logger.Error("remove pods by name failed", zap.Any("evict pods", realEvictPods), zap.Error(err))
		return false
	}

	logger.Info("migrate self tenant because pod not suitable for sharing",
		zap.Any("evict pods", realEvictPods), zap.Any("after evict pods num", tenantDesc.GetPodsNum()))

	if _, err := t.SyncAddPodsByNum(tenantDesc, uint(len(realEvictPods)), "tenantMgr", traceId); err != nil {
		logger.Error("failed to add new pods after evict sharing pods",
			zap.Error(err), zap.Any("cur pods cnt", tenantDesc.GetPodsNum()))
	} else {
		logger.Info("get pod successfully because of evict", zap.Any("cur pods cnt", tenantDesc.GetPodsNum()))
	}
	tenantDesc.UpdateLastEvictTs(time.Now())
	return true
}

func (t *TenantMgr) tryScale(config *config.Config, tenantDesc *meta.TenantDesc) (trigger bool) {
	tenantName := tenantDesc.GetName()
	rgName := tenantDesc.GetRGName()
	logger := t.logger.With(zap.Any("tenant", tenantDesc.GetName()), zap.Any("rgName", rgName), zap.Any("policy", ScalePolicyStr))
	oriPodNum := tenantDesc.GetPodsNum()

	clusterConfig := config.GetClusterConfig()
	tenantConfig := config.GetTenantConfig(tenantName)

	var (
		ignoreCPUDim     bool
		ignoreMemDim     bool
		bestPodNumCPUDim int
		bestPodNumMemDim int
	)
	if *tenantConfig.EnableScale {
		// CPU Dim
		ignoreCPUDim, bestPodNumCPUDim = tryScaleCPUDim(oriPodNum, clusterConfig, tenantConfig, tenantDesc, logger)

		// OOM Dim
		// NOTE: for now mem dim is disabled, because tidb will do MemLimit recovery.
		// TODO: but the policy is relative simple.
		//    1. Merge/Evict should also consider Mem.
		//    2. should add a policy for mem dim just like cpu dim.
		ignoreMemDim, bestPodNumMemDim = tryScaleMemDim(oriPodNum, clusterConfig, tenantDesc, logger)
	} else {
		logger.Info("tenant disable scale, skip cpu/mem dim")
	}

	// Static Dim
	coreNumPerPod := *config.GetClusterConfig().CoreNumPerPod
	minPodNum := int((*tenantConfig.TenantMinCoreNum) / coreNumPerPod)
	maxPodNum := int((*tenantConfig.TenantMaxCoreNum) / coreNumPerPod)

	var bestPodNum int
	if ignoreCPUDim && ignoreMemDim {
		bestPodNumCPUDim = oriPodNum
		bestPodNumMemDim = oriPodNum
		bestPodNum = oriPodNum
	} else if !ignoreCPUDim && !ignoreMemDim {
		bestPodNum = bestPodNumCPUDim
		if bestPodNum < bestPodNumMemDim {
			bestPodNum = bestPodNumMemDim
		}
	} else if !ignoreCPUDim && ignoreMemDim {
		bestPodNum = bestPodNumCPUDim
	} else if ignoreCPUDim && !ignoreMemDim {
		bestPodNum = bestPodNumMemDim
	}

	if bestPodNum < minPodNum {
		bestPodNum = minPodNum
	} else if bestPodNum > maxPodNum {
		bestPodNum = maxPodNum
	}

	removed, _ := tenantDesc.RemoveUnusablePods()
	logger.Info("remove unused pods", zap.Any("removed", removed))
	var scaled bool
	if oriPodNum != bestPodNum {
		scaled = true
	}

	logger.Info("start to SyncResizePods",
		zap.Any("cpu dim", bestPodNumCPUDim),
		zap.Any("mem dim", bestPodNumMemDim),
		zap.Any("static dim", fmt.Sprintf("%v/%v", minPodNum, maxPodNum)),
		zap.Any("oriPodNum", oriPodNum),
		zap.Any("final", bestPodNum),
		zap.Any("scaled", scaled))

	delta, err := t.SyncResizePods(tenantDesc, uint(bestPodNum), "tenantMgr", fmt.Sprintf("tryScale-%s", rgName))
	if err != nil {
		logger.Info("resize pod failed", zap.Any("delta", delta),
			zap.Any("cur num", tenantDesc.GetPodsNum()), zap.Error(err))
		return false
	}
	if delta != 0 {
		logger.Info("resize pod succeed", zap.Any("delta", delta), zap.Any("cur num", tenantDesc.GetPodsNum()))
	} else {
		logger.Info("no need to resize", zap.Any("cur num", tenantDesc.GetPodsNum()))
	}
	return true
}

func tryScaleCPUDim(oriPodNum int,
	clusterConfig *config.ClusterConfig,
	tenantConfig *config.TenantConfig,
	tenantDesc *meta.TenantDesc,
	logger *zap.Logger) (ignoreCPUDim bool, bestPodNumCPUDim int) {

	ok, msg := tenantDesc.MemLimitRecoveryCoolDown(*clusterConfig.MemLimitRecoveryCoolDownSec)
	if !ok {
		logger.Info("ignore scale cpu dim, because mem limit cool down", zap.Any("msg", msg))
		return true, 0
	}

	var tenantTotalCPU float64
	podsCPUWindow := tenantDesc.GetCPUMetricWindow()
	podsRUWindow := tenantDesc.GetRUMetricWindow()
	var podCPU string
	for podName, window := range podsCPUWindow {
		// TODO: it means if a new pod added to this tenant and there is no enough metric points.
		// This tenant will no scale by CPU.
		if err := metric.IsWindowValid(*clusterConfig.MetricWindowSec,
			*clusterConfig.MetricIntervalSec, len(window)); err != nil {
			logger.Info("failed to scale cpu dim, ignore cpu dim, because cpu metric point error",
				zap.String("pod", podName), zap.Error(err))
			ignoreCPUDim = true
			break
		}

		ruWindow, ok := podsRUWindow[podName]
		if !ok {
			logger.Info("failed to scale cpu dim, because ru metric not found", zap.String("pod", podName))
			ignoreCPUDim = true
			break
		}

		validNum, err := metric.IsRUWindowValid(*clusterConfig.MetricWindowSec,
			*clusterConfig.MetricIntervalSec, tenantDesc.GetRGName(), ruWindow)
		if err != nil {
			logger.Info("failed to scale cpu dim, ignore cpu dim, because ru metric point error",
				zap.String("pod", podName), zap.Error(err))
			ignoreCPUDim = true
			break
		}
		if validNum == 0 {
			logger.Info("failed to scale cpu dim, because no valid metric point for this tenant")
			ignoreCPUDim = true
			break
		}

		ruRate, msg := metric.GetRURate(tenantDesc.GetRGName(), ruWindow)
		if len(msg) != 0 {
			logger.Info("failed to scale cpu dim, ignore cpu dim",
				zap.String("pod", podName), zap.String("msg", msg))
		}

		avgCPU := window.GetAvgCPU()
		avgCPU1 := avgCPU * ruRate
		tenantTotalCPU += avgCPU1
		podCPU += fmt.Sprintf("%v*%v=%v;", avgCPU, ruRate, avgCPU1)
	}

	if !ignoreCPUDim {
		avgCPUPerPod := tenantTotalCPU / float64(len(podsCPUWindow))
		low := (*tenantConfig.ScaleLowCPULimit) * float64(*clusterConfig.CoreNumPerPod)
		high := (*tenantConfig.ScaleHighCPULimit) * float64(*clusterConfig.CoreNumPerPod)
		var skipScale bool
		if avgCPUPerPod >= low && avgCPUPerPod <= high {
			skipScale = true
		}
		if !skipScale {
			half := (low + high) / 2
			bestPodNumCPUDim = int(tenantTotalCPU / half)
			if tenantTotalCPU-float64(bestPodNumCPUDim)*half >= half || bestPodNumCPUDim == 0 {
				bestPodNumCPUDim++
			}
		}

		// podCPU * podNum == tenantTotalCPU
		// tenantTotalCPU / podNum == avgCPUPerPod(cpu per pod)
		logger.Info("cpu dim", zap.Bool("skip scale by cpu dim", skipScale),
			zap.Any("perPodCPU", podCPU),
			zap.Any("tenantTotalCPU", tenantTotalCPU),
			zap.Any("avgCPUPerPod", avgCPUPerPod),
			zap.Any("low", low),
			zap.Any("high", high),
			zap.Any("best in cpu dim", bestPodNumCPUDim))
	}
	return ignoreCPUDim, bestPodNumCPUDim
}

func tryScaleMemDim(oriPodNum int,
	clusterConfig *config.ClusterConfig,
	tenantDesc *meta.TenantDesc,
	logger *zap.Logger) (ignoreMemDim bool, bestPodNumMemDim int) {

	return true, 0

	// if !tenantDesc.MemLimitRecoveryCoolDown(*clusterConfig.MemLimitRecoveryCoolDownSec) {
	// 	return true, 0
	// }
	// podsMemWindow := tenantDesc.GetOOMMetricWindow()
	// for podName, window := range podsMemWindow {
	// 	if err := metric.IsWindowValid(*clusterConfig.MetricWindowSec,
	// 		*clusterConfig.MetricIntervalSec, len(window)); err != nil {
	// 		logger.Info("failed to scale mem dim, ignore mem dim",
	// 			zap.String("pod", podName), zap.Error(err))
	// 		ignoreMemDim = true
	// 		break
	// 	}
	// }

	// if !ignoreMemDim {
	// 	var oomPod string
	// 	var oomWindow metric.OOMMetricWindow
	// 	for podName, window := range podsMemWindow {
	// 		last := len(window)
	// 		if window[0].GetOOM() < window[last-1].GetOOM() {
	// 			oomPod = podName
	// 			oomWindow = window
	// 			break
	// 		}
	// 	}

	// 	if len(oomPod) > 0 {
	// 		bestPodNumMemDim = oriPodNum * 2
	// 	}

	// 	logger.Info("mem dim", zap.Bool("skip scale by mem dim", len(oomPod) == 0),
	// 		zap.Any("oom pod", oomPod),
	// 		zap.Any("oom window", oomWindow.ToString()))
	// }
	// return ignoreMemDim, bestPodNumMemDim
}

// called by resumed tenant to resize pods. difference between TenantDesc.SyncResume:
//  1. this func can also deref pods.
//  2. this func is only for resumed tenants.
//
// called by TenantMgr.analyzeTaskLoop()
func (t *TenantMgr) SyncResizePods(tenantDesc *meta.TenantDesc, expectedNum uint,
	sourceMod string, traceId string) (delta int, err error) {
	needResize, delta := tenantDesc.CheckResumedAndGetPodsNum(expectedNum)
	if !needResize {
		return 0, nil
	}

	if expectedNum == 0 {
		return 0, fmt.Errorf("should not resize pod to zero for resumed tenant")
	}

	if delta == 0 {
		return 0, nil
	} else if delta > 0 {
		newPods, err := t.SyncAddPodsByNum(tenantDesc, uint(delta), sourceMod, traceId)
		return len(newPods), err
	} else {
		removed, errMsg := tenantDesc.RemovePodsByNum(uint(-delta), "tenantMgr SyncResizePods")
		if len(errMsg) != 0 {
			return -int(removed), errors.New(errMsg)
		}
		return -int(removed), nil
	}
}

// call podMgr to take pods from podMgr.
// called by ResizePods(called by tryScale) and tryEvict(to get new pods after tenant be evicted)
func (t *TenantMgr) SyncAddPodsByNum(tenantDesc *meta.TenantDesc, delta uint,
	sourceMod string, traceId string) ([]*meta.PodDesc, error) {
	res := t.podMgr.SyncTakePods(delta, tenantDesc.GetName(), tenantDesc.GetEnableShare(), sourceMod, traceId)
	tenantDesc.AddPodsByDesc(res)
	return res, nil
}

// GetTopo return fqdn addr for all pods of one tenant.
func (t *TenantMgr) GetTopo(tenantName string) ([]string, []string) {
	tenantDesc := t.GetTenantDescByName(tenantName)
	if tenantDesc == nil {
		return []string{}, []string{}
	}

	tenantDesc.RemoveUnusablePods()
	return tenantDesc.GetTopo()
}

// GetTopoFQDN return fqdn addr for all pods of one tenant.
func (t *TenantMgr) GetTopoFQDN(tenantName string) ([]string, []string) {
	tenantDesc := t.GetTenantDescByName(tenantName)
	if tenantDesc == nil {
		return []string{}, []string{}
	}

	tenantDesc.RemoveUnusablePods()
	return tenantDesc.GetTopoFQDN()
}

func (t *TenantMgr) WaitAllocPodsFinish(name string, timeout time.Duration) {
	tenantDesc := t.GetTenantDescByName(name)
	if tenantDesc == nil {
		return
	}
	tenantDesc.WaitAllocPodsFinish(timeout)
}