Merge branch 'master' into keyspace-pick

.github/workflows/pd-tests.yaml

@@ -23,9 +23,9 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        worker_id: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+        worker_id: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]
     outputs:
-      job-total: 10
+      job-total: 13
     steps:
       - uses: actions/setup-go@v3
         with:
@@ -44,9 +44,10 @@ jobs:
       - name: Make Test
         env:
           WORKER_ID: ${{ matrix.worker_id }}
-          WORKER_COUNT: 10
+          WORKER_COUNT: 13
+          JOB_COUNT: 10 # 11, 12 13 are for other integrations jobs
         run: |
-          make ci-test-job JOB_COUNT=$(($WORKER_COUNT)) JOB_INDEX=$WORKER_ID
+          make ci-test-job JOB_COUNT=$(($JOB_COUNT)) JOB_INDEX=$WORKER_ID
           mv covprofile covprofile_$WORKER_ID
           sed -i "/failpoint_binding/d" covprofile_$WORKER_ID
       - name: Upload coverage result ${{ matrix.worker_id }}

Makefile

@@ -215,8 +215,11 @@ basic-test: install-tools
 
 ci-test-job: install-tools dashboard-ui
 	@$(FAILPOINT_ENABLE)
-	CGO_ENABLED=1 go test -timeout=15m -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=./... $(shell ./scripts/ci-subtask.sh $(JOB_COUNT) $(JOB_INDEX))
-	@ for mod in $(shell ./scripts/ci-subtask.sh $(JOB_COUNT) $(JOB_INDEX) 1); do cd $$mod && $(MAKE) ci-test-job && cd $(ROOT_PATH) > /dev/null && cat $$mod/covprofile >> covprofile; done	
+	if [[ $(JOB_INDEX) -le 10 ]]; then \
+	CGO_ENABLED=1 go test -timeout=15m -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=./... $(shell ./scripts/ci-subtask.sh $(JOB_COUNT) $(JOB_INDEX)); \
+	else \
+	for mod in $(shell ./scripts/ci-subtask.sh $(JOB_COUNT) $(JOB_INDEX)); do cd $$mod && $(MAKE) ci-test-job && cd $(ROOT_PATH) > /dev/null && cat $$mod/covprofile >> covprofile; done; \
+	fi
 
 TSO_INTEGRATION_TEST_PKGS := $(PD_PKG)/tests/server/tso
 

pkg/election/leadership.go

