diff --git a/cmd/tidb-server/BUILD.bazel b/cmd/tidb-server/BUILD.bazel index 9b5396426f607..1f0b1ee0befe1 100644 --- a/cmd/tidb-server/BUILD.bazel +++ b/cmd/tidb-server/BUILD.bazel @@ -76,6 +76,7 @@ go_library( "@com_github_tikv_client_go_v2//tikv", "@com_github_tikv_client_go_v2//tikvrpc", "@com_github_tikv_client_go_v2//txnkv/transaction", + "@com_github_tikv_pd_client//:client", "@org_uber_go_automaxprocs//maxprocs", "@org_uber_go_zap//:zap", ], @@ -111,16 +112,20 @@ go_test( srcs = ["main_test.go"], embed = [":tidb-server_lib"], flaky = True, - shard_count = 11, + shard_count = 12, deps = [ "//pkg/config", "//pkg/config/deploymode", "//pkg/config/kerneltype", + "//pkg/extworkload", "//pkg/parser/mysql", "//pkg/sessionctx/vardef", "//pkg/sessionctx/variable", + "//pkg/testkit", "//pkg/testkit/testsetup", + "@com_github_pingcap_kvproto//pkg/keyspacepb", "@com_github_stretchr_testify//require", + "@com_github_tikv_pd_client//:client", "@io_opencensus_go//stats/view", "@org_uber_go_goleak//:goleak", ], diff --git a/cmd/tidb-server/main.go b/cmd/tidb-server/main.go index e117e3e138629..2cb170a83c942 100644 --- a/cmd/tidb-server/main.go +++ b/cmd/tidb-server/main.go @@ -98,6 +98,7 @@ import ( "github.com/tikv/client-go/v2/tikv" "github.com/tikv/client-go/v2/tikvrpc" "github.com/tikv/client-go/v2/txnkv/transaction" + pd "github.com/tikv/pd/client" "go.uber.org/automaxprocs/maxprocs" "go.uber.org/zap" ) @@ -321,33 +322,81 @@ func initDeployMode(cfg *config.Config) error { } func initExternalWorkloadManager(ctx context.Context, storage kv.Storage) extworkload.Manager { + if !deploymode.IsStarter() { + return nil + } cfg := config.GetGlobalConfig().ExternalWorkload if !cfg.Enable { return nil } - // Manager init is non-fatal; TiDB can continue without external workload coordination. + // Non-GCV2 roles can continue without coordination, but a dedicated GCV2 + // worker must not run without the controller. meta := storage.GetCodec().GetKeyspaceMeta() if meta == nil { + if cfg.Role == config.RoleGCV2Worker { + logutil.BgLogger().Fatal("external workload GCV2 role requires keyspace meta") + } logutil.BgLogger().Warn("external workload controller enabled but keyspace meta is unavailable; TiDB will continue without external workload coordination") return nil } + if cfg.Role == config.RoleGCV2Worker && !pd.IsKeyspaceUsingKeyspaceLevelGC(meta) { + logutil.BgLogger().Fatal("external workload GCV2 role requires keyspace-level GC") + } mgr, err := extworkload.NewManager(ctx, meta, cfg) if err != nil { + if cfg.Role == config.RoleGCV2Worker { + logutil.BgLogger().Fatal("failed to initialize external workload manager for GCV2 role", zap.Error(err)) + } logutil.BgLogger().Warn("failed to initialize external workload manager; TiDB will continue without external workload coordination", zap.Error(err)) return nil } + extworkload.SetManagerForStore(storage, mgr) return mgr } -func closeExternalWorkloadManager(mgr extworkload.Manager) { +func closeExternalWorkloadManager(storage kv.Storage, mgr extworkload.Manager) { if mgr == nil { return } + extworkload.SetManagerForStore(storage, nil) if err := mgr.Close(); err != nil { logutil.BgLogger().Warn("failed to close external workload manager", zap.Error(err)) } } +func loadExternalWorkloadGCLifeTime(ctx context.Context, storage kv.Storage) (time.Duration, error) { + se, err := session.CreateSession(storage) + if err != nil { + return 0, err + } + defer se.Close() + gcLifeTimeVal, err := variable.GetSysVar(vardef.TiDBGCLifetime).GetGlobalFromHook(ctx, se.GetSessionVars()) + if err != nil { + return 0, err + } + gcLifeTime, err := time.ParseDuration(gcLifeTimeVal) + if err != nil { + return 0, err + } + return gcLifeTime, nil +} + +func initializeExternalWorkloadGCV2(ctx context.Context, storage kv.Storage, mgr extworkload.Manager) { + if !extworkload.IsMaster(mgr) || !pd.IsKeyspaceUsingKeyspaceLevelGC(mgr.Meta()) { + return + } + gcLifeTime, err := loadExternalWorkloadGCLifeTime(ctx, storage) + if err != nil { + logutil.BgLogger().Warn("failed to load GC life time for external workload GCV2 task; TiDB will continue without external workload coordination", zap.Error(err)) + closeExternalWorkloadManager(storage, mgr) + return + } + if err := mgr.InitializeGCV2(ctx, gcLifeTime); err != nil { + logutil.BgLogger().Warn("failed to initialize external workload GCV2 task; TiDB will continue without external workload coordination", zap.Error(err)) + closeExternalWorkloadManager(storage, mgr) + } +} + func main() { fset := initFlagSet() if args := fset.Args(); len(args) != 0 { @@ -480,9 +529,8 @@ func main() { storage, dom, err := createStoreDDLOwnerMgrAndDomain(keyspaceName) terror.MustNil(err) repository.SetupRepository(dom) - if deploymode.IsStarter() { - externalWorkloadManager := initExternalWorkloadManager(context.Background(), storage) - defer closeExternalWorkloadManager(externalWorkloadManager) + if externalWorkloadManager := extworkload.GetManagerFromStore(storage); externalWorkloadManager != nil { + defer closeExternalWorkloadManager(storage, externalWorkloadManager) } svr := createServer(storage, dom) if standbyController != nil { @@ -631,6 +679,7 @@ func createStoreDDLOwnerMgrAndDomain(keyspaceName string) (kv.Storage, *domain.D } } } + externalWorkloadManager := initExternalWorkloadManager(context.Background(), storage) copr.GlobalMPPFailedStoreProber.Run() mppcoordmanager.InstanceMPPCoordinatorManager.Run() // Bootstrap a session to load information schema. @@ -642,6 +691,7 @@ func createStoreDDLOwnerMgrAndDomain(keyspaceName string) (kv.Storage, *domain.D if err != nil { return nil, nil, err } + initializeExternalWorkloadGCV2(context.Background(), storage, externalWorkloadManager) return storage, dom, nil } diff --git a/cmd/tidb-server/main_test.go b/cmd/tidb-server/main_test.go index 5808ccc14bdf9..971f5be1959cb 100644 --- a/cmd/tidb-server/main_test.go +++ b/cmd/tidb-server/main_test.go @@ -15,18 +15,24 @@ package main import ( + "context" "os" "syscall" "testing" + "time" + "github.com/pingcap/kvproto/pkg/keyspacepb" "github.com/pingcap/tidb/pkg/config" "github.com/pingcap/tidb/pkg/config/deploymode" "github.com/pingcap/tidb/pkg/config/kerneltype" + "github.com/pingcap/tidb/pkg/extworkload" "github.com/pingcap/tidb/pkg/parser/mysql" "github.com/pingcap/tidb/pkg/sessionctx/vardef" "github.com/pingcap/tidb/pkg/sessionctx/variable" + "github.com/pingcap/tidb/pkg/testkit" "github.com/pingcap/tidb/pkg/testkit/testsetup" "github.com/stretchr/testify/require" + pd "github.com/tikv/pd/client" "go.opencensus.io/stats/view" "go.uber.org/goleak" ) @@ -182,6 +188,42 @@ func TestInitDeployMode(t *testing.T) { }) } +type gcv2InitManager struct { + extworkload.Manager + gcLifeTime time.Duration + initCnt int +} + +func (m *gcv2InitManager) Role() config.ExternalWorkloadRole { + return config.RoleMaster +} + +func (*gcv2InitManager) Meta() *keyspacepb.KeyspaceMeta { + return &keyspacepb.KeyspaceMeta{ + Config: map[string]string{ + pd.KeyspaceConfigGCManagementType: pd.KeyspaceConfigGCManagementTypeKeyspaceLevel, + }, + } +} + +func (m *gcv2InitManager) InitializeGCV2(_ context.Context, gcLifeTime time.Duration) error { + m.initCnt++ + m.gcLifeTime = gcLifeTime + return nil +} + +func TestInitializeExternalWorkloadGCV2UsesEffectiveGCLifeTime(t *testing.T) { + store := testkit.CreateMockStore(t) + tk := testkit.NewTestKit(t, store) + tk.MustExec("set global tidb_gc_life_time = '24h'") + + mgr := &gcv2InitManager{} + initializeExternalWorkloadGCV2(context.Background(), store, mgr) + + require.Equal(t, 1, mgr.initCnt) + require.Equal(t, 24*time.Hour, mgr.gcLifeTime) +} + func TestCreateMgrClientRequiresPodIdentityInStarter(t *testing.T) { if kerneltype.IsClassic() { t.Skip("only for nextgen kernel") diff --git a/pkg/executor/BUILD.bazel b/pkg/executor/BUILD.bazel index 9b26968e75bc8..dfaa693142ff2 100644 --- a/pkg/executor/BUILD.bazel +++ b/pkg/executor/BUILD.bazel @@ -147,6 +147,7 @@ go_library( "//pkg/expression/exprctx", "//pkg/expression/sessionexpr", "//pkg/extension", + "//pkg/extworkload", "//pkg/infoschema", "//pkg/infoschema/context", "//pkg/keyspace", @@ -450,6 +451,7 @@ go_test( "//pkg/expression/aggregation", "//pkg/expression/exprstatic", "//pkg/extension", + "//pkg/extworkload", "//pkg/infoschema", "//pkg/keyspace", "//pkg/kv", diff --git a/pkg/executor/set.go b/pkg/executor/set.go index 2ea3d5e1e1ea5..2fd6714d5d910 100644 --- a/pkg/executor/set.go +++ b/pkg/executor/set.go @@ -17,12 +17,14 @@ package executor import ( "context" "strings" + "time" "github.com/pingcap/errors" "github.com/pingcap/tidb/pkg/domain" "github.com/pingcap/tidb/pkg/dxf/framework/storage" "github.com/pingcap/tidb/pkg/executor/internal/exec" "github.com/pingcap/tidb/pkg/expression" + "github.com/pingcap/tidb/pkg/extworkload" "github.com/pingcap/tidb/pkg/parser/ast" "github.com/pingcap/tidb/pkg/parser/charset" "github.com/pingcap/tidb/pkg/parser/mysql" @@ -42,6 +44,7 @@ import ( sem "github.com/pingcap/tidb/pkg/util/sem/compat" semv2 "github.com/pingcap/tidb/pkg/util/sem/v2" "github.com/tikv/client-go/v2/oracle/oracles" + pd "github.com/tikv/pd/client" "go.uber.org/zap" ) @@ -189,6 +192,9 @@ func (e *SetExecutor) setSysVariable(ctx context.Context, name string, v *expres logstr = "set instance var" } logutil.BgLogger().Info(logstr, zap.Uint64("conn", sessionVars.ConnectionID), zap.String("name", name), zap.String("val", showValStr)) + if v.IsGlobal && name == vardef.TiDBGCLifetime { + notifyExternalWorkloadGCLifeTime(ctx, e.Ctx(), showValStr) + } if name == vardef.TiDBServiceScope { dom := domain.GetDomain(e.Ctx()) // SetInstanceSysVar has already updated vardef.ServiceScope in the sysvar hook. @@ -275,6 +281,36 @@ func (e *SetExecutor) setSysVariable(ctx context.Context, name string, v *expres return nil } +func notifyExternalWorkloadGCLifeTime(ctx context.Context, sctx sessionctx.Context, setValue string) { + mgr := extworkload.GetManagerFromStore(sctx.GetStore()) + if !extworkload.IsEnabled(mgr) || !pd.IsKeyspaceUsingKeyspaceLevelGC(mgr.Meta()) { + return + } + + gcLifeTimeVal, err := variable.GetSysVar(vardef.TiDBGCLifetime).GetGlobalFromHook(ctx, sctx.GetSessionVars()) + if err != nil { + logutil.BgLogger().Warn("failed to load effective external workload GC life time", + zap.String("name", vardef.TiDBGCLifetime), + zap.String("val", setValue), + zap.Error(err)) + return + } + gcLifeTime, err := time.ParseDuration(gcLifeTimeVal) + if err != nil { + logutil.BgLogger().Warn("failed to parse effective external workload GC life time", + zap.String("name", vardef.TiDBGCLifetime), + zap.String("val", gcLifeTimeVal), + zap.Error(err)) + return + } + if err := mgr.UpdateGCLifeTime(ctx, gcLifeTime); err != nil { + logutil.BgLogger().Warn("failed to update external workload GC life time", + zap.String("name", vardef.TiDBGCLifetime), + zap.String("val", gcLifeTimeVal), + zap.Error(err)) + } +} + func (e *SetExecutor) setCharset(cs, co string, isSetName bool) error { var err error sessionVars := e.Ctx().GetSessionVars() diff --git a/pkg/executor/set_test.go b/pkg/executor/set_test.go index 1cb6d0f669e0a..79a8b8d446998 100644 --- a/pkg/executor/set_test.go +++ b/pkg/executor/set_test.go @@ -23,12 +23,15 @@ import ( "net/http" "strconv" "testing" + "time" "github.com/pingcap/failpoint" + "github.com/pingcap/kvproto/pkg/keyspacepb" "github.com/pingcap/tidb/pkg/config" "github.com/pingcap/tidb/pkg/errno" "github.com/pingcap/tidb/pkg/executor" "github.com/pingcap/tidb/pkg/expression" + "github.com/pingcap/tidb/pkg/extworkload" "github.com/pingcap/tidb/pkg/infoschema" "github.com/pingcap/tidb/pkg/parser/mysql" "github.com/pingcap/tidb/pkg/parser/terror" @@ -40,8 +43,67 @@ import ( "github.com/pingcap/tidb/pkg/util/mock" topsqlstate "github.com/pingcap/tidb/pkg/util/topsql/state" "github.com/stretchr/testify/require" + pd "github.com/tikv/pd/client" ) +type setGCLifeTimeManager struct { + extworkload.Manager + gcLifeTime time.Duration + updateCnt int + role config.ExternalWorkloadRole + meta *keyspacepb.KeyspaceMeta +} + +func (m *setGCLifeTimeManager) Role() config.ExternalWorkloadRole { + return m.role +} + +func (m *setGCLifeTimeManager) Meta() *keyspacepb.KeyspaceMeta { + return m.meta +} + +func (m *setGCLifeTimeManager) UpdateGCLifeTime(_ context.Context, gcLifeTime time.Duration) error { + m.updateCnt++ + m.gcLifeTime = gcLifeTime + return nil +} + +func TestSetGCLifeTimeNotifiesExternalWorkloadWithEffectiveValue(t *testing.T) { + keyspaceLevelMeta := &keyspacepb.KeyspaceMeta{Config: map[string]string{ + pd.KeyspaceConfigGCManagementType: pd.KeyspaceConfigGCManagementTypeKeyspaceLevel, + }} + cases := []struct { + name string + role config.ExternalWorkloadRole + meta *keyspacepb.KeyspaceMeta + setValue string + expectedGlobal time.Duration + expectedNotify time.Duration + expectedUpdate int + }{ + {name: "master", role: config.RoleMaster, meta: keyspaceLevelMeta, setValue: "24h", expectedGlobal: 24 * time.Hour, expectedNotify: 24 * time.Hour, expectedUpdate: 1}, + {name: "GCV2 worker", role: config.RoleGCV2Worker, meta: keyspaceLevelMeta, setValue: "24h", expectedGlobal: 24 * time.Hour, expectedNotify: 24 * time.Hour, expectedUpdate: 1}, + {name: "TTL worker", role: config.RoleTTLTaskWorker, meta: keyspaceLevelMeta, setValue: "24h", expectedGlobal: 24 * time.Hour, expectedNotify: 24 * time.Hour, expectedUpdate: 1}, + {name: "auto analyze worker", role: config.RoleAutoAnalyzeWorker, meta: keyspaceLevelMeta, setValue: "24h", expectedGlobal: 24 * time.Hour, expectedNotify: 24 * time.Hour, expectedUpdate: 1}, + {name: "minimum value", role: config.RoleMaster, meta: keyspaceLevelMeta, setValue: "1m", expectedGlobal: 10 * time.Minute, expectedNotify: 10 * time.Minute, expectedUpdate: 1}, + {name: "unified GC", role: config.RoleMaster, meta: &keyspacepb.KeyspaceMeta{}, setValue: "24h", expectedGlobal: 24 * time.Hour, expectedUpdate: 0}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + store := testkit.CreateMockStore(t) + mgr := &setGCLifeTimeManager{role: tc.role, meta: tc.meta} + extworkload.SetManagerForStore(store, mgr) + + tk := testkit.NewTestKit(t, store) + tk.MustExec(fmt.Sprintf("set global tidb_gc_life_time = '%s'", tc.setValue)) + tk.MustQuery("select @@global.tidb_gc_life_time").Check(testkit.Rows(tc.expectedGlobal.String())) + require.Equal(t, tc.expectedUpdate, mgr.updateCnt) + require.Equal(t, tc.expectedNotify, mgr.gcLifeTime) + }) + } +} + func TestSetVar(t *testing.T) { store := testkit.CreateMockStore(t) tk := testkit.NewTestKit(t, store) diff --git a/pkg/extworkload/BUILD.bazel b/pkg/extworkload/BUILD.bazel index d50f2725e7d4b..406e091a3d1cd 100644 --- a/pkg/extworkload/BUILD.bazel +++ b/pkg/extworkload/BUILD.bazel @@ -12,6 +12,7 @@ go_library( deps = [ "//pkg/config", "//pkg/extworkload/client", + "//pkg/kv", "//pkg/metrics", "//pkg/util/logutil", "@com_github_pingcap_errors//:errors", @@ -30,7 +31,7 @@ go_test( ], embed = [":extworkload"], flaky = True, - shard_count = 6, + shard_count = 7, deps = [ "//pkg/config", "//pkg/extworkload/client", diff --git a/pkg/extworkload/external_workload.go b/pkg/extworkload/external_workload.go index 53514ae617a1e..6261c0201ae8e 100644 --- a/pkg/extworkload/external_workload.go +++ b/pkg/extworkload/external_workload.go @@ -20,6 +20,7 @@ package extworkload import ( "context" + "time" "github.com/pingcap/kvproto/pkg/keyspacepb" "github.com/pingcap/tidb/pkg/config" @@ -35,15 +36,15 @@ type Manager interface { Meta() *keyspacepb.KeyspaceMeta // InitializeGCV2 seeds the controller with an initial keyspace-level GC task. - InitializeGCV2(ctx context.Context) error + InitializeGCV2(ctx context.Context, gcLifeTime time.Duration) error // AbortGCV2 asks the controller to abort all outstanding keyspace-level GC tasks. AbortGCV2(ctx context.Context) error // RegisterGCV2 reports that a keyspace-level GC round at safePoint has completed. - RegisterGCV2(ctx context.Context, safePoint uint64, gcLifeTime int64) error + RegisterGCV2(ctx context.Context, safePoint uint64, gcLifeTime time.Duration) error // RecycleGCV2 reports that keyspace-level GC up to safePoint has been processed. RecycleGCV2(ctx context.Context, safePoint uint64) error // UpdateGCLifeTime reports that the user's gc_life_time has changed. - UpdateGCLifeTime(ctx context.Context, gcLifeTime int64) error + UpdateGCLifeTime(ctx context.Context, gcLifeTime time.Duration) error // RegisterTTLTask reports that a table with TTL has been created or altered. RegisterTTLTask(ctx context.Context, tableID int64, ttlJobEnable bool) error diff --git a/pkg/extworkload/manager.go b/pkg/extworkload/manager.go index 7f699afbeeb18..0bd1be0e6de72 100644 --- a/pkg/extworkload/manager.go +++ b/pkg/extworkload/manager.go @@ -30,10 +30,6 @@ import ( "google.golang.org/grpc" ) -// defGCLifeTimeSec is the gc_life_time (seconds) seeded to the controller -// at InitializeGCV2 time, before the local gc_life_time variable is loaded. -const defGCLifeTimeSec = 600 - // dialTimeout bounds the synchronous dial + Ping during manager creation. const dialTimeout = 30 * time.Second @@ -138,11 +134,11 @@ func withRequestTimeout(ctx context.Context) (context.Context, context.CancelFun return context.WithTimeout(ctx, requestTimeout) } -func (m *manager) InitializeGCV2(ctx context.Context) error { +func (m *manager) InitializeGCV2(ctx context.Context, gcLifeTime time.Duration) error { ctx, cancel := withRequestTimeout(ctx) defer cancel() ctx = withMetric(ctx, string(config.RoleGCV2Worker), metrics.WorkerActionInit) - return m.cli.RegisterGCV2(ctx, 0, defGCLifeTimeSec) + return m.cli.RegisterGCV2(ctx, 0, int64(gcLifeTime.Seconds())) } func (m *manager) AbortGCV2(ctx context.Context) error { @@ -153,11 +149,11 @@ func (m *manager) AbortGCV2(ctx context.Context) error { return m.cli.RecycleGCV2(ctx, math.MaxUint64) } -func (m *manager) RegisterGCV2(ctx context.Context, safePoint uint64, gcLifeTime int64) error { +func (m *manager) RegisterGCV2(ctx context.Context, safePoint uint64, gcLifeTime time.Duration) error { ctx, cancel := withRequestTimeout(ctx) defer cancel() ctx = withMetric(ctx, string(config.RoleGCV2Worker), metrics.WorkerActionRegister) - return m.cli.RegisterGCV2(ctx, safePoint, gcLifeTime) + return m.cli.RegisterGCV2(ctx, safePoint, int64(gcLifeTime.Seconds())) } func (m *manager) RecycleGCV2(ctx context.Context, safePoint uint64) error { @@ -167,10 +163,10 @@ func (m *manager) RecycleGCV2(ctx context.Context, safePoint uint64) error { return m.cli.RecycleGCV2(ctx, safePoint) } -func (m *manager) UpdateGCLifeTime(ctx context.Context, gcLifeTime int64) error { +func (m *manager) UpdateGCLifeTime(ctx context.Context, gcLifeTime time.Duration) error { ctx, cancel := withRequestTimeout(ctx) defer cancel() - return m.cli.UpdateGCLifeTime(ctx, gcLifeTime) + return m.cli.UpdateGCLifeTime(ctx, int64(gcLifeTime.Seconds())) } func (m *manager) RegisterTTLTask(ctx context.Context, tableID int64, ttlJobEnable bool) error { diff --git a/pkg/extworkload/manager_test.go b/pkg/extworkload/manager_test.go index 017ef7cf1faff..86476a82d7b61 100644 --- a/pkg/extworkload/manager_test.go +++ b/pkg/extworkload/manager_test.go @@ -20,6 +20,7 @@ import ( "math" "net" "testing" + "time" pb "github.com/pingcap/kvproto/pkg/externalworkloadpb" "github.com/pingcap/kvproto/pkg/keyspacepb" @@ -184,11 +185,11 @@ func TestManagerMethodsSetDeadlineAndMetrics(t *testing.T) { }{ { name: "InitializeGCV2", - call: func(m *manager) error { return m.InitializeGCV2(context.Background()) }, + call: func(m *manager) error { return m.InitializeGCV2(context.Background(), time.Hour) }, check: func(cli *fakeClient) { require.Equal(t, "RegisterGCV2", cli.call) require.Equal(t, uint64(0), cli.safePoint) - require.Equal(t, int64(defGCLifeTimeSec), cli.gcLifeTime) + require.Equal(t, int64(3600), cli.gcLifeTime) requireLabels(t, cli, string(config.RoleGCV2Worker), metrics.WorkerActionInit) }, }, @@ -203,7 +204,7 @@ func TestManagerMethodsSetDeadlineAndMetrics(t *testing.T) { }, { name: "RegisterGCV2", - call: func(m *manager) error { return m.RegisterGCV2(context.Background(), 10, 600) }, + call: func(m *manager) error { return m.RegisterGCV2(context.Background(), 10, 10*time.Minute) }, check: func(cli *fakeClient) { require.Equal(t, "RegisterGCV2", cli.call) require.Equal(t, uint64(10), cli.safePoint) @@ -222,7 +223,7 @@ func TestManagerMethodsSetDeadlineAndMetrics(t *testing.T) { }, { name: "UpdateGCLifeTime", - call: func(m *manager) error { return m.UpdateGCLifeTime(context.Background(), 60) }, + call: func(m *manager) error { return m.UpdateGCLifeTime(context.Background(), time.Minute) }, check: func(cli *fakeClient) { require.Equal(t, "UpdateGCLifeTime", cli.call) require.Equal(t, int64(60), cli.gcLifeTime) diff --git a/pkg/extworkload/util.go b/pkg/extworkload/util.go index db17bb6a4a82f..d73a68903f2b0 100644 --- a/pkg/extworkload/util.go +++ b/pkg/extworkload/util.go @@ -14,7 +14,12 @@ package extworkload -import "github.com/pingcap/tidb/pkg/config" +import ( + "context" + + "github.com/pingcap/tidb/pkg/config" + "github.com/pingcap/tidb/pkg/kv" +) // IsEnabled reports whether a Manager is present. func IsEnabled(m Manager) bool { return m != nil } @@ -31,6 +36,41 @@ func IsTTLTaskWorker(m Manager) bool { return roleIs(m, config.RoleTTLTaskWorker // IsAutoAnalyzeWorker reports whether this TiDB should run auto-analyze jobs. func IsAutoAnalyzeWorker(m Manager) bool { return roleIs(m, config.RoleAutoAnalyzeWorker) } +type managerStoreKey struct{} + +// SetManagerForStore binds the manager to store. Passing nil removes it. +func SetManagerForStore(store kv.Storage, mgr Manager) { + if store == nil { + return + } + store.SetOption(managerStoreKey{}, mgr) +} + +// GetManagerFromStore returns the manager bound to store. +func GetManagerFromStore(store kv.Storage) Manager { + if store == nil { + return nil + } + v, ok := store.GetOption(managerStoreKey{}) + if !ok { + return nil + } + mgr, _ := v.(Manager) + return mgr +} + +// AbortGCV2ForUpgrade aborts GCV2 work when this TiDB is the dedicated GCV2 +// worker. It returns true when the caller must terminate after the abort. +func AbortGCV2ForUpgrade(ctx context.Context, mgr Manager) (bool, error) { + if !IsGCV2Worker(mgr) { + return false, nil + } + if err := mgr.AbortGCV2(ctx); err != nil { + return false, err + } + return true, nil +} + func roleIs(m Manager, role config.ExternalWorkloadRole) bool { return IsEnabled(m) && m.Role() == role } diff --git a/pkg/extworkload/util_test.go b/pkg/extworkload/util_test.go index 613a3e3d8315c..27a0b193fd6ba 100644 --- a/pkg/extworkload/util_test.go +++ b/pkg/extworkload/util_test.go @@ -15,21 +15,27 @@ package extworkload import ( + "context" + "errors" "testing" - "github.com/pingcap/kvproto/pkg/keyspacepb" "github.com/pingcap/tidb/pkg/config" "github.com/stretchr/testify/require" ) type stubManager struct { Manager - role config.ExternalWorkloadRole + role config.ExternalWorkloadRole + abortCount int + abortErr error } func (s *stubManager) Role() config.ExternalWorkloadRole { return s.role } -func (s *stubManager) Meta() *keyspacepb.KeyspaceMeta { return nil } -func (s *stubManager) Close() error { return nil } + +func (s *stubManager) AbortGCV2(context.Context) error { + s.abortCount++ + return s.abortErr +} func TestRolePredicatesWhenDisabled(t *testing.T) { require.False(t, IsEnabled(nil)) @@ -59,3 +65,30 @@ func TestRolePredicatesDedicated(t *testing.T) { }) } } + +func TestAbortGCV2ForUpgrade(t *testing.T) { + t.Run("other role", func(t *testing.T) { + mgr := &stubManager{role: config.RoleMaster} + shouldTerminate, err := AbortGCV2ForUpgrade(context.Background(), mgr) + require.NoError(t, err) + require.False(t, shouldTerminate) + require.Zero(t, mgr.abortCount) + }) + + t.Run("GCV2 worker", func(t *testing.T) { + mgr := &stubManager{role: config.RoleGCV2Worker} + shouldTerminate, err := AbortGCV2ForUpgrade(context.Background(), mgr) + require.NoError(t, err) + require.True(t, shouldTerminate) + require.Equal(t, 1, mgr.abortCount) + }) + + t.Run("abort error", func(t *testing.T) { + abortErr := errors.New("abort failed") + mgr := &stubManager{role: config.RoleGCV2Worker, abortErr: abortErr} + shouldTerminate, err := AbortGCV2ForUpgrade(context.Background(), mgr) + require.ErrorIs(t, err, abortErr) + require.False(t, shouldTerminate) + require.Equal(t, 1, mgr.abortCount) + }) +} diff --git a/pkg/session/BUILD.bazel b/pkg/session/BUILD.bazel index 278f832530b28..f7d98732abf0b 100644 --- a/pkg/session/BUILD.bazel +++ b/pkg/session/BUILD.bazel @@ -47,6 +47,7 @@ go_library( "//pkg/expression/sessionexpr", "//pkg/extension", "//pkg/extension/extensionimpl", + "//pkg/extworkload", "//pkg/infoschema", "//pkg/infoschema/context", "//pkg/infoschema/issyncer", diff --git a/pkg/session/session.go b/pkg/session/session.go index a414b90cd3941..bc8e956137010 100644 --- a/pkg/session/session.go +++ b/pkg/session/session.go @@ -66,6 +66,7 @@ import ( "github.com/pingcap/tidb/pkg/expression/sessionexpr" "github.com/pingcap/tidb/pkg/extension" "github.com/pingcap/tidb/pkg/extension/extensionimpl" + "github.com/pingcap/tidb/pkg/extworkload" "github.com/pingcap/tidb/pkg/infoschema" infoschemactx "github.com/pingcap/tidb/pkg/infoschema/context" "github.com/pingcap/tidb/pkg/infoschema/issyncer" @@ -4614,6 +4615,18 @@ func runInBootstrapSession(store kv.Storage, ver int64) { // TODO remove this after we can refactor below code out in this case. logutil.BgLogger().Info("[upgrade] already upgraded by other nodes, switch to normal mode") startMode = ddl.Normal + } else if deploymode.IsStarter() { + shouldTerminate, err := extworkload.AbortGCV2ForUpgrade(context.Background(), extworkload.GetManagerFromStore(store)) + if err != nil { + logutil.BgLogger().Fatal("abort GCV2 worker failed", zap.Error(err)) + } + if shouldTerminate { + if intest.InTest { + return + } + releaseFn() + logutil.BgLogger().Fatal("GCV2 worker aborted before bootstrap upgrade") + } } } s, err := createSession(store) diff --git a/pkg/session/session_nextgen_test.go b/pkg/session/session_nextgen_test.go index 9b1d9f8384cb9..5a5beb439b596 100644 --- a/pkg/session/session_nextgen_test.go +++ b/pkg/session/session_nextgen_test.go @@ -20,11 +20,27 @@ import ( "context" "testing" + "github.com/pingcap/tidb/pkg/config" "github.com/pingcap/tidb/pkg/config/deploymode" + "github.com/pingcap/tidb/pkg/extworkload" "github.com/pingcap/tidb/pkg/sessionctx/variable" "github.com/stretchr/testify/require" ) +type upgradeGCV2Manager struct { + extworkload.Manager + abortCount int +} + +func (*upgradeGCV2Manager) Role() config.ExternalWorkloadRole { + return config.RoleGCV2Worker +} + +func (m *upgradeGCV2Manager) AbortGCV2(context.Context) error { + m.abortCount++ + return nil +} + func TestUsePipelinedDMLDisabledInStarter(t *testing.T) { originalMode := deploymode.Get() require.NoError(t, deploymode.Set(deploymode.Starter)) @@ -41,3 +57,24 @@ func TestUsePipelinedDMLDisabledInStarter(t *testing.T) { require.Len(t, warnings, 1) require.EqualError(t, warnings[0].Err, "Pipelined DML is not supported in this deployment. Fallback to standard mode") } + +func TestUpgradeGCV2AbortUsesPostLockBootstrapVersion(t *testing.T) { + originalMode := deploymode.Get() + require.NoError(t, deploymode.Set(deploymode.Starter)) + t.Cleanup(func() { + require.NoError(t, deploymode.Set(originalMode)) + }) + + store, dom := CreateStoreAndBootstrap(t) + t.Cleanup(func() { + require.NoError(t, store.Close()) + }) + dom.Close() + domap.Delete(store) + + mgr := &upgradeGCV2Manager{} + extworkload.SetManagerForStore(store, mgr) + runInBootstrapSession(store, currentBootstrapVersion-1) + + require.Zero(t, mgr.abortCount) +} diff --git a/pkg/store/gcworker/BUILD.bazel b/pkg/store/gcworker/BUILD.bazel index 812e7efe25672..c2b5521265669 100644 --- a/pkg/store/gcworker/BUILD.bazel +++ b/pkg/store/gcworker/BUILD.bazel @@ -6,12 +6,14 @@ go_library( importpath = "github.com/pingcap/tidb/pkg/store/gcworker", visibility = ["//visibility:public"], deps = [ + "//pkg/config", "//pkg/config/deploymode", "//pkg/ddl", "//pkg/ddl/label", "//pkg/ddl/placement", "//pkg/ddl/util", "//pkg/domain/infosync", + "//pkg/extworkload", "//pkg/kv", "//pkg/meta/model", "//pkg/metrics", @@ -58,7 +60,7 @@ go_test( embed = [":gcworker"], flaky = True, race = "on", - shard_count = 37, + shard_count = 38, deps = [ "//pkg/config", "//pkg/config/deploymode", @@ -67,6 +69,7 @@ go_test( "//pkg/ddl/util", "//pkg/domain", "//pkg/domain/infosync", + "//pkg/extworkload", "//pkg/kv", "//pkg/meta/model", "//pkg/session", diff --git a/pkg/store/gcworker/gc_worker.go b/pkg/store/gcworker/gc_worker.go index 34396ef4d0e13..f38b5a7c0ac46 100644 --- a/pkg/store/gcworker/gc_worker.go +++ b/pkg/store/gcworker/gc_worker.go @@ -31,12 +31,14 @@ import ( "github.com/pingcap/kvproto/pkg/keyspacepb" "github.com/pingcap/kvproto/pkg/kvrpcpb" "github.com/pingcap/kvproto/pkg/metapb" + "github.com/pingcap/tidb/pkg/config" "github.com/pingcap/tidb/pkg/config/deploymode" "github.com/pingcap/tidb/pkg/ddl" "github.com/pingcap/tidb/pkg/ddl/label" "github.com/pingcap/tidb/pkg/ddl/placement" "github.com/pingcap/tidb/pkg/ddl/util" "github.com/pingcap/tidb/pkg/domain/infosync" + "github.com/pingcap/tidb/pkg/extworkload" "github.com/pingcap/tidb/pkg/kv" "github.com/pingcap/tidb/pkg/meta/model" "github.com/pingcap/tidb/pkg/metrics" @@ -837,10 +839,42 @@ func (w *GCWorker) runGCJob(ctx context.Context, safePoint uint64, concurrency g return errors.Trace(err) } + w.notifyGCV2AfterGC(ctx, gcSafePoint) metrics.GCHistogram.WithLabelValues(metrics.StageTotal).Observe(time.Since(startTime).Seconds()) return nil } +func (w *GCWorker) notifyGCV2AfterGC(ctx context.Context, safePoint uint64) { + mgr := extworkload.GetManagerFromStore(w.store) + if !extworkload.IsEnabled(mgr) || !pd.IsKeyspaceUsingKeyspaceLevelGC(mgr.Meta()) { + return + } + + role := mgr.Role() + if role == config.RoleMaster || role == config.RoleTTLTaskWorker || role == config.RoleGCV2Worker { + if err := mgr.RecycleGCV2(ctx, safePoint); err != nil { + logutil.Logger(ctx).Warn("failed to recycle GCV2 task", + zap.String("category", "gc worker"), + zap.Uint64("safePoint", safePoint), + zap.Error(err)) + } + } + + if role == config.RoleMaster || role == config.RoleTTLTaskWorker { + gcLifeTime, err := w.loadDurationWithDefault(gcLifeTimeKey, gcDefaultLifeTime) + if err != nil { + logutil.Logger(ctx).Warn("failed to load GC life time for external workload", + zap.String("category", "gc worker"), + zap.Error(err)) + } else if err := mgr.RegisterGCV2(ctx, safePoint, *gcLifeTime); err != nil { + logutil.Logger(ctx).Warn("failed to register GCV2 task", + zap.String("category", "gc worker"), + zap.Uint64("safePoint", safePoint), + zap.Error(err)) + } + } +} + // deleteRanges processes all delete range records whose ts < safePoint in table `gc_delete_range` // `concurrency` specifies the concurrency to send NotifyDeleteRange. func (w *GCWorker) deleteRanges( diff --git a/pkg/store/gcworker/gc_worker_test.go b/pkg/store/gcworker/gc_worker_test.go index 72ca5d6e3004a..937cd3c98a06e 100644 --- a/pkg/store/gcworker/gc_worker_test.go +++ b/pkg/store/gcworker/gc_worker_test.go @@ -41,6 +41,7 @@ import ( "github.com/pingcap/tidb/pkg/ddl/util" "github.com/pingcap/tidb/pkg/domain" "github.com/pingcap/tidb/pkg/domain/infosync" + "github.com/pingcap/tidb/pkg/extworkload" "github.com/pingcap/tidb/pkg/kv" "github.com/pingcap/tidb/pkg/meta/model" "github.com/pingcap/tidb/pkg/session" @@ -68,6 +69,31 @@ type mockGCWorkerLockResolver struct { batchResolveLocks func([]*txnlock.Lock, *tikv.KeyLocation) (*tikv.KeyLocation, error) } +type stubGCV2Manager struct { + extworkload.Manager + + role config.ExternalWorkloadRole + meta *keyspacepb.KeyspaceMeta + err error + recycleSafePoint uint64 + registerCount int + recycleCount int +} + +func (m *stubGCV2Manager) Role() config.ExternalWorkloadRole { return m.role } +func (m *stubGCV2Manager) Meta() *keyspacepb.KeyspaceMeta { return m.meta } + +func (m *stubGCV2Manager) RegisterGCV2(_ context.Context, _ uint64, _ time.Duration) error { + m.registerCount++ + return m.err +} + +func (m *stubGCV2Manager) RecycleGCV2(_ context.Context, safePoint uint64) error { + m.recycleSafePoint = safePoint + m.recycleCount++ + return m.err +} + func (l *mockGCWorkerLockResolver) ScanLocksInOneRegion(bo *tikv.Backoffer, key []byte, endKey []byte, maxVersion uint64, limit uint32) ([]*txnlock.Lock, *tikv.KeyLocation, error) { locks, loc, err := l.RegionLockResolver.ScanLocksInOneRegion(bo, key, endKey, maxVersion, limit) if err != nil { @@ -224,6 +250,60 @@ func createGCWorkerSuite(t *testing.T, opts ...mockGCWorkerSuiteOption) *mockGCW return s } +func TestNotifyGCV2AfterGCForDedicatedWorker(t *testing.T) { + const safePoint = 123 + for _, tc := range []struct { + name string + gcManagementType string + recycleErr error + wantRecycleCount int + }{ + { + name: "keyspace level GC", + gcManagementType: pd.KeyspaceConfigGCManagementTypeKeyspaceLevel, + wantRecycleCount: 1, + }, + { + name: "recycle failure is best effort", + gcManagementType: pd.KeyspaceConfigGCManagementTypeKeyspaceLevel, + recycleErr: errors.New("mock recycle GCV2 failure"), + wantRecycleCount: 1, + }, + { + name: "unified GC", + gcManagementType: pd.KeyspaceConfigGCManagementTypeUnified, + wantRecycleCount: 0, + }, + } { + t.Run(tc.name, func(t *testing.T) { + store, err := mockstore.NewMockStore(mockstore.WithCurrentKeyspaceMeta(&keyspacepb.KeyspaceMeta{ + Id: 1, + Name: "ks", + Config: map[string]string{pd.KeyspaceConfigGCManagementType: tc.gcManagementType}, + })) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, store.Close()) + }) + + mgr := &stubGCV2Manager{ + role: config.RoleGCV2Worker, + meta: store.GetCodec().GetKeyspaceMeta(), + err: tc.recycleErr, + } + extworkload.SetManagerForStore(store, mgr) + worker := &GCWorker{store: store} + + worker.notifyGCV2AfterGC(context.Background(), safePoint) + require.Zero(t, mgr.registerCount) + require.Equal(t, tc.wantRecycleCount, mgr.recycleCount) + if tc.wantRecycleCount > 0 { + require.Equal(t, uint64(safePoint), mgr.recycleSafePoint) + } + }) + } +} + func (s *mockGCWorkerSuite) mustPut(t *testing.T, key, value string) { txn, err := s.store.Begin() require.NoError(t, err)