package oltppkg

import (
	"context"
	"errors"
	"fmt"
	"io"
	"math"
	"net/url"
	"reflect"
	"strings"
	"time"

	"github.com/onsi/gomega"
	. "github.com/onsi/gomega"
	pdclient "github.com/pingcap/endless/pkg/client/pd"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/matcher/metrics"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/prometheus/client_golang/api"
	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
	"github.com/prometheus/common/model"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

type WorkloadCheckParameters struct {
	IsCheck   bool
	IgnoreErr []string
}
type QpsJitterCheckParameters struct {
	IsCheck      bool
	Offset       string
	ToleratePerc float64
	TolerateDur  time.Duration
}
type DurJitterCheckParameters struct {
	IsCheck      bool
	Offset       string
	ToleratePerc float64
	TolerateDur  time.Duration
}
type ApplyDbMutexLockDurCheckParameters struct {
	IsCheck        bool
	TolerateMaxDur float64
}
type RegionSizeCheckParameters struct {
	IsCheck      bool
	TolerateSize int
	TolerateDur  time.Duration
}
type RestartCheckParameters struct {
	AllComponentRestartCheck bool
	TiKVRestartCheck         bool
	PDRestartCheck           bool
	TiDBRestartCheck         bool
}

type OomCheckParameters struct {
	AllComponentOomCheck bool
	TiKVOomCheck         bool
	PDOomCheck           bool
	TiDBOomCheck         bool
}
type EmptyRegionCheckParameters struct {
	IsCheck          bool
	RatioThreshhold  float64
	NumberThreshhold int64
}

type RaftEngineAppendFileCountParameters struct {
	IsCheck       bool
	TolerateCount int
	TolerateDur   time.Duration
}

type TidbPanicAndCriticalErrCheckParameter struct {
	IsPanicCheck    bool
	IsCriticalCheck bool
}

func FunCommonCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, pr *io.PipeReader, params ...interface{}) {

	workloadCheckParms := WorkloadCheckParameters{
		IsCheck: true,
		//Error 8022,8028,9004,9007(https://docs.pingcap.com/tidb/stable/error-codes/)
		//Error 1213(https://docs.pingcap.com/tidb/stable/pessimistic-transaction/)
		//Error 1105(https://github.com/tikv/tikv/issues/11649)
		//Error 1205(https://docs.pingcap.com/zh/tidb/stable/troubleshoot-lock-conflicts#lock-wait-timeout-exceeded)
		//Error 9007(https://github.com/tikv/tikv/issues/11612, https://github.com/pingcap/tidb/issues/27197)
		//Error 1062(https://github.com/pingcap/tidb/issues/30938)
		//For ycsb workload, ignore %_error and panic message
		//todo: check if all error is reasonable
		IgnoreErr: []string{"1213", "1105", "1205", "8022", "8028", "9004", "9007", "1062", "_error", "panic"},
	}
	qpsJitterCheckParms := QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.8,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := DurJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	applyDbMutexLockDurCheckParms := ApplyDbMutexLockDurCheckParameters{
		TolerateMaxDur: 0.5, //0.5(s)
	}
	regionSizeCheckParms := RegionSizeCheckParameters{
		IsCheck:      false,
		TolerateSize: 1073741824,
		TolerateDur:  10 * time.Minute,
	}
	restartCheckParms := RestartCheckParameters{
		AllComponentRestartCheck: true,
		TiKVRestartCheck:         true,
		PDRestartCheck:           true,
		TiDBRestartCheck:         true,
	}
	oomCheckParms := OomCheckParameters{
		AllComponentOomCheck: true,
		TiKVOomCheck:         true,
		PDOomCheck:           true,
		TiDBOomCheck:         true,
	}
	emptyRegionCheckParms := EmptyRegionCheckParameters{
		IsCheck:          false,
		RatioThreshhold:  0.1,
		NumberThreshhold: 1000,
	}
	raftEngineAppendFileCountCheckParms := RaftEngineAppendFileCountParameters{
		IsCheck:       true,
		TolerateCount: 200,
		TolerateDur:   300 * time.Second,
	}
	tidbPanicAndCriticalErrCheckParms := TidbPanicAndCriticalErrCheckParameter{
		IsPanicCheck:    true,
		IsCriticalCheck: true,
	}
	for _, param := range params {
		switch reflect.TypeOf(param).Name() {
		case "WorkloadCheckParameters":
			workloadCheckParms = param.(WorkloadCheckParameters)
		case "QpsJitterCheckParameters":
			qpsJitterCheckParms = param.(QpsJitterCheckParameters)
		case "DurJitterCheckParameters":
			durJitterCheckParms = param.(DurJitterCheckParameters)
		case "ApplyDbMutexLockDurCheckParameters":
			applyDbMutexLockDurCheckParms = param.(ApplyDbMutexLockDurCheckParameters)
		case "RegionSizeCheckParameters":
			regionSizeCheckParms = param.(RegionSizeCheckParameters)
		case "RestartCheckParameters":
			restartCheckParms = param.(RestartCheckParameters)
		case "OomCheckParameters":
			oomCheckParms = param.(OomCheckParameters)
		case "EmptyRegionCheckParameters":
			emptyRegionCheckParms = param.(EmptyRegionCheckParameters)
		case "RaftEngineAppendFileCountParameters":
			raftEngineAppendFileCountCheckParms = param.(RaftEngineAppendFileCountParameters)
		case "TidbPanicAndCriticalErrCheckParameter":
			tidbPanicAndCriticalErrCheckParms = param.(TidbPanicAndCriticalErrCheckParameter)
		default:
			fmt.Printf("param:%v is an unknown type, using default parameters.\n", param)
		}
	}
	if pr != nil {
		g.Go(WithGinkgoRecover(func() {
			log.Info("workload log check", zap.Strings("IgnoreErr", workloadCheckParms.IgnoreErr))
			err := WorkLoadErrLogCheck(ctx, pr, timeoutDuration, workloadCheckParms.IsCheck, workloadCheckParms.IgnoreErr)
			Expect(err).Should(Succeed())
		}))
	}
	ClusterRestartCheck(ctx, tc, g, timeoutDuration, restartCheckParms)
	_, ok := interface{}(tc).(resource.TiDBCluster)
	if ok {
		ClusterOomCheck(ctx, interface{}(tc).(resource.TiDBCluster), g, timeoutDuration, oomCheckParms)
	}

	TiDBPanicAndCriticalErrCheck(ctx, tc, g, timeoutDuration, tidbPanicAndCriticalErrCheckParms)
	RegionSizeCheck(ctx, tc, g, timeoutDuration, regionSizeCheckParms)
	RaftEngineGcCheck(ctx, tc, g, timeoutDuration, raftEngineAppendFileCountCheckParms)
	ApplyDbMutexLockDurCheck(ctx, tc, g, timeoutDuration, applyDbMutexLockDurCheckParms)
	QpsJitterCheck(ctx, tc, g, timeoutDuration, qpsJitterCheckParms)
	DurJitterCheck(ctx, tc, g, timeoutDuration, durJitterCheckParms)
	EmptyRegionCheck(ctx, tc, g, timeoutDuration, emptyRegionCheckParms)

}

func TiDBPanicAndCriticalErrCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, tidbPanicAndCriticalErrCheckParms TidbPanicAndCriticalErrCheckParameter) {
	if tidbPanicAndCriticalErrCheckParms.IsPanicCheck {
		g.Go(WithGinkgoRecover(func() {
			log.Info("TiDB connection panic check.")
			ConsistentlyWithMetrics(tc, timeoutDuration).
				ShouldNot(TiDBConnectionPanic().WithCtx(ctx))
			log.Info("TiDB connection panic check done.")
		}))
	}
	if tidbPanicAndCriticalErrCheckParms.IsCriticalCheck {
		g.Go(WithGinkgoRecover(func() {
			log.Info("TiDB critical error check.")
			ConsistentlyWithMetrics(tc, timeoutDuration).
				ShouldNot(TiDBCriticalErrorCheck().WithCtx(ctx))
			log.Info("TiDB critical error check done.")
		}))
	}
}

func RaftEngineGcCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, raftEngineAppendFileCountCheckParams RaftEngineAppendFileCountParameters) {
	tikvVer := GetCompVersion(ctx, tc, "tikv")
	curVerShort := strings.TrimSuffix(strings.TrimPrefix(tikvVer, "v"), "-pre")
	var raftEngineEnable = "false"
	if CompareVersion(curVerShort, "6.0.0") >= 0 {
		configRows := GetConfigByName(tc, "tikv", "raft-engine.enable")
		if len(configRows) > 0 {
			raftEngineEnable = configRows[0].Value
		}
	}
	if raftEngineAppendFileCountCheckParams.IsCheck && raftEngineEnable == "true" {
		g.Go(WithGinkgoRecover(func() {
			log.Info("Raft engine file append count check", zap.Int("TolerateCount", raftEngineAppendFileCountCheckParams.TolerateCount), zap.Duration("TolerateDur", raftEngineAppendFileCountCheckParams.TolerateDur))
			ConsistentlyWithMetrics(tc, timeoutDuration).
				ShouldNot(RaftEngineAppendFileCountMoreThan(raftEngineAppendFileCountCheckParams.TolerateCount).For(raftEngineAppendFileCountCheckParams.TolerateDur).WithCtx(ctx))
			log.Info("Raft engine file append count check done.")
		}))
	}
}

func RegionSizeCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, regionSizeCheckParams RegionSizeCheckParameters) {
	tikvVer := GetCompVersion(ctx, tc, "tikv")
	curVerShort := strings.TrimSuffix(strings.TrimPrefix(tikvVer, "v"), "-pre")
	var dynamicRegionsEnable = "false"
	if CompareVersion(curVerShort, "6.1.0") >= 0 {
		configRows := GetConfigByName(tc, "tikv", "coprocessor.enable-region-bucket")
		if len(configRows) > 0 {
			dynamicRegionsEnable = configRows[0].Value
		}
	}
	if regionSizeCheckParams.IsCheck && dynamicRegionsEnable == "false" {
		g.Go(WithGinkgoRecover(func() {
			log.Info("region size check", zap.Int("regionTolerateSize", regionSizeCheckParams.TolerateSize), zap.Duration("regionSizeTolerateDur", regionSizeCheckParams.TolerateDur))
			ConsistentlyWithMetrics(tc, timeoutDuration).
				ShouldNot(RegionSizeMoreThan(regionSizeCheckParams.TolerateSize).For(regionSizeCheckParams.TolerateDur).WithCtx(ctx))
			log.Info("region size check done.")
		}))
	}
}

func ApplyDbMutexLockDurCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, applyDbMutexLockDurCheckParams ApplyDbMutexLockDurCheckParameters) {
	perfJitterCheckWarmUpTime := 5 * time.Minute
	if applyDbMutexLockDurCheckParams.IsCheck {
		g.Go(WithGinkgoRecover(func() {
			WaitFor(ctx, "Warm Up before ApplyDbMutexLockDuration check", perfJitterCheckWarmUpTime)
			log.Info("max ApplyDbMutexLockDuration check", zap.Float64("TolerateMaxDur", applyDbMutexLockDurCheckParams.TolerateMaxDur))
			ConsistentlyWithMetrics(tc, timeoutDuration, 15*time.Second).
				ShouldNot(ApplyDbMutexLockDurationMoreThan(applyDbMutexLockDurCheckParams.TolerateMaxDur).WithCtx(ctx))
			log.Info("max ApplyDbMutexLockDuration check done.")
		}))
	}
}

func QpsJitterCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, qpsJitterCheckParams QpsJitterCheckParameters) {
	perfJitterCheckWarmUpTime := 5 * time.Minute
	if qpsJitterCheckParams.IsCheck {
		g.Go(WithGinkgoRecover(func() {
			WaitFor(ctx, "Warm Up before QPS check", perfJitterCheckWarmUpTime)
			log.Info("qps jitter check", zap.String("Offset", qpsJitterCheckParams.Offset), zap.Float64("ToleratePerc", qpsJitterCheckParams.ToleratePerc), zap.Duration("TolerateDur", qpsJitterCheckParams.TolerateDur))
			ConsistentlyWithMetrics(tc, timeoutDuration, 15*time.Second).
				ShouldNot(QPSDropMoreThanOffset(qpsJitterCheckParams.ToleratePerc, qpsJitterCheckParams.Offset).For(qpsJitterCheckParams.TolerateDur).WithCtx(ctx))
			log.Info("qps check done.")
		}))
	}
}

func DurJitterCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, durJitterCheckParams DurJitterCheckParameters) {
	perfJitterCheckWarmUpTime := 5 * time.Minute
	if durJitterCheckParams.IsCheck {
		g.Go(WithGinkgoRecover(func() {
			WaitFor(ctx, "Warm Up before Duration check", perfJitterCheckWarmUpTime)
			log.Info("duration jitter check", zap.String("Offset", durJitterCheckParams.Offset), zap.Float64("ToleratePerc", durJitterCheckParams.ToleratePerc), zap.Duration("TolerateDur", durJitterCheckParams.TolerateDur))
			ConsistentlyWithMetrics(tc, timeoutDuration, 15*time.Second).
				ShouldNot(DurationIncreasedMoreThanOffset(durJitterCheckParams.ToleratePerc, durJitterCheckParams.Offset).For(durJitterCheckParams.TolerateDur).WithCtx(ctx))
			log.Info("duration check done.")
		}))
	}
}

func EmptyRegionCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, emptyRegionCheckParams EmptyRegionCheckParameters) {
	if emptyRegionCheckParams.IsCheck {
		g.Go(WithGinkgoRecover(func() {
			log.Info("empty region number check", zap.Int64("empty region number", emptyRegionCheckParams.NumberThreshhold))
			ConsistentlyWithMetrics(tc, timeoutDuration, 15*time.Second).
				ShouldNot(EmptyRegionsMoreThan(emptyRegionCheckParams.NumberThreshhold).For(3 * time.Minute).WithCtx(ctx))
			log.Info("empty region number check done.")
		}))
		g.Go(WithGinkgoRecover(func() {
			log.Info("empty region ratio", zap.Float64("empty region ratio", emptyRegionCheckParams.RatioThreshhold))
			ConsistentlyWithMetrics(tc, timeoutDuration, 15*time.Second).
				ShouldNot(EmptyRegionsRatioMoreThan(emptyRegionCheckParams.RatioThreshhold).For(3 * time.Minute).WithCtx(ctx))
			log.Info("empty region ratio check done.")
		}))
	}
}

func ClusterRestartCheck(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, timeoutDuration time.Duration, restartCheckParams RestartCheckParameters) {
	if restartCheckParams.AllComponentRestartCheck {
		g.Go(WithGinkgoRecover(func() {
			log.Info("AllComponent restart check.")
			ConsistentlyWithMetrics(tc, timeoutDuration).
				ShouldNot(AllComponentRestart().WithCtx(ctx))
			log.Info("AllComponent restart check done.")
		}))
	} else {
		if restartCheckParams.TiKVRestartCheck {
			g.Go(WithGinkgoRecover(func() {
				log.Info("TiKV restart check.")
				ConsistentlyWithMetrics(tc, timeoutDuration).
					ShouldNot(TiKVRestart().WithCtx(ctx))
				log.Info("TiKV restart check done.")
			}))
		}
		if restartCheckParams.PDRestartCheck {
			g.Go(WithGinkgoRecover(func() {
				log.Info("PD restart check.")
				ConsistentlyWithMetrics(tc, timeoutDuration).
					ShouldNot(PDRestart().WithCtx(ctx))
				log.Info("PD restart check done.")
			}))
		}
		if restartCheckParams.TiDBRestartCheck {
			g.Go(WithGinkgoRecover(func() {
				log.Info("TiDB restart check.")
				ConsistentlyWithMetrics(tc, timeoutDuration).
					ShouldNot(TiDBRestart().WithCtx(ctx))
				log.Info("TiDB restart check done.")
			}))
		}
	}
}

