diff --git a/pkg/keyspace/keyspace_test.go b/pkg/keyspace/keyspace_test.go index 61c6c44fdfd..e7806b39e3a 100644 --- a/pkg/keyspace/keyspace_test.go +++ b/pkg/keyspace/keyspace_test.go @@ -47,9 +47,11 @@ func TestMain(m *testing.M) { } const ( - testConfig = "test config" - testConfig1 = "config_entry_1" - testConfig2 = "config_entry_2" + testConfig = "test config" + testConfig1 = "config_entry_1" + testConfig2 = "config_entry_2" + testGroupID = "tso_keyspace_group_id" + testUserKind = "user_kind" ) type keyspaceTestSuite struct { @@ -128,8 +130,10 @@ func makeCreateKeyspaceRequests(count int) []*CreateKeyspaceRequest { requests[i] = &CreateKeyspaceRequest{ Name: fmt.Sprintf("test_keyspace_%d", i), Config: map[string]string{ - testConfig1: "100", - testConfig2: "200", + testConfig1: "100", + testConfig2: "200", + testGroupID: "0", + testUserKind: "basic", }, CreateTime: now, } diff --git a/pkg/mcs/discovery/register.go b/pkg/mcs/discovery/register.go index 1610c1a25a0..dc17b768889 100644 --- a/pkg/mcs/discovery/register.go +++ b/pkg/mcs/discovery/register.go @@ -63,6 +63,7 @@ func (sr *ServiceRegister) Register() error { sr.cancel() return fmt.Errorf("put the key with lease %s failed: %v", sr.key, err) } + log.Info("register service", zap.String("key", sr.key)) kresp, err := sr.cli.KeepAlive(sr.ctx, id) if err != nil { sr.cancel() diff --git a/pkg/mcs/registry/registry.go b/pkg/mcs/registry/registry.go index 6c5e8ed07b1..0c8c44687d1 100644 --- a/pkg/mcs/registry/registry.go +++ b/pkg/mcs/registry/registry.go @@ -80,18 +80,18 @@ func (r *ServiceRegistry) InstallAllRESTHandler(srv bs.Server, h map[string]http serviceName := createServiceName(prefix, name) if l, ok := r.services[serviceName]; ok { if err := l.RegisterRESTHandler(h); err != nil { - log.Error("register restful PD failed", zap.String("prefix", prefix), zap.String("service-name", name), zap.Error(err)) + log.Error("register restful service failed", zap.String("prefix", prefix), zap.String("service-name", name), zap.Error(err)) } else { - log.Info("restful PD already registered", zap.String("prefix", prefix), zap.String("service-name", name)) + log.Info("restful service already registered", zap.String("prefix", prefix), zap.String("service-name", name)) } continue } l := builder(srv) r.services[serviceName] = l if err := l.RegisterRESTHandler(h); err != nil { - log.Error("register restful PD failed", zap.String("prefix", prefix), zap.String("service-name", name), zap.Error(err)) + log.Error("register restful service failed", zap.String("prefix", prefix), zap.String("service-name", name), zap.Error(err)) } else { - log.Info("restful PD registered successfully", zap.String("prefix", prefix), zap.String("service-name", name)) + log.Info("restful service registered successfully", zap.String("prefix", prefix), zap.String("service-name", name)) } } } diff --git a/pkg/storage/endpoint/tso.go b/pkg/storage/endpoint/tso.go index 66a11daeb3b..9087a8716c9 100644 --- a/pkg/storage/endpoint/tso.go +++ b/pkg/storage/endpoint/tso.go @@ -21,10 +21,13 @@ import ( "go.uber.org/zap" "github.com/pingcap/errors" + "github.com/pingcap/failpoint" "github.com/pingcap/log" "github.com/tikv/pd/pkg/election" "github.com/tikv/pd/pkg/errs" + keyspaceconstant "github.com/tikv/pd/pkg/keyspace/constant" + mcsconstant "github.com/tikv/pd/pkg/mcs/utils/constant" "github.com/tikv/pd/pkg/storage/kv" "github.com/tikv/pd/pkg/utils/keypath" "github.com/tikv/pd/pkg/utils/typeutil" @@ -33,12 +36,16 @@ import ( // TSOStorage is the interface for timestamp storage. type TSOStorage interface { LoadTimestamp(groupID uint32) (time.Time, error) - SaveTimestamp(ctx context.Context, groupID uint32, ts time.Time, leadership *election.Leadership) error + SaveTimestamp(ctx context.Context, groupID uint32, ts time.Time, leadership *election.Leadership, checkTSOPrimary bool) error DeleteTimestamp(ctx context.Context, groupID uint32) error } var _ TSOStorage = (*StorageEndpoint)(nil) +// ErrTSOServicePrimaryExists indicates that embedded PD TSO must stop updating +// its timestamp window because a TSO service primary has been elected. +var ErrTSOServicePrimaryExists = errors.New("tso microservice primary exists, pd must yield embedded tso") + // LoadTimestamp retrieves the last saved TSO timestamp from etcd. // Before switching back from the TSO microservice to the PD leader, // we must ensure that all keyspace groups are merged into the default @@ -68,12 +75,13 @@ func (se *StorageEndpoint) LoadTimestamp(groupID uint32) (time.Time, error) { // SaveTimestamp saves the timestamp to the storage. The leadership is used to check if the current server is leader // before saving the timestamp to ensure a strong consistency for persistence of the TSO timestamp window. -func (se *StorageEndpoint) SaveTimestamp(ctx context.Context, groupID uint32, ts time.Time, leadership *election.Leadership) error { +func (se *StorageEndpoint) SaveTimestamp(ctx context.Context, groupID uint32, ts time.Time, leadership *election.Leadership, checkTSOPrimary bool) error { logFilds := []zap.Field{ zap.Uint32("group-id", groupID), zap.Time("ts", ts), zap.String("leader-key", leadership.GetLeaderKey()), zap.String("expected-leader-value", leadership.GetLeaderValue()), + zap.Bool("check-tso-primary", checkTSOPrimary), } log.Debug("saving timestamp to the storage", logFilds...) // The PD leadership or TSO primary will always be granted first before the TSO timestamp window is saved. @@ -81,7 +89,14 @@ func (se *StorageEndpoint) SaveTimestamp(ctx context.Context, groupID uint32, ts if len(leadership.GetLeaderValue()) == 0 { return errors.Errorf("%s due to leadership has not been granted yet", errs.NotLeaderErr) } - return se.RunInTxn(ctx, func(txn kv.Txn) error { + var tsoPrimaryPath string + if checkTSOPrimary { + tsoPrimaryPath = keypath.ElectionPath(&keypath.MsParam{ + ServiceName: mcsconstant.TSOServiceName, + GroupID: keyspaceconstant.DefaultKeyspaceGroupID, + }) + } + err := se.RunInTxn(ctx, func(txn kv.Txn) error { // Ensure the current server is leader by reading and comparing the leader value. leaderValue, err := txn.Load(leadership.GetLeaderKey()) if err != nil { @@ -92,6 +107,16 @@ func (se *StorageEndpoint) SaveTimestamp(ctx context.Context, groupID uint32, ts return errors.Errorf("%s due to leader value does not match, current: %s, expected: %s", errs.NotLeaderErr, leaderValue, expected) } + if checkTSOPrimary { + tsoPrimary, err := txn.Load(tsoPrimaryPath) + if err != nil { + return err + } + if len(tsoPrimary) != 0 { + return ErrTSOServicePrimaryExists + } + } + value, err := txn.Load(keypath.TimestampPath(groupID)) if err != nil { return err @@ -108,9 +133,17 @@ func (se *StorageEndpoint) SaveTimestamp(ctx context.Context, groupID uint32, ts if previousTS != typeutil.ZeroTime && typeutil.SubRealTimeByWallClock(ts, previousTS) <= 0 { return errors.Errorf("saving timestamp %d is less than or equal to the previous one %d", ts.UnixNano(), previousTS.UnixNano()) } + failpoint.InjectCall("beforeSaveTimestampTxnCommit") data := typeutil.Uint64ToBytes(uint64(ts.UnixNano())) return txn.Save(keypath.TimestampPath(groupID), string(data)) }) + if checkTSOPrimary && errors.ErrorEqual(err, errs.ErrEtcdTxnConflict) { + tsoPrimary, loadErr := se.Load(tsoPrimaryPath) + if loadErr == nil && len(tsoPrimary) != 0 { + return ErrTSOServicePrimaryExists + } + } + return err } // DeleteTimestamp deletes the timestamp from the storage. diff --git a/pkg/storage/storage_tso_test.go b/pkg/storage/storage_tso_test.go index 808984a9718..2c69dad9553 100644 --- a/pkg/storage/storage_tso_test.go +++ b/pkg/storage/storage_tso_test.go @@ -25,7 +25,11 @@ import ( "github.com/tikv/pd/pkg/election" "github.com/tikv/pd/pkg/errs" + keyspaceconstant "github.com/tikv/pd/pkg/keyspace/constant" + mcsconstant "github.com/tikv/pd/pkg/mcs/utils/constant" + storageendpoint "github.com/tikv/pd/pkg/storage/endpoint" "github.com/tikv/pd/pkg/utils/etcdutil" + "github.com/tikv/pd/pkg/utils/keypath" ) const ( @@ -55,7 +59,7 @@ func TestSaveTimestampWithTimeout(t *testing.T) { defer clean() ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - err := storage.SaveTimestamp(ctx, testGroupID, time.Now().Round(0), leadership) + err := storage.SaveTimestamp(ctx, testGroupID, time.Now().Round(0), leadership, false) re.ErrorIs(err, context.DeadlineExceeded) } @@ -64,7 +68,7 @@ func TestSaveLoadTimestamp(t *testing.T) { storage, clean, leadership := prepare(t) defer clean() expectedTS := time.Now().Round(0) - err := storage.SaveTimestamp(defaultContext, testGroupID, expectedTS, leadership) + err := storage.SaveTimestamp(defaultContext, testGroupID, expectedTS, leadership, false) re.NoError(err) ts, err := storage.LoadTimestamp(testGroupID) re.NoError(err) @@ -76,11 +80,11 @@ func TestTimestampTxn(t *testing.T) { storage, clean, leadership := prepare(t) defer clean() globalTS1 := time.Now().Round(0) - err := storage.SaveTimestamp(defaultContext, testGroupID, globalTS1, leadership) + err := storage.SaveTimestamp(defaultContext, testGroupID, globalTS1, leadership, false) re.NoError(err) globalTS2 := globalTS1.Add(-time.Millisecond).Round(0) - err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS2, leadership) + err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS2, leadership, false) re.Error(err) ts, err := storage.LoadTimestamp(testGroupID) @@ -95,13 +99,13 @@ func TestSaveTimestampWithLeaderCheck(t *testing.T) { // testLeaderKey -> testLeaderValue globalTS := time.Now().Round(0) - err := storage.SaveTimestamp(defaultContext, testGroupID, globalTS, leadership) + err := storage.SaveTimestamp(defaultContext, testGroupID, globalTS, leadership, false) re.NoError(err) ts, err := storage.LoadTimestamp(testGroupID) re.NoError(err) re.Equal(globalTS, ts) - err = storage.SaveTimestamp(context.Background(), testGroupID, globalTS.Add(time.Second), &election.Leadership{}) + err = storage.SaveTimestamp(context.Background(), testGroupID, globalTS.Add(time.Second), &election.Leadership{}, false) re.True(errs.IsLeaderChanged(err)) ts, err = storage.LoadTimestamp(testGroupID) re.NoError(err) @@ -110,7 +114,7 @@ func TestSaveTimestampWithLeaderCheck(t *testing.T) { // testLeaderKey -> "" err = storage.Save(leadership.GetLeaderKey(), "") re.NoError(err) - err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS.Add(time.Second), leadership) + err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS.Add(time.Second), leadership, false) re.True(errs.IsLeaderChanged(err)) ts, err = storage.LoadTimestamp(testGroupID) re.NoError(err) @@ -119,7 +123,7 @@ func TestSaveTimestampWithLeaderCheck(t *testing.T) { // testLeaderKey -> non-existent err = storage.Remove(leadership.GetLeaderKey()) re.NoError(err) - err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS.Add(time.Second), leadership) + err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS.Add(time.Second), leadership, false) re.True(errs.IsLeaderChanged(err)) ts, err = storage.LoadTimestamp(testGroupID) re.NoError(err) @@ -129,9 +133,77 @@ func TestSaveTimestampWithLeaderCheck(t *testing.T) { err = storage.Save(leadership.GetLeaderKey(), testLeaderValue) re.NoError(err) globalTS = globalTS.Add(time.Second) - err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS, leadership) + err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS, leadership, false) re.NoError(err) ts, err = storage.LoadTimestamp(testGroupID) re.NoError(err) re.Equal(globalTS, ts) } + +func TestSaveTimestampCheckTSOPrimary(t *testing.T) { + re := require.New(t) + storage, clean, leadership := prepare(t) + defer clean() + + globalTS := time.Now().Round(0) + err := storage.SaveTimestamp(defaultContext, testGroupID, globalTS, leadership, true) + re.NoError(err) + + tsoPrimaryPath := keypath.ElectionPath(&keypath.MsParam{ + ServiceName: mcsconstant.TSOServiceName, + GroupID: keyspaceconstant.DefaultKeyspaceGroupID, + }) + err = storage.Save(tsoPrimaryPath, "tso-primary") + re.NoError(err) + + err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS.Add(time.Second), leadership, true) + re.ErrorIs(err, storageendpoint.ErrTSOServicePrimaryExists) + + ts, err := storage.LoadTimestamp(testGroupID) + re.NoError(err) + re.Equal(globalTS, ts) + + err = storage.SaveTimestamp(defaultContext, testGroupID, globalTS.Add(time.Second), leadership, false) + re.NoError(err) +} + +func TestSaveTimestampFencedByConcurrentTSOPrimary(t *testing.T) { + re := require.New(t) + storage, clean, leadership := prepare(t) + defer clean() + + globalTS := time.Now().Round(0) + re.NoError(storage.SaveTimestamp(defaultContext, testGroupID, globalTS, leadership, true)) + + entered := make(chan struct{}) + resume := make(chan struct{}) + re.NoError(failpoint.EnableCall("github.com/tikv/pd/pkg/storage/endpoint/beforeSaveTimestampTxnCommit", func() { + close(entered) + <-resume + })) + defer func() { + re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/storage/endpoint/beforeSaveTimestampTxnCommit")) + }() + + errCh := make(chan error, 1) + go func() { + errCh <- storage.SaveTimestamp(defaultContext, testGroupID, globalTS.Add(time.Second), leadership, true) + }() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("timestamp transaction did not reach the commit fence") + } + + tsoPrimaryPath := keypath.ElectionPath(&keypath.MsParam{ + ServiceName: mcsconstant.TSOServiceName, + GroupID: keyspaceconstant.DefaultKeyspaceGroupID, + }) + re.NoError(storage.Save(tsoPrimaryPath, "tso-primary")) + close(resume) + re.ErrorIs(<-errCh, storageendpoint.ErrTSOServicePrimaryExists) + + ts, err := storage.LoadTimestamp(testGroupID) + re.NoError(err) + re.Equal(globalTS, ts) +} diff --git a/pkg/tso/allocator.go b/pkg/tso/allocator.go index 287ebe1d09c..7f1140120e7 100644 --- a/pkg/tso/allocator.go +++ b/pkg/tso/allocator.go @@ -78,6 +78,7 @@ func NewAllocator( member member.Election, storage endpoint.TSOStorage, cfg Config, + checkTSOPrimary func() bool, ) *Allocator { ctx, cancel := context.WithCancel(ctx) keyspaceGroupIDStr := strconv.FormatUint(uint64(keyspaceGroupID), 10) @@ -95,6 +96,7 @@ func NewAllocator( updatePhysicalInterval: cfg.GetTSOUpdatePhysicalInterval(), maxResetTSGap: cfg.GetMaxResetTSGap, tsoMux: &tsoObject{}, + checkTSOPrimary: checkTSOPrimary, metrics: newTSOMetrics(keyspaceGroupIDStr), }, tsoAllocatorRoleGauge: tsoAllocatorRole.WithLabelValues(keyspaceGroupIDStr), @@ -131,6 +133,11 @@ func (a *Allocator) allocatorUpdater() { continue } if err := a.UpdateTSO(); err != nil { + if errors.Is(err, endpoint.ErrTSOServicePrimaryExists) { + log.Info("stop embedded TSO allocator because TSO service primary exists", a.logFields...) + a.Reset(false) + continue + } log.Warn("failed to update allocator's timestamp, resetting the TSO allocator with leadership resignation", append(a.logFields, errs.ZapError(err))...) a.Reset(true) @@ -173,6 +180,9 @@ func (a *Allocator) UpdateTSO() (err error) { if err == nil { return nil } + if errors.Is(err, endpoint.ErrTSOServicePrimaryExists) { + return err + } log.Warn("try to update the tso but failed", zap.Int("retry-count", i), zap.Duration("retry-interval", updateTSORetryInterval), diff --git a/pkg/tso/keyspace_group_manager.go b/pkg/tso/keyspace_group_manager.go index 2e810e9e12c..04ea0a6d7bf 100644 --- a/pkg/tso/keyspace_group_manager.go +++ b/pkg/tso/keyspace_group_manager.go @@ -804,7 +804,7 @@ func (kgm *KeyspaceGroupManager) updateKeyspaceGroup(group *endpoint.KeyspaceGro }) } // Initialize all kinds of maps. - allocator := NewAllocator(kgm.ctx, group.ID, participant, kgm.storage, kgm.cfg) + allocator := NewAllocator(kgm.ctx, group.ID, participant, kgm.storage, kgm.cfg, nil) allocator.startPrimaryElectionLoop() log.Info("created tso allocator", zap.Uint32("keyspace-group-id", group.ID)) diff --git a/pkg/tso/tso.go b/pkg/tso/tso.go index 637327b78e0..f6a4f0699fc 100644 --- a/pkg/tso/tso.go +++ b/pkg/tso/tso.go @@ -71,6 +71,8 @@ type timestampOracle struct { tsoMux *tsoObject // last timestamp window stored in etcd lastSavedTime atomic.Value // stored as time.Time + // checkTSOPrimary fences PD embedded TSO writes when a TSO microservice primary exists. + checkTSOPrimary func() bool // pre-initialized metrics metrics *tsoMetrics @@ -87,7 +89,11 @@ func (t *timestampOracle) getStorageTimeout() time.Duration { func (t *timestampOracle) saveTimestamp(ts time.Time) error { ctx, cancel := context.WithTimeout(context.Background(), t.getStorageTimeout()) defer cancel() - return t.storage.SaveTimestamp(ctx, t.keyspaceGroupID, ts, t.member.GetLeadership()) + checkTSOPrimary := false + if t.checkTSOPrimary != nil { + checkTSOPrimary = t.checkTSOPrimary() + } + return t.storage.SaveTimestamp(ctx, t.keyspaceGroupID, ts, t.member.GetLeadership(), checkTSOPrimary) } // setTSOCondition is the condition for updating physical time. diff --git a/server/api/member.go b/server/api/member.go index b4b3a0a6103..709aa4e2e62 100644 --- a/server/api/member.go +++ b/server/api/member.go @@ -81,7 +81,6 @@ func getMembers(svr *server.Server, forceRefresh bool) (*pdpb.GetMembersResponse if members.GetHeader().GetError() != nil { return nil, errors.WithStack(errors.New(members.GetHeader().GetError().String())) } - for _, m := range members.GetMembers() { var e error m.BinaryVersion, e = svr.GetMember().GetMemberBinaryVersion(m.GetMemberId()) diff --git a/server/apiv2/handlers/microservice.go b/server/apiv2/handlers/microservice.go index af77570473b..82556ff46a2 100644 --- a/server/apiv2/handlers/microservice.go +++ b/server/apiv2/handlers/microservice.go @@ -40,17 +40,16 @@ func RegisterMicroservice(r *gin.RouterGroup) { // @Router /ms/members/{service} [get] func GetMembers(c *gin.Context) { svr := c.MustGet(middlewares.ServerContextKey).(*server.Server) - if !svr.IsKeyspaceGroupEnabled() { - c.AbortWithStatusJSON(http.StatusNotFound, "not support microservice") - return - } - if service := c.Param("service"); len(service) > 0 { entries, err := discovery.GetMSMembers(service, svr.GetClient()) if err != nil { c.AbortWithStatusJSON(http.StatusInternalServerError, err.Error()) return } + if len(entries) == 0 { + c.AbortWithStatusJSON(http.StatusNotFound, "no members found") + return + } c.IndentedJSON(http.StatusOK, entries) return } @@ -67,13 +66,12 @@ func GetMembers(c *gin.Context) { // @Router /ms/primary/{service} [get] func GetPrimary(c *gin.Context) { svr := c.MustGet(middlewares.ServerContextKey).(*server.Server) - if !svr.IsKeyspaceGroupEnabled() { - c.AbortWithStatusJSON(http.StatusNotFound, "not support microservice") - return - } - if service := c.Param("service"); len(service) > 0 { - addr, _ := svr.GetServicePrimaryAddr(c.Request.Context(), service) + addr, exist := svr.GetServicePrimaryAddr(c.Request.Context(), service) + if !exist { + c.AbortWithStatusJSON(http.StatusNotFound, "no primary found") + return + } c.IndentedJSON(http.StatusOK, addr) return } diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 6ca3b4f57d9..aeabdc6497a 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -47,6 +47,7 @@ import ( "github.com/tikv/pd/pkg/gctuner" "github.com/tikv/pd/pkg/id" "github.com/tikv/pd/pkg/keyspace" + keyspaceconstant "github.com/tikv/pd/pkg/keyspace/constant" "github.com/tikv/pd/pkg/mcs/discovery" "github.com/tikv/pd/pkg/mcs/utils/constant" "github.com/tikv/pd/pkg/member" @@ -100,6 +101,7 @@ const ( gcTombstoneInterval = 30 * 24 * time.Hour schedulingServiceCheckInterval = 10 * time.Second tsoServiceCheckInterval = 100 * time.Millisecond + tsoServiceReadyCheckTimeout = time.Second // persistLimitRetryTimes is used to reduce the probability of the persistent error // since the once the store is added or removed, we shouldn't return an error even if the store limit is failed to persist. persistLimitRetryTimes = 5 @@ -194,6 +196,8 @@ type RaftCluster struct { changedRegions chan *core.RegionInfo keyspaceGroupManager *keyspace.GroupManager independentServices sync.Map + tsoSwitchMu sync.RWMutex + tsoServiceReadyFunc func(context.Context) bool hbstreams *hbstream.HeartbeatStreams tsoAllocator *tso.Allocator @@ -525,21 +529,19 @@ func (c *RaftCluster) checkTSOService() { log.Info("TSO dynamic switching is enabled, resuming TSO service checks") } servers, err := discovery.Discover(c.etcdClient, constant.TSOServiceName) - if err != nil || len(servers) == 0 { - if err := c.startTSOJobsIfNeeded(); err != nil { + if err != nil { + log.Warn("failed to discover TSO service, keep current TSO provider", errs.ZapError(err)) + return + } + if len(servers) == 0 { + if err := c.switchTSOProviderToPD(); err != nil { log.Error("failed to start TSO jobs", errs.ZapError(err)) return } - if c.IsServiceIndependent(constant.TSOServiceName) { - log.Info("TSO is provided by PD") - c.UnsetServiceIndependent(constant.TSOServiceName) - } - } else { - c.stopTSOJobsIfNeeded() - if !c.IsServiceIndependent(constant.TSOServiceName) { - log.Info("TSO is provided by TSO server") - c.SetServiceIndependent(constant.TSOServiceName) - } + } else if !c.isDefaultTSOPrimaryEtcdReady() { + log.Debug("TSO service is discovered but not ready, keep current TSO provider") + } else if c.IsServiceIndependent(constant.TSOServiceName) || c.isDefaultTSOServiceReady() { + c.switchTSOProviderToTSO() } } else { prev := c.tsoDynamicSwitchingState.Swap(tsoDynamicSwitchingStateDisabled) @@ -550,12 +552,166 @@ func (c *RaftCluster) checkTSOService() { return } - if err := c.startTSOJobsIfNeeded(); err != nil { + if err := c.switchTSOProviderToPD(); err != nil { log.Error("failed to start TSO jobs", errs.ZapError(err)) return } } +func (c *RaftCluster) isDefaultTSOPrimaryReady() bool { + if !c.isDefaultTSOPrimaryEtcdReady() { + return false + } + return c.isDefaultTSOServiceReady() +} + +func (c *RaftCluster) isDefaultTSOServiceReady() bool { + if c.tsoServiceReadyFunc == nil { + return true + } + ctx := c.ctx + if ctx == nil { + ctx = context.Background() + } + checkCtx, cancel := context.WithTimeout(ctx, tsoServiceReadyCheckTimeout) + defer cancel() + return c.tsoServiceReadyFunc(checkCtx) +} + +func (c *RaftCluster) isDefaultTSOPrimaryEtcdReady() bool { + msParam := &keypath.MsParam{ + ServiceName: constant.TSOServiceName, + GroupID: keyspaceconstant.DefaultKeyspaceGroupID, + } + primaryPath := keypath.ElectionPath(msParam) + expectedPrimaryPath := keypath.ExpectedPrimaryPath(msParam) + // The TSO primary election key is written before its allocator is initialized. + // expected_primary is kept alive only after initialization, so require both + // keys to exist and to point to the same primary. + primary, ok := c.getEtcdValue(primaryPath, "TSO primary") + if !ok { + return false + } + expectedPrimary, ok := c.getEtcdValue(expectedPrimaryPath, "TSO expected primary") + if !ok { + return false + } + if primary != expectedPrimary { + log.Debug("TSO primary is not ready", + zap.String("primary-path", primaryPath), + zap.String("expected-primary-path", expectedPrimaryPath)) + return false + } + return true +} + +func (c *RaftCluster) getEtcdValue(path, name string) (string, bool) { + resp, err := etcdutil.EtcdKVGet(c.etcdClient, path) + if err != nil { + log.Warn("failed to check "+name, zap.String("path", path), errs.ZapError(err)) + return "", false + } + if len(resp.Kvs) == 0 { + return "", false + } + return string(resp.Kvs[0].Value), true +} + +// SetTSOServiceReadyFunc sets the callback used to probe whether the default +// TSO service is ready to serve requests before switching traffic to it. +func (c *RaftCluster) SetTSOServiceReadyFunc(fn func(context.Context) bool) { + c.tsoServiceReadyFunc = fn +} + +// RunEmbeddedTSORequest runs a local TSO request only when TSO is currently +// provided by PD. It shares the switch lock with provider transitions so no +// request can observe a half-switched allocator/provider state. +func (c *RaftCluster) RunEmbeddedTSORequest(fn func() error) (bool, error) { + c.tsoSwitchMu.RLock() + defer c.tsoSwitchMu.RUnlock() + if c.isTSOServiceIndependent() { + return false, nil + } + return true, fn() +} + +// SwitchTSOProviderToPD switches the TSO provider back to PD and keeps new +// local TSO requests blocked until the embedded allocator is initialized. +func (c *RaftCluster) SwitchTSOProviderToPD() error { + if c.isKeyspaceGroupEnabled && c.opt != nil && !c.opt.GetMicroserviceConfig().IsTSODynamicSwitchingEnabled() { + return nil + } + return c.switchTSOProviderToPD() +} + +func (c *RaftCluster) isTSOServiceIndependent() bool { + if c.isKeyspaceGroupEnabled && c.opt != nil && !c.opt.GetMicroserviceConfig().IsTSODynamicSwitchingEnabled() { + return true + } + return c.IsServiceIndependent(constant.TSOServiceName) +} + +func (c *RaftCluster) switchTSOProviderToPD() error { + if c.isTSOAllocatorInitialized() && !c.IsServiceIndependent(constant.TSOServiceName) { + return nil + } + if c.isKeyspaceGroupEnabled { + return c.switchTSOProviderToPDIfNoTSOService() + } + return c.switchTSOProviderToPDLocked() +} + +func (c *RaftCluster) switchTSOProviderToPDLocked() error { + c.tsoSwitchMu.Lock() + defer c.tsoSwitchMu.Unlock() + return c.enableEmbeddedTSO() +} + +func (c *RaftCluster) switchTSOProviderToPDIfNoTSOService() error { + c.tsoSwitchMu.Lock() + defer c.tsoSwitchMu.Unlock() + servers, err := discovery.Discover(c.etcdClient, constant.TSOServiceName) + if err != nil { + return err + } + if len(servers) != 0 { + return nil + } + return c.enableEmbeddedTSO() +} + +// enableEmbeddedTSO must be called with tsoSwitchMu locked. +func (c *RaftCluster) enableEmbeddedTSO() error { + if err := c.startTSOJobsIfNeeded(); err != nil { + return err + } + if c.IsServiceIndependent(constant.TSOServiceName) { + log.Info("TSO is provided by PD") + c.UnsetServiceIndependent(constant.TSOServiceName) + } + return nil +} + +func (c *RaftCluster) switchTSOProviderToTSO() { + if c.IsServiceIndependent(constant.TSOServiceName) && !c.isTSOAllocatorInitialized() { + return + } + c.tsoSwitchMu.Lock() + defer c.tsoSwitchMu.Unlock() + if !c.isDefaultTSOPrimaryEtcdReady() { + return + } + c.stopTSOJobsIfNeeded() + if !c.IsServiceIndependent(constant.TSOServiceName) { + log.Info("TSO is provided by TSO server") + c.SetServiceIndependent(constant.TSOServiceName) + } +} + +func (c *RaftCluster) isTSOAllocatorInitialized() bool { + return c.tsoAllocator != nil && c.tsoAllocator.IsInitialize() +} + func (c *RaftCluster) runServiceCheckJob() { defer logutil.LogPanic() defer c.wg.Done() diff --git a/server/cluster/cluster_test.go b/server/cluster/cluster_test.go index b8080d7a45f..f1549bf90bb 100644 --- a/server/cluster/cluster_test.go +++ b/server/cluster/cluster_test.go @@ -44,6 +44,8 @@ import ( "github.com/tikv/pd/pkg/core/storelimit" "github.com/tikv/pd/pkg/errs" "github.com/tikv/pd/pkg/id" + keyspaceconstant "github.com/tikv/pd/pkg/keyspace/constant" + mcsconstant "github.com/tikv/pd/pkg/mcs/utils/constant" "github.com/tikv/pd/pkg/mock/mockhbstream" "github.com/tikv/pd/pkg/mock/mockid" "github.com/tikv/pd/pkg/progress" @@ -62,6 +64,8 @@ import ( "github.com/tikv/pd/pkg/statistics" "github.com/tikv/pd/pkg/statistics/utils" "github.com/tikv/pd/pkg/storage" + "github.com/tikv/pd/pkg/utils/etcdutil" + "github.com/tikv/pd/pkg/utils/keypath" "github.com/tikv/pd/pkg/utils/keyutil" "github.com/tikv/pd/pkg/utils/logutil" "github.com/tikv/pd/pkg/utils/operatorutil" @@ -2359,6 +2363,158 @@ func newTestRaftCluster( return rc } +func TestRunEmbeddedTSORequestFencesProviderSwitch(t *testing.T) { + re := require.New(t) + c := &RaftCluster{} + entered := make(chan struct{}) + releaseRequest := make(chan struct{}) + resultCh := make(chan struct { + handled bool + err error + }, 1) + go func() { + handled, err := c.RunEmbeddedTSORequest(func() error { + close(entered) + <-releaseRequest + return nil + }) + resultCh <- struct { + handled bool + err error + }{handled: handled, err: err} + }() + + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("embedded TSO request did not start") + } + + switchAcquired := make(chan struct{}) + releaseSwitch := make(chan struct{}) + go func() { + c.tsoSwitchMu.Lock() + defer c.tsoSwitchMu.Unlock() + close(switchAcquired) + <-releaseSwitch + }() + + select { + case <-switchAcquired: + t.Fatal("provider switch acquired the lock while an embedded TSO request was running") + case <-time.After(50 * time.Millisecond): + } + + close(releaseRequest) + select { + case result := <-resultCh: + re.True(result.handled) + re.NoError(result.err) + case <-time.After(time.Second): + t.Fatal("embedded TSO request did not finish") + } + select { + case <-switchAcquired: + case <-time.After(time.Second): + t.Fatal("provider switch did not acquire the lock after the request finished") + } + close(releaseSwitch) +} + +func TestRunEmbeddedTSORequestSkipsIndependentTSO(t *testing.T) { + re := require.New(t) + c := &RaftCluster{} + c.SetServiceIndependent(mcsconstant.TSOServiceName) + called := false + handled, err := c.RunEmbeddedTSORequest(func() error { + called = true + return nil + }) + re.NoError(err) + re.False(handled) + re.False(called) +} + +func TestRunEmbeddedTSORequestSkipsWhenDynamicSwitchingDisabled(t *testing.T) { + re := require.New(t) + _, opt, err := newTestScheduleConfig() + re.NoError(err) + opt.GetMicroserviceConfig().EnableTSODynamicSwitching = false + c := &RaftCluster{ + isKeyspaceGroupEnabled: true, + opt: opt, + } + called := false + handled, err := c.RunEmbeddedTSORequest(func() error { + called = true + return nil + }) + re.NoError(err) + re.False(handled) + re.False(called) +} + +func TestDefaultTSOPrimaryReadyRequiresExpectedPrimaryMatch(t *testing.T) { + re := require.New(t) + _, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil) + defer clean() + c := &RaftCluster{etcdClient: client} + msParam := &keypath.MsParam{ + ServiceName: mcsconstant.TSOServiceName, + GroupID: keyspaceconstant.DefaultKeyspaceGroupID, + } + primaryPath := keypath.ElectionPath(msParam) + expectedPrimaryPath := keypath.ExpectedPrimaryPath(msParam) + + re.False(c.isDefaultTSOPrimaryReady()) + _, err := client.Put(context.Background(), primaryPath, "primary-a") + re.NoError(err) + re.False(c.isDefaultTSOPrimaryReady()) + _, err = client.Put(context.Background(), expectedPrimaryPath, "primary-b") + re.NoError(err) + re.False(c.isDefaultTSOPrimaryReady()) + _, err = client.Put(context.Background(), expectedPrimaryPath, "primary-a") + re.NoError(err) + re.True(c.isDefaultTSOPrimaryReady()) + + called := 0 + c.tsoServiceReadyFunc = func(context.Context) bool { + called++ + return false + } + re.False(c.isDefaultTSOPrimaryReady()) + re.Equal(1, called) + c.tsoServiceReadyFunc = func(context.Context) bool { + called++ + return true + } + re.True(c.isDefaultTSOPrimaryReady()) + re.Equal(2, called) +} + +func TestSwitchTSOProviderToPDKeepsDiscoveredTSOService(t *testing.T) { + re := require.New(t) + _, client, clean := etcdutil.NewTestEtcdCluster(t, 1, nil) + defer clean() + _, opt, err := newTestScheduleConfig() + re.NoError(err) + opt.GetMicroserviceConfig().EnableTSODynamicSwitching = true + c := &RaftCluster{ + etcdClient: client, + isKeyspaceGroupEnabled: true, + opt: opt, + } + c.SetServiceIndependent(mcsconstant.TSOServiceName) + _, err = client.Put(context.Background(), + keypath.RegistryPath(mcsconstant.TSOServiceName, "127.0.0.1:3379"), "127.0.0.1:3379") + re.NoError(err) + + re.NoError(c.SwitchTSOProviderToPD()) + re.True(c.IsServiceIndependent(mcsconstant.TSOServiceName)) + c.checkTSOService() + re.True(c.IsServiceIndependent(mcsconstant.TSOServiceName)) +} + // Create n stores (0..n). func newTestStores(n uint64, version string) []*core.StoreInfo { stores := make([]*core.StoreInfo, 0, n) diff --git a/server/forward.go b/server/forward.go index a5d97bfe81e..026b4fb97de 100644 --- a/server/forward.go +++ b/server/forward.go @@ -47,21 +47,18 @@ func (s *GrpcServer) forwardToTSOService(stream pdpb.PD_TsoServer) error { forwarder = newTSOForwarder(server) tsoStreamErr error ) + releaseForwardSlot, err := s.acquireTSOForwardingSlot() + if err != nil { + return err + } defer func() { - s.concurrentTSOProxyStreamings.Add(-1) + releaseForwardSlot() forwarder.cancel() if grpcutil.NeedRebuildConnection(tsoStreamErr) { s.closeDelegateClient(forwarder.host) } }() - maxConcurrentTSOProxyStreamings := int32(s.GetMaxConcurrentTSOProxyStreamings()) - if maxConcurrentTSOProxyStreamings >= 0 { - if newCount := s.concurrentTSOProxyStreamings.Add(1); newCount > maxConcurrentTSOProxyStreamings { - return errors.WithStack(errs.ErrMaxCountTSOProxyRoutinesExceeded) - } - } - tsDeadlineCh := make(chan *tsoutil.TSDeadline, 1) go tsoutil.WatchTSDeadline(stream.Context(), tsDeadlineCh) @@ -120,6 +117,17 @@ func (f *tsoForwarder) cancel() { } } +func (f *tsoForwarder) reset() { + if f == nil { + return + } + f.cancel() + f.stream = nil + f.ctx = nil + f.canceller = nil + f.host = "" +} + // forwardTSORequest sends the TSO request with the current forward stream. func (f *tsoForwarder) forwardTSORequest( request *pdpb.TsoRequest, @@ -475,9 +483,48 @@ func (s *GrpcServer) isLocalRequest(host string) bool { return false } +func (s *GrpcServer) checkDefaultTSOServiceReady(ctx context.Context) bool { + targetHost, ok := s.GetServicePrimaryAddr(ctx, mcs.TSOServiceName) + if !ok || len(targetHost) == 0 { + return false + } + clientConn, err := s.getDelegateClient(ctx, targetHost) + if err != nil { + log.Debug("failed to probe TSO service", zap.String("tso-addr", targetHost), errs.ZapError(err)) + return false + } + forwardStream, _, cancelForward, err := createTSOForwardStream(ctx, clientConn) + if err != nil { + s.closeDelegateClient(targetHost) + log.Debug("failed to create TSO probe stream", zap.String("tso-addr", targetHost), errs.ZapError(err)) + return false + } + defer cancelForward() + request := &tsopb.TsoRequest{ + Header: &tsopb.RequestHeader{ + ClusterId: keypath.ClusterID(), + KeyspaceId: keyspace.GetBootstrapKeyspaceID(), + KeyspaceGroupId: constant.DefaultKeyspaceGroupID, + }, + Count: 1, + } + if err := forwardStream.Send(request); err != nil { + s.closeDelegateClient(targetHost) + log.Debug("failed to send TSO probe request", zap.String("tso-addr", targetHost), errs.ZapError(err)) + return false + } + resp, err := forwardStream.Recv() + if err != nil { + s.closeDelegateClient(targetHost) + log.Debug("failed to receive TSO probe response", zap.String("tso-addr", targetHost), errs.ZapError(err)) + return false + } + return resp.GetTimestamp() != nil +} + func (s *GrpcServer) getGlobalTSO(ctx context.Context) (pdpb.Timestamp, error) { - if !s.IsServiceIndependent(mcs.TSOServiceName) { - return s.tsoAllocator.GenerateTSO(ctx, 1) + if ts, handled, err := s.generateEmbeddedTSOWithGate(ctx, 1); handled || err != nil { + return ts, err } request := &tsopb.TsoRequest{ Header: &tsopb.RequestHeader{ @@ -509,16 +556,32 @@ func (s *GrpcServer) getGlobalTSO(ctx context.Context) (pdpb.Timestamp, error) { } return false } + tryFallbackToPD := func() (pdpb.Timestamp, bool) { + ts, handled, err := s.switchTSOProviderToPDAndGenerateEmbeddedTSO(ctx, 1) + if handled && err == nil { + return ts, true + } + if err != nil { + log.Warn("failed to switch TSO provider back to PD after TSO forwarding failed", errs.ZapError(err)) + } + return pdpb.Timestamp{}, false + } for i := range maxRetryTimesRequestTSOServer { if i > 0 { time.Sleep(retryIntervalRequestTSOServer) } forwardedHost, ok = s.GetServicePrimaryAddr(ctx, mcs.TSOServiceName) if !ok || forwardedHost == "" { + if ts, ok := tryFallbackToPD(); ok { + return ts, nil + } return pdpb.Timestamp{}, errs.ErrNotFoundTSOAddr } forwardStream, err = s.getTSOForwardStream(forwardedHost) if err != nil { + if ts, ok := tryFallbackToPD(); ok { + return ts, nil + } return pdpb.Timestamp{}, err } start := time.Now() @@ -531,6 +594,9 @@ func (s *GrpcServer) getGlobalTSO(ctx context.Context) (pdpb.Timestamp, error) { } log.Error("send request to tso primary server failed", zap.Error(err), zap.String("tso-addr", forwardedHost)) forwardStream.Unlock() + if ts, ok := tryFallbackToPD(); ok { + return ts, nil + } return pdpb.Timestamp{}, err } ts, err = forwardStream.Recv() @@ -541,11 +607,17 @@ func (s *GrpcServer) getGlobalTSO(ctx context.Context) (pdpb.Timestamp, error) { continue } log.Error("receive response from tso primary server failed", zap.Error(err), zap.String("tso-addr", forwardedHost)) + if ts, ok := tryFallbackToPD(); ok { + return ts, nil + } return pdpb.Timestamp{}, err } return *ts.GetTimestamp(), nil } log.Error("get global tso from tso primary server failed after retry", zap.Error(err), zap.String("tso-addr", forwardedHost)) + if ts, ok := tryFallbackToPD(); ok { + return ts, nil + } return pdpb.Timestamp{}, err } diff --git a/server/grpc_service.go b/server/grpc_service.go index 1a159e6afcf..55e68ed1763 100644 --- a/server/grpc_service.go +++ b/server/grpc_service.go @@ -293,15 +293,17 @@ func (s *GrpcServer) GetMinTS( } var minTS *pdpb.Timestamp - if s.IsServiceIndependent(constant.TSOServiceName) { - minTS, err = s.GetMinTSFromTSOService() + if ts, handled, internalErr := s.generateEmbeddedTSOWithGate(ctx, 1); handled || internalErr != nil { + minTS = &ts + err = internalErr } else { - start := time.Now() - ts, internalErr := s.tsoAllocator.GenerateTSO(ctx, 1) - if internalErr == nil { - tsoHandleDuration.Observe(time.Since(start).Seconds()) + minTS, err = s.GetMinTSFromTSOService() + if err != nil { + if ts, handled, internalErr := s.switchTSOProviderToPDAndGenerateEmbeddedTSO(ctx, 1); handled && internalErr == nil { + minTS = &ts + err = nil + } } - minTS = &ts } if err != nil { return &pdpb.GetMinTSResponse{ @@ -495,10 +497,6 @@ func (s *GrpcServer) Tso(stream pdpb.PD_TsoServer) error { if done != nil { defer done() } - if s.IsServiceIndependent(constant.TSOServiceName) { - tsoForwardStreamCounter.Inc() - return s.forwardToTSOService(stream) - } tsDeadlineCh := make(chan *tsoutil.TSDeadline, 1) go tsoutil.WatchTSDeadline(stream.Context(), tsDeadlineCh) @@ -506,11 +504,16 @@ func (s *GrpcServer) Tso(stream pdpb.PD_TsoServer) error { var ( // The following are tso forward stream related variables. tsoRequestProxyCtx context.Context - forwarder = newTSOForwarder(stream) + proxyStream = &tsoServer{stream: stream} + forwarder = newTSOForwarder(proxyStream) tsoStreamErr error + releaseForwardSlot func() ) defer func() { + if releaseForwardSlot != nil { + releaseForwardSlot() + } forwarder.cancel() if grpcutil.NeedRebuildConnection(tsoStreamErr) { s.closeDelegateClient(forwarder.host) @@ -519,6 +522,7 @@ func (s *GrpcServer) Tso(stream pdpb.PD_TsoServer) error { ctx, cancel := context.WithCancel(stream.Context()) defer cancel() +requestLoop: for { var ( request *pdpb.TsoRequest @@ -526,7 +530,11 @@ func (s *GrpcServer) Tso(stream pdpb.PD_TsoServer) error { ) if tsoRequestProxyCtx == nil { - request, err = stream.Recv() + if s.IsServiceIndependent(constant.TSOServiceName) { + request, err = proxyStream.recv(s.GetTSOProxyRecvFromClientTimeout()) + } else { + request, err = stream.Recv() + } } else { // if we forward requests to TSO proxy we can't block on the next request in the stream // as proxy might fail on the previous request, and we need to return the error to client @@ -577,45 +585,156 @@ func (s *GrpcServer) Tso(stream pdpb.PD_TsoServer) error { continue } - if s.IsServiceIndependent(constant.TSOServiceName) { - if request.GetCount() == 0 { - err = errs.ErrGenerateTimestamp.FastGenByArgs("tso count should be positive") - return errs.ErrUnknown(err) - } - tsoStreamErr, err = s.handleTSOForwarding(stream.Context(), forwarder, request, tsDeadlineCh) - if tsoStreamErr != nil { - return tsoStreamErr + raftCluster := s.DirectlyGetRaftCluster() + if raftCluster == nil { + return errs.ErrNotStarted + } + var response *pdpb.TsoResponse + handled, err := raftCluster.RunEmbeddedTSORequest(func() error { + var generateErr error + response, generateErr = s.generateEmbeddedTSOResponse(ctx, request) + return generateErr + }) + if handled { + if releaseForwardSlot != nil { + releaseForwardSlot() + releaseForwardSlot = nil } if err != nil { return err } + if err := stream.Send(response); err != nil { + return errors.WithStack(err) + } continue } - start := time.Now() - if clusterID := keypath.ClusterID(); request.GetHeader().GetClusterId() != clusterID { - return errs.ErrMismatchClusterID(clusterID, request.GetHeader().GetClusterId()) + if releaseForwardSlot == nil { + releaseForwardSlot, err = s.acquireTSOForwardingSlot() + if err != nil { + return err + } + tsoForwardStreamCounter.Inc() } - count := request.GetCount() - tsoBatchSize.Observe(float64(count)) - ctx, task := trace.NewTask(ctx, "tso") - ts, err := s.tsoAllocator.GenerateTSO(ctx, count) - task.End() - tsoHandleDuration.Observe(time.Since(start).Seconds()) - if err != nil { + if request.GetCount() == 0 { + err = errs.ErrGenerateTimestamp.FastGenByArgs("tso count should be positive") return errs.ErrUnknown(err) } - response := &pdpb.TsoResponse{ - Header: grpcutil.WrapHeader(), - Timestamp: &ts, - Count: count, - } - if err := stream.Send(response); err != nil { - return errors.WithStack(err) + for retry := range maxRetryTimesRequestTSOServer { + if retry > 0 { + time.Sleep(retryIntervalRequestTSOServer) + } + tsoStreamErr, err = s.handleTSOForwarding(stream.Context(), forwarder, request, tsDeadlineCh) + if tsoStreamErr == nil { + if err != nil { + return err + } + continue requestLoop + } + if err := raftCluster.SwitchTSOProviderToPD(); err != nil { + log.Warn("failed to switch TSO provider back to PD after TSO forwarding failed", errs.ZapError(err)) + } + needRebuildConnection := grpcutil.NeedRebuildConnection(tsoStreamErr) + forwardHost := forwarder.host + handled, err = raftCluster.RunEmbeddedTSORequest(func() error { + forwarder.reset() + var generateErr error + response, generateErr = s.generateEmbeddedTSOResponse(ctx, request) + return generateErr + }) + if handled { + if releaseForwardSlot != nil { + releaseForwardSlot() + releaseForwardSlot = nil + } + if needRebuildConnection { + s.closeDelegateClient(forwardHost) + } + tsoStreamErr = nil + if err != nil { + return err + } + if err := stream.Send(response); err != nil { + return errors.WithStack(err) + } + continue requestLoop + } + if needRebuildConnection { + s.closeDelegateClient(forwardHost) + } + forwarder.reset() + if retry == maxRetryTimesRequestTSOServer-1 { + return tsoStreamErr + } } } } +func (s *GrpcServer) acquireTSOForwardingSlot() (func(), error) { + maxConcurrentTSOProxyStreamings := int32(s.GetMaxConcurrentTSOProxyStreamings()) + if maxConcurrentTSOProxyStreamings < 0 { + return func() {}, nil + } + if newCount := s.concurrentTSOProxyStreamings.Add(1); newCount > maxConcurrentTSOProxyStreamings { + s.concurrentTSOProxyStreamings.Add(-1) + return nil, errors.WithStack(errs.ErrMaxCountTSOProxyRoutinesExceeded) + } + return func() { + s.concurrentTSOProxyStreamings.Add(-1) + }, nil +} + +func (s *GrpcServer) generateEmbeddedTSOResponse(ctx context.Context, request *pdpb.TsoRequest) (*pdpb.TsoResponse, error) { + if clusterID := keypath.ClusterID(); request.GetHeader().GetClusterId() != clusterID { + return nil, errs.ErrMismatchClusterID(clusterID, request.GetHeader().GetClusterId()) + } + count := request.GetCount() + tsoBatchSize.Observe(float64(count)) + ts, err := s.generateEmbeddedTSO(ctx, count) + if err != nil { + return nil, errs.ErrUnknown(err) + } + return &pdpb.TsoResponse{ + Header: grpcutil.WrapHeader(), + Timestamp: &ts, + Count: count, + }, nil +} + +func (s *GrpcServer) generateEmbeddedTSOWithGate(ctx context.Context, count uint32) (pdpb.Timestamp, bool, error) { + raftCluster := s.DirectlyGetRaftCluster() + if raftCluster == nil { + return pdpb.Timestamp{}, false, errs.ErrNotStarted + } + var ts pdpb.Timestamp + handled, err := raftCluster.RunEmbeddedTSORequest(func() error { + var generateErr error + ts, generateErr = s.generateEmbeddedTSO(ctx, count) + return generateErr + }) + return ts, handled, err +} + +func (s *GrpcServer) switchTSOProviderToPDAndGenerateEmbeddedTSO(ctx context.Context, count uint32) (pdpb.Timestamp, bool, error) { + raftCluster := s.DirectlyGetRaftCluster() + if raftCluster == nil { + return pdpb.Timestamp{}, false, errs.ErrNotStarted + } + if err := raftCluster.SwitchTSOProviderToPD(); err != nil { + return pdpb.Timestamp{}, true, err + } + return s.generateEmbeddedTSOWithGate(ctx, count) +} + +func (s *GrpcServer) generateEmbeddedTSO(ctx context.Context, count uint32) (pdpb.Timestamp, error) { + start := time.Now() + ctx, task := trace.NewTask(ctx, "tso") + ts, err := s.tsoAllocator.GenerateTSO(ctx, count) + task.End() + tsoHandleDuration.Observe(time.Since(start).Seconds()) + return ts, err +} + // Bootstrap implements gRPC PDServer. func (s *GrpcServer) Bootstrap(ctx context.Context, request *pdpb.BootstrapRequest) (*pdpb.BootstrapResponse, error) { done, err := s.rateLimitCheck() diff --git a/server/server.go b/server/server.go index fe0a2a37d1c..efbc4afbc68 100644 --- a/server/server.go +++ b/server/server.go @@ -522,9 +522,15 @@ func (s *Server) startServer(ctx context.Context) error { s.tsoProtoFactory = &tsoutil.TSOProtoFactory{} s.pdProtoFactory = &tsoutil.PDProtoFactory{} tsoStorage := storage.NewStorageWithEtcdBackend(s.electionClient) - s.tsoAllocator = tso.NewAllocator(s.ctx, constant.DefaultKeyspaceGroupID, s.member, tsoStorage, s) + checkTSOPrimary := func() bool { + return s.IsKeyspaceGroupEnabled() && s.GetMicroserviceConfig().IsTSODynamicSwitchingEnabled() + } + s.tsoAllocator = tso.NewAllocator(s.ctx, constant.DefaultKeyspaceGroupID, s.member, tsoStorage, s, checkTSOPrimary) s.basicCluster = core.NewBasicCluster() s.cluster = cluster.NewRaftCluster(ctx, s.GetMember(), s.GetBasicCluster(), s.GetStorage(), syncer.NewRegionSyncer(s), s.client, s.httpClient, s.tsoAllocator) + s.cluster.SetTSOServiceReadyFunc(func(ctx context.Context) bool { + return (&GrpcServer{Server: s}).checkDefaultTSOServiceReady(ctx) + }) keyspaceIDAllocator := id.NewAllocator(&id.AllocatorParams{ Client: s.client, Label: id.KeyspaceLabel, diff --git a/server/util.go b/server/util.go index a1fce6a8689..7e46dcb1b0a 100644 --- a/server/util.go +++ b/server/util.go @@ -118,7 +118,7 @@ func combineBuilderServerHTTPService(ctx context.Context, svr *Server, serviceBu } } - // Combine the pd service to the router. the extension service will be added to the userHandlers. + // Combine the pd to the router. the extension service will be added to the userHandlers. for pathPrefix, handler := range registerMap { if strings.Contains(pathPrefix, apiutil.CorePath) || strings.Contains(pathPrefix, apiutil.ExtensionsPath) { router.PathPrefix(pathPrefix).Handler(handler) diff --git a/tests/integrations/client/client_test.go b/tests/integrations/client/client_test.go index f7e093e54b1..eea482de531 100644 --- a/tests/integrations/client/client_test.go +++ b/tests/integrations/client/client_test.go @@ -369,7 +369,9 @@ func TestTSOFollowerProxyWithTSOService(t *testing.T) { re.NoError(failpoint.Enable("github.com/tikv/pd/client/servicediscovery/fastUpdateServiceMode", `return(true)`)) ctx, cancel := context.WithCancel(context.Background()) defer cancel() - cluster, err := tests.NewTestClusterWithKeyspaceGroup(ctx, 1) + cluster, err := tests.NewTestCluster(ctx, 1, func(conf *config.Config, _ string) { + conf.Microservice.EnableMultiTimelines = true + }) re.NoError(err) defer cluster.Destroy() err = cluster.RunInitialServers() diff --git a/tests/integrations/mcs/discovery/register_test.go b/tests/integrations/mcs/discovery/register_test.go index 0aec6cfa9ee..29ae6195433 100644 --- a/tests/integrations/mcs/discovery/register_test.go +++ b/tests/integrations/mcs/discovery/register_test.go @@ -54,7 +54,7 @@ func (suite *serverRegisterTestSuite) SetupSuite() { re := suite.Require() suite.ctx, suite.cancel = context.WithCancel(context.Background()) - suite.cluster, err = tests.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + suite.cluster, err = tests.NewTestCluster(suite.ctx, 1) re.NoError(err) err = suite.cluster.RunInitialServers() diff --git a/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go b/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go index e4f42f12b19..56507212926 100644 --- a/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go +++ b/tests/integrations/mcs/keyspace/tso_keyspace_group_test.go @@ -77,7 +77,7 @@ func (suite *keyspaceGroupTestSuite) SetupTest() { re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/keyspace/skipSplitRegion", "return(true)")) ctx, cancel := context.WithCancel(context.Background()) suite.ctx = ctx - cluster, err := tests.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + cluster, err := tests.NewTestCluster(suite.ctx, 1) suite.cluster = cluster re.NoError(err) re.NoError(cluster.RunInitialServers()) diff --git a/tests/integrations/mcs/members/member_test.go b/tests/integrations/mcs/members/member_test.go index 94795aea025..397a4042409 100644 --- a/tests/integrations/mcs/members/member_test.go +++ b/tests/integrations/mcs/members/member_test.go @@ -75,7 +75,7 @@ func (suite *memberTestSuite) SetupTest() { re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/keyspace/acceleratedAllocNodes", `return(true)`)) ctx, cancel := context.WithCancel(context.Background()) suite.ctx = ctx - cluster, err := tests.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + cluster, err := tests.NewTestCluster(suite.ctx, 1) suite.cluster = cluster re.NoError(err) re.NoError(cluster.RunInitialServers()) diff --git a/tests/integrations/mcs/scheduling/api_test.go b/tests/integrations/mcs/scheduling/api_test.go index c03451168c5..b4c8f9c805a 100644 --- a/tests/integrations/mcs/scheduling/api_test.go +++ b/tests/integrations/mcs/scheduling/api_test.go @@ -433,8 +433,8 @@ func (suite *apiTestSuite) checkConfigForward(cluster *tests.TestCluster) { addr := cluster.GetLeaderServer().GetAddr() urlPrefix := fmt.Sprintf("%s/pd/api/v1/config", addr) - // Test config forward - // Expect to get same config in scheduling server and PD + // Test config forward. + // Expect to get same config in scheduling server and PD. testutil.Eventually(re, func() bool { err := testutil.ReadGetJSON(re, tests.TestDialClient, urlPrefix, &cfg) re.NoError(err) @@ -443,8 +443,8 @@ func (suite *apiTestSuite) checkConfigForward(cluster *tests.TestCluster) { return cfg["replication"].(map[string]any)["max-replicas"] == float64(opts.GetReplicationConfig().MaxReplicas) }) - // Test to change config in PD - // Expect to get new config in scheduling server and PD + // Test to change config in PD. + // Expect to get new config in scheduling server and PD. reqData, err := json.Marshal(map[string]any{ "max-replicas": 4, }) @@ -458,8 +458,8 @@ func (suite *apiTestSuite) checkConfigForward(cluster *tests.TestCluster) { opts.GetReplicationConfig().MaxReplicas == 4. }) - // Test to change config only in scheduling server - // Expect to get new config in scheduling server but not old config in PD + // Test to change config only in scheduling server. + // Expect to get new config in scheduling server but not old config in PD. scheCfg := opts.GetScheduleConfig().Clone() scheCfg.LeaderScheduleLimit = 100 opts.SetScheduleConfig(scheCfg) diff --git a/tests/integrations/mcs/scheduling/meta_test.go b/tests/integrations/mcs/scheduling/meta_test.go index f03adfef42e..659de77de69 100644 --- a/tests/integrations/mcs/scheduling/meta_test.go +++ b/tests/integrations/mcs/scheduling/meta_test.go @@ -53,7 +53,7 @@ func (suite *metaTestSuite) SetupSuite() { re.NoError(failpoint.Enable("github.com/tikv/pd/server/cluster/highFrequencyClusterJobs", `return(true)`)) var err error suite.ctx, suite.cancel = context.WithCancel(context.Background()) - suite.cluster, err = tests.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + suite.cluster, err = tests.NewTestCluster(suite.ctx, 1) re.NoError(err) err = suite.cluster.RunInitialServers() re.NoError(err) diff --git a/tests/integrations/mcs/scheduling/rule_test.go b/tests/integrations/mcs/scheduling/rule_test.go index b741f147061..f866bccee1f 100644 --- a/tests/integrations/mcs/scheduling/rule_test.go +++ b/tests/integrations/mcs/scheduling/rule_test.go @@ -54,7 +54,7 @@ func (suite *ruleTestSuite) SetupSuite() { var err error suite.ctx, suite.cancel = context.WithCancel(context.Background()) - suite.cluster, err = tests.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + suite.cluster, err = tests.NewTestCluster(suite.ctx, 1) re.NoError(err) err = suite.cluster.RunInitialServers() re.NoError(err) diff --git a/tests/integrations/mcs/scheduling/server_test.go b/tests/integrations/mcs/scheduling/server_test.go index 055044aa7d4..6f14cb4d16a 100644 --- a/tests/integrations/mcs/scheduling/server_test.go +++ b/tests/integrations/mcs/scheduling/server_test.go @@ -70,7 +70,7 @@ func (suite *serverTestSuite) SetupSuite() { re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/schedule/changeCoordinatorTicker", `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.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + suite.cluster, err = tests.NewTestCluster(suite.ctx, 1) re.NoError(err) err = suite.cluster.RunInitialServers() @@ -937,7 +937,7 @@ func (suite *multipleServerTestSuite) SetupSuite() { re := suite.Require() 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.NewTestClusterWithKeyspaceGroup(suite.ctx, 2) + suite.cluster, err = tests.NewTestCluster(suite.ctx, 2) re.NoError(err) err = suite.cluster.RunInitialServers() diff --git a/tests/integrations/tso/consistency_test.go b/tests/integrations/tso/consistency_test.go index 766106993a1..8d4335de795 100644 --- a/tests/integrations/tso/consistency_test.go +++ b/tests/integrations/tso/consistency_test.go @@ -73,11 +73,7 @@ func (suite *tsoConsistencyTestSuite) SetupSuite() { var err error suite.ctx, suite.cancel = context.WithCancel(context.Background()) - if suite.legacy { - suite.cluster, err = tests.NewTestCluster(suite.ctx, serverCount) - } else { - suite.cluster, err = tests.NewTestClusterWithKeyspaceGroup(suite.ctx, serverCount) - } + suite.cluster, err = tests.NewTestCluster(suite.ctx, serverCount) re.NoError(err) err = suite.cluster.RunInitialServers() re.NoError(err) diff --git a/tests/integrations/tso/server_test.go b/tests/integrations/tso/server_test.go index 6553591c931..58df7652633 100644 --- a/tests/integrations/tso/server_test.go +++ b/tests/integrations/tso/server_test.go @@ -72,11 +72,7 @@ func (suite *tsoServerTestSuite) SetupSuite() { var err error suite.ctx, suite.cancel = context.WithCancel(context.Background()) - if suite.legacy { - suite.cluster, err = tests.NewTestCluster(suite.ctx, serverCount) - } else { - suite.cluster, err = tests.NewTestClusterWithKeyspaceGroup(suite.ctx, serverCount) - } + suite.cluster, err = tests.NewTestCluster(suite.ctx, serverCount) re.NoError(err) err = suite.cluster.RunInitialServers() re.NoError(err) diff --git a/tests/server/apiv2/handlers/keyspace_test.go b/tests/server/apiv2/handlers/keyspace_test.go index 6e881542e6c..1ac6d31efb0 100644 --- a/tests/server/apiv2/handlers/keyspace_test.go +++ b/tests/server/apiv2/handlers/keyspace_test.go @@ -305,8 +305,10 @@ func (suite *keyspaceTestSuite) TestLoadRangeKeyspace() { func mustMakeTestKeyspaces(re *require.Assertions, server *tests.TestServer, count int) []*keyspacepb.KeyspaceMeta { testConfig := map[string]string{ - "config1": "100", - "config2": "200", + "config1": "100", + "config2": "200", + "tso_keyspace_group_id": "0", + "user_kind": "basic", } resultMeta := make([]*keyspacepb.KeyspaceMeta, count) base := uint64(time.Now().UnixNano()) diff --git a/tests/server/apiv2/handlers/tso_keyspace_group_test.go b/tests/server/apiv2/handlers/tso_keyspace_group_test.go index 9677534c0a6..69c250676e1 100644 --- a/tests/server/apiv2/handlers/tso_keyspace_group_test.go +++ b/tests/server/apiv2/handlers/tso_keyspace_group_test.go @@ -51,7 +51,7 @@ func TestKeyspaceGroupTestSuite(t *testing.T) { func (suite *keyspaceGroupTestSuite) SetupTest() { re := suite.Require() suite.ctx, suite.cancel = context.WithCancel(context.Background()) - cluster, err := tests.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + cluster, err := tests.NewTestCluster(suite.ctx, 1) suite.cluster = cluster re.NoError(err) re.NoError(cluster.RunInitialServers()) diff --git a/tests/server/server_test.go b/tests/server/server_test.go index a579531e156..b0e71f3d915 100644 --- a/tests/server/server_test.go +++ b/tests/server/server_test.go @@ -79,7 +79,7 @@ func TestUpdateAdvertiseUrls(t *testing.T) { for _, conf := range cluster.GetConfig().InitialServers { serverConf, err := conf.Generate() re.NoError(err) - s, err := tests.NewTestServer(ctx, serverConf, nil) + s, err := tests.NewTestServer(ctx, serverConf) re.NoError(err) cluster.GetServers()[conf.Name] = s } diff --git a/tools/pd-ctl/tests/config/config_test.go b/tools/pd-ctl/tests/config/config_test.go index 289ff31b3ba..3319297c76d 100644 --- a/tools/pd-ctl/tests/config/config_test.go +++ b/tools/pd-ctl/tests/config/config_test.go @@ -366,9 +366,9 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu f.Close() defer os.RemoveAll(fname) - checkScheduleConfig := func(scheduleCfg *sc.ScheduleConfig, isFromPDService bool) { + checkScheduleConfig := func(scheduleCfg *sc.ScheduleConfig, isFromPD bool) { if schedulingServer := cluster.GetSchedulingPrimaryServer(); schedulingServer != nil { - if isFromPDService { + if isFromPD { re.Equal(scheduleCfg.LeaderScheduleLimit, leaderServer.GetPersistOptions().GetLeaderScheduleLimit()) re.NotEqual(scheduleCfg.LeaderScheduleLimit, schedulingServer.GetPersistConfig().GetLeaderScheduleLimit()) } else { @@ -380,9 +380,9 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu } } - checkReplicateConfig := func(replicationCfg *sc.ReplicationConfig, isFromPDService bool) { + checkReplicateConfig := func(replicationCfg *sc.ReplicationConfig, isFromPD bool) { if schedulingServer := cluster.GetSchedulingPrimaryServer(); schedulingServer != nil { - if isFromPDService { + if isFromPD { re.Equal(replicationCfg.MaxReplicas, uint64(leaderServer.GetPersistOptions().GetMaxReplicas())) re.NotEqual(int(replicationCfg.MaxReplicas), schedulingServer.GetPersistConfig().GetMaxReplicas()) } else { @@ -394,11 +394,11 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu } } - checkRules := func(rules []*placement.Rule, isFromPDService bool) { + checkRules := func(rules []*placement.Rule, isFromPD bool) { apiRules := leaderServer.GetRaftCluster().GetRuleManager().GetAllRules() if schedulingServer := cluster.GetSchedulingPrimaryServer(); schedulingServer != nil { schedulingRules := schedulingServer.GetCluster().GetRuleManager().GetAllRules() - if isFromPDService { + if isFromPD { re.Len(apiRules, len(rules)) re.NotEqual(len(schedulingRules), len(rules)) } else { @@ -410,11 +410,11 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu } } - checkGroup := func(group placement.RuleGroup, isFromPDService bool) { + checkGroup := func(group placement.RuleGroup, isFromPD bool) { apiGroup := leaderServer.GetRaftCluster().GetRuleManager().GetRuleGroup(placement.DefaultGroupID) if schedulingServer := cluster.GetSchedulingPrimaryServer(); schedulingServer != nil { schedulingGroup := schedulingServer.GetCluster().GetRuleManager().GetRuleGroup(placement.DefaultGroupID) - if isFromPDService { + if isFromPD { re.Equal(apiGroup.Index, group.Index) re.NotEqual(schedulingGroup.Index, group.Index) } else { @@ -427,11 +427,11 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu } testConfig := func(options ...string) { - for _, isFromPDService := range []bool{true, false} { + for _, isFromPD := range []bool{true, false} { cmd := ctl.GetRootCmd() args := []string{"-u", pdAddr, "config", "show"} args = append(args, options...) - if isFromPDService { + if isFromPD { args = append(args, "--from_pd") } output, err := tests.ExecuteCommand(cmd, args...) @@ -439,16 +439,16 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu if len(options) == 0 || options[0] == "all" { cfg := config.Config{} re.NoError(json.Unmarshal(output, &cfg)) - checkReplicateConfig(&cfg.Replication, isFromPDService) - checkScheduleConfig(&cfg.Schedule, isFromPDService) + checkReplicateConfig(&cfg.Replication, isFromPD) + checkScheduleConfig(&cfg.Schedule, isFromPD) } else if options[0] == "replication" { replicationCfg := &sc.ReplicationConfig{} re.NoError(json.Unmarshal(output, replicationCfg)) - checkReplicateConfig(replicationCfg, isFromPDService) + checkReplicateConfig(replicationCfg, isFromPD) } else if options[0] == "schedule" { scheduleCfg := &sc.ScheduleConfig{} re.NoError(json.Unmarshal(output, scheduleCfg)) - checkScheduleConfig(scheduleCfg, isFromPDService) + checkScheduleConfig(scheduleCfg, isFromPD) } else { re.Fail("no implement") } @@ -456,11 +456,11 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu } testRules := func(options ...string) { - for _, isFromPDService := range []bool{true, false} { + for _, isFromPD := range []bool{true, false} { cmd := ctl.GetRootCmd() args := []string{"-u", pdAddr, "config", "placement-rules"} args = append(args, options...) - if isFromPDService { + if isFromPD { args = append(args, "--from_pd") } output, err := tests.ExecuteCommand(cmd, args...) @@ -468,27 +468,27 @@ func (suite *configTestSuite) checkConfigForwardControl(cluster *pdTests.TestClu if options[0] == "show" { var rules []*placement.Rule re.NoError(json.Unmarshal(output, &rules)) - checkRules(rules, isFromPDService) + checkRules(rules, isFromPD) } else if options[0] == "load" { var rules []*placement.Rule b, err := os.ReadFile(fname) re.NoError(err) re.NoError(json.Unmarshal(b, &rules)) - checkRules(rules, isFromPDService) + checkRules(rules, isFromPD) } else if options[0] == "rule-group" { var group placement.RuleGroup re.NoError(json.Unmarshal(output, &group), string(output)) - checkGroup(group, isFromPDService) + checkGroup(group, isFromPD) } else if options[0] == "rule-bundle" && options[1] == "get" { var bundle placement.GroupBundle re.NoError(json.Unmarshal(output, &bundle), string(output)) - checkRules(bundle.Rules, isFromPDService) + checkRules(bundle.Rules, isFromPD) } else if options[0] == "rule-bundle" && options[1] == "load" { var bundles []placement.GroupBundle b, err := os.ReadFile(fname) re.NoError(err) re.NoError(json.Unmarshal(b, &bundles), string(output)) - checkRules(bundles[0].Rules, isFromPDService) + checkRules(bundles[0].Rules, isFromPD) } else { re.Fail("no implement") } diff --git a/tools/pd-ctl/tests/keyspace/keyspace_test.go b/tools/pd-ctl/tests/keyspace/keyspace_test.go index 1435c1d31b3..87548db51e2 100644 --- a/tools/pd-ctl/tests/keyspace/keyspace_test.go +++ b/tools/pd-ctl/tests/keyspace/keyspace_test.go @@ -57,7 +57,7 @@ func TestKeyspace(t *testing.T) { for i := 1; i < 10; i++ { keyspaces = append(keyspaces, fmt.Sprintf("keyspace_%d", i)) } - tc, err := pdTests.NewTestClusterWithKeyspaceGroup(ctx, 3, func(conf *config.Config, _ string) { + tc, err := pdTests.NewTestCluster(ctx, 3, func(conf *config.Config, _ string) { conf.Keyspace.PreAlloc = keyspaces conf.Keyspace.WaitRegionSplit = false }) @@ -136,7 +136,7 @@ func (suite *keyspaceTestSuite) SetupTest() { suite.ctx, suite.cancel = context.WithCancel(context.Background()) re.NoError(failpoint.Enable("github.com/tikv/pd/server/delayStartServerLoop", `return(true)`)) re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/keyspace/skipSplitRegion", "return(true)")) - tc, err := pdTests.NewTestClusterWithKeyspaceGroup(suite.ctx, 1) + tc, err := pdTests.NewTestCluster(suite.ctx, 1) re.NoError(err) re.NoError(tc.RunInitialServers()) tc.WaitLeader() diff --git a/tools/pd-simulator/main.go b/tools/pd-simulator/main.go index 44c80d478c1..15528f2f330 100644 --- a/tools/pd-simulator/main.go +++ b/tools/pd-simulator/main.go @@ -118,7 +118,7 @@ func NewSingleServer(ctx context.Context, simConfig *sc.SimConfig) (*server.Serv log.Fatal("setup logger error", zap.Error(err)) } - s, err := server.CreateServer(ctx, simConfig.ServerConfig, nil, api.NewHandler) + s, err := server.CreateServer(ctx, simConfig.ServerConfig, api.NewHandler) if err != nil { panic("create server failed") } diff --git a/tools/pd-simulator/simulator/client.go b/tools/pd-simulator/simulator/client.go index f221bce0599..24649df5c27 100644 --- a/tools/pd-simulator/simulator/client.go +++ b/tools/pd-simulator/simulator/client.go @@ -455,7 +455,6 @@ func (rc *retryClient) Close() { } // Bootstrap bootstraps the cluster and using the given PD address firstly. -// because before bootstrapping the cluster, PDServiceDiscovery can not been started. func Bootstrap(ctx context.Context, pdAddrs string, store *metapb.Store, region *metapb.Region) ( leaderURL string, pdCli pdpb.PDClient, err error) { urls := strings.Split(pdAddrs, ",")