ci-subtask: exit ci-subtask.sh when execute ci failed

Makefile

@@ -242,7 +242,7 @@ basic-test: install-tools
 
 ci-test-job: install-tools dashboard-ui
 	@$(FAILPOINT_ENABLE)
-	./scripts/ci-subtask.sh $(JOB_COUNT) $(JOB_INDEX)
+	./scripts/ci-subtask.sh $(JOB_COUNT) $(JOB_INDEX) || { $(FAILPOINT_DISABLE); exit 1; }
 	@$(FAILPOINT_DISABLE)
 
 TSO_INTEGRATION_TEST_PKGS := $(PD_PKG)/tests/server/tso

pkg/mcs/scheduling/server/cluster.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/pdpb"
 	"github.com/pingcap/kvproto/pkg/schedulingpb"
 	"github.com/pingcap/log"
@@ -454,7 +455,11 @@ func (c *Cluster) runCoordinator() {
 	defer logutil.LogPanic()
 	defer c.wg.Done()
 	// force wait for 1 minute to make prepare checker won't be directly skipped
-	c.coordinator.RunUntilStop(collectWaitTime)
+	runCollectWaitTime := collectWaitTime
+	failpoint.Inject("changeRunCollectWaitTime", func() {
+		runCollectWaitTime = 1 * time.Second
+	})
+	c.coordinator.RunUntilStop(runCollectWaitTime)
 }
 
 func (c *Cluster) runMetricsCollectionJob() {

pkg/schedule/config/config.go

@@ -557,7 +557,7 @@ var DefaultSchedulers = SchedulerConfigs{
 	{Type: "evict-slow-store"},
 }
 
-// IsDefaultScheduler checks whether the scheduler is enable by default.
+// IsDefaultScheduler checks whether the scheduler is enabled by default.
 func IsDefaultScheduler(typ string) bool {
 	for _, c := range DefaultSchedulers {
 		if typ == c.Type {

pkg/schedule/schedulers/scheduler_controller.go

@@ -48,7 +48,7 @@ type Controller struct {
 	ctx     context.Context
 	cluster sche.SchedulerCluster
 	storage endpoint.ConfigStorage
-	// schedulers is used to manage all schedulers, which will only be initialized
+	// schedulers are used to manage all schedulers, which will only be initialized
 	// and used in the PD leader service mode now.
 	schedulers map[string]*ScheduleController
 	// schedulerHandlers is used to manage the HTTP handlers of schedulers,

scripts/ci-subtask.sh

@@ -8,20 +8,14 @@ if [[ $2 -gt 10 ]]; then
     integrations_dir=./tests/integrations
     integrations_tasks=($(find "$integrations_dir" -mindepth 1 -maxdepth 1 -type d))
     # Currently, we only have 3 integration tests, so we can hardcode the task index.
-    for t in ${integrations_tasks[@]}; do
-        if [[ "$t" = "$integrations_dir/client" && "$2" = 11 ]]; then
-            cd ./client && make ci-test-job && cd .. && cat ./client/covprofile >> covprofile
-            cd $integrations_dir && make ci-test-job test_name=client
-            cd $ROOT_PATH && cat $integrations_dir/client/covprofile >> covprofile
-            break
-        elif [[ "$t" = "$integrations_dir/tso" && "$2" = 12 ]]; then
-            cd $integrations_dir && make ci-test-job test_name=tso
-            cd $ROOT_PATH && cat $integrations_dir/tso/covprofile >> covprofile
-            break
-        elif [[ "$t" = "$integrations_dir/mcs" && "$2" = 13 ]]; then
-            cd $integrations_dir && make ci-test-job test_name=mcs
-            cd $ROOT_PATH && cat $integrations_dir/mcs/covprofile >> covprofile
-            break
+    for t in "${integrations_tasks[@]}"; do
+        if [[ "$t" = "$integrations_dir/client" && $2 -eq 11 ]]; then
+            cd ./client && make ci-test-job && cd .. && cat ./covprofile >> covprofile || exit 1
+            cd $integrations_dir && make ci-test-job test_name=client && cat ./client/covprofile >> "$ROOT_PATH/covprofile" || exit 1
+        elif [[ "$t" = "$integrations_dir/tso" && $2 -eq 12 ]]; then
+            cd $integrations_dir && make ci-test-job test_name=tso && cat ./tso/covprofile >> "$ROOT_PATH/covprofile" || exit 1
+        elif [[ "$t" = "$integrations_dir/mcs" && $2 -eq 13 ]]; then
+            cd $integrations_dir && make ci-test-job test_name=mcs && cat ./mcs/covprofile >> "$ROOT_PATH/covprofile" || exit 1
         fi
     done
 else

tests/integrations/client/client_test.go

@@ -77,21 +77,19 @@ func TestClientClusterIDCheck(t *testing.T) {
 	defer cluster2.Destroy()
 	endpoints2 := runServer(re, cluster2)
 	// Try to create a client with the mixed endpoints.
-	cli, err := pd.NewClientWithContext(
+	_, err = pd.NewClientWithContext(
 		ctx, append(endpoints1, endpoints2...),
 		pd.SecurityOption{}, pd.WithMaxErrorRetry(1),
 	)
 	re.Error(err)
-	defer cli.Close()
 	re.Contains(err.Error(), "unmatched cluster id")
 	// updateMember should fail due to unmatched cluster ID found.
 	re.NoError(failpoint.Enable("github.com/tikv/pd/client/skipClusterIDCheck", `return(true)`))
 	re.NoError(failpoint.Enable("github.com/tikv/pd/client/skipFirstUpdateMember", `return(true)`))
-	cli, err = pd.NewClientWithContext(ctx, []string{endpoints1[0], endpoints2[0]},
+	_, err = pd.NewClientWithContext(ctx, []string{endpoints1[0], endpoints2[0]},
 		pd.SecurityOption{}, pd.WithMaxErrorRetry(1),
 	)
 	re.Error(err)
-	defer cli.Close()
 	re.Contains(err.Error(), "ErrClientGetMember")
 	re.NoError(failpoint.Disable("github.com/tikv/pd/client/skipFirstUpdateMember"))
 	re.NoError(failpoint.Disable("github.com/tikv/pd/client/skipClusterIDCheck"))

tests/integrations/client/client_tls_test.go

@@ -173,12 +173,12 @@ func testTLSReload(
 				CertPath: testClientTLSInfo.CertFile,
 				KeyPath:  testClientTLSInfo.KeyFile,
 			}, pd.WithGRPCDialOptions(grpc.WithBlock()))
-			cli.Close()
 			if err != nil {
 				errc <- err
 				dcancel()
 				return
 			}
+			cli.Close()
 			dcancel()
 		}
 	}()

tests/integrations/client/http_client_test.go

@@ -508,6 +508,12 @@ func (suite *httpClientTestSuite) checkConfig(mode mode, client pd.Client) {
 	resp, err := env.cluster.GetEtcdClient().Get(env.ctx, sc.TTLConfigPrefix+"/schedule.leader-schedule-limit")
 	re.NoError(err)
 	re.Equal([]byte("16"), resp.Kvs[0].Value)
+	// delete the config with TTL.
+	err = client.SetConfig(env.ctx, newConfig, 0)
+	re.NoError(err)
+	resp, err = env.cluster.GetEtcdClient().Get(env.ctx, sc.TTLConfigPrefix+"/schedule.leader-schedule-limit")
+	re.NoError(err)
+	re.Empty(resp.Kvs)
 }
 
 func (suite *httpClientTestSuite) TestScheduleConfig() {
@@ -520,14 +526,14 @@ func (suite *httpClientTestSuite) checkScheduleConfig(mode mode, client pd.Clien
 
 	config, err := client.GetScheduleConfig(env.ctx)
 	re.NoError(err)
-	re.Equal(float64(4), config["leader-schedule-limit"])
+	re.Equal(float64(4), config["hot-region-schedule-limit"])
 	re.Equal(float64(2048), config["region-schedule-limit"])
-	config["leader-schedule-limit"] = float64(8)
+	config["hot-region-schedule-limit"] = float64(8)
 	err = client.SetScheduleConfig(env.ctx, config)
 	re.NoError(err)
 	config, err = client.GetScheduleConfig(env.ctx)
 	re.NoError(err)
-	re.Equal(float64(8), config["leader-schedule-limit"])
+	re.Equal(float64(8), config["hot-region-schedule-limit"])
 	re.Equal(float64(2048), config["region-schedule-limit"])
 }
 

tests/integrations/mcs/scheduling/api_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/stretchr/testify/suite"
 	"github.com/tikv/pd/pkg/core"
 	_ "github.com/tikv/pd/pkg/mcs/scheduling/server/apis/v1"
@@ -43,11 +44,18 @@ func TestAPI(t *testing.T) {
 }
 
 func (suite *apiTestSuite) SetupSuite() {
+	re := suite.Require()
+	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/schedule/changeCoordinatorTicker", `return(true)`))
+	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/mcs/scheduling/server/changeRunCollectWaitTime", `return(true)`))
 	suite.env = tests.NewSchedulingTestEnvironment(suite.T())
 }
 
 func (suite *apiTestSuite) TearDownSuite() {
 	suite.env.Cleanup()
+	re := suite.Require()
+	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/schedule/changeCoordinatorTicker"))
+	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/mcs/scheduling/server/changeRunCollectWaitTime"))
+	testDialClient.CloseIdleConnections()
 }
 
 func (suite *apiTestSuite) TestGetCheckerByName() {
@@ -190,12 +198,10 @@ func (suite *apiTestSuite) checkAPIForward(cluster *tests.TestCluster) {
 		testutil.WithHeader(re, apiutil.XForwardedToMicroServiceHeader, "true"))
 	re.NoError(err)
 	re.Contains(resp, "balance-leader-scheduler")
-	re.Contains(resp, "balance-witness-scheduler")
 	re.Contains(resp, "balance-hot-region-scheduler")
 
 	schedulers := []string{
 		"balance-leader-scheduler",
-		"balance-witness-scheduler",
 		"balance-hot-region-scheduler",
 	}
 	for _, schedulerName := range schedulers {
@@ -397,13 +403,11 @@ func (suite *apiTestSuite) checkConfig(cluster *tests.TestCluster) {
 	re.Equal(cfg.DataDir, s.GetConfig().DataDir)
 	testutil.Eventually(re, func() bool {
 		// wait for all schedulers to be loaded in scheduling server.
-		return len(cfg.Schedule.SchedulersPayload) == 6
+		return len(cfg.Schedule.SchedulersPayload) == 4
 	})
 	re.Contains(cfg.Schedule.SchedulersPayload, "balance-leader-scheduler")
 	re.Contains(cfg.Schedule.SchedulersPayload, "balance-region-scheduler")
 	re.Contains(cfg.Schedule.SchedulersPayload, "balance-hot-region-scheduler")
-	re.Contains(cfg.Schedule.SchedulersPayload, "balance-witness-scheduler")
-	re.Contains(cfg.Schedule.SchedulersPayload, "transfer-witness-leader-scheduler")
 	re.Contains(cfg.Schedule.SchedulersPayload, "evict-slow-store-scheduler")
 }
 
@@ -428,7 +432,7 @@ func (suite *apiTestSuite) checkConfigForward(cluster *tests.TestCluster) {
 		re.Equal(cfg["replication"].(map[string]interface{})["max-replicas"],
 			float64(opts.GetReplicationConfig().MaxReplicas))
 		schedulers := cfg["schedule"].(map[string]interface{})["schedulers-payload"].(map[string]interface{})
-		return len(schedulers) == 6
+		return len(schedulers) == 4
 	})
 
 	// Test to change config in api server

tests/integrations/mcs/scheduling/server_test.go

@@ -59,6 +59,8 @@ func TestServerTestSuite(t *testing.T) {
 func (suite *serverTestSuite) SetupSuite() {
 	var err error
 	re := suite.Require()
+	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/schedule/changeCoordinatorTicker", `return(true)`))
+	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/mcs/scheduling/server/changeRunCollectWaitTime", `return(true)`))
 	re.NoError(failpoint.Enable("github.com/tikv/pd/server/cluster/highFrequencyClusterJobs", `return(true)`))
 	suite.ctx, suite.cancel = context.WithCancel(context.Background())
 	suite.cluster, err = tests.NewTestAPICluster(suite.ctx, 1)
@@ -78,6 +80,8 @@ func (suite *serverTestSuite) TearDownSuite() {
 	suite.cluster.Destroy()
 	suite.cancel()
 	re.NoError(failpoint.Disable("github.com/tikv/pd/server/cluster/highFrequencyClusterJobs"))
+	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/schedule/changeCoordinatorTicker"))
+	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/mcs/scheduling/server/changeRunCollectWaitTime"))
 }
 
 func (suite *serverTestSuite) TestAllocID() {
@@ -137,7 +141,7 @@ func (suite *serverTestSuite) TestPrimaryChange() {
 	testutil.Eventually(re, func() bool {
 		watchedAddr, ok := suite.pdLeader.GetServicePrimaryAddr(suite.ctx, mcs.SchedulingServiceName)
 		return ok && oldPrimaryAddr == watchedAddr &&
-			len(primary.GetCluster().GetCoordinator().GetSchedulersController().GetSchedulerNames()) == 6
+			len(primary.GetCluster().GetCoordinator().GetSchedulersController().GetSchedulerNames()) == 4
 	})
 	// change primary
 	primary.Close()
@@ -148,7 +152,7 @@ func (suite *serverTestSuite) TestPrimaryChange() {
 	testutil.Eventually(re, func() bool {
 		watchedAddr, ok := suite.pdLeader.GetServicePrimaryAddr(suite.ctx, mcs.SchedulingServiceName)
 		return ok && newPrimaryAddr == watchedAddr &&
-			len(primary.GetCluster().GetCoordinator().GetSchedulersController().GetSchedulerNames()) == 6
+			len(primary.GetCluster().GetCoordinator().GetSchedulersController().GetSchedulerNames()) == 4
 	})
 }