func ClusterOomCheck(ctx context.Context, tc resource.TiDBCluster, g *errgroup.Group, timeoutDuration time.Duration, oomCheckParams OomCheckParameters) {
	if oomCheckParams.AllComponentOomCheck {
		g.Go(WithGinkgoRecover(func() {
			log.Info("PD oom check.")
			InstanceOomCheck(ctx, tc, timeoutDuration, resource.PD)
			log.Info("PD oom check done.")
		}))
		g.Go(WithGinkgoRecover(func() {
			log.Info("TiKV oom check.")
			InstanceOomCheck(ctx, tc, timeoutDuration, resource.TiKV)
			log.Info("TiKV oom check done.")
		}))
		g.Go(WithGinkgoRecover(func() {
			log.Info("TiDB oom check.")
			InstanceOomCheck(ctx, tc, timeoutDuration, resource.TiDB)
			log.Info("TiDB oom check done.")
		}))
	} else {
		if oomCheckParams.TiKVOomCheck {
			g.Go(WithGinkgoRecover(func() {
				log.Info("TiKV oom check.")
				InstanceOomCheck(ctx, tc, timeoutDuration, resource.TiKV)
				log.Info("TiKV oom check done.")
			}))
		}
		if oomCheckParams.PDOomCheck {
			g.Go(WithGinkgoRecover(func() {
				log.Info("PD oom check.")
				InstanceOomCheck(ctx, tc, timeoutDuration, resource.PD)
				log.Info("PD oom check done.")
			}))
		}
		if oomCheckParams.TiDBOomCheck {
			g.Go(WithGinkgoRecover(func() {
				log.Info("TiDB oom check.")
				InstanceOomCheck(ctx, tc, timeoutDuration, resource.TiDB)
				log.Info("TiDB oom check done.")
			}))
		}
	}
}

func InstanceOomCheck(ctx context.Context, tc resource.TiDBCluster, duration time.Duration, comp resource.TiDBCompType) {
	runCtx, cancel := context.WithTimeout(ctx, duration)
	defer cancel()
	oomEventCh, err := tc.WatchOOM(runCtx, resource.TiDBClusterOOMWatchOptions{Target: comp})
	gomega.Expect(err).Should(gomega.BeNil())
	event := <-oomEventCh
	gomega.Expect(event.InstanceName).To(gomega.Equal(""))
}

func WaitForRaftLogLagLessThan(ctx context.Context, tc resource.MySQLCompatible, duration time.Duration, pollingInterval time.Duration, threshhold int) {
	if duration < pollingInterval {
		pollingInterval = duration
	}
	Eventually(tc, duration, pollingInterval).Should(
		RaftLogLaglessThan(threshhold).WithCtx(ctx))
}

func WaitForRaftEngineAppendFileCountLessThan(ctx context.Context, tc resource.MySQLCompatible, duration time.Duration, pollingInterval time.Duration, threshhold int, tolerateDur time.Duration) {
	if duration < pollingInterval {
		pollingInterval = duration
	}
	Eventually(tc, duration, pollingInterval).ShouldNot(
		RaftEngineAppendFileCountMoreThan(threshhold).For(tolerateDur).WithCtx(ctx))
}

func WaitForUndersizedRegionsLessThan(ctx context.Context, tc resource.TiDBCluster, duration time.Duration, pollingInterval time.Duration, threshhold int64) {
	if duration < pollingInterval {
		pollingInterval = duration
	}
	Eventually(tc, duration, pollingInterval).ShouldNot(
		UndersizedRegionsMoreThan(threshhold).WithCtx(ctx))
}

func WaitForOversizedRegionsLessThan(ctx context.Context, tc resource.TiDBCluster, duration time.Duration, pollingInterval time.Duration, threshhold int64) {
	if duration < pollingInterval {
		pollingInterval = duration
	}
	Eventually(tc, duration, pollingInterval).ShouldNot(
		OversizedRegionsMoreThan(threshhold).WithCtx(ctx))
}

func BalanceCheckAftWl(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, duration time.Duration, threshhold float64) {
	log.Info("Begin to static balance check")
	g.Go(WithGinkgoRecover(func() {
		ConsistentlyWithMetrics(tc, duration).
			ShouldNot(SatisfyAny(
				StoreRegionSizeDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				LeaderScoreDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				StoreScoreDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				RegionLeaderDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				RegionNumberDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
			))
	}))
}
func BalanceCheckDurWl(ctx context.Context, tc resource.MySQLCompatible, g *errgroup.Group, duration time.Duration, threshhold float64) {
	log.Info("Begin to dynamic balance check")
	g.Go(WithGinkgoRecover(func() {
		ConsistentlyWithMetrics(tc, duration).
			ShouldNot(SatisfyAny(
				StoreReadByteDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				StoreReadKeyDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				StoreWriteByteDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				StoreWriteKeyDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				StoreReadQueryDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
				StoreWriteQueryDifferentExceeds(threshhold).For(30*time.Second).WithCtx(ctx),
			))
	}))
}

