package pdrel

import (
	"context"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/chaos"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/matcher/metrics"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/pkg/workloadnode/cluster"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/systemtest/query"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	k8swait "k8s.io/apimachinery/pkg/util/wait"
)

func OLTP_PD_REL_LEADER_DOWN(
	ctx core.TestContext,
	tc cluster.BaseCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	pdrelOptions := &PdRelOptions{
		FaultType:       "failure",
		IsLeader:        true,
		RunTime:         60 * time.Minute,
		WarmUpTime:      10 * time.Minute,
		FaultDuration:   5 * time.Minute,
		RTO:             2 * time.Minute,
		RecoveryTime:    2 * time.Minute,
		PollingInterval: 20 * time.Minute,
	}
	pdrelOptions.PdRelTest(ctx, tc, caseCmdParms, workloadRunParms)
}
func OLTP_PD_REL_LEADER_IO_DELAY(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	experimentSpec := &resource.IODelaySpec{VolumePath: "/var/lib/pd", Delay: oltppkg.GetRandIntStr(100, 500) + "ms", Percent: int64(oltppkg.GetRandInt(10, 90))}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: GetPdIngoreErrors(caseCmdParms, "rel"),
	}
	RTO := 2 * time.Minute
	PdRelTest(ctx, tc, caseCmdParms, workloadRunParms, "pd io delay", true, experimentSpec, RTO, workloadCheckParms)
}
func OLTP_PD_REL_LEADER_NETWORK_LOSS(
	ctx core.TestContext,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	pdrelOptions := &PdRelOptions{
		FaultType:       "network-loss-100",
		IsLeader:        true,
		RunTime:         60 * time.Minute,
		WarmUpTime:      10 * time.Minute,
		FaultDuration:   5 * time.Minute,
		RTO:             2 * time.Minute,
		RecoveryTime:    3 * time.Minute,
		PollingInterval: 20 * time.Minute,
	}
	pdrelOptions.PdRelTest(ctx, tc, caseCmdParms, workloadRunParms)
}
func OLTP_PD_REL_LEADER_NETWORK_LATENCY(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	experimentSpec := &resource.NetworkLatencySpec{
		Latency:     "5ms", //oltppkg.GetRandIntStr(9, 10) + "ms",
		Correlation: "25",
		Jitter:      "0ms", //"2ms"
	}

	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 0.6,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      false,
		Offset:       "2m",
		ToleratePerc: 2,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: GetPdIngoreErrors(caseCmdParms, "rel"),
	}
	RTO := 2 * time.Minute
	PdRelTest(ctx, tc, caseCmdParms, workloadRunParms, "pd network latency", true, experimentSpec, RTO, workloadCheckParms, qpsJitterCheckParms, durJitterCheckParms)
}

func OLTP_PD_REL_LEADER_TIME_OFFSET(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	//timeOffset := fmt.Sprintf("%dh", oltppkg.GetRandInt(-10000, 10000))
	timeOffset := fmt.Sprintf("%dms", oltppkg.GetRandInt(-1000, 1000))
	experimentSpec := &resource.TimeChaosSpec{TimeOffset: timeOffset}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: GetPdIngoreErrors(caseCmdParms, "rel"),
	}
	RTO := 2 * time.Minute
	PdRelTest(ctx, tc, caseCmdParms, workloadRunParms, "pd leader time offset", true, experimentSpec, RTO, workloadCheckParms)
}
func OLTP_PD_REL_FOLLOWER_DOWN(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	experimentSpec := &resource.FailureExperimentSpec{}
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         true,
		PDRestartCheck:           false,
		TiDBRestartCheck:         true,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: GetPdIngoreErrors(caseCmdParms, "rel"),
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.2,
		TolerateDur:  60 * time.Second,
	}
	RTO := 2 * time.Minute
	PdRelTest(ctx, tc, caseCmdParms, workloadRunParms, "pd follower down", false, experimentSpec, RTO, workloadCheckParms, restartCheckParms, qpsJitterCheckParms)
}
func OLTP_PD_REL_FOLLOWER_NETWORK_LOSS(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	experimentSpec := &resource.NetworkLossSpec{
		Duration:    oltppkg.GetRandIntStr(1, 3) + "s",
		Scheduler:   &resource.ExperimentScheduler{Cron: "@every " + oltppkg.GetRandIntStr(5, 60) + "s"},
		Loss:        "100",
		Correlation: "70"}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.2,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 1.5,
		TolerateDur:  600 * time.Second,
	}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: GetPdIngoreErrors(caseCmdParms, "fun"),
	}
	RTO := 2 * time.Minute
	PdRelTest(ctx, tc, caseCmdParms, workloadRunParms, "pd follower network loss", false, experimentSpec, RTO, workloadCheckParms, qpsJitterCheckParms, durJitterCheckParms)
}
func OLTP_PD_REL_FOLLOWER_NETWORK_LATENCY(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	experimentSpec := &resource.NetworkLatencySpec{Latency: oltppkg.GetRandIntStr(5, 20) + "ms", Correlation: "25", Jitter: "5ms"}
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: GetPdIngoreErrors(caseCmdParms, "rel"),
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 0.2,
		TolerateDur:  120 * time.Second,
	}
	durJitterCheckParms := oltppkg.DurJitterCheckParameters{
		IsCheck:      true,
		Offset:       "2m",
		ToleratePerc: 1.2,
		TolerateDur:  600 * time.Second,
	}
	RTO := 2 * time.Minute
	PdRelTest(ctx, tc, caseCmdParms, workloadRunParms, "pd follower latency", false, experimentSpec, RTO, workloadCheckParms, qpsJitterCheckParms, durJitterCheckParms)
}

