api(ticdc): add /api/v2/tso: get tso from pd

cdc/api/v1/api.go

@@ -46,32 +46,32 @@ const (
 	forWardFromCapture = "TiCDC-ForwardFromCapture"
 )
 
-// openAPI provides capture APIs.
-type openAPI struct {
+// OpenAPI provides capture APIs.
+type OpenAPI struct {
 	capture *capture.Capture
 	// use for unit test only
 	testStatusProvider owner.StatusProvider
 }
 
-// NewOpenAPI creates a new openAPI.
-func NewOpenAPI(c *capture.Capture) openAPI {
-	return openAPI{capture: c}
+// NewOpenAPI creates a new OpenAPI.
+func NewOpenAPI(c *capture.Capture) OpenAPI {
+	return OpenAPI{capture: c}
 }
 
-// NewOpenAPI4Test return a openAPI for test
-func NewOpenAPI4Test(c *capture.Capture, p owner.StatusProvider) openAPI {
-	return openAPI{capture: c, testStatusProvider: p}
+// NewOpenAPI4Test return a OpenAPI for test
+func NewOpenAPI4Test(c *capture.Capture, p owner.StatusProvider) OpenAPI {
+	return OpenAPI{capture: c, testStatusProvider: p}
 }
 
-func (h *openAPI) statusProvider() owner.StatusProvider {
+func (h *OpenAPI) statusProvider() owner.StatusProvider {
 	if h.testStatusProvider != nil {
 		return h.testStatusProvider
 	}
 	return h.capture.StatusProvider()
 }
 
 // RegisterOpenAPIRoutes registers routes for OpenAPI
-func RegisterOpenAPIRoutes(router *gin.Engine, api openAPI) {
+func RegisterOpenAPIRoutes(router *gin.Engine, api OpenAPI) {
 	v1 := router.Group("/api/v1")
 
 	v1.Use(middleware.LogMiddleware())
@@ -118,7 +118,7 @@ func RegisterOpenAPIRoutes(router *gin.Engine, api openAPI) {
 // @Success 200 {array} model.ChangefeedCommonInfo
 // @Failure 500 {object} model.HTTPError
 // @Router /api/v1/changefeeds [get]
-func (h *openAPI) ListChangefeed(c *gin.Context) {
+func (h *OpenAPI) ListChangefeed(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -184,7 +184,7 @@ func (h *openAPI) ListChangefeed(c *gin.Context) {
 // @Success 200 {object} model.ChangefeedDetail
 // @Failure 500,400 {object} model.HTTPError
 // @Router /api/v1/changefeeds/{changefeed_id} [get]
-func (h *openAPI) GetChangefeed(c *gin.Context) {
+func (h *OpenAPI) GetChangefeed(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -253,7 +253,7 @@ func (h *openAPI) GetChangefeed(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/changefeeds [post]
-func (h *openAPI) CreateChangefeed(c *gin.Context) {
+func (h *OpenAPI) CreateChangefeed(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -302,7 +302,7 @@ func (h *openAPI) CreateChangefeed(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router /api/v1/changefeeds/{changefeed_id}/pause [post]
-func (h *openAPI) PauseChangefeed(c *gin.Context) {
+func (h *OpenAPI) PauseChangefeed(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -345,7 +345,7 @@ func (h *openAPI) PauseChangefeed(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/changefeeds/{changefeed_id}/resume [post]
-func (h *openAPI) ResumeChangefeed(c *gin.Context) {
+func (h *OpenAPI) ResumeChangefeed(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -393,7 +393,7 @@ func (h *openAPI) ResumeChangefeed(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router /api/v1/changefeeds/{changefeed_id} [put]
-func (h *openAPI) UpdateChangefeed(c *gin.Context) {
+func (h *OpenAPI) UpdateChangefeed(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -450,7 +450,7 @@ func (h *openAPI) UpdateChangefeed(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/changefeeds/{changefeed_id} [delete]
-func (h *openAPI) RemoveChangefeed(c *gin.Context) {
+func (h *OpenAPI) RemoveChangefeed(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -492,7 +492,7 @@ func (h *openAPI) RemoveChangefeed(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router /api/v1/changefeeds/{changefeed_id}/tables/rebalance_table [post]
-func (h *openAPI) RebalanceTables(c *gin.Context) {
+func (h *OpenAPI) RebalanceTables(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -532,7 +532,7 @@ func (h *openAPI) RebalanceTables(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router /api/v1/changefeeds/{changefeed_id}/tables/move_table [post]
-func (h *openAPI) MoveTable(c *gin.Context) {
+func (h *OpenAPI) MoveTable(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -585,7 +585,7 @@ func (h *openAPI) MoveTable(c *gin.Context) {
 // @Success 202
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/owner/resign [post]
-func (h *openAPI) ResignOwner(c *gin.Context) {
+func (h *OpenAPI) ResignOwner(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -608,7 +608,7 @@ func (h *openAPI) ResignOwner(c *gin.Context) {
 // @Success 200 {object} model.ProcessorDetail
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/processors/{changefeed_id}/{capture_id} [get]
-func (h *openAPI) GetProcessor(c *gin.Context) {
+func (h *OpenAPI) GetProcessor(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -688,7 +688,7 @@ func (h *openAPI) GetProcessor(c *gin.Context) {
 // @Success 200 {array} model.ProcessorCommonInfo
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/processors [get]
-func (h *openAPI) ListProcessor(c *gin.Context) {
+func (h *OpenAPI) ListProcessor(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -720,7 +720,7 @@ func (h *openAPI) ListProcessor(c *gin.Context) {
 // @Success 200 {array} model.Capture
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/captures [get]
-func (h *openAPI) ListCapture(c *gin.Context) {
+func (h *OpenAPI) ListCapture(c *gin.Context) {
 	if !h.capture.IsOwner() {
 		h.forwardToOwner(c)
 		return
@@ -754,7 +754,7 @@ func (h *openAPI) ListCapture(c *gin.Context) {
 // @Success 200 {object} model.ServerStatus
 // @Failure 500,400 {object} model.HTTPError
 // @Router	/api/v1/status [get]
-func (h *openAPI) ServerStatus(c *gin.Context) {
+func (h *OpenAPI) ServerStatus(c *gin.Context) {
 	status := model.ServerStatus{
 		Version: version.ReleaseVersion,
 		GitHash: version.GitHash,
@@ -774,7 +774,7 @@ func (h *openAPI) ServerStatus(c *gin.Context) {
 // @Success 200
 // @Failure 500 {object} model.HTTPError
 // @Router	/api/v1/health [get]
-func (h *openAPI) Health(c *gin.Context) {
+func (h *OpenAPI) Health(c *gin.Context) {
 	ctx := c.Request.Context()
 
 	if _, err := h.capture.GetOwnerCaptureInfo(ctx); err != nil {
@@ -815,7 +815,7 @@ func SetLogLevel(c *gin.Context) {
 }
 
 // forwardToOwner forward an request to owner
-func (h *openAPI) forwardToOwner(c *gin.Context) {
+func (h *OpenAPI) forwardToOwner(c *gin.Context) {
 	ctx := c.Request.Context()
 	// every request can only forward to owner one time
 	if len(c.GetHeader(forWardFromCapture)) != 0 {

cdc/api/v2/api.go

@@ -24,15 +24,18 @@ type OpenAPIV2 struct {
 	capture *capture.Capture
 }
 
-// NewOpenAPI creates a new openAPIs.
-func NewOpenAPI(c *capture.Capture) *OpenAPIV2 {
-	return &OpenAPIV2{capture: c}
+// NewOpenAPIV2 creates a new OpenAPIV2.
+func NewOpenAPIV2(c *capture.Capture) OpenAPIV2 {
+	return OpenAPIV2{capture: c}
 }
 
-// RegisterOpenAPIRoutes registers routes for OpenAPI
-func RegisterOpenAPIRoutes(router *gin.Engine, api *OpenAPIV2) {
+// RegisterOpenAPIV2Routes registers routes for OpenAPI
+func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {
 	v2 := router.Group("/api/v2")
 
 	v2.Use(middleware.LogMiddleware())
 	v2.Use(middleware.ErrorHandleMiddleware())
+
+	// common APIs
+	v2.GET("/tso", api.GetTso)
 }

cdc/api/v2/api_test.go

@@ -0,0 +1,44 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v2
+
+import (
+	"context"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/pingcap/tiflow/cdc/capture"
+	"github.com/tikv/client-go/v2/oracle"
+	pd "github.com/tikv/pd/client"
+)
+
+// MockPDClient mocks pd.Client to facilitate unit testing.
+type MockPDClient struct {
+	pd.Client
+}
+
+func (m *MockPDClient) GetTS(ctx context.Context) (int64, int64, error) {
+	return oracle.GetPhysical(time.Now()), 0, nil
+}
+
+type testCase struct {
+	url    string
+	method string
+}
+
+func newRouter(c *capture.Capture) *gin.Engine {
+	router := gin.New()
+	RegisterOpenAPIV2Routes(router, NewOpenAPIV2(c))
+	return router
+}

cdc/api/v2/tso.go

@@ -0,0 +1,46 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v2
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/upstream"
+)
+
+// GetTso request and returns a TSO from PD
+func (h *OpenAPIV2) GetTso(c *gin.Context) {
+	ctx := c.Request.Context()
+	if h.capture.UpstreamManager == nil {
+		c.Status(http.StatusServiceUnavailable)
+		return
+	}
+	pdClient := h.capture.UpstreamManager.Get(upstream.DefaultUpstreamID).PDClient
+	if pdClient == nil {
+		c.Status(http.StatusServiceUnavailable)
+		return
+	}
+
+	timestamp, logicalTime, err := pdClient.GetTS(ctx)
+	if err != nil {
+		_ = c.Error(err)
+		c.IndentedJSON(http.StatusInternalServerError, model.NewHTTPError(err))
+		return
+	}
+
+	resp := Tso{timestamp, logicalTime}
+	c.IndentedJSON(http.StatusOK, resp)
+}

cdc/api/v2/tso_test.go

@@ -0,0 +1,46 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v2
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/pingcap/tiflow/cdc/capture"
+	mock_owner "github.com/pingcap/tiflow/cdc/owner/mock"
+	"github.com/pingcap/tiflow/pkg/upstream"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetTso(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	mockOwner := mock_owner.NewMockOwner(ctrl)
+	mockPDClient := &MockPDClient{}
+	mockManager := upstream.NewManager4Test(mockPDClient)
+	cp := capture.NewCaptureWithManager4Test(mockOwner, mockManager)
+
+	router := newRouter(cp)
+	w := httptest.NewRecorder()
+
+	tc := testCase{url: "/api/v2/tso", method: "GET"}
+	req, err := http.NewRequestWithContext(context.Background(), tc.method, tc.url, nil)
+	require.Nil(t, err)
+	router.ServeHTTP(w, req)
+	require.Equal(t, 200, w.Code)
+}

cdc/capture/capture.go

@@ -106,6 +106,15 @@ func NewCapture4Test(o owner.Owner) *Capture {
 	return res
 }
 
+// NewCaptureWithManager4Test returns a new Capture instance for test.
+func NewCaptureWithManager4Test(o owner.Owner, m *upstream.Manager) *Capture {
+	res := &Capture{
+		UpstreamManager: m,
+	}
+	res.owner = o
+	return res
+}
+
 func (c *Capture) reset(ctx context.Context) error {
 	conf := config.GetGlobalServerConfig()
 	sess, err := concurrency.NewSession(c.EtcdClient.Client.Unwrap(),

cdc/http.go

@@ -46,7 +46,7 @@ func RegisterRoutes(
 	// Open API V1
 	v1.RegisterOpenAPIRoutes(router, v1.NewOpenAPI(capture))
 	// Open API V2
-	v2.RegisterOpenAPIRoutes(router, v2.NewOpenAPI(capture))
+	v2.RegisterOpenAPIV2Routes(router, v2.NewOpenAPIV2(capture))
 
 	// Owner API
 	owner.RegisterOwnerAPIRoutes(router, capture)

tests/integration_tests/http_api/run.sh

@@ -109,6 +109,7 @@ function run() {
 		"set_log_level"
 		"remove_changefeed"
 		"resign_owner"
+		"get_tso"
 	)
 
 	for case in ${sequential_cases[@]}; do