@@ -16,6 +16,7 @@ package election
 
 import (
 	"context"
+	"sync"
 	"sync/atomic"
 
 	"github.com/pingcap/failpoint"
@@ -54,8 +55,9 @@ type Leadership struct {
 	leaderKey   string
 	leaderValue string
 
-	keepAliveCtx        context.Context
-	keepAliveCancelFunc context.CancelFunc
+	keepAliveCtx            context.Context
+	keepAliveCancelFunc     context.CancelFunc
+	keepAliveCancelFuncLock sync.Mutex
 }
 
 // NewLeadership creates a new Leadership.
@@ -137,7 +139,9 @@ func (ls *Leadership) Keep(ctx context.Context) {
 	if ls == nil {
 		return
 	}
+	ls.keepAliveCancelFuncLock.Lock()
 	ls.keepAliveCtx, ls.keepAliveCancelFunc = context.WithCancel(ctx)
+	ls.keepAliveCancelFuncLock.Unlock()
 	go ls.getLease().KeepAlive(ls.keepAliveCtx)
 }
 
@@ -230,8 +234,10 @@ func (ls *Leadership) Reset() {
 	if ls == nil || ls.getLease() == nil {
 		return
 	}
+	ls.keepAliveCancelFuncLock.Lock()
 	if ls.keepAliveCancelFunc != nil {
 		ls.keepAliveCancelFunc()
 	}
+	ls.keepAliveCancelFuncLock.Unlock()
 	ls.getLease().Close()
 }

scripts/ci-subtask.sh

@@ -1,21 +1,21 @@
 #!/bin/bash
 
-# ./ci-subtask.sh <TOTAL_TASK_N> <TASK_INDEX> <INTEGRATION_TEST_ONLY>
+# ./ci-subtask.sh <TOTAL_TASK_N> <TASK_INDEX>
 
-if [[ $3 ]]; then
+if [[ $2 -gt 10 ]]; then
     # Get integration test list.
     makefile_dirs=($(find . -iname "Makefile" -exec dirname {} \; | sort -u))
     submod_dirs=($(find . -iname "go.mod" -exec dirname {} \; | sort -u))
     integration_tasks=$(comm -12 <(printf "%s\n" "${makefile_dirs[@]}") <(printf "%s\n" "${submod_dirs[@]}") | grep "./tests/integrations/*")
     # Currently, we only have 3 integration tests, so we can hardcode the task index.
     for t in ${integration_tasks[@]}; do
-        if [[ "$t" = "./tests/integrations/client" && "$2" = 9 ]]; then
+        if [[ "$t" = "./tests/integrations/client" && "$2" = 11 ]]; then
             printf "%s " "$t"
             break
-        elif [[ "$t" = "./tests/integrations/tso" && "$2" = 7 ]]; then
+        elif [[ "$t" = "./tests/integrations/tso" && "$2" = 12 ]]; then
             printf "%s " "$t"
             break
-        elif [[ "$t" = "./tests/integrations/mcs" && "$2" = 2 ]]; then
+        elif [[ "$t" = "./tests/integrations/mcs" && "$2" = 13 ]]; then
             printf "%s " "$t"
             break
         fi
@@ -45,6 +45,5 @@ else
         scores[$min_i]=$((${scores[$min_i]} + $?))
         [[ $(($min_i + 1)) -eq $2 ]] && res+=($t)
     done
-
     printf "%s " "${res[@]}"
 fi

tests/cluster.go

@@ -62,6 +62,8 @@ var (
 	WaitLeaderReturnDelay = 20 * time.Millisecond
 	// WaitLeaderCheckInterval represents the time interval of WaitLeader running check.
 	WaitLeaderCheckInterval = 500 * time.Millisecond
+	// WaitLeaderRetryTimes represents the maximum number of loops of WaitLeader.
+	WaitLeaderRetryTimes = 100
 )
 
 // TestServer is only for test.
@@ -440,7 +442,7 @@ func (s *TestServer) BootstrapCluster() error {
 // make a test know the PD leader has been elected as soon as possible.
 // If it exceeds the maximum number of loops, it will return nil.
 func (s *TestServer) WaitLeader() bool {
-	for i := 0; i < 100; i++ {
+	for i := 0; i < WaitLeaderRetryTimes; i++ {
 		if s.server.GetMember().IsLeader() {
 			return true
 		}
@@ -649,7 +651,7 @@ func (c *TestCluster) GetFollower() string {
 // If it exceeds the maximum number of loops, it will return an empty string.
 func (c *TestCluster) WaitLeader(ops ...WaitOption) string {
 	option := &WaitOp{
-		retryTimes:   100,
+		retryTimes:   WaitLeaderRetryTimes,
 		waitInterval: WaitLeaderCheckInterval,
 	}
 	for _, op := range ops {
@@ -716,7 +718,7 @@ func (c *TestCluster) ResignLeader() error {
 // If it exceeds the maximum number of loops, it will return an empty string.
 func (c *TestCluster) WaitAllocatorLeader(dcLocation string, ops ...WaitOption) string {
 	option := &WaitOp{
-		retryTimes:   100,
+		retryTimes:   WaitLeaderRetryTimes,
 		waitInterval: WaitLeaderCheckInterval,
 	}
 	for _, op := range ops {

tests/integrations/client/Makefile

@@ -32,7 +32,7 @@ test: failpoint-enable
 	$(MAKE) failpoint-disable
 
 ci-test-job:
-	CGO_ENABLED=1 go test -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=$(ROOT_PATH)/... github.com/tikv/pd/tests/integrations/client
+	CGO_ENABLED=1 go test -v -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=$(ROOT_PATH)/... github.com/tikv/pd/tests/integrations/client
 
 install-tools:
 	cd $(ROOT_PATH) && $(MAKE) install-tools

tests/integrations/mcs/Makefile

@@ -32,7 +32,7 @@ test: failpoint-enable
 	$(MAKE) failpoint-disable
 
 ci-test-job:
-	CGO_ENABLED=1 go test ./... -timeout=15m -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=$(ROOT_PATH)/... github.com/tikv/pd/tests/integrations/mcs
+	CGO_ENABLED=1 go test ./... -v -timeout=15m -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=$(ROOT_PATH)/... github.com/tikv/pd/tests/integrations/mcs
 
 install-tools:
 	cd $(ROOT_PATH) && $(MAKE) install-tools

tests/integrations/mcs/keyspace/tso_keyspace_group_test.go

@@ -24,6 +24,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/suite"
 	bs "github.com/tikv/pd/pkg/basicserver"
 	"github.com/tikv/pd/pkg/mcs/utils"
@@ -54,6 +55,7 @@ func TestKeyspaceGroupTestSuite(t *testing.T) {
 }
 
 func (suite *keyspaceGroupTestSuite) SetupTest() {
+	suite.NoError(failpoint.Enable("github.com/tikv/pd/pkg/keyspace/acceleratedAllocNodes", `return(true)`))
 	ctx, cancel := context.WithCancel(context.Background())
 	suite.ctx = ctx
 	cluster, err := tests.NewTestAPICluster(suite.ctx, 1)
@@ -77,6 +79,7 @@ func (suite *keyspaceGroupTestSuite) SetupTest() {
 func (suite *keyspaceGroupTestSuite) TearDownTest() {
 	suite.cleanupFunc()
 	suite.cluster.Destroy()
+	suite.NoError(failpoint.Disable("github.com/tikv/pd/pkg/keyspace/acceleratedAllocNodes"))
 }
 
 func (suite *keyspaceGroupTestSuite) TestAllocNodesUpdate() {
@@ -292,9 +295,12 @@ func (suite *keyspaceGroupTestSuite) TestDefaultKeyspaceGroup() {
 	mcs.WaitForPrimaryServing(suite.Require(), nodes)
 
 	// the default keyspace group is exist.
-	time.Sleep(2 * time.Second)
-	kg, code := suite.tryGetKeyspaceGroup(utils.DefaultKeyspaceGroupID)
-	suite.Equal(http.StatusOK, code)
+	var kg *endpoint.KeyspaceGroup
+	var code int
+	testutil.Eventually(suite.Require(), func() bool {
+		kg, code = suite.tryGetKeyspaceGroup(utils.DefaultKeyspaceGroupID)
+		return code == http.StatusOK && kg != nil
+	}, testutil.WithWaitFor(time.Second*1))
 	suite.Equal(utils.DefaultKeyspaceGroupID, kg.ID)
 	suite.Len(kg.Members, utils.DefaultKeyspaceGroupReplicaCount)
 	for _, member := range kg.Members {

tests/integrations/tso/Makefile

@@ -32,7 +32,7 @@ test: failpoint-enable
 	$(MAKE) failpoint-disable
 
 ci-test-job:
-	CGO_ENABLED=1 go test -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=$(ROOT_PATH)/... github.com/tikv/pd/tests/integrations/tso
+	CGO_ENABLED=1 go test -v -tags deadlock -race -covermode=atomic -coverprofile=covprofile -coverpkg=$(ROOT_PATH)/... github.com/tikv/pd/tests/integrations/tso
 
 install-tools:
 	cd $(ROOT_PATH) && $(MAKE) install-tools

tests/pdctl/keyspace/keyspace_group_test.go

@@ -21,7 +21,6 @@ import (
 	"strconv"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/require"
@@ -117,7 +116,6 @@ func TestSplitKeyspaceGroup(t *testing.T) {
 	re.NoError(err)
 	cmd := pdctlCmd.GetRootCmd()
 
-	time.Sleep(2 * time.Second)
 	tc.WaitLeader()
 	leaderServer := tc.GetServer(tc.GetLeader())
 	re.NoError(leaderServer.BootstrapCluster())
@@ -157,7 +155,6 @@ func TestExternalAllocNodeWhenStart(t *testing.T) {
 
 	cmd := pdctlCmd.GetRootCmd()
 
-	time.Sleep(2 * time.Second)
 	tc.WaitLeader()
 	leaderServer := tc.GetServer(tc.GetLeader())
 	re.NoError(leaderServer.BootstrapCluster())