package cc_rolling_update import ( "context" "flag" "fmt" "net/url" "time" _ "github.com/go-sql-driver/mysql" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" "github.com/pingcap/log" "github.com/pingcap/test-infra/caselib/pkg/host" "github.com/pingcap/test-infra/caselib/pkg/model/workload" "github.com/pingcap/test-infra/caselib/pkg/utils" "github.com/pingcap/test-infra/sdk/resource" _ "github.com/pingcap/test-infra/sdk/resource/impl/k8s" "golang.org/x/sync/errgroup" . "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" "go.uber.org/zap" ) var tiupTopo = ` global: user: "tidb" ssh_port: 22 deploy_dir: "/tiup/deploy" data_dir: "/tiup/data" arch: "amd64" monitored: node_exporter_port: 19100 blackbox_exporter_port: 19115 server_configs: tikv: raftstore.region-max-size: 16MB raftstore.region-split-size: 4MB raftstore.region-split-check-diff: 1MB storage.block-cache.capacity: 1GB pd_servers: - host: pd-peer client_port: 12379 peer_port: 12380 tidb_servers: - host: tidb-peer port: 14000 status_port: 20080 tikv_servers: - host: tikv1-peer port: 30160 status_port: 30180 - host: tikv2-peer port: 30160 status_port: 30180 - host: tikv3-peer port: 30160 status_port: 30180 - host: tikv4-peer port: 30160 status_port: 30180 - host: tikv5-peer port: 30160 status_port: 30180 monitoring_servers: - host: tiup-peer port: 19090 grafana_servers: - host: tiup-peer port: 13000 ` func SleepWithCtx(ctx context.Context, timeout time.Duration) { select { case <-time.After(timeout): return case <-ctx.Done(): return } } func WaitFor(ctx context.Context, reason string, duration time.Duration) { log.Info(reason, zap.Duration("wait", duration)) SleepWithCtx(ctx, duration) } // 1 tiup // 1 wkload // 1 pd // 5 tikv // 1 tidb var _ = ParameterizedDescribe("TiKV", func(flagSet *flag.FlagSet) { var ( cntlNode resource.WorkloadNode cntlHost *host.WorkloadHost workloadNode resource.WorkloadNode workloadHost *host.WorkloadHost cluster *host.TiUPCluster dbURL *url.URL pdURL *url.URL version string s3_endpoint string s3_key string s3_passwd string storage string concurrency uint warmup uint reloadLimitPerNode uint qpsJitter float64 latencyJitter float64 ) flagSet.StringVar(&version, "version", "v5.3.0", "") flagSet.StringVar(&s3_endpoint, "s3_endpoint", "http://minio.pingcap.net:9000", "") flagSet.StringVar(&s3_key, "s3_key", "minioadmin", "") flagSet.StringVar(&s3_passwd, "s3_passwd", "minioadmin", "") flagSet.StringVar(&storage, "storage", "s3://patrick/workloads/cc-rolling-update/baseline-lite", "") flagSet.UintVar(&concurrency, "concurrency", 5, "workload concurrency") flagSet.UintVar(&warmup, "warmup", 10, "workload warm up(in minutes)") flagSet.UintVar(&reloadLimitPerNode, "reload-per-node-limit", 4, "limit a node reloading cost(in minutes)") flagSet.Float64Var(&qpsJitter, "qps-jitter", 0.2, "percentage of qps jitter") flagSet.Float64Var(&latencyJitter, "latency-jitter", 0.2, "percentage of qps jitter") ctx, _ := utils.GetCtxAndLogger() BeforeEach(func() { cntlNode = ctx.Resource("tiup").(resource.WorkloadNode) cntlHost = host.NewWorkloadHost(cntlNode) log.Info("set tiup mirror to staging server") updateMirrorCmd := "tiup mirror set http://staging.tiup-server.pingcap.net/" Try(cntlHost.Run( workload.NewCMD(updateMirrorCmd, 3*time.Minute))) workloadNode = ctx.Resource("wkload").(resource.WorkloadNode) workloadHost = host.NewWorkloadHost(workloadNode) cluster = host.NewTiUPCluster(cntlNode, "tiup-peer", tiupTopo, "debug") workloadHost.Exec( "pkill br >/dev/null 2>&1 || true", 3*time.Minute) workloadHost.Exec( "pkill go-ycsb >/dev/null 2>&1 || true", 3*time.Minute) cntlHost.Exec( fmt.Sprintf("tiup cluster destroy %s -y >/root/destroy.log 2>&1 || true", "debug"), 3*time.Hour) Try(cluster.Deploy(version)) Try(cluster.Start()) info := Try(cluster.Display()).(host.TiUPClusterInfo) log.Info("install cluster", zap.String("current version:", info.ClusterMeta.ClusterVersion)) pdURL = Try(cluster.ServiceURL(resource.PDAddr)).(*url.URL) dbURL = Try(cluster.ServiceURL(resource.DBAddr)).(*url.URL) log.Info("db info", zap.String("url", dbURL.String()), zap.String("ip:", dbURL.Hostname()), zap.String("port:", dbURL.Port()), zap.String("host:", dbURL.Hostname())) log.Info("restore workload baseline data") start := time.Now() restoreCmd := fmt.Sprintf(` export AWS_ACCESS_KEY_ID=%s AWS_SECRET_ACCESS_KEY=%s tiup br restore full --pd "%s" \ --storage "%s" \ --s3.endpoint "%s" \ --send-credentials-to-tikv=true >/dev/null 2>&1 `, s3_key, s3_passwd, pdURL.Host, storage, s3_endpoint, ) Try(cntlHost.Run( workload.NewCMD(restoreCmd, 5*time.Hour))) log.Info("restore workload baseline date successed", zap.Duration("elapsed", time.Since(start))) }) AfterEach(func() { workloadHost.Exec( "pkill br >/dev/null 2>&1 || true", 3*time.Minute) workloadHost.Exec( "pkill go-ycsb >/dev/null 2>&1 || true", 3*time.Minute) Try(cluster.ServiceURL(resource.DBAddr)) Try(cluster.Destroy()) }) Context("Under CloudChat heavy workload", func() { TestCase("rolling update should fast and less jitter", "oltp_tikv_cc_smooth_rolling_update", func() { ctx, cancel := context.WithTimeout(ctx, 5*time.Hour) defer cancel() g, groupCtx := errgroup.WithContext(ctx) workloadCtx, workloadCancel := context.WithCancel(groupCtx) metricCtx, metricCancel := context.WithCancel(groupCtx) By("Running workload asynchronously") g.Go(WithGinkgoRecover(func() { s := fmt.Sprintf(` /root/wkload/bin/go-ycsb run mysql \ -P /root/wkload/workloads/betting \ -p operationcount=200000000 \ -p recordcount=200000000 \ -p mysql.host=%s \ -p mysql.port=%s \ -p dbnameprefix=test_betting_ \ -p databaseproportions=1.0 \ -p unitnameprefix=unit1_ \ -p unitscount=10 \ -p tablecount=100 \ --threads=%d | tee /root/ycsb.log`, dbURL.Hostname(), dbURL.Port(), concurrency) cmd := workloadnode.CommandContext(workloadCtx, s) Try(cmd.Start(workloadNode)) defer cmd.Stop() if err := cmd.Wait(); err != nil { Expect(cmd.ExitDueToCtxDone).To(BeTrue()) } log.Info("async ycsb task is finished") })) WaitFor(workloadCtx, "workload warm up", time.Duration(warmup)*time.Minute) reloadDone := make(chan struct{}, 1) reloadStart := time.Now() By("Reload test cluster asynchronously") g.Go(WithGinkgoRecover(func() { // Five node cmd := fmt.Sprintf("tiup cluster reload %s -R %s -y >/root/reload.log", cluster.ClusterName, resource.TiKV.String()) Try(cntlHost.Run(workload.NewCMD(cmd, 5*time.Duration(reloadLimitPerNode)*time.Minute))) reloadDone <- struct{}{} log.Info("async reload cluster task is finished") })) By("Running metrics checker asynchronously") g.Go(WithGinkgoRecover(func() { SleepWithCtx(metricCtx, 5*time.Minute) ConsistentlyWithMetrics(cluster). ShouldNot( QPSDropMoreThanOffset(qpsJitter, "5m").For(1 * time.Minute).WithCtx(metricCtx)) log.Info("async QPS metrics checking task is finished") })) g.Go(WithGinkgoRecover(func() { SleepWithCtx(metricCtx, 5*time.Minute) ConsistentlyWithMetrics(cluster).ShouldNot( DurationIncreasedMoreThanOffset(1.+latencyJitter, "5m").For(1 * time.Minute).WithCtx(metricCtx)) log.Info("async latency metrics checking task is finished") })) select { case <-reloadDone: elapsed := time.Since(reloadStart) log.Info("Reload cluster successed", zap.Duration("cost", elapsed)) case <-groupCtx.Done(): elapsed := time.Since(reloadStart) log.Warn("Reload cluster failed", zap.Duration("cost", elapsed)) } workloadCancel() metricCancel() Expect(g.Wait()).Should(BeNil()) }) }) })