package switch_test

import (
	"fmt"
	"testing"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"

	"github.com/pingcap/endless/pkg/matcher/metrics"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/topsql"

	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"
)

const MaxToleranceTimeDrift = 30

var (
	suiteTestCtx core.TestContext
)

func init() {
	suiteTestCtx = AutoConf()
}

func TestTopSQL(t *testing.T) {
	RegisterFailHandler(Fail)
	RunSpecs(t, "TopSQL Switch Suite")
}

var _ = Describe("TopSQL Switch Test", func() {
	var deployErr error
	var b *topsql.ClusterBundle
	var n *topsql.NgmBundle

	BeforeSuite(func() {
		tiupNode := suiteTestCtx.Resource("tiup").(resource.WorkloadNode)
		if deployErr = topsql.CheckTiUPAvailable(tiupNode); deployErr != nil {
			return
		}

		sysbenchNode := suiteTestCtx.Resource("sysbench").(resource.WorkloadNode)
		cfg := topsql.ClusterConfig{
			Version:        suiteTestCtx.Param("target_version"),
			Name:           "topsql-switch",
			PDClientPort:   2379,
			PDPeerPort:     2380,
			TiDBPort:       4000,
			TiDBStatusPort: 10080,
			TiKVPort:       20160,
			TiKVStatusPort: 20180,
			PrometheusPort: 9090,
		}
		if b, deployErr = topsql.NewClusterBundle(tiupNode, sysbenchNode, cfg); deployErr != nil {
			return
		}

		ngmNode := suiteTestCtx.Resource("ngm").(resource.WorkloadNode)
		n, deployErr = topsql.NewNgmBundle(ngmNode, tiupNode, 12020, cfg.PDClientPort, cfg.PrometheusPort, cfg.Name)
	})

	AfterSuite(func() {
		if n != nil {
			n.Close()
			n = nil
		}
		if b != nil {
			b.Close()
			b = nil
		}
	})

	TestCase("TopSQL Switch Test", "TEST_SWITCH", func() {
		if deployErr != nil {
			Skip(fmt.Sprintf("tiup is unavailable: %s", deployErr.Error()))
		}

		By(`
1. Turn on TopSQL and start ngm
	* Targets begin to profile`)
		// ============================
		// Turn on TopSQL and start ngm
		// ============================
		Expect(b.TurnOnTopSQL()).Should(Succeed())
		Expect(n.StartNGM()).Should(Succeed())
		Expect(topsql.TurnOffConprof(n.NgmURL())).Should(Succeed())
		Expect(b.PrepareSysbench(4000)).Should(Succeed())

		Eventually(func(g Gomega) {
			// ------------------------
			// Targets begin to profile
			// ------------------------
			g.Expect(b.Cluster).Should(metrics.TiDBProfilingCPU("tidb-peer:10080"))
			g.Expect(b.Cluster).Should(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180"))
		}, "5m", "10s").Should(Succeed())

		By(`
2. Stop ngm
	* Targets stop profiling`)
		// ========
		// Stop ngm
		// ========
		Expect(n.StopNGM()).Should(Succeed())

		Eventually(func(g Gomega) {
			// ----------------------
			// Targets stop profiling
			// ----------------------
			g.Expect(b.Cluster).Should(Not(metrics.TiDBProfilingCPU("tidb-peer:10080")))
			g.Expect(b.Cluster).Should(Not(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180")))
		}, "5m", "10s").Should(Succeed())

		By(`
3. Restart ngm
	* Targets begin to profile`)
		// ===========
		// Restart ngm
		// ===========
		Expect(n.StartNGM()).Should(Succeed())

		Eventually(func(g Gomega) {
			// ------------------------
			// Targets begin to profile
			// ------------------------
			g.Expect(b.Cluster).Should(metrics.TiDBProfilingCPU("tidb-peer:10080"))
			g.Expect(b.Cluster).Should(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180"))
		}, "5m", "10s").Should(Succeed())

		By(`
4. Run sysbench
	* Data are available`)
		// ============
		// Run sysbench
		// ============
		Expect(b.StartSysbench(4000)).Should(Succeed())

		Eventually(func(g Gomega) {
			// ------------------
			// Data are available
			// ------------------
			now := time.Now().Unix()
			bytes, err := topsql.FetchInstances(n.NgmURL(), now-120, now)
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(ConsistOf(
				map[string]interface{}{"instance_type": "tidb", "instance": "tidb-peer:10080"},
				map[string]interface{}{"instance_type": "tikv", "instance": "tikv-peer:20160"},
			))

			bytes, err = topsql.FetchSummary(n.NgmURL(), now-120, now, "tidb-peer:10080", "tidb")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			// 6 items: top5 and others
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(HaveLen(6))

			bytes, err = topsql.FetchSummary(n.NgmURL(), now-120, now, "tikv-peer:20160", "tikv")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			// 6 items: top5 and others
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(HaveLen(6))
		}, "5m", "10s").Should(Succeed())

		By(`
5. Stop ngm
	* Ngm not found`)
		// ========
		// Stop ngm
		// ========
		Expect(n.StopNGM()).Should(Succeed())

		// -------------
		// Ngm not found
		// -------------
		Eventually(b.NgmTopology, "2m", "10s").Should(BeEmpty())

		By(`
6. Restart ngm
	* Data are unavailable during the downtime
	* Data are available after ngm started`)
		Consistently(b.NgmTopology, "2m", "30s").Should(BeEmpty())

		// ===========
		// Restart ngm
		// ===========
		Expect(n.StartNGM()).Should(Succeed())

		elapsed := time.Now().Unix() - n.LatestNgmDownUnixTime - 2*MaxToleranceTimeDrift
		Eventually(func(g Gomega) {
			// ----------------------------------------
			// Data are unavailable during the downtime
			// ----------------------------------------
			base := time.Now().Unix() - MaxToleranceTimeDrift
			bytes, err := topsql.FetchInstances(n.NgmURL(), base-elapsed, base)
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tidb-peer:10080", "tidb")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tikv-peer:20160", "tikv")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())
		}, "2m", "10s").Should(Succeed())

		Eventually(func(g Gomega) {
			// ------------------------------------
			// Data are available after ngm started
			// ------------------------------------
			now := time.Now().Unix()
			bytes, err := topsql.FetchInstances(n.NgmURL(), now-120, now)
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(ConsistOf(
				map[string]interface{}{"instance_type": "tidb", "instance": "tidb-peer:10080"},
				map[string]interface{}{"instance_type": "tikv", "instance": "tikv-peer:20160"},
			))

			bytes, err = topsql.FetchSummary(n.NgmURL(), now-120, now, "tidb-peer:10080", "tidb")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			// 6 items: top5 and others
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(HaveLen(6))

			bytes, err = topsql.FetchSummary(n.NgmURL(), now-120, now, "tikv-peer:20160", "tikv")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			// 6 items: top5 and others
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(HaveLen(6))
		}, "5m", "10s").Should(Succeed())

		By(`
7. Turn off TopSQL
	* Variable tidb_enable_top_sql is off
	* Targets stop profiling
	* Data are unavailable during offtime`)
		// ===============
		// Turn off TopSQL
		// ===============
		Expect(b.TurnOffTopSQL()).Should(Succeed())

		// -----------------------------------
		// Variable tidb_enable_top_sql is off
		// -----------------------------------
		Consistently(b.TopSQLIsOn, "2m", "30s").Should(BeFalse())

		elapsed = time.Now().Unix() - b.LatestTopSQLOffUnixTime
		Eventually(func(g Gomega) {
			// ----------------------
			// Targets stop profiling
			// ----------------------
			g.Expect(b.Cluster).Should(Not(metrics.TiDBProfilingCPU("tidb-peer:10080")))
			g.Expect(b.Cluster).Should(Not(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180")))

			// -----------------------------------
			// Data are unavailable during offtime
			// -----------------------------------
			base := time.Now().Unix()
			bytes, err := topsql.FetchInstances(n.NgmURL(), base-elapsed, base)
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tidb-peer:10080", "tidb")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tikv-peer:20160", "tikv")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())
		}, "5m", "10s").Should(Succeed())

		By(`
8. Stop ngm
	* Ngm not found
	* Targets stop profiling`)
		// ========
		// Stop ngm
		// ========
		Expect(n.StopNGM()).Should(Succeed())

		Eventually(func(g Gomega) {
			// -------------
			// Ngm not found
			// -------------
			g.Expect(b.NgmTopology()).Should(BeEmpty())

			// ----------------------
			// Targets stop profiling
			// ----------------------
			g.Expect(b.Cluster).Should(Not(metrics.TiDBProfilingCPU("tidb-peer:10080")))
			g.Expect(b.Cluster).Should(Not(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180")))
		}, "5m", "10s").Should(Succeed())

		By(`
9. Turn on TopSQL but ngm is still down
	* Ngm not found
	* Targets stop profiling`)
		// ====================================
		// Turn on TopSQL but ngm is still down
		// ====================================
		Expect(b.TurnOnTopSQL()).Should(Succeed())

		Consistently(func(g Gomega) {
			// -------------
			// Ngm not found
			// -------------
			g.Expect(b.NgmTopology()).Should(BeEmpty())

			// ----------------------
			// Targets stop profiling
			// ----------------------
			g.Expect(b.Cluster).Should(Not(metrics.TiDBProfilingCPU("tidb-peer:10080")))
			g.Expect(b.Cluster).Should(Not(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180")))
		}, "2m", "30s").Should(Succeed())

		By(`
10. Restart ngm
	* Data unavailable during offtime
	* Targets begin to profile
	* Data are available after ngm started`)
		// ===========
		// Restart ngm
		// ===========
		Expect(n.StartNGM()).Should(Succeed())

		elapsed = time.Now().Unix() - b.LatestTopSQLOffUnixTime - 2*MaxToleranceTimeDrift
		Eventually(func(g Gomega) {
			// -------------------------------
			// Data unavailable during offtime
			// -------------------------------
			base := time.Now().Unix() - MaxToleranceTimeDrift
			bytes, err := topsql.FetchInstances(n.NgmURL(), base-elapsed, base)
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tidb-peer:10080", "tidb")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tikv-peer:20160", "tikv")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())
		}, "2m", "10s").Should(Succeed())

		Eventually(func(g Gomega) {
			// ------------------------
			// Targets begin to profile
			// ------------------------
			g.Expect(b.Cluster).Should(metrics.TiDBProfilingCPU("tidb-peer:10080"))
			g.Expect(b.Cluster).Should(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180"))

			// ------------------------------------
			// Data are available after ngm started
			// ------------------------------------
			now := time.Now().Unix()
			bytes, err := topsql.FetchInstances(n.NgmURL(), now-120, now)
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(ConsistOf(
				map[string]interface{}{"instance_type": "tidb", "instance": "tidb-peer:10080"},
				map[string]interface{}{"instance_type": "tikv", "instance": "tikv-peer:20160"},
			))

			bytes, err = topsql.FetchSummary(n.NgmURL(), now-120, now, "tidb-peer:10080", "tidb")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(HaveLen(6))

			bytes, err = topsql.FetchSummary(n.NgmURL(), now-120, now, "tikv-peer:20160", "tikv")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(HaveLen(6))
		}, "5m", "10s").Should(Succeed())

		By(`
11. Turn off TopSQL
	* Variable tidb_enable_top_sql is off
	* Targets stop profiling
	* Data unavailable during offtime`)
		// ===============
		// Turn off TopSQL
		// ===============
		Expect(b.TurnOffTopSQL()).Should(Succeed())

		// -----------------------------------
		// Variable tidb_enable_top_sql is off
		// -----------------------------------
		Consistently(b.TopSQLIsOn, "2m", "30s").Should(BeFalse())

		elapsed = time.Now().Unix() - b.LatestTopSQLOffUnixTime
		Eventually(func(g Gomega) {
			// ----------------------
			// Targets stop profiling
			// ----------------------
			g.Expect(b.Cluster).Should(Not(metrics.TiDBProfilingCPU("tidb-peer:10080")))
			g.Expect(b.Cluster).Should(Not(metrics.TiKVRsmeterStatingThreads("tikv-peer:20180")))

			// -------------------------------
			// Data unavailable during offtime
			// -------------------------------
			base := time.Now().Unix()
			bytes, err := topsql.FetchInstances(n.NgmURL(), base-elapsed, base)
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tidb-peer:10080", "tidb")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())

			bytes, err = topsql.FetchSummary(n.NgmURL(), base-elapsed, base, "tikv-peer:20160", "tikv")
			g.Expect(err).ShouldNot(HaveOccurred())
			g.Expect(topsql.QueryJSON(bytes, "status")).Should(Equal("ok"))
			g.Expect(topsql.QueryJSON(bytes, "data")).Should(BeEmpty())
		}, "5m", "10s").Should(Succeed())
	})
})