api: reject changefeed creation during upstream import or restore (#6525)

api/v2/changefeed.go

@@ -197,6 +197,11 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) {
 		return
 	}
 
+	if err = hasRunningImport(ctx, h.server.GetEtcdClient().GetEtcdClient(), h.server.GetPdClient().GetClusterID(ctx)); err != nil {
+		_ = c.Error(err)
+		return
+	}
+
 	// Ensure the start ts is valid in the next 3600 seconds, aka 1 hour
 	const ensureTTL = 60 * 60
 	createGcServiceID := h.server.GetEtcdClient().GetEnsureGCServiceID(gc.EnsureGCServiceCreating)

api/v2/changefeed_test.go

@@ -21,6 +21,7 @@ import (
 	"net/url"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/golang/mock/gomock"
@@ -49,8 +50,62 @@ import (
 	"github.com/stretchr/testify/require"
 	pd "github.com/tikv/pd/client"
 	"github.com/twmb/franz-go/pkg/kfake"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
+func TestCreateChangefeedRejectsRunningImport(t *testing.T) {
+	_, etcdServer, err := etcd.SetupEmbedEtcd(t.TempDir())
+	require.NoError(t, err)
+	t.Cleanup(etcdServer.Close)
+	cli, err := clientv3.New(clientv3.Config{
+		Endpoints:   []string{etcdServer.Clients[0].Addr().String()},
+		DialTimeout: 3 * time.Second,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, cli.Close()) })
+	wrappedClient := etcd.NewWrappedClient(cli)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+
+	for _, taskType := range []string{"lightning", "restore", "import-into"} {
+		t.Run(taskType, func(t *testing.T) {
+			lease, err := cli.Grant(ctx, 180)
+			require.NoError(t, err)
+			_, err = cli.Put(ctx, registerImportTaskPrefix+"/"+taskType+"/test", "", clientv3.WithLease(lease.ID))
+			require.NoError(t, err)
+
+			ctrl := gomock.NewController(t)
+			cdcEtcdClient := etcd.NewMockCDCEtcdClient(ctrl)
+			cdcEtcdClient.EXPECT().GetEtcdClient().Return(wrappedClient)
+			srv := mock.NewMockServer(ctrl)
+			srv.EXPECT().GetCoordinator().Return(&resumeNormalCoordinator{info: &config.ChangeFeedInfo{}}, nil)
+			srv.EXPECT().GetPdClient().Return(&gc.MockPDClient{ClusterID: 123}).AnyTimes()
+			srv.EXPECT().GetEtcdClient().Return(cdcEtcdClient)
+			h := OpenAPIV2{server: srv}
+			router := gin.New()
+			router.Use(middleware.ErrorHandleMiddleware())
+			router.POST("/changefeeds", func(c *gin.Context) {
+				c.Set("ctx-keyspace", &keyspacepb.KeyspaceMeta{State: keyspacepb.KeyspaceState_ENABLED})
+				h.CreateChangefeed(c)
+			})
+			w := httptest.NewRecorder()
+			request := `{"changefeed_id":"test","sink_uri":"blackhole://"}`
+			router.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/changefeeds", strings.NewReader(request)))
+
+			require.Equal(t, http.StatusInternalServerError, w.Code, w.Body.String())
+			var response api.HTTPError
+			require.NoError(t, json.Unmarshal(w.Body.Bytes(), &response))
+			require.Equal(t, "CDC:ErrUpstreamHasRunningImport", response.Code)
+			require.Contains(t, response.Error, "upstream-id: 123")
+			require.Contains(t, response.Error, "please stop or wait")
+
+			_, err = cli.Revoke(ctx, lease.ID)
+			require.NoError(t, err)
+			require.NoError(t, hasRunningImport(ctx, wrappedClient, 123))
+		})
+	}
+}
+
 func TestDeleteMissingChangefeedRequiresAuthentication(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 

api/v2/helper.go

@@ -26,11 +26,35 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/etcd"
 	"github.com/pingcap/ticdc/pkg/httputil"
 	"github.com/pingcap/ticdc/pkg/server"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
 )
 
+// registerImportTaskPrefix is where Lightning and BR register running import
+// and restore tasks in the upstream PD's embedded etcd.
+const registerImportTaskPrefix = "/tidb/brie/import"
+
+func hasRunningImport(ctx context.Context, cli etcd.Client, upstreamID uint64) error {
+	resp, err := cli.Get(ctx, registerImportTaskPrefix, clientv3.WithPrefix())
+	if err != nil {
+		return errors.WrapError(errors.ErrPDEtcdAPIError, err)
+	}
+
+	for _, kv := range resp.Kvs {
+		lease, err := cli.TimeToLive(ctx, clientv3.LeaseID(kv.Lease))
+		if err != nil {
+			return errors.WrapError(errors.ErrPDEtcdAPIError, err)
+		}
+		if lease.TTL > 0 {
+			return errors.ErrUpstreamHasRunningImport.GenWithStackByArgs(upstreamID)
+		}
+	}
+	return nil
+}
+
 func (h *OpenAPIV2) getChangeFeed(ctx context.Context, keyspaceName, cfName string) (ChangeFeedInfo, error) {
 	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 	defer cancel()

api/v2/helper_test.go

@@ -27,12 +27,66 @@ import (
 	"github.com/pingcap/ticdc/api/middleware"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/etcd"
 	"github.com/pingcap/ticdc/pkg/node"
 	"github.com/pingcap/ticdc/pkg/server/mock"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
+func TestHasRunningImport(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		ttls     []int64
+		getErr   error
+		leaseErr error
+		wantCode string
+	}{
+		{name: "no tasks"},
+		{name: "running task", ttls: []int64{180}, wantCode: "CDC:ErrUpstreamHasRunningImport"},
+		{name: "expired lease", ttls: []int64{-1}},
+		{name: "zero ttl", ttls: []int64{0}},
+		{name: "expired then running", ttls: []int64{-1, 180}, wantCode: "CDC:ErrUpstreamHasRunningImport"},
+		{name: "list failure", getErr: context.Canceled, wantCode: "CDC:ErrPDEtcdAPIError"},
+		{name: "lease failure", ttls: []int64{180}, leaseErr: context.Canceled, wantCode: "CDC:ErrPDEtcdAPIError"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			cli := etcd.NewMockClient(gomock.NewController(t))
+			resp := &clientv3.GetResponse{}
+			for i, ttl := range tc.ttls {
+				leaseID := int64(i + 1)
+				resp.Kvs = append(resp.Kvs, &mvccpb.KeyValue{Lease: leaseID})
+				cli.EXPECT().TimeToLive(ctx, clientv3.LeaseID(leaseID)).Return(
+					&clientv3.LeaseTimeToLiveResponse{TTL: ttl}, tc.leaseErr)
+			}
+			cli.EXPECT().Get(ctx, registerImportTaskPrefix, gomock.Any()).DoAndReturn(
+				func(_ context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
+					op := clientv3.OpGet(key, opts...)
+					require.Equal(t, []byte(clientv3.GetPrefixRangeEnd(registerImportTaskPrefix)), op.RangeBytes())
+					return resp, tc.getErr
+				})
+
+			err := hasRunningImport(ctx, cli, 123)
+			if tc.wantCode == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			code, ok := errors.RFCCode(err)
+			require.True(t, ok)
+			require.Equal(t, tc.wantCode, string(code))
+			if tc.wantCode == "CDC:ErrUpstreamHasRunningImport" {
+				require.Contains(t, err.Error(), "upstream-id: 123")
+			} else {
+				require.ErrorIs(t, err, context.Canceled)
+			}
+		})
+	}
+}
+
 func TestGetChangeFeedCoordinatorError(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	srv := mock.NewMockServer(ctrl)

pkg/errors/error.go

@@ -571,6 +571,11 @@ var (
 		"upstream not found, cluster-id: %d",
 		errors.RFCCodeText("CDC:ErrUpstreamNotFound"),
 	)
+	ErrUpstreamHasRunningImport = errors.Normalize(
+		"upstream has running import tasks, upstream-id: %d; please stop or wait for "+
+			"Lightning/restore tasks to finish, or wait for the task lease to expire if it terminated abnormally",
+		errors.RFCCodeText("CDC:ErrUpstreamHasRunningImport"),
+	)
 	ErrUpdateServiceSafepointFailed = errors.Normalize(
 		"updating service safepoint failed",
 		errors.RFCCodeText("CDC:ErrUpdateServiceSafepointFailed"),