support stale-read testcase

pkg/test-infra/tidb/recommend.go

@@ -162,6 +162,9 @@ func RecommendedTiDBCluster(ns, name string, clusterConfig fixture.TiDBClusterCo
 							MountPath:   "/var/log/pdlog",
 						},
 					},
+					Service: &v1alpha1.ServiceSpec{
+						Type: corev1.ServiceTypeNodePort,
+					},
 				},
 				TiKV: &v1alpha1.TiKVSpec{
 					Replicas: int32(clusterConfig.TiKVReplicas),
@@ -260,6 +263,9 @@ func RecommendedTiDBCluster(ns, name string, clusterConfig fixture.TiDBClusterCo
 						Version:   "v2.11.1",
 					},
 					LogLevel: "info",
+					Service: v1alpha1.ServiceSpec{
+						Type: corev1.ServiceTypeNodePort,
+					},
 				},
 				Grafana: &v1alpha1.GrafanaSpec{
 					Service: v1alpha1.ServiceSpec{

run/lib/case.libsonnet

@@ -58,6 +58,10 @@
       '/bin/cross-region',
        '-tso-request-count=%s' % args.tso_request_count,
     ],
+  staleread(args={})::
+    [
+      '/bin/stale-read',
+    ],
   'example'(args={})::
     [
       '/bin/example',

run/workflow/stale-read.jsonnet

@@ -0,0 +1,15 @@
+{
+  _config+:: {
+    case_name: 'staleread',
+    image_name: 'hub.pingcap.net/qa/tipocket',
+    args+: {
+      // k8s configurations
+      // 'storage-class': 'local-storage',
+      // tidbcluster configurations
+      // 'pd-storage-class': 'shared-local-storage',
+      // 'tikv-storage-class': 'local-storage',
+      // 'log-storage-class': 'shared-sas',
+    },
+    command: {},
+  },
+}

testcase/stale-read/Dockerfile

@@ -0,0 +1,5 @@
+FROM alpine:3.9
+
+RUN apk update && apk add --no-cache wget bash sed
+COPY ./bin/stale-read /bin/stale-read
+ENTRYPOINT [ "/bin/stale-read" ]

testcase/stale-read/Makefile

@@ -0,0 +1,56 @@
+GOARCH := $(if $(GOARCH),$(GOARCH),amd64)
+GO=GO15VENDOREXPERIMENT="1" CGO_ENABLED=0 GOOS=$(GOOS) GOARCH=$(GOARCH) GO111MODULE=on go
+GOTEST=GO15VENDOREXPERIMENT="1" CGO_ENABLED=1 GO111MODULE=on go test # go race detector requires cgo
+VERSION   := $(if $(VERSION),$(VERSION),latest)
+
+PACKAGES := go list ./...| grep -vE 'vendor'
+
+LDFLAGS += -s -w
+LDFLAGS += -X "github.com/pingcap/tipocket/pkg/test-infra/fixture.BuildTS=$(shell date -u '+%Y-%m-%d %I:%M:%S')"
+LDFLAGS += -X "github.com/pingcap/tipocket/pkg/test-infra/fixture.BuildHash=$(shell git rev-parse HEAD)"
+
+GOBUILD=$(GO) build -ldflags '$(LDFLAGS)'
+
+DOCKER_REGISTRY_PREFIX := $(if $(DOCKER_REGISTRY),$(DOCKER_REGISTRY)/,)
+
+default: tidy fmt lint build
+
+build: stale-read
+
+stale-read:
+	$(GOBUILD) $(GOMOD) -o bin/stale-read cmd/*.go
+
+fmt: groupimports
+	go fmt ./...
+
+tidy:
+	@echo "go mod tidy"
+	GO111MODULE=on go mod tidy
+	@git diff --exit-code -- go.mod
+
+lint: revive
+	@echo "linting"
+	revive -formatter friendly -config revive.toml $$($(PACKAGES))
+
+revive:
+ifeq (,$(shell which revive))
+	@echo "installing revive"
+	$(GO) get github.com/mgechev/revive@v1.0.2
+endif
+
+groupimports: install-goimports
+	goimports -w -l -local github.com/pingcap/tipocket .
+
+install-goimports:
+ifeq (,$(shell which goimports))
+	@echo "installing goimports"
+	go get golang.org/x/tools/cmd/goimports
+endif
+
+clean:
+	@rm -rf bin/*
+
+test:
+	$(GOTEST) ./...
+
+.PHONY: all clean build

testcase/stale-read/bench.go

@@ -0,0 +1,163 @@
+package staleread
+
+import (
+	"bytes"
+	"database/sql"
+	"fmt"
+	"math/rand"
+	"time"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	"github.com/tiancaiamao/sysbench"
+	"go.uber.org/zap"
+)
+
+const createTableTemplate = `create table if not exists sbtest%d (
+id int(11) not null primary key,
+k int(11) not null,
+c char(120) not null default '',
+pad char(255) not null default '')`
+
+const splitTableTemplate = `SPLIT TABLE sbtest%d BETWEEN (0) AND (1000000000) REGIONS 100;`
+
+// SysbenchCase indicates a sysbench case
+type SysbenchCase struct {
+	insertCount    int
+	rowsEachInsert int
+	preSec         int
+}
+
+// CreateTable ...
+func (c *SysbenchCase) CreateTable(db *sql.DB) error {
+	if err := c.DropTable(db); err != nil {
+		log.Error("fail to drop table", zap.Error(err))
+		return err
+	}
+	if _, err := db.Exec(fmt.Sprintf(createTableTemplate, 0)); err != nil {
+		log.Error("fail to create table", zap.Error(err))
+		return err
+	}
+	if _, err := db.Exec(fmt.Sprintf(splitTableTemplate, 0)); err != nil {
+		log.Error("fail to split table", zap.Error(err))
+		return err
+	}
+	return nil
+}
+
+// InsertData ...
+func (c *SysbenchCase) InsertData(worker *sysbench.Worker, db *sql.DB) error {
+	var buf bytes.Buffer
+	pkID := worker.ID
+	for i := 0; i < c.insertCount; i++ {
+		buf.Reset()
+		fmt.Fprintf(&buf, "insert into sbtest%d (id, k, c, pad) values ", 0)
+		for i := 0; i < c.rowsEachInsert; i++ {
+			pkID = nextPrimaryID(worker.Count, pkID)
+			dot := ""
+			if i > 0 {
+				dot = ", "
+			}
+			fmt.Fprintf(&buf, "%s(%d, %d, '%s', '%s')", dot, pkID, rand.Intn(1<<11), randString(32), randString(32))
+		}
+
+		_, err := db.Exec(buf.String())
+		if err != nil {
+			log.Info("Insert data error", zap.Error(err))
+			return errors.WithStack(err)
+		}
+	}
+	log.Info("insert data finish")
+	// wait some seconds to push safe_ts
+	time.Sleep(5 * time.Second)
+	return nil
+}
+
+// Execute ...
+// TODO: fulfill workload in future
+func (c *SysbenchCase) Execute(worker *sysbench.Worker, db *sql.DB) error {
+	log.Info("worker start execute")
+	err := c.executeSET(db)
+	if err != nil {
+		log.Info("execute set transaction read only as of testcase fail", zap.Error(err))
+		return err
+	}
+	err = c.executeSelect(db)
+	if err != nil {
+		log.Info("execute select as of timestamp fail", zap.Error(err))
+		return err
+	}
+	log.Info("worker start success")
+	return nil
+}
+
+func (c *SysbenchCase) executeSET(db *sql.DB) error {
+	num := c.insertCount * c.rowsEachInsert
+	now := time.Now()
+	previous := now.Add(time.Duration(-c.preSec) * time.Second)
+	nowStr := now.Format("2006-1-2 15:04:05.000")
+	previousStr := previous.Format("2006-1-2 15:04:05.000")
+	setSQL := fmt.Sprintf(`SET TRANSACTION READ ONLY as of timestamp tidb_bounded_staleness('%v', '%v')`, previousStr, nowStr)
+	// set transaction as of and select
+	mustExec(db, setSQL)
+	mustExec(db, fmt.Sprintf("select id, k, c, pad from sbtest0 where k in (%v, %v, %v)", rand.Intn(num), rand.Intn(num), rand.Intn(num)))
+
+	// set transaction as of and begin
+	mustExec(db, setSQL)
+	mustExec(db, "begin")
+	mustExec(db, fmt.Sprintf("select id, k, c, pad from sbtest0 where k in (%v, %v, %v)", rand.Intn(num), rand.Intn(num), rand.Intn(num)))
+	mustExec(db, "commit")
+	return nil
+}
+
+func (c *SysbenchCase) executeSelect(db *sql.DB) error {
+	num := c.insertCount * c.rowsEachInsert
+	now := time.Now()
+	previous := now.Add(time.Duration(-c.preSec) * time.Second)
+	nowStr := now.Format("2006-1-2 15:04:05.000")
+	previousStr := previous.Format("2006-1-2 15:04:05.000")
+	selectSQL := fmt.Sprintf("select id, k, c, pad from sbtest0 as of timestamp tidb_bounded_staleness('%v','%v') where k in (%v, %v, %v)", previousStr, nowStr, rand.Intn(num), rand.Intn(num), rand.Intn(num))
+	mustExec(db, selectSQL)
+	return nil
+}
+
+func (c *SysbenchCase) executeStart(db *sql.DB) {
+	num := c.insertCount * c.rowsEachInsert
+	now := time.Now()
+	previous := now.Add(time.Duration(-c.preSec) * time.Second)
+	nowStr := now.Format("2006-1-2 15:04:05.000")
+	previousStr := previous.Format("2006-1-2 15:04:05.000")
+	startSQL := fmt.Sprintf(`START TRANSACTION READ ONLY as of timestamp tidb_bounded_staleness('%v', '%v')`, previousStr, nowStr)
+	mustExec(db, startSQL)
+	mustExec(db, fmt.Sprintf("select id, k, c, pad from sbtest0 where k in (%v, %v, %v)", rand.Intn(num), rand.Intn(num), rand.Intn(num)))
+	mustExec(db, "commit")
+}
+
+// DropTable ...
+func (c *SysbenchCase) DropTable(db *sql.DB) error {
+	_, err := db.Exec("drop table if exists sbtest0")
+	return err
+}
+
+const ascii = "abcdefghijklmnopqrstuvwxyz1234567890"
+
+func randString(n int) string {
+	var buf bytes.Buffer
+	for i := 0; i < n; i++ {
+		pos := rand.Intn(len(ascii))
+		buf.WriteByte(ascii[pos])
+	}
+	return buf.String()
+}
+
+func nextPrimaryID(workerCount int, current int) int {
+	return current + workerCount
+}
+
+func mustExec(db *sql.DB, sql string) {
+	_, err := db.Exec(sql)
+	if err != nil {
+		log.Error("get err for sql", zap.String("sql", sql), zap.Error(err))
+		panic(fmt.Sprintf("get err for sql: %v, err: %v", sql, err.Error()))
+	}
+}

testcase/stale-read/cmd/main.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"time"
+
+	// use mysql
+	_ "github.com/go-sql-driver/mysql"
+
+	"github.com/pingcap/tipocket/cmd/util"
+	logs "github.com/pingcap/tipocket/logsearch/pkg/logs"
+	"github.com/pingcap/tipocket/pkg/cluster"
+	"github.com/pingcap/tipocket/pkg/control"
+	test_infra "github.com/pingcap/tipocket/pkg/test-infra"
+	"github.com/pingcap/tipocket/pkg/test-infra/fixture"
+	staleread "github.com/pingcap/tipocket/testcase/stale-read"
+)
+
+var (
+	workerCount    = flag.Int("sysbench-worker-count", 1, "the worker count of the sysbench")
+	runDuration    = flag.Duration("sysbench-duration", 1*time.Minute, "the duration of the sysbench running")
+	rowsEachInsert = flag.Int("rows-each-insert", 50, "rows each time insert")
+	insertCount    = flag.Int("insert-count", 20, "count of the inserting")
+	preSecs        = flag.Int("pre-secs", 3, "previous seconds of stale read query")
+)
+
+func main() {
+	flag.Parse()
+	fixture.Context.ClusterName = "stale-read"
+	cfg := control.Config{
+		Mode:        control.ModeStandard,
+		ClientCount: 1,
+		RunTime:     fixture.Context.RunTime,
+	}
+	c := fixture.Context
+	c.TiDBClusterConfig.PDReplicas = 1
+	c.TiDBClusterConfig.TiDBReplicas = 1
+	c.TiDBClusterConfig.TiKVReplicas = 3
+	c.TiDBClusterConfig.LogStorageClassName = "shared-sas-disks"
+	c.TiDBClusterConfig.PDStorageClassName = "shared-nvme-disks"
+	c.TiDBClusterConfig.TiKVStorageClassName = "nvme-disks"
+	c.TiDBClusterConfig.PDImage = "hub.pingcap.net/gaosong/pd:newly-master"
+	c.TiDBClusterConfig.TiDBImage = "hub.pingcap.net/gaosong/tidb:newly-master"
+	c.TiDBClusterConfig.TiKVImage = "hub.pingcap.net/gaosong/tikv:newly-master"
+	suit := util.Suit{
+		Config:   &cfg,
+		Provider: cluster.NewDefaultClusterProvider(),
+		ClientCreator: staleread.ClientCreator{
+			Config: staleread.Config{
+				SysBenchWorkerCount: *workerCount,
+				SysBenchDuration:    *runDuration,
+				RowsEachInsert:      *rowsEachInsert,
+				InsertCount:         *insertCount,
+				PreSec:              *preSecs,
+			},
+		},
+		NemesisGens: util.ParseNemesisGenerators(fixture.Context.Nemesis),
+		ClusterDefs: test_infra.NewDefaultCluster(c.Namespace, c.ClusterName, c.TiDBClusterConfig),
+		LogsClient:  logs.NewDiagnosticLogClient(),
+	}
+	suit.Run(context.Background())
+}

testcase/stale-read/go.mod

@@ -0,0 +1,53 @@
+module github.com/pingcap/tipocket/testcase/stale-read
+
+go 1.16
+
+require (
+	github.com/go-sql-driver/mysql v1.5.0
+	github.com/pingcap/errors v0.11.5-0.20201126102027-b0a155152ca3
+	github.com/pingcap/kvproto v0.0.0-20210219064844-c1844a4775d6 // indirect
+	github.com/pingcap/log v0.0.0-20201112100606-8f1e84a3abc8
+	github.com/pingcap/tipocket v1.0.0
+	github.com/pingcap/tipocket/logsearch v0.0.0-20210602095541-45d321986652
+	github.com/prometheus/client_golang v1.5.0
+	github.com/prometheus/common v0.9.1
+	github.com/stretchr/testify v1.6.0 // indirect
+	github.com/tiancaiamao/sysbench v0.0.0-20200214034607-ee9d97eabd23
+	go.uber.org/zap v1.15.0
+)
+
+replace google.golang.org/grpc => google.golang.org/grpc v1.26.0
+
+replace github.com/prometheus/prometheus => github.com/prometheus/prometheus v1.8.2-0.20200213233353-b90be6f32a33
+
+replace github.com/uber-go/atomic => go.uber.org/atomic v1.5.0
+
+replace (
+	k8s.io/api => k8s.io/api v0.17.0
+	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.17.0
+	k8s.io/apimachinery => k8s.io/apimachinery v0.17.0
+	k8s.io/apiserver => k8s.io/apiserver v0.17.0
+	k8s.io/cli-runtime => k8s.io/cli-runtime v0.17.0
+	k8s.io/client-go => k8s.io/client-go v0.17.0
+	k8s.io/cloud-provider => k8s.io/cloud-provider v0.17.0
+	k8s.io/cluster-bootstrap => k8s.io/cluster-bootstrap v0.17.0
+	k8s.io/code-generator => k8s.io/code-generator v0.17.0
+	k8s.io/component-base => k8s.io/component-base v0.17.0
+	k8s.io/cri-api => k8s.io/cri-api v0.17.0
+	k8s.io/csi-translation-lib => k8s.io/csi-translation-lib v0.17.0
+	k8s.io/kube-aggregator => k8s.io/kube-aggregator v0.17.0
+	k8s.io/kube-controller-manager => k8s.io/kube-controller-manager v0.17.0
+	k8s.io/kube-proxy => k8s.io/kube-proxy v0.17.0
+	k8s.io/kube-scheduler => k8s.io/kube-scheduler v0.17.0
+	k8s.io/kubectl => k8s.io/kubectl v0.17.0
+	k8s.io/kubelet => k8s.io/kubelet v0.17.0
+	k8s.io/legacy-cloud-providers => k8s.io/legacy-cloud-providers v0.17.0
+	k8s.io/metrics => k8s.io/metrics v0.17.0
+	k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.17.0
+)
+
+replace github.com/Azure/go-autorest => github.com/Azure/go-autorest v12.2.0+incompatible
+
+replace golang.org/x/net v0.0.0-20190813000000-74dc4d7220e7 => golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7
+
+replace github.com/pingcap/tipocket => ../../.