package ccb_application_router

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	. "github.com/onsi/gomega"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"golang.org/x/sync/errgroup"
)

func LaunchCCBLoadClient(PdClientAddr string, workloadClient resource.WorkloadNode, PeerStart int, PeerCount int, APIVersion int, ExecTime int, StartUpInterval int, ctx context.Context) {
	// JVM Monitor
	AGENT := "-javaagent:libs/jmx_prometheus_javaagent-0.16.1.jar=8081:configs/config.yaml"
	JavaOPTS := "-XX:CICompilerCount=6 " +
		"-XX:GCLogFileSize=104857600 " +
		"-XX:GCPauseIntervalMillis=200 " +
		"-XX:InitialHeapSize=8589934592 " +
		"-XX:MaxGCPauseMillis=100 " +
		"-XX:MaxHeapSize=8589934592 " +
		"-XX:NumberOfGCLogFiles=20 " +
		"-XX:+PrintGCDetails " +
		"-XX:+PrintGCDateStamps " +
		"-XX:+UseCompressedClassPointers " +
		"-XX:+UseCompressedOops " +
		"-XX:+UseG1GC " +
		"-XX:+UseGCLogFileRotation " +
		"-verbose:gc " +
		"-Xloggc:gc.log"
	CCBClientPara := fmt.Sprintf("-Dtikv.grpc.timeout_in_ms=150 "+
		"-Dtikv.enable_grpc_forward=true "+
		"-Dtikv.grpc.forward_timeout_in_ms=200 "+
		"-Dtikv.rawkv.read_timeout_in_ms=400 "+
		"-Dtikv.rawkv.write_timeout_in_ms=400 "+
		"-Dtikv.rawkv.batch_read_timeout_in_ms=400 "+
		"-Dtikv.rawkv.batch_write_timeout_in_ms=400 "+
		"-Dtikv.rawkv.read_slowlog_in_ms=10 "+
		"-Dtikv.rawkv.write_slowlog_in_ms=10 "+
		"-Dtikv.rawkv.batch_read_slowlog_in_ms=10 "+
		"-Dtikv.rawkv.batch_write_slowlog_in_ms=10 "+
		"-Dtikv.circuit_break.enable=true "+
		"-Dtikv.circuit_break.trigger.availability.window_in_seconds=60 "+
		"-Dtikv.circuit_break.trigger.availability.error_threshold_percentage=100 "+
		"-Dtikv.circuit_break.trigger.availability.request_volumn_threshold=10 "+
		"-Dtikv.circuit_break.trigger.sleep_window_in_seconds=20 "+
		"-Dtikv.circuit_break.trigger.attempt_request_count=10 "+
		"-Dtikv.enable_atomic_for_cas=true "+
		"-Dtikv.pd.addresses=%s "+
		"-Dtikv.api_version=%d ", PdClientAddr, APIVersion)
	CCBLoadParaTemplate := "-Drawkvworkload.idx-trx.threads=300 " +
		"-Drawkvworkload.tmp-ttl=604800 " +
		"-Drawkvworkload.idx-trx.rate=1500 " +
		"-Drawkvworkload.idx-trx.clients=1 " +
		"-Drawkvworkload.time=%v " +
		"-Drawkvworkload.only.read=1 " +
		"-Drawkvworkload.altercluster=1 " +
		"-Drawkvworkload.metrics.port=10000 " +
		"-Drawkvworkload.process=%v"
	for i := 0; i < PeerCount; i++ {
		Peer := PeerStart + i
		// Kill old ccbload process.
		// options := resource.WorkloadNodeExecOptions{Command: "ps aux | grep rawkvworkload | grep -v grep | awk -F' ' '{print $2}' | xargs kill || true", Peer: Peer}
		// _, _, exitCode, err := workloadClient.Exec(options)
		// log.S().Infof("Kill old ccbload client with execcode: %v, err: %v", exitCode, err)
		// time.Sleep(3 * time.Second)
		CCBLoad_PARA := fmt.Sprintf(CCBLoadParaTemplate, ExecTime, Peer)
		cmd := fmt.Sprintf("nohup java %s %s -cp libs/rawkvworkload.jar %s %s org.pingcap.rawkvworkload.Main > log/ccbload.log 2>&1 &", AGENT, JavaOPTS, CCBClientPara, CCBLoad_PARA)
		options := resource.WorkloadNodeExecOptions{Command: cmd, Peer: Peer}
		_, _, exitCode, err := workloadClient.Exec(options)
		log.S().Infof("start a new ccbload client with execcode: %v, err: %v", exitCode, err)
		sleepT := time.Duration(rand.Intn(5)+StartUpInterval) * time.Second
		time.Sleep(sleepT)
	}
}