func OLTP_PD_REL_RESTART_ALL_PODS(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	defer func() {
		log.Info("start upload profiling")
		for i := 0; i < 3; i++ {
			TikvProfilingUpload(ctx, tc, workloadRunParms.WorkloadNode)
		}
		log.Info("finished upload profiling")
	}()

	runTimeParms := oltppkg.NewRunTimePara(60*time.Minute, 5*time.Minute, 2*time.Minute, 5*time.Minute)
	if caseCmdParms.RunTime.TotalRunTime != 0 {
		runTimeParms.TotalRunTime = caseCmdParms.RunTime.TotalRunTime
	} else {
		runTimeParms.TotalRunTime = 1 * time.Hour
	}
	log.Info(fmt.Sprintf("Will run %v", runTimeParms))
	runTimeParms.RunInterval = 15 * time.Minute
	ctxAll, cancelAll := context.WithTimeout(ctx, runTimeParms.TotalRunTime)
	defer cancelAll()

	g, ctxg := errgroup.WithContext(ctxAll)

	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   false,
		IgnoreErr: append(GetPdIngoreErrors(caseCmdParms, "rel"), "9005", "unable to connect to MySQL"),
	}
	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         false,
		PDRestartCheck:           false,
		TiDBRestartCheck:         false,
	}
	qpsJitterCheckParms := oltppkg.QpsJitterCheckParameters{
		IsCheck: false,
	}
	startTime := time.Now().Add(runTimeParms.WarmUpTime / 2)
	oltppkg.FunCommonCheck(ctxg, tc, g, runTimeParms.TotalRunTime, workloadRunParms.Cmdpr, workloadCheckParms, restartCheckParms, qpsJitterCheckParms)
	oltppkg.WaitFor(ctxg, "waiting for workload running", runTimeParms.WarmUpTime)

	//Inject chaos
	oltppkg.LogAndBy("Inject all pod restart")
	g.Go(WithGinkgoRecover(func() {
		conditonWrapper := func(ctx context.Context) (bool, error) {
			go workloadRunParms.Pause(ctx, runTimeParms.RecoveryTime)
			RestartAllNodes(tc)
			oltppkg.WaitFor(ctx, "wait for cluster", runTimeParms.RecoveryTime)
			Eventually(func() string {
				return oltppkg.TryGetPdLeaderName(tc)
			}, 2*time.Minute, 2*time.Second).ShouldNot(Equal(""))
			Eventually(tc, 5*time.Minute, 1*time.Minute).ShouldNot(RegionLeaderDifferentExceeds(0.1).WithCtx(ctxg))
			oltppkg.CheckClusterHealth(tc)
			oltppkg.WaitFor(ctx, "workload recover", runTimeParms.RecoveryTime)
			oltppkg.RepeatQpsDurCheckWithAbsTime(ctx, 5*time.Minute, startTime, tc, 0.8)
			oltppkg.WaitForBalance(ctx, tc, 5*time.Minute, 1*time.Minute, 0.2)
			log.Info("one run finished.")
			return false, nil
		}
		k8swait.PollImmediateUntilWithContext(ctxg, runTimeParms.RunInterval, conditonWrapper)
		log.Info("inject pod restart done")
	}))
	Expect(g.Wait()).Should(BeNil())
	log.Info("Test finished")
}

