add long flag for tests

Signed-off-by: Yang Keao <[email protected]>

Makefile

@@ -297,9 +297,9 @@ failpoint-disable: tools/bin/failpoint-ctl
 	@$(FAILPOINT_DISABLE)
 
 .PHONY: tools/bin/ut
-tools/bin/ut: tools/check/ut.go
+tools/bin/ut: tools/check/ut.go tools/check/longtests.go
 	cd tools/check; \
-	$(GO) build -o ../bin/ut ut.go
+	$(GO) build -o ../bin/ut ut.go longtests.go
 
 .PHONY: tools/bin/xprog
 tools/bin/xprog: tools/check/xprog/xprog.go

pkg/testkit/testflag/flag.go

@@ -0,0 +1,24 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package testflag
+
+import "flag"
+
+var long = flag.Bool("long", false, "run long tests")
+
+// Long returns whether the -long flag is set.
+func Long() bool {
+	return *long
+}

pkg/ttl/ttlworker/job_manager.go

@@ -559,11 +559,13 @@ j:
 				zap.Int64("tableID", job.tbl.ID),
 				zap.String("table", job.tbl.FullName()),
 			)
-			logger.Info("job has finished")
 			summary, err := summarizeTaskResult(allTasks)
 			if err != nil {
 				logger.Info("fail to summarize job", zap.Error(err))
 			}