func StopCCBLoadClient(workloadClient resource.WorkloadNode, PeerStart int, PeerCount int) {
	for i := 0; i < PeerCount; i++ {
		Peer := PeerStart + i
		options := resource.WorkloadNodeExecOptions{Command: "ps aux | grep rawkvworkload | grep -v grep | awk -F' ' '{print $2}' | xargs kill || true", Peer: Peer}
		_, _, exitCode, err := workloadClient.Exec(options)
		log.S().Infof("Kill old ccbload client with execcode: %v, err: %v", exitCode, err)
		Expect(err).Should(BeNil())
	}
	time.Sleep(3 * time.Second)
}

// Import 100e KV Data
func ImportDataByBatchPut(PdClientAddr string, workloadClient resource.WorkloadNode, PeerStart int, PeerCount int, APIVersion int, KVTotalCount int64, ctx context.Context) (*errgroup.Group, context.Context) {
	JavaOPTS := "-XX:CICompilerCount=6 " +
		"-XX:GCPauseIntervalMillis=200 " +
		"-XX:InitialHeapSize=8589934592 " +
		"-XX:MaxGCPauseMillis=100 " +
		"-XX:MaxHeapSize=8589934592 " +
		"-XX:+UseCompressedClassPointers " +
		"-XX:+UseCompressedOops " +
		"-XX:+UseG1GC "
	CCBClientPara := fmt.Sprintf("-Dtikv.grpc.timeout_in_ms=1500 "+
		"-Dtikv.enable_grpc_forward=true "+
		"-Dtikv.grpc.forward_timeout_in_ms=10000 "+
		"-Dtikv.rawkv.read_timeout_in_ms=10000 "+
		"-Dtikv.rawkv.write_timeout_in_ms=10000 "+
		"-Dtikv.rawkv.batch_read_timeout_in_ms=40000 "+
		"-Dtikv.rawkv.batch_write_timeout_in_ms=40000 "+
		"-Dtikv.rawkv.read_slowlog_in_ms=10000 "+
		"-Dtikv.rawkv.write_slowlog_in_ms=10000 "+
		"-Dtikv.rawkv.batch_read_slowlog_in_ms=10000 "+
		"-Dtikv.rawkv.batch_write_slowlog_in_ms=10000 "+
		"-Dtikv.circuit_break.enable=true "+
		"-Dtikv.circuit_break.trigger.availability.window_in_seconds=600 "+
		"-Dtikv.circuit_break.trigger.availability.error_threshold_percentage=100 "+
		"-Dtikv.circuit_break.trigger.availability.request_volumn_threshold=10 "+
		"-Dtikv.circuit_break.trigger.sleep_window_in_seconds=200 "+
		"-Dtikv.circuit_break.trigger.attempt_request_count=10 "+
		"-Dtikv.enable_atomic_for_cas=true "+
		"-Dtikv.pd.addresses=%s "+
		"-Dtikv.api_version=%d ", PdClientAddr, APIVersion)
	CCBLoadParaTemplate := "-Drawkvworkload.idx-trx.threads=100 " +
		"-Drawkvworkload.tmp-ttl=0 " +
		"-Drawkvworkload.idx-trx.rate=1500 " +
		"-Drawkvworkload.idx-trx.clients=1 " +
		"-Drawkvworkload.time=0 " +
		"-Drawkvworkload.altercluster=1 " +
		"-Drawkvworkload.metrics.port=10000 " +
		"-Drawkvworkload.process=%v " +
		"-Drawkvworkload.idx-init.count=%d"
	g, gctx := errgroup.WithContext(ctx)
	KVCountPerWorkload := KVTotalCount / int64(PeerCount)
	for i := 0; i < PeerCount; i++ {
		Peer := PeerStart + i
		g.Go(func() error {
			CCBLoad_PARA := fmt.Sprintf(CCBLoadParaTemplate, Peer, KVCountPerWorkload)
			cmd := fmt.Sprintf("java %s -cp libs/rawkvworkload.jar %s %s org.pingcap.rawkvworkload.InitIdx > log/initidx.log 2>&1", JavaOPTS, CCBClientPara, CCBLoad_PARA)
			options := resource.WorkloadNodeExecOptions{Command: cmd, Peer: Peer}
			_, _, _, err := workloadClient.Exec(options)
			return err
		})
		log.S().Info("start a new importing data client")
		sleepT := time.Duration(rand.Intn(20)) * time.Second
		time.Sleep(sleepT)
	}
	return g, gctx
}