//transfer leader
func OLTP_PD_REL_TRANSFER_LEADER(
	ctx context.Context,
	tc resource.TiDBCluster,
	caseCmdParms oltppkg.CaseCmdParms,
	workloadRunParms oltppkg.WorkloadRunParms,
) {
	runTimeParms := oltppkg.NewRunTimePara(30*time.Minute, 10*time.Minute, 10*time.Minute, 2*time.Minute)
	ctxAll, cancelAll := context.WithTimeout(ctx, runTimeParms.TotalRunTime)
	defer cancelAll()
	g, ctxg := errgroup.WithContext(ctxAll)

	log.Info("Will inject chaos for all pod restart")
	workloadCheckParms := oltppkg.WorkloadCheckParameters{
		IsCheck:   true,
		IgnoreErr: append(GetPdIngoreErrors(caseCmdParms, "rel"), "9005", "unable to connect to MySQL"),
	}
	oltppkg.FunCommonCheck(ctxg, tc, g, runTimeParms.TotalRunTime, workloadRunParms.Cmdpr, workloadCheckParms)

	//Inject chaos
	oltppkg.LogAndBy("Transfer leader manullay")
	g.Go(WithGinkgoRecover(func() {
		interval := 5 * time.Minute
	Loop:
		for {
			select {
			case <-time.After(interval):
				log.Info("Start to transfer leader")
			case <-ctxg.Done():
				break Loop
			}
			pdLeader := oltppkg.GetPdLeaderName(tc)
			t := strings.Split(pdLeader, "-")
			ins := Try(strconv.Atoi(t[len(t)-1])).(int) + 1
			pdNodes := Try(tc.GetNodes(resource.PD)).([]resource.TiDBComponentNode)
			targetLeader := fmt.Sprintf("tc-pd-%d", ins%(len(pdNodes)))
			startTime := time.Now()
			oltppkg.ExecPdCtl(tc, fmt.Sprintf("member leader transfer %s", targetLeader))
			log.Info("Member Transfer Cost", zap.Duration("TimeCost", time.Since(startTime)))
			Eventually(func() string {
				return oltppkg.TryGetPdLeaderName(tc)
			}, 1*time.Minute, 2*time.Second).ShouldNot(Equal(targetLeader))
			log.Info("New leader work Cost", zap.Duration("TimeCost", time.Since(startTime)))
		}
	}))
	Expect(g.Wait()).Should(BeNil())
}

func OLTP_FULL_SCAN_SYSTEM_TALBES_REL_001(
	ctx context.Context,
	tc resource.TiDBCluster,
	pr *io.PipeReader,
	caseCmdParms oltppkg.CaseCmdParms,
) {
	runTimeParms := oltppkg.RunTimePara{
		TotalRunTime:  30 * time.Minute,
		WarmUpTime:    5 * time.Minute,
		RecoveryTime:  10 * time.Minute,
		RunInterval:   30 * time.Minute,
		FaultDuration: oltppkg.GetRandDuration(time.Minute, 5),
	}
	ctxAll, cancel := context.WithTimeout(ctx, runTimeParms.TotalRunTime)
	defer cancel()
	g, ctxg := errgroup.WithContext(ctxAll)

	restartCheckParms := oltppkg.RestartCheckParameters{
		AllComponentRestartCheck: false,
		TiKVRestartCheck:         true,
		PDRestartCheck:           false,
		TiDBRestartCheck:         true,
	}

	oltppkg.FunCommonCheck(ctxg, tc, g, runTimeParms.TotalRunTime, pr, restartCheckParms)

	useDb := "INFORMATION_SCHEMA"
	g.Go(WithGinkgoRecover(func() {
		//oltppkg.WaitFor(ctxg, "wait for full scan start, which is 2min later than qps check", 7*time.Minute)
		filterOut := []string{"CLUSTER_LOG", "METRICS_SUMMARY",
			"TABLE_STORAGE_STATS", "INSPECTION_SUMMARY", "METRICS_SUMMARY_BY_LABEL", "TIDB_HOT_REGIONS_HISTORY"}
		err := query.FullScanAllTablesRepeatedly(ctxg, tc, runTimeParms.TotalRunTime, useDb, filterOut, nil)
		mysqlErr, ok := err.(*mysql.MySQLError)
		Expect(ok).Should(BeTrue())
		if mysqlErr.Number == 1105 && !strings.Contains(mysqlErr.Message, "denied") {
			oltppkg.WaitFor(ctxg, "waiting for leader changed", 2*time.Minute)
			err := query.FullScanAllTablesRepeatedly(ctxg, tc, runTimeParms.TotalRunTime, useDb, filterOut, nil)
			Expect(err).Should(BeNil())
		}
	}))
	g.Go(WithGinkgoRecover(func() {
		oltppkg.WaitFor(ctxg, "wait for full scan start, which is 2min later than qps check", runTimeParms.WarmUpTime)
		c := chaos.Chaos{
			Name:                  "pd-leader-down",
			Tc:                    tc,
			ExperimentSpec:        &resource.FailureExperimentSpec{},
			SelectorsRetainPolicy: chaos.RetainRandomly(1),
			SelectorOpts: resource.TiDBClusterSelectorOpts{
				TiDBClusterPDSelectorSpec: &resource.TiDBClusterPDSelectorSpec{
					IsLeader: oltppkg.ToBoolPtr(true),
				},
			},
		}
		err := oltppkg.InjectFaultByChaos(&c, context.Background(), runTimeParms.FaultDuration)
		Expect(err).Should(Succeed())
	}))

	Expect(g.Wait()).Should(BeNil())
}