+			logger.Info("job has finished", zap.String("summary", summary.SummaryText),
+				zap.Uint64("totalRows", summary.TotalRows), zap.Uint64("successRows", summary.SuccessRows), zap.Uint64("errorRows", summary.ErrorRows),
+				zap.String("scanTaskError", summary.ScanTaskErr))
 			err = job.finish(se, se.Now(), summary)
 			if err != nil {
 				logger.Warn("fail to finish job", zap.Error(err))

pkg/ttl/ttlworker/job_manager_integration_test.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/rand/v2"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -38,6 +39,7 @@ import (
 	"github.com/pingcap/tidb/pkg/statistics"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
+	"github.com/pingcap/tidb/pkg/testkit/testflag"
 	timerapi "github.com/pingcap/tidb/pkg/timer/api"
 	timertable "github.com/pingcap/tidb/pkg/timer/tablestore"
 	"github.com/pingcap/tidb/pkg/ttl/cache"
@@ -47,6 +49,7 @@ import (
 	"github.com/pingcap/tidb/pkg/ttl/ttlworker"
 	"github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/logutil"
+	"github.com/pingcap/tidb/pkg/util/skip"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
@@ -107,6 +110,7 @@ func TestGetSession(t *testing.T) {
 func TestParallelLockNewJob(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	waitAndStopTTLManager(t, dom)
+	tk := testkit.NewTestKit(t, store)
 
 	sessionFactory := sessionFactory(t, dom)
 
@@ -116,21 +120,31 @@ func TestParallelLockNewJob(t *testing.T) {
 	m.InfoSchemaCache().Tables[testTable.ID] = testTable
 
 	se := sessionFactory()
+	defer se.Close()
 	job, err := m.LockJob(context.Background(), se, testTable, se.Now(), uuid.NewString(), false)
 	require.NoError(t, err)
 	job.Finish(se, se.Now(), &ttlworker.TTLSummary{})
 
 	// lock one table in parallel, only one of them should lock successfully
-	testTimes := 100
+	testDuration := time.Second
 	concurrency := 5
-	now := se.Now()
-	for i := 0; i < testTimes; i++ {
+	if testflag.Long() {
+		testDuration = 5 * time.Minute
+		concurrency = 50
+	}
+
+	testStart := time.Now()
+	for time.Since(testStart) < testDuration {
+		now := se.Now()
+
+		// reset the table status.
+		tk.MustExec("delete from mysql.tidb_ttl_table_status")
+
 		successCounter := atomic.NewUint64(0)
 		successJob := &ttlworker.TTLJob{}
 
-		now = now.Add(time.Hour * 48)
-
 		wg := sync.WaitGroup{}
+		stopErr := atomic.NewError(nil)
 		for j := 0; j < concurrency; j++ {
 			jobManagerID := fmt.Sprintf("test-ttl-manager-%d", j)
 			wg.Add(1)
@@ -139,19 +153,25 @@ func TestParallelLockNewJob(t *testing.T) {
 				m.InfoSchemaCache().Tables[testTable.ID] = testTable
 
 				se := sessionFactory()
+				defer se.Close()
 				job, err := m.LockJob(context.Background(), se, testTable, now, uuid.NewString(), false)
 				if err == nil {
 					successCounter.Add(1)
 					successJob = job
 				} else {
 					logutil.BgLogger().Info("lock new job with error", zap.Error(err))
 				}
+
+				m.Stop()
+				err = m.WaitStopped(context.Background(), 5*time.Second)
+				stopErr.CompareAndSwap(nil, err)
 				wg.Done()
 			}()
 		}
 		wg.Wait()
 
 		require.Equal(t, uint64(1), successCounter.Load())
+		require.Nil(t, stopErr.Load())
 		successJob.Finish(se, se.Now(), &ttlworker.TTLSummary{})
 	}
 }
@@ -1654,7 +1674,7 @@ func accelerateHeartBeat(t *testing.T, tk *testkit.TestKit) func() {
 
 func TestJobManagerWithFault(t *testing.T) {
 	// TODO: add a flag `-long` to enable this test
-	t.Skip("skip this test because it'll need to run for a long time")
+	skip.NotUnderLong(t)
 
 	defer boostJobScheduleForTest(t)()
 
@@ -1665,7 +1685,8 @@ func TestJobManagerWithFault(t *testing.T) {
 	defer accelerateHeartBeat(t, tk)()
 	tk.MustExec("set @@global.tidb_ttl_running_tasks=32")
 
-	managerCount := 20
+	// don't run too many manager, or the test will be unstable
+	managerCount := runtime.GOMAXPROCS(0) * 2
 	testDuration := 10 * time.Minute
 	faultPercent := 0.5
 
@@ -1774,7 +1795,7 @@ func TestJobManagerWithFault(t *testing.T) {
 
 			logutil.BgLogger().Info("get row count", zap.String("count", rows[0][0].(string)))
 			return false
-		}, time.Second*5, time.Millisecond*100)
+		}, time.Second*30, time.Millisecond*100)
 
 		require.Eventually(t, func() bool {
 			rows := tk.MustQuery("SELECT current_job_state FROM mysql.tidb_ttl_table_status").Rows()
@@ -1785,7 +1806,7 @@ func TestJobManagerWithFault(t *testing.T) {
 			tableStatus := tk.MustQuery("SELECT * FROM mysql.tidb_ttl_table_status").String()
 			logutil.BgLogger().Info("get job state", zap.String("tidb_ttl_table_status", tableStatus))
 			return false
-		}, time.Second*5, time.Millisecond*100)
+		}, time.Second*30, time.Millisecond*100)
 
 		logutil.BgLogger().Info("finish workload", zap.Duration("duration", time.Since(start)))
 	}

pkg/ttl/ttlworker/task_manager_integration_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/store/mockstore"
 	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/testkit/testflag"
 	"github.com/pingcap/tidb/pkg/ttl/cache"
 	"github.com/pingcap/tidb/pkg/ttl/metrics"
 	"github.com/pingcap/tidb/pkg/ttl/ttlworker"
@@ -51,6 +52,7 @@ func TestParallelLockNewTask(t *testing.T) {
 
 	sessionFactory := sessionFactory(t, dom)
 	se := sessionFactory()
+	defer se.Close()
 
 	now := se.Now()
 
@@ -72,9 +74,16 @@ func TestParallelLockNewTask(t *testing.T) {
 	tk.MustExec("DELETE FROM mysql.tidb_ttl_task")
 
 	// lock one table in parallel, only one of them should lock successfully
-	testTimes := 100
+	testDuration := time.Second
 	concurrency := 5
-	for i := 0; i < testTimes; i++ {
+	if testflag.Long() {
+		testDuration = 5 * time.Minute
+		concurrency = 50
+	}
+
+	testStart := time.Now()
+	for time.Since(testStart) < testDuration {
+		now := se.Now()
 		sql, args, err := cache.InsertIntoTTLTask(tk.Session(), "test-job", testTable.Meta().ID, 1, nil, nil, now, now)
 		require.NoError(t, err)
 		_, err = tk.Session().ExecuteInternal(ctx, sql, args...)
@@ -90,6 +99,7 @@ func TestParallelLockNewTask(t *testing.T) {
 			wg.Add(1)
 			go func() {
 				se := sessionFactory()
+				defer se.Close()
 
 				isc := cache.NewInfoSchemaCache(time.Minute)
 				require.NoError(t, isc.Update(se))

pkg/util/skip/skip.go

@@ -14,12 +14,24 @@
 
 package skip
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/testkit/testflag"
+)
 
 // UnderShort skips the test if the -short flag is set.
 func UnderShort(t *testing.T, args ...any) {
 	t.Helper()
 	if testing.Short() {
-		t.Skip(append([]any{"disabled under -short"}, args...))
+		t.Skip(append([]any{"disabled under -short"}, args...)...)
+	}
+}
+
+// NotUnderLong skips the test if the -long flag is not set
+func NotUnderLong(t *testing.T, args ...any) {
+	t.Helper()
+	if !testflag.Long() {
+		t.Skip(append([]any{"disabled not under -short"}, args...)...)
 	}
 }

tools/check/longtests.go

@@ -0,0 +1,25 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+var longTests = map[string][]string{
+	"pkg/ttl/ttlworker": {
+		"TestParallelLockNewJob",
+		"TestParallelLockNewTask",
+		"TestJobManagerWithFault",
+	},
+}
+
+var longTestWorkerCount = 2