func WaitForBalance(ctx context.Context, tc resource.MySQLCompatible, duration time.Duration, pollingInterval time.Duration, threshhold float64) {
	log.Info("Waiting for workload balance", zap.Duration("TimeOut", duration))
	if duration < pollingInterval {
		pollingInterval = duration
	}
	Eventually(tc, duration, pollingInterval).ShouldNot(SatisfyAny(
		StoreScoreDifferentExceeds(threshhold).WithCtx(ctx),
		LeaderScoreDifferentExceeds(threshhold).WithCtx(ctx),
		StoreRegionSizeDifferentExceeds(threshhold*2).WithCtx(ctx),
		RegionLeaderDifferentExceeds(threshhold*2).WithCtx(ctx),
		RegionNumberDifferentExceeds(threshhold*2).WithCtx(ctx)))
}
func WaitForEmptyRegionMerge(ctx context.Context, tc resource.MySQLCompatible, duration time.Duration, pollingInterval time.Duration, emptyRegionsNumberTrd int64, emptyRegionsRatioTrd float64) {
	client := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
	Try(client.SetConfig("split-merge-interval", "10m"))
	Try(BasicClusterExecPdCtl(tc, "config set merge-schedule-limit 200"))

	defer func() {
		Try(BasicClusterExecPdCtl(tc, "config set merge-schedule-limit 8"))
		Try(client.SetConfig("split-merge-interval", "30m"))
	}()
	if duration < pollingInterval {
		pollingInterval = duration
	}
	Eventually(tc, duration, pollingInterval).ShouldNot(SatisfyAny(
		EmptyRegionsMoreThan(emptyRegionsNumberTrd).WithCtx(ctx),
		EmptyRegionsRatioMoreThan(emptyRegionsRatioTrd).WithCtx(ctx),
	))
}

func QPSDropTolerateByDuration(ctx context.Context, tc resource.TiDBCluster, timeoutDuration time.Duration, percentage float64, TolerateDuration time.Duration) {
	ConsistentlyWithMetrics(tc, timeoutDuration, 15*time.Second).
		ShouldNot(QPSDropMoreThan(percentage).For(TolerateDuration).WithCtx(ctx))
}

func WaitForWorkloadBalance(ctx context.Context, tc resource.MySQLCompatible, workloadType string, g *errgroup.Group, checkTime time.Duration, timeOut time.Duration, threshHold float64) {
	g.Go(WithGinkgoRecover(func() {
		Eventually(tc, timeOut, 1*time.Minute).ShouldNot(
			TiKVCPUDifferentExceeds(threshHold / 2).For(30 * time.Second).WithCtx(ctx),
		)
	}))
	g.Go(WithGinkgoRecover(func() {
		ioType := Try(GetWorkloadIoType(workloadType)).(string)
		switch ioType {
		case "wr":
			Eventually(tc, timeOut, 1*time.Minute).ShouldNot(
				SatisfyAny(
					StoreReadByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					StoreReadQueryDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					StoreWriteByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					StoreWriteKeyDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
				))

		case "w":
			Eventually(tc, timeOut, 1*time.Minute).ShouldNot(
				SatisfyAny(
					StoreWriteByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					StoreWriteKeyDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
				))
		case "r":
			Eventually(tc, timeOut, 1*time.Minute).ShouldNot(
				SatisfyAny(
					StoreReadByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					StoreReadQueryDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
				))
		default:
			log.Warn("Got invalid io type.")
		}
	}))
}
func MonitorWorkloadBalance(ctx context.Context, tc resource.MySQLCompatible, workloadType string, g *errgroup.Group, checkTime time.Duration, threshHold float64) {
	g.Go(WithGinkgoRecover(func() {
		ConsistentlyWithMetrics(tc, checkTime).ShouldNot(
			TiKVCPUDifferentExceeds(threshHold / 2).For(30 * time.Second).WithCtx(ctx),
		)
	}))
	g.Go(WithGinkgoRecover(func() {
		ioType := Try(GetWorkloadIoType(workloadType)).(string)
		switch ioType {
		case "wr":
			ConsistentlyWithMetrics(tc, checkTime).
				ShouldNot(
					SatisfyAny(
						StoreReadByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
						StoreReadQueryDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
						StoreWriteByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
						StoreWriteKeyDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					))
		case "w":
			ConsistentlyWithMetrics(tc, checkTime).
				ShouldNot(
					SatisfyAny(
						StoreWriteByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
						StoreWriteKeyDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					))
		case "r":
			ConsistentlyWithMetrics(tc, checkTime).
				ShouldNot(
					SatisfyAny(
						StoreReadByteDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
						StoreReadQueryDifferentExceeds(threshHold).For(30*time.Second).WithCtx(ctx),
					))
		default:
			log.Warn("Got invalid io type.")
		}
	}))
}

func RepeatQpsDurCheckWithAbsTime(ctx context.Context,
	timeout time.Duration,
	startTime time.Time,
	tc resource.MySQLCompatible,
	qpsDropLessThanThreshhold float64,
) {
	log.Info("sla check parameters",
		zap.Time("checkStartTime", startTime),
		zap.Duration("timeOut", timeout),
		zap.Float64("qpsDropLessThanThreshhold", qpsDropLessThanThreshhold),
		zap.Float64("durationIncreaseLessThan", 3*1/qpsDropLessThanThreshhold))
	timer := time.NewTimer(timeout)
Loop:
	for {
		select {
		case <-ctx.Done():
			break Loop
		case <-timer.C:
			break Loop
		default:
		}
		checkTime := 2 * time.Minute
		timeCost := time.Since(startTime)
		min := int(timeCost.Minutes())
		offset := fmt.Sprintf("%dm", min)
		log.Info("Check offset", zap.String("offset", offset))
		if qpsDropLessThanThreshhold < float64(1.0) {
			ConsistentlyWithMetrics(tc, checkTime).ShouldNot(
				SatisfyAny(QPSDropMoreThanOffset(qpsDropLessThanThreshhold, offset).For(30*time.Second).WithCtx(ctx),
					DurationIncreasedMoreThanOffset(3*(1/qpsDropLessThanThreshhold), offset).For(60*time.Second).WithCtx(ctx),
				))
		} else {
			ConsistentlyWithMetrics(tc, checkTime).ShouldNot(
				SatisfyAny(QPSIncreaseLessThanOffset(qpsDropLessThanThreshhold-1, offset).For(30 * time.Second).WithCtx(ctx)))
		}

	}
}

func GetMetric(promURL *url.URL, query string, ts time.Time) (float64, error) {
	//promURL := Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
	log.Info("get metric time", zap.Time("getTime", ts))
	prometheusAddr := promURL.String()
	client, err := api.NewClient(api.Config{
		Address: prometheusAddr,
	})
	if err != nil {
		panic(err)
	}

	httpAPI := v1.NewAPI(client)

	// Record the current value of the query results.
	val, _, err := httpAPI.Query(context.Background(), query, ts)
	if err != nil {
		panic(err)
	}
	valMatrix := val.(model.Vector)
	if len(valMatrix) == 0 {
		return 0, errors.New("empty")
	}
	// fix the Error 1690: constant NaN overflows double, the value from prom exepression is not a number
	// http://172.16.4.180:31714/artifacts/testground/plan-exec-600556-st7xs/plan-exec-600556-st7xs-428117755/main-logs
	value := float64(valMatrix[0].Value)
	if math.IsNaN(value) {
		return 0, nil
	} else {
		return value, nil
	}
}
func WaitForLeaderToZero(promURL *url.URL, address string, pollingInterval time.Duration, timeOut time.Duration) {
	query := fmt.Sprintf(`avg_over_time(pd_scheduler_store_status{type="leader_count",address=~"%s.*"} [30s:])`, address)
	Eventually(func() float64 {
		curLeader := Try(GetMetric(promURL, query, time.Now())).(float64)
		log.Info("current leader:", zap.Float64(address, curLeader))
		return curLeader
	}, timeOut, pollingInterval).Should(Equal(float64(0)))
}