From 05be96c8b60bd7f52f1828c794a253abcbd3b393 Mon Sep 17 00:00:00 2001 From: nhsmw Date: Wed, 14 Oct 2026 18:23:46 +0800 Subject: [PATCH] This is an automated cherry-pick of #6525 Signed-off-by: ti-chi-bot --- api/v2/changefeed.go | 21 +++++++ api/v2/changefeed_test.go | 116 ++++++++++++++++++++++++++++++++++++++ api/v2/helper.go | 24 ++++++++ api/v2/helper_test.go | 54 ++++++++++++++++++ pkg/errors/error.go | 5 ++ 5 files changed, 220 insertions(+) diff --git a/api/v2/changefeed.go b/api/v2/changefeed.go index be080312d8..7cba789ad2 100644 --- a/api/v2/changefeed.go +++ b/api/v2/changefeed.go @@ -197,6 +197,27 @@ func (h *OpenAPIV2) CreateChangefeed(c *gin.Context) { return } +<<<<<<< HEAD +======= + // do not shadow err after this point + pdClient := h.server.GetPdClient() + if err = hasRunningImport(ctx, h.server.GetEtcdClient().GetEtcdClient(), pdClient.GetClusterID(ctx)); err != nil { + _ = c.Error(err) + return + } + + tmpInfo := &config.ChangeFeedInfo{ + ChangefeedID: changefeedID, + SinkURI: cfg.SinkURI, + Config: replicaCfg, + } + err = check.ValidateActiveActiveTSOIndexes(ctx, pdClient, tmpInfo.ToChangefeedConfig()) + if err != nil { + _ = c.Error(err) + return + } + +>>>>>>> be38e502a (api: reject changefeed creation during upstream import or restore (#6525)) // 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) diff --git a/api/v2/changefeed_test.go b/api/v2/changefeed_test.go index 1f5ec5e1a3..bd99f15bd0 100644 --- a/api/v2/changefeed_test.go +++ b/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" @@ -48,8 +49,123 @@ 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" ) +<<<<<<< HEAD +======= +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(&deleteMissingCoordinator{}, 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) + + originalConfig := config.GetGlobalServerConfig() + t.Cleanup(func() { + config.StoreGlobalServerConfig(originalConfig) + }) + cfg := originalConfig.Clone() + cfg.Security.ClientUserRequired = true + cfg.Security.ClientAllowedUser = []string{"alice"} + config.StoreGlobalServerConfig(cfg) + + ctrl := gomock.NewController(t) + etcdClient := etcd.NewMockCDCEtcdClient(ctrl) + etcdClient.EXPECT().GetEtcdClient().Return(nil) + + coordinator := &deleteMissingCoordinator{} + handler := &OpenAPIV2{server: &deleteMissingServer{ + coordinator: coordinator, + etcdClient: etcdClient, + }} + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest( + http.MethodDelete, + "/api/v2/changefeeds/missing?keyspace=test", + nil, + ) + c.Params = gin.Params{{Key: api.APIOpVarChangefeedID, Value: "missing"}} + + handler.DeleteChangefeed(c) + + require.Equal(t, http.StatusUnauthorized, w.Code) + require.False(t, coordinator.removeCalled) +} + +func TestLoadKeyspaceInContext(t *testing.T) { + if !kerneltype.IsNextGen() { + t.Skip("keyspace authentication context is only needed in next-gen") + } + + ctrl := gomock.NewController(t) + keyspaceManager := keyspace.NewMockManager(ctrl) + keyspaceManager.EXPECT().LoadKeyspace(gomock.Any(), "test").Return(&keyspacepb.KeyspaceMeta{ + Keyspace: &keyspacepb.KeyspaceMeta_Id{Id: 1}, + Name: "test", + }, nil) + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodDelete, "/api/v2/changefeeds/missing?keyspace=test", nil) + + loadKeyspaceInContext(c, keyspaceManager) + + require.Equal(t, uint32(1), middleware.GetKeyspaceFromContext(c).GetId()) +} + +>>>>>>> be38e502a (api: reject changefeed creation during upstream import or restore (#6525)) // TestValidateResumeChangefeedState covers the API-side guard that runs before // resume GC safepoint/barrier setup. Running states must fail fast, while states // that are actually stopped can proceed to the remaining resume validation. diff --git a/api/v2/helper.go b/api/v2/helper.go index 5dd9c4a40a..fdbac3b486 100644 --- a/api/v2/helper.go +++ b/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() diff --git a/api/v2/helper_test.go b/api/v2/helper_test.go index 8d90da5581..3b8091c381 100644 --- a/api/v2/helper_test.go +++ b/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) diff --git a/pkg/errors/error.go b/pkg/errors/error.go index 2b1769e1b5..2afa1c33be 100644 --- a/pkg/errors/error.go +++ b/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"),