mcs: refactor tso handler

pkg/mcs/resourcemanager/server/apis/v1/api.go

@@ -124,7 +124,7 @@ func (s *Service) postResourceGroup(c *gin.Context) {
 		c.String(http.StatusInternalServerError, err.Error())
 		return
 	}
-	c.JSON(http.StatusOK, "Success!")
+	c.String(http.StatusOK, "Success!")
 }
 
 // putResourceGroup
@@ -146,7 +146,7 @@ func (s *Service) putResourceGroup(c *gin.Context) {
 		c.String(http.StatusInternalServerError, err.Error())
 		return
 	}
-	c.JSON(http.StatusOK, "Success!")
+	c.String(http.StatusOK, "Success!")
 }
 
 // getResourceGroup
@@ -162,7 +162,7 @@ func (s *Service) getResourceGroup(c *gin.Context) {
 	if group == nil {
 		c.String(http.StatusNotFound, errors.New("resource group not found").Error())
 	}
-	c.JSON(http.StatusOK, group)
+	c.IndentedJSON(http.StatusOK, group)
 }
 
 // getResourceGroupList
@@ -174,7 +174,7 @@ func (s *Service) getResourceGroup(c *gin.Context) {
 //	@Router		/config/groups [GET]
 func (s *Service) getResourceGroupList(c *gin.Context) {
 	groups := s.manager.GetResourceGroupList()
-	c.JSON(http.StatusOK, groups)
+	c.IndentedJSON(http.StatusOK, groups)
 }
 
 // deleteResourceGroup
@@ -189,5 +189,5 @@ func (s *Service) deleteResourceGroup(c *gin.Context) {
 	if err := s.manager.DeleteResourceGroup(c.Param("name")); err != nil {
 		c.String(http.StatusNotFound, err.Error())
 	}
-	c.JSON(http.StatusOK, "Success!")
+	c.String(http.StatusOK, "Success!")
 }

pkg/mcs/scheduling/server/apis/v1/api.go

@@ -142,7 +142,7 @@ func getOperatorByID(c *gin.Context) {
 		return
 	}
 
-	c.JSON(http.StatusOK, opController.GetOperatorStatus(regionID))
+	c.IndentedJSON(http.StatusOK, opController.GetOperatorStatus(regionID))
 }
 
 // @Tags     operators
@@ -184,7 +184,7 @@ func getOperators(c *gin.Context) {
 		}
 	}
 
-	c.JSON(http.StatusOK, results)
+	c.IndentedJSON(http.StatusOK, results)
 }
 
 // @Tags     checkers
@@ -206,7 +206,7 @@ func getCheckerByName(c *gin.Context) {
 	output := map[string]bool{
 		"paused": isPaused,
 	}
-	c.JSON(http.StatusOK, output)
+	c.IndentedJSON(http.StatusOK, output)
 }
 
 type schedulerPausedPeriod struct {
@@ -266,9 +266,9 @@ func getSchedulers(c *gin.Context) {
 			}
 		}
 		if needTS {
-			c.JSON(http.StatusOK, pausedPeriods)
+			c.IndentedJSON(http.StatusOK, pausedPeriods)
 		} else {
-			c.JSON(http.StatusOK, pausedSchedulers)
+			c.IndentedJSON(http.StatusOK, pausedSchedulers)
 		}
 		return
 	case "disabled":
@@ -284,8 +284,8 @@ func getSchedulers(c *gin.Context) {
 				disabledSchedulers = append(disabledSchedulers, scheduler)
 			}
 		}
-		c.JSON(http.StatusOK, disabledSchedulers)
+		c.IndentedJSON(http.StatusOK, disabledSchedulers)
 	default:
-		c.JSON(http.StatusOK, schedulers)
+		c.IndentedJSON(http.StatusOK, schedulers)
 	}
 }

pkg/mcs/tso/server/apis/v1/api.go

@@ -16,6 +16,7 @@ package apis
 
 import (
 	"net/http"
+	"strconv"
 	"sync"
 
 	"github.com/gin-contrib/cors"
@@ -25,10 +26,10 @@ import (
 	"github.com/joho/godotenv"
 	"github.com/pingcap/kvproto/pkg/tsopb"
 	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/errs"
 	tsoserver "github.com/tikv/pd/pkg/mcs/tso/server"
 	"github.com/tikv/pd/pkg/mcs/utils"
 	"github.com/tikv/pd/pkg/storage/endpoint"
-	"github.com/tikv/pd/pkg/tso"
 	"github.com/tikv/pd/pkg/utils/apiutil"
 	"github.com/tikv/pd/pkg/utils/apiutil/multiservicesapi"
 	"github.com/unrolled/render"
@@ -105,8 +106,7 @@ func NewService(srv *tsoserver.Service) *Service {
 // RegisterAdminRouter registers the router of the TSO admin handler.
 func (s *Service) RegisterAdminRouter() {
 	router := s.root.Group("admin")
-	tsoAdminHandler := tso.NewAdminHandler(s.srv.GetHandler(), s.rd)
-	router.POST("/reset-ts", gin.WrapF(tsoAdminHandler.ResetTS))
+	router.POST("/reset-ts", ResetTS)
 }
 
 // RegisterKeyspaceGroupRouter registers the router of the TSO keyspace group handler.
@@ -115,6 +115,66 @@ func (s *Service) RegisterKeyspaceGroupRouter() {
 	router.GET("/members", GetKeyspaceGroupMembers)
 }
 
+// ResetTSParams is the input json body params of ResetTS
+type ResetTSParams struct {
+	TSO           string `json:"tso"`
+	ForceUseLarge bool   `json:"force-use-larger"`
+}
+
+// ResetTS is the http.HandlerFunc of ResetTS
+// FIXME: details of input json body params
+// @Tags     admin
+// @Summary  Reset the ts.
+// @Accept   json
+// @Param    body  body  object  true  "json params"
+// @Produce  json
+// @Success  200  {string}  string  "Reset ts successfully."
+// @Failure  400  {string}  string  "The input is invalid."
+// @Failure  403  {string}  string  "Reset ts is forbidden."
+// @Failure  500  {string}  string  "TSO server failed to proceed the request."
+// @Router   /admin/reset-ts [post]
+// if force-use-larger=true:
+//
+//	reset ts to max(current ts, input ts).
+//
+// else:
+//
+//	reset ts to input ts if it > current ts and < upper bound, error if not in that range
+//
+// during EBS based restore, we call this to make sure ts of pd >= resolved_ts in backup.
+func ResetTS(c *gin.Context) {
+	svr := c.MustGet(multiservicesapi.ServiceContextKey).(*tsoserver.Service)
+	var param ResetTSParams
+	if err := c.ShouldBindJSON(&param); err != nil {
+		c.String(http.StatusBadRequest, err.Error())
+		return
+	}
+	if len(param.TSO) == 0 {
+		c.String(http.StatusBadRequest, "invalid tso value")
+		return
+	}
+	ts, err := strconv.ParseUint(param.TSO, 10, 64)
+	if err != nil {
+		c.String(http.StatusBadRequest, "invalid tso value")
+		return
+	}
+
+	var ignoreSmaller, skipUpperBoundCheck bool
+	if param.ForceUseLarge {
+		ignoreSmaller, skipUpperBoundCheck = true, true
+	}
+
+	if err = svr.ResetTS(ts, ignoreSmaller, skipUpperBoundCheck, 0); err != nil {
+		if err == errs.ErrServerNotStarted {
+			c.String(http.StatusInternalServerError, err.Error())
+		} else {
+			c.String(http.StatusForbidden, err.Error())
+		}
+		return
+	}
+	c.String(http.StatusOK, "Reset ts successfully.")
+}
+
 // KeyspaceGroupMember contains the keyspace group and its member information.
 type KeyspaceGroupMember struct {
 	Group     *endpoint.KeyspaceGroup

pkg/mcs/tso/server/server.go

@@ -74,8 +74,6 @@ type Server struct {
 	serverLoopCancel func()
 	serverLoopWg     sync.WaitGroup
 
-	handler *Handler
-
 	cfg       *Config
 	clusterID uint64
 	listenURL *url.URL
@@ -121,11 +119,6 @@ func (s *Server) Context() context.Context {
 	return s.ctx
 }
 
-// GetHandler returns the handler.
-func (s *Server) GetHandler() *Handler {
-	return s.handler
-}
-
 // GetBasicServer returns the basic server.
 func (s *Server) GetBasicServer() bs.Server {
 	return s
@@ -411,6 +404,28 @@ func (s *Server) SetExternalTS(externalTS uint64) error {
 	return nil
 }
 
+// ResetTS resets the TSO with the specified one.
+func (s *Server) ResetTS(ts uint64, ignoreSmaller, skipUpperBoundCheck bool, keyspaceGroupID uint32) error {
+	log.Info("reset-ts",
+		zap.Uint64("new-ts", ts),
+		zap.Bool("ignore-smaller", ignoreSmaller),
+		zap.Bool("skip-upper-bound-check", skipUpperBoundCheck),
+		zap.Uint32("keyspace-group-id", keyspaceGroupID))
+	tsoAllocatorManager, err := s.GetTSOAllocatorManager(keyspaceGroupID)
+	if err != nil {
+		log.Error("failed to get allocator manager", errs.ZapError(err))
+		return err
+	}
+	tsoAllocator, err := tsoAllocatorManager.GetAllocator(tso.GlobalDCLocation)
+	if err != nil {
+		return err
+	}
+	if tsoAllocator == nil {
+		return errs.ErrServerNotStarted
+	}
+	return tsoAllocator.SetTSO(ts, ignoreSmaller, skipUpperBoundCheck)
+}
+
 // GetConfig gets the config.
 func (s *Server) GetConfig() *Config {
 	return s.cfg
@@ -503,7 +518,6 @@ func CreateServer(ctx context.Context, cfg *Config) *Server {
 		cfg:               cfg,
 		ctx:               ctx,
 	}
-	svr.handler = newHandler(svr)
 	return svr
 }
 

tests/integrations/mcs/resourcemanager/resource_manager_test.go

@@ -941,7 +941,7 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() {
 		re.NoError(err)
 		re.Contains(string(respString), tcase.name)
 		if tcase.modifySuccess {
-			re.Equal(string(respString), tcase.expectMarshal)
+			re.JSONEq(string(respString), tcase.expectMarshal)
 		}
 
 		// Last one, Check list and delete all resource groups

tests/integrations/mcs/resourcemanager/server_test.go

@@ -68,7 +68,7 @@ func TestResourceManagerServer(t *testing.T) {
 		re.Equal(http.StatusOK, resp.StatusCode)
 		respString, err := io.ReadAll(resp.Body)
 		re.NoError(err)
-		re.Equal(`[{"name":"default","mode":1,"r_u_settings":{"r_u":{"settings":{"fill_rate":2147483647,"burst_limit":-1},"state":{"initialized":false}}},"priority":8}]`, string(respString))
+		re.JSONEq(`[{"name":"default","mode":1,"r_u_settings":{"r_u":{"settings":{"fill_rate":2147483647,"burst_limit":-1},"state":{"initialized":false}}},"priority":8}]`, string(respString))
 	}
 	{
 		group := &rmpb.ResourceGroup{
@@ -89,7 +89,7 @@ func TestResourceManagerServer(t *testing.T) {
 		re.Equal(http.StatusOK, resp.StatusCode)
 		respString, err := io.ReadAll(resp.Body)
 		re.NoError(err)
-		re.Equal("{\"name\":\"pingcap\",\"mode\":1,\"r_u_settings\":{\"r_u\":{\"state\":{\"initialized\":false}}},\"priority\":0}", string(respString))
+		re.JSONEq("{\"name\":\"pingcap\",\"mode\":1,\"r_u_settings\":{\"r_u\":{\"state\":{\"initialized\":false}}},\"priority\":0}", string(respString))
 	}
 
 	// Test metrics handler

tests/integrations/mcs/tso/keyspace_group_manager_test.go

@@ -262,7 +262,7 @@ func (suite *tsoKeyspaceGroupManagerTestSuite) TestTSOKeyspaceGroupSplit() {
 	})
 	ts.Physical += time.Hour.Milliseconds()
 	// Set the TSO of the keyspace group 1 to a large value.
-	err = suite.tsoCluster.GetPrimaryServer(222, 1).GetHandler().ResetTS(tsoutil.GenerateTS(&ts), false, true, 1)
+	err = suite.tsoCluster.GetPrimaryServer(222, 1).ResetTS(tsoutil.GenerateTS(&ts), false, true, 1)
 	re.NoError(err)
 	// Split the keyspace group 1 to 2.
 	handlersutil.MustSplitKeyspaceGroup(re, suite.pdLeaderServer, 1, &handlers.SplitKeyspaceGroupByIDParams{
@@ -588,7 +588,7 @@ func (suite *tsoKeyspaceGroupManagerTestSuite) TestTSOKeyspaceGroupMerge() {
 	})
 	ts.Physical += time.Hour.Milliseconds()
 	// Set the TSO of the keyspace group 1 to a large value.
-	err = suite.tsoCluster.GetPrimaryServer(222, 1).GetHandler().ResetTS(tsoutil.GenerateTS(&ts), false, true, 1)
+	err = suite.tsoCluster.GetPrimaryServer(222, 1).ResetTS(tsoutil.GenerateTS(&ts), false, true, 1)
 	re.NoError(err)
 	// Merge the keyspace group 1 and 2 to the default keyspace group.
 	handlersutil.MustMergeKeyspaceGroup(re, suite.pdLeaderServer, mcsutils.DefaultKeyspaceGroupID, &handlers.MergeKeyspaceGroupsParams{

tests/integrations/tso/server_test.go

@@ -108,7 +108,7 @@ func (suite *tsoServerTestSuite) resetTS(ts uint64, ignoreSmaller, skipUpperBoun
 	if suite.legacy {
 		err = suite.pdLeaderServer.GetServer().GetHandler().ResetTS(ts, ignoreSmaller, skipUpperBoundCheck, 0)
 	} else {
-		err = suite.tsoServer.GetHandler().ResetTS(ts, ignoreSmaller, skipUpperBoundCheck, 0)
+		err = suite.tsoServer.ResetTS(ts, ignoreSmaller, skipUpperBoundCheck, 0)
 	}
 	// Only this error is acceptable.
 	if err != nil {