api(ticdc): refactor api packages

Makefile

@@ -255,7 +255,7 @@ data-flow-diagram: docs/data-flow.dot
 	dot -Tsvg docs/data-flow.dot > docs/data-flow.svg
 
 swagger-spec: tools/bin/swag
-	tools/bin/swag init --parseVendor -generalInfo cdc/api/open.go --output docs/swagger
+	tools/bin/swag init --parseVendor -generalInfo cdc/api/v1/api.go --output docs/swagger
 
 generate_mock: tools/bin/mockgen
 	tools/bin/mockgen -source cdc/owner/owner.go -destination cdc/owner/mock/owner_mock.go

cdc/api/middleware.go → cdc/api/middleware/middleware.go

@@ -11,19 +11,21 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package api
+package middleware
 
 import (
 	"net/http"
 	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tiflow/cdc/api"
 	"github.com/pingcap/tiflow/cdc/model"
 	"go.uber.org/zap"
 )
 
-func logMiddleware() gin.HandlerFunc {
+// LogMiddleware logs the api requests
+func LogMiddleware() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		start := time.Now()
 		path := c.Request.URL.Path
@@ -51,7 +53,8 @@ func logMiddleware() gin.HandlerFunc {
 	}
 }
 
-func errorHandleMiddleware() gin.HandlerFunc {
+// ErrorHandleMiddleware puts the error into response
+func ErrorHandleMiddleware() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		c.Next()
 		// because we will return immediately after an error occurs in http_handler
@@ -60,7 +63,7 @@ func errorHandleMiddleware() gin.HandlerFunc {
 		if lastError != nil {
 			err := lastError.Err
 			// put the error into response
-			if IsHTTPBadRequestError(err) {
+			if api.IsHTTPBadRequestError(err) {
 				c.IndentedJSON(http.StatusBadRequest, model.NewHTTPError(err))
 			} else {
 				c.IndentedJSON(http.StatusInternalServerError, model.NewHTTPError(err))

cdc/api/owner/main_test.go

@@ -0,0 +1,24 @@
+// 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 owner
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/pkg/leakutil"
+)
+
+func TestMain(m *testing.M) {
+	leakutil.SetUpLeakTest(m)
+}

cdc/api/owner.go → cdc/api/owner/owner.go

@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package api
+package owner
 
 import (
 	"context"
@@ -24,6 +24,8 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tiflow/cdc/api"
+	"github.com/pingcap/tiflow/cdc/api/middleware"
 	"github.com/tikv/client-go/v2/oracle"
 	"go.etcd.io/etcd/client/v3/concurrency"
 	"go.uber.org/zap"
@@ -84,8 +86,8 @@ func RegisterOwnerAPIRoutes(router *gin.Engine, capture *capture.Capture) {
 	ownerAPI := ownerAPI{capture: capture}
 	owner := router.Group("/capture/owner")
 
-	owner.Use(errorHandleMiddleware())
-	owner.Use(logMiddleware())
+	owner.Use(middleware.ErrorHandleMiddleware())
+	owner.Use(middleware.LogMiddleware())
 
 	owner.POST("/resign", gin.WrapF(ownerAPI.handleResignOwner))
 	owner.POST("/admin", gin.WrapF(ownerAPI.handleChangefeedAdmin))
@@ -97,13 +99,13 @@ func RegisterOwnerAPIRoutes(router *gin.Engine, capture *capture.Capture) {
 func handleOwnerResp(w http.ResponseWriter, err error) {
 	if err != nil {
 		if errors.Cause(err) == concurrency.ErrElectionNotLeader {
-			writeError(w, http.StatusBadRequest, err)
+			api.WriteError(w, http.StatusBadRequest, err)
 			return
 		}
-		writeError(w, http.StatusInternalServerError, err)
+		api.WriteError(w, http.StatusInternalServerError, err)
 		return
 	}
-	writeData(w, commonResp{Status: true})
+	api.WriteData(w, commonResp{Status: true})
 }
 
 func (h *ownerAPI) handleResignOwner(w http.ResponseWriter, req *http.Request) {
@@ -128,20 +130,21 @@ func (h *ownerAPI) handleChangefeedAdmin(w http.ResponseWriter, req *http.Reques
 
 	err := req.ParseForm()
 	if err != nil {
-		writeError(w, http.StatusInternalServerError, err)
+		api.WriteError(w, http.StatusInternalServerError, err)
 		return
 	}
 	typeStr := req.Form.Get(OpVarAdminJob)
 	typ, err := strconv.ParseInt(typeStr, 10, 64)
 	if err != nil {
-		writeError(w, http.StatusBadRequest, cerror.ErrAPIInvalidParam.GenWithStack("invalid admin job type: %s", typeStr))
+		api.WriteError(w, http.StatusBadRequest,
+			cerror.ErrAPIInvalidParam.GenWithStack("invalid admin job type: %s", typeStr))
 		return
 	}
 	opts := &model.AdminJobOption{}
 	if forceRemoveStr := req.Form.Get(OpForceRemoveChangefeed); forceRemoveStr != "" {
 		forceRemoveOpt, err := strconv.ParseBool(forceRemoveStr)
 		if err != nil {
-			writeError(w, http.StatusBadRequest,
+			api.WriteError(w, http.StatusBadRequest,
 				cerror.ErrAPIInvalidParam.GenWithStack("invalid force remove option: %s", forceRemoveStr))
 			return
 		}
@@ -153,7 +156,7 @@ func (h *ownerAPI) handleChangefeedAdmin(w http.ResponseWriter, req *http.Reques
 		Opts: opts,
 	}
 
-	err = handleOwnerJob(req.Context(), h.capture, job)
+	err = api.HandleOwnerJob(req.Context(), h.capture, job)
 	handleOwnerResp(w, err)
 }
 
@@ -166,17 +169,17 @@ func (h *ownerAPI) handleRebalanceTrigger(w http.ResponseWriter, req *http.Reque
 
 	err := req.ParseForm()
 	if err != nil {
-		writeError(w, http.StatusInternalServerError, err)
+		api.WriteError(w, http.StatusInternalServerError, err)
 		return
 	}
 	changefeedID := model.DefaultChangeFeedID(req.Form.Get(OpVarChangefeedID))
 	if err := model.ValidateChangefeedID(changefeedID.ID); err != nil {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed id: %s", changefeedID.ID))
 		return
 	}
 
-	err = handleOwnerRebalance(req.Context(), h.capture, changefeedID)
+	err = api.HandleOwnerBalance(req.Context(), h.capture, changefeedID)
 	handleOwnerResp(w, err)
 }
 
@@ -189,31 +192,31 @@ func (h *ownerAPI) handleMoveTable(w http.ResponseWriter, req *http.Request) {
 
 	err := req.ParseForm()
 	if err != nil {
-		writeError(w, http.StatusInternalServerError,
+		api.WriteError(w, http.StatusInternalServerError,
 			cerror.WrapError(cerror.ErrInternalServerError, err))
 		return
 	}
 	changefeedID := model.DefaultChangeFeedID(req.Form.Get(OpVarChangefeedID))
 	if err := model.ValidateChangefeedID(changefeedID.ID); err != nil {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed id: %s", changefeedID.ID))
 		return
 	}
 	to := req.Form.Get(OpVarTargetCaptureID)
 	if err := model.ValidateChangefeedID(to); err != nil {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid target capture id: %s", to))
 		return
 	}
 	tableIDStr := req.Form.Get(OpVarTableID)
 	tableID, err := strconv.ParseInt(tableIDStr, 10, 64)
 	if err != nil {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid tableID: %s", tableIDStr))
 		return
 	}
 
-	err = handleOwnerScheduleTable(
+	err = api.HandleOwnerScheduleTable(
 		req.Context(), h.capture, changefeedID, to, tableID)
 	handleOwnerResp(w, err)
 }
@@ -227,26 +230,26 @@ func (h *ownerAPI) handleChangefeedQuery(w http.ResponseWriter, req *http.Reques
 
 	err := req.ParseForm()
 	if err != nil {
-		writeError(w, http.StatusInternalServerError, err)
+		api.WriteError(w, http.StatusInternalServerError, err)
 		return
 	}
 	changefeedID := model.DefaultChangeFeedID(req.Form.Get(OpVarChangefeedID))
 	if err := model.ValidateChangefeedID(changefeedID.ID); err != nil {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed id: %s", changefeedID.ID))
 		return
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	cfInfo, err := h.capture.EtcdClient.GetChangeFeedInfo(ctx, changefeedID)
 	if err != nil && cerror.ErrChangeFeedNotExists.NotEqual(err) {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed id: %s", changefeedID))
 		return
 	}
 	cfStatus, _, err := h.capture.EtcdClient.GetChangeFeedStatus(ctx, changefeedID)
 	if err != nil && cerror.ErrChangeFeedNotExists.NotEqual(err) {
-		writeError(w, http.StatusBadRequest, err)
+		api.WriteError(w, http.StatusBadRequest, err)
 		return
 	}
 
@@ -260,7 +263,7 @@ func (h *ownerAPI) handleChangefeedQuery(w http.ResponseWriter, req *http.Reques
 		tm := oracle.GetTimeFromTS(cfStatus.CheckpointTs)
 		resp.Checkpoint = tm.Format("2006-01-02 15:04:05.000")
 	}
-	writeData(w, resp)
+	api.WriteData(w, resp)
 }
 
 // HandleAdminLogLevel handles requests to set the log level.
@@ -269,23 +272,23 @@ func HandleAdminLogLevel(w http.ResponseWriter, r *http.Request) {
 	data, err := io.ReadAll(r.Body)
 	r.Body.Close()
 	if err != nil {
-		writeError(w, http.StatusInternalServerError, err)
+		api.WriteError(w, http.StatusInternalServerError, err)
 		return
 	}
 	err = json.Unmarshal(data, &level)
 	if err != nil {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("invalid log level: %s", err))
 		return
 	}
 
 	err = logutil.SetLogLevel(level)
 	if err != nil {
-		writeError(w, http.StatusBadRequest,
+		api.WriteError(w, http.StatusBadRequest,
 			cerror.ErrAPIInvalidParam.GenWithStack("fail to change log level: %s", err))
 		return
 	}
 	log.Warn("log level changed", zap.String("level", level))
 
-	writeData(w, struct{}{})
+	api.WriteData(w, struct{}{})
 }

cdc/api/owner_test.go → cdc/api/owner/owner_test.go

@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package api
+package owner
 
 import (
 	"context"

cdc/api/status.go → cdc/api/status/status.go

@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package api
+package status
 
 import (
 	"context"
@@ -21,6 +21,7 @@ import (
 	"os"
 
 	"github.com/gin-gonic/gin"
+	"github.com/pingcap/tiflow/cdc/api"
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/version"
@@ -77,5 +78,5 @@ func (h *statusAPI) handleStatus(w http.ResponseWriter, req *http.Request) {
 		st.ID = h.capture.Info().ID
 		st.IsOwner = h.capture.IsOwner()
 	}
-	writeData(w, st)
+	api.WriteData(w, st)
 }

cdc/api/util.go

@@ -57,19 +57,21 @@ func IsHTTPBadRequestError(err error) bool {
 	return false
 }
 
-func writeError(w http.ResponseWriter, statusCode int, err error) {
+// WriteError write error message to response
+func WriteError(w http.ResponseWriter, statusCode int, err error) {
 	w.WriteHeader(statusCode)
 	_, err = w.Write([]byte(err.Error()))
 	if err != nil {
 		log.Error("write error", zap.Error(err))
 	}
 }
 
-func writeData(w http.ResponseWriter, data interface{}) {
+// WriteData write data to response with http status code 200
+func WriteData(w http.ResponseWriter, data interface{}) {
 	js, err := json.MarshalIndent(data, "", " ")
 	if err != nil {
 		log.Error("invalid json data", zap.Reflect("data", data), zap.Error(err))
-		writeError(w, http.StatusInternalServerError, err)
+		WriteError(w, http.StatusInternalServerError, err)
 		return
 	}
 	w.Header().Set("Content-Type", "application/json")
@@ -80,7 +82,8 @@ func writeData(w http.ResponseWriter, data interface{}) {
 	}
 }
 
-func handleOwnerJob(
+// HandleOwnerJob enqueue the admin job
+func HandleOwnerJob(
 	ctx context.Context, capture *capture.Capture, job model.AdminJob,
 ) error {
 	// Use buffered channel to prevernt blocking owner.
@@ -98,7 +101,8 @@ func handleOwnerJob(
 	}
 }
 
-func handleOwnerRebalance(
+// HandleOwnerBalance balance the changefeed tables
+func HandleOwnerBalance(
 	ctx context.Context, capture *capture.Capture, changefeedID model.ChangeFeedID,
 ) error {
 	// Use buffered channel to prevernt blocking owner.
@@ -116,7 +120,8 @@ func handleOwnerRebalance(
 	}
 }
 
-func handleOwnerScheduleTable(
+// HandleOwnerScheduleTable schedule tables
+func HandleOwnerScheduleTable(
 	ctx context.Context, capture *capture.Capture,
 	changefeedID model.ChangeFeedID, captureID string, tableID int64,
 ) error {