api (ticdc): check http request context in api helper (#11165) (#11398)

close #11046

cdc/api/v2/api_helpers.go

@@ -57,6 +57,8 @@ const RegisterImportTaskPrefix = "/tidb/brie/import"
 
 // APIV2Helpers is a collections of helper functions of OpenAPIV2.
 // Defining it as an interface to make APIs more testable.
+// Note: Every method in this interface should check the context before returning.
+// If the context is canceled, the method should return an error immediately.
 type APIV2Helpers interface {
 	// verifyCreateChangefeedConfig verifies the changefeedConfig,
 	// and yield a valid changefeedInfo or error
@@ -105,18 +107,22 @@ type APIV2Helpers interface {
 	// getEtcdClient returns an Etcd client given the PD endpoints and
 	// tls config
 	getEtcdClient(
+		ctx context.Context,
 		pdAddrs []string,
 		tlsConfig *tls.Config,
 	) (*clientv3.Client, error)
 
 	// getKVCreateTiStore wraps kv.createTiStore method to increase testability
 	createTiStore(
+		ctx context.Context,
 		pdAddrs []string,
 		credential *security.Credential,
 	) (tidbkv.Storage, error)
 
 	// getVerifiedTables wraps entry.VerifyTables to increase testability
-	getVerifiedTables(replicaConfig *config.ReplicaConfig,
+	getVerifiedTables(
+		ctx context.Context,
+		replicaConfig *config.ReplicaConfig,
 		storage tidbkv.Storage, startTs uint64,
 		scheme string, topic string, protocol config.Protocol,
 	) (ineligibleTables,
@@ -258,7 +264,8 @@ func (APIV2HelpersImpl) verifyCreateChangefeedConfig(
 }
 
 // verifyUpstream verifies the upstream config before updating a changefeed
-func (h APIV2HelpersImpl) verifyUpstream(ctx context.Context,
+func (h APIV2HelpersImpl) verifyUpstream(
+	ctx context.Context,
 	changefeedConfig *ChangefeedConfig,
 	cfInfo *model.ChangeFeedInfo,
 ) error {
@@ -402,7 +409,8 @@ func (APIV2HelpersImpl) verifyUpdateChangefeedConfig(
 // overrideCheckpointTs is the checkpointTs of the changefeed that specified by the user.
 // or it is the checkpointTs of the changefeed before it is paused.
 // we need to check weather the resuming changefeed is gc safe or not.
-func (APIV2HelpersImpl) verifyResumeChangefeedConfig(ctx context.Context,
+func (APIV2HelpersImpl) verifyResumeChangefeedConfig(
+	ctx context.Context,
 	pdClient pd.Client,
 	gcServiceID string,
 	changefeedID model.ChangeFeedID,
@@ -441,7 +449,8 @@ func (APIV2HelpersImpl) verifyResumeChangefeedConfig(ctx context.Context,
 }
 
 // getPDClient returns a PDClient given the PD cluster addresses and a credential
-func (APIV2HelpersImpl) getPDClient(ctx context.Context,
+func (APIV2HelpersImpl) getPDClient(
+	ctx context.Context,
 	pdAddrs []string,
 	credential *security.Credential,
 ) (pd.Client, error) {
@@ -473,7 +482,9 @@ func (APIV2HelpersImpl) getPDClient(ctx context.Context,
 }
 
 func (h APIV2HelpersImpl) getEtcdClient(
-	pdAddrs []string, tlsCfg *tls.Config,
+	ctx context.Context,
+	pdAddrs []string,
+	tlsCfg *tls.Config,
 ) (*clientv3.Client, error) {
 	conf := config.GetGlobalServerConfig()
 	grpcTLSOption, err := conf.Security.ToGRPCDialOption()
@@ -482,7 +493,7 @@ func (h APIV2HelpersImpl) getEtcdClient(
 	}
 	logConfig := &logutil.DefaultZapLoggerConfig
 	logConfig.Level = zap.NewAtomicLevelAt(zapcore.ErrorLevel)
-	return clientv3.New(
+	res, err := clientv3.New(
 		clientv3.Config{
 			Endpoints:   pdAddrs,
 			TLS:         tlsCfg,
@@ -505,16 +516,33 @@ func (h APIV2HelpersImpl) getEtcdClient(
 			},
 		},
 	)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	if ctx.Err() != nil {
+		return nil, errors.Trace(ctx.Err())
+	}
+	return res, nil
 }
 
 // getTiStore wrap the kv.createTiStore method to increase testability
-func (h APIV2HelpersImpl) createTiStore(pdAddrs []string,
+func (h APIV2HelpersImpl) createTiStore(
+	ctx context.Context,
+	pdAddrs []string,
 	credential *security.Credential,
 ) (tidbkv.Storage, error) {
-	return kv.CreateTiStore(strings.Join(pdAddrs, ","), credential)
+	res, err := kv.CreateTiStore(strings.Join(pdAddrs, ","), credential)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	if ctx.Err() != nil {
+		return nil, errors.Trace(ctx.Err())
+	}
+	return res, nil
 }
 
 func (h APIV2HelpersImpl) getVerifiedTables(
+	ctx context.Context,
 	replicaConfig *config.ReplicaConfig,
 	storage tidbkv.Storage, startTs uint64,
 	scheme string, topic string, protocol config.Protocol,
@@ -551,5 +579,9 @@ func (h APIV2HelpersImpl) getVerifiedTables(
 		return nil, nil, err
 	}
 
+	if ctx.Err() != nil {
+		return nil, nil, errors.Trace(ctx.Err())
+	}
+
 	return ineligibleTables, eligibleTables, nil
 }

cdc/api/v2/changefeed.go

@@ -86,7 +86,7 @@ func (h *OpenAPIV2) createChangefeed(c *gin.Context) {
 	defer pdClient.Close()
 
 	// verify tables todo: del kvstore
-	kvStorage, err := h.helpers.createTiStore(cfg.PDAddrs, credential)
+	kvStorage, err := h.helpers.createTiStore(ctx, cfg.PDAddrs, credential)
 	if err != nil {
 		_ = c.Error(cerror.WrapError(cerror.ErrNewStore, err))
 		return
@@ -143,7 +143,7 @@ func (h *OpenAPIV2) createChangefeed(c *gin.Context) {
 		return
 	}
 
-	cli, err := h.helpers.getEtcdClient(cfg.PDAddrs, tlsCfg)
+	cli, err := h.helpers.getEtcdClient(ctx, cfg.PDAddrs, tlsCfg)
 	if err != nil {
 		_ = c.Error(err)
 		return
@@ -332,8 +332,8 @@ func (h *OpenAPIV2) verifyTable(c *gin.Context) {
 		cfg.PDConfig = getUpstreamPDConfig(up)
 	}
 	credential := cfg.PDConfig.toCredential()
-
-	kvStore, err := h.helpers.createTiStore(cfg.PDAddrs, credential)
+	ctx := c.Request.Context()
+	kvStore, err := h.helpers.createTiStore(ctx, cfg.PDAddrs, credential)
 	if err != nil {
 		_ = c.Error(err)
 		return
@@ -351,7 +351,7 @@ func (h *OpenAPIV2) verifyTable(c *gin.Context) {
 	protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(replicaCfg.Sink.Protocol))
 
 	ineligibleTables, eligibleTables, err := h.helpers.
-		getVerifiedTables(replicaCfg, kvStore, cfg.StartTs, scheme, topic, protocol)
+		getVerifiedTables(ctx, replicaCfg, kvStore, cfg.StartTs, scheme, topic, protocol)
 	if err != nil {
 		_ = c.Error(err)
 		return
@@ -468,7 +468,7 @@ func (h *OpenAPIV2) updateChangefeed(c *gin.Context) {
 	if len(updateCfConfig.PDAddrs) != 0 || upManager == nil {
 		pdAddrs := updateCfConfig.PDAddrs
 		credentials := updateCfConfig.PDConfig.toCredential()
-		storage, err = h.helpers.createTiStore(pdAddrs, credentials)
+		storage, err = h.helpers.createTiStore(ctx, pdAddrs, credentials)
 		if err != nil {
 			_ = c.Error(errors.Trace(err))
 		}

cdc/api/v2/changefeed_test.go

@@ -134,7 +134,7 @@ func TestCreateChangefeed(t *testing.T) {
 		getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(pdClient, nil).AnyTimes()
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, cerrors.ErrNewStore).
 		Times(1)
 	cfConfig.PDAddrs = []string{"http://127.0.0.1:2379", "http://127.0.0.1:2382"}
@@ -152,7 +152,7 @@ func TestCreateChangefeed(t *testing.T) {
 
 	// case 4: failed to verify tables
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
 	helpers.EXPECT().
@@ -176,9 +176,9 @@ func TestCreateChangefeed(t *testing.T) {
 
 	// case 5:
 	helpers.EXPECT().
-		getEtcdClient(gomock.Any(), gomock.Any()).
+		getEtcdClient(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(testEtcdCluster.RandClient(), nil)
-	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil, nil).
 		AnyTimes()
@@ -221,7 +221,7 @@ func TestCreateChangefeed(t *testing.T) {
 
 	// case 6: success
 	helpers.EXPECT().
-		getEtcdClient(gomock.Any(), gomock.Any()).
+		getEtcdClient(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(testEtcdCluster.RandClient(), nil)
 	ctrl.EXPECT().
 		CreateChangefeed(gomock.Any(), gomock.Any(), gomock.Any()).
@@ -445,7 +445,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		verifyUpstream(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).AnyTimes()
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
 	mockCapture.EXPECT().GetUpstreamManager().Return(nil, nil).AnyTimes()
@@ -649,7 +649,7 @@ func TestVerifyTable(t *testing.T) {
 	body, err := json.Marshal(&updateCfg)
 	require.Nil(t, err)
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, cerrors.ErrNewStore).
 		Times(1)
 
@@ -664,10 +664,10 @@ func TestVerifyTable(t *testing.T) {
 
 	// case 3: getVerifiedTables failed
 	helpers.EXPECT().
-		createTiStore(gomock.Any(), gomock.Any()).
+		createTiStore(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
-	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil, cerrors.ErrFilterRuleInvalid).
 		Times(1)
@@ -689,7 +689,7 @@ func TestVerifyTable(t *testing.T) {
 	ineligible := []model.TableName{
 		{Schema: "test", Table: "invalidTable"},
 	}
-	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
 		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(eligible, ineligible, nil)
 
@@ -701,6 +701,21 @@ func TestVerifyTable(t *testing.T) {
 	err = json.NewDecoder(w.Body).Decode(&resp)
 	require.Nil(t, err)
 	require.Equal(t, http.StatusOK, w.Code)
+
+	// case 5: context canceled
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
+		gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(nil, nil, ctx.Err()).
+		Times(1)
+	w = httptest.NewRecorder()
+	req, _ = http.NewRequestWithContext(ctx, verify.method, verify.url, bytes.NewReader(body))
+	router.ServeHTTP(w, req)
+	respErr = model.HTTPError{}
+	err = json.NewDecoder(w.Body).Decode(&respErr)
+	require.Nil(t, err)
+	require.Contains(t, respErr.Error, "context canceled")
 }
 
 func TestResumeChangefeed(t *testing.T) {