remove tidb dependency

Signed-off-by: Ping Yu <[email protected]>

cdc/cdc/capture/capture.go

@@ -24,7 +24,6 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
-	tidbkv "github.com/pingcap/tidb/kv"
 	"github.com/tikv/client-go/v2/tikv"
 	pd "github.com/tikv/pd/client"
 	"go.etcd.io/etcd/client/v3/concurrency"
@@ -61,7 +60,7 @@ type Capture struct {
 	election *concurrency.Election
 
 	pdClient         pd.Client
-	kvStorage        tidbkv.Storage
+	kvStorage        tikv.Storage
 	createEtcdClient createEtcdClientFunc
 	etcdClient       *etcd.CDCEtcdClient
 	grpcPool         kv.GrpcPool
@@ -75,7 +74,7 @@ type Capture struct {
 }
 
 // NewCapture returns a new Capture instance
-func NewCapture(pdClient pd.Client, kvStorage tidbkv.Storage, createEtcdClient createEtcdClientFunc) *Capture {
+func NewCapture(pdClient pd.Client, kvStorage tikv.Storage, createEtcdClient createEtcdClientFunc) *Capture {
 	return &Capture{
 		pdClient:         pdClient,
 		kvStorage:        kvStorage,

cdc/cdc/kv/region_worker.go

@@ -312,7 +312,7 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 				log.Warn("failed to get current version from PD", zap.Error(err))
 				continue
 			}
-			currentTimeFromPD := oracle.GetTimeFromTS(version.Ver)
+			currentTimeFromPD := oracle.GetTimeFromTS(version)
 			expired := make([]*regionTsInfo, 0)
 			for w.rtsManager.Len() > 0 {
 				item := w.rtsManager.Pop()
@@ -330,7 +330,7 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 			if len(expired) == 0 {
 				continue
 			}
-			maxVersion := oracle.ComposeTS(oracle.GetPhysical(currentTimeFromPD.Add(-10*time.Second)), 0)
+			// maxVersion := oracle.ComposeTS(oracle.GetPhysical(currentTimeFromPD.Add(-10*time.Second)), 0)
 			for _, rts := range expired {
 				state, ok := w.getRegionState(rts.regionID)
 				if !ok || state.isStopped() {
@@ -359,18 +359,19 @@ func (w *regionWorker) resolveLock(ctx context.Context) error {
 						w.rtsManager.Upsert(rts)
 						continue
 					}
-					log.Warn("region not receiving resolved event from tikv or resolved ts is not pushing for too long time, try to resolve lock",
+					log.Warn("region not receiving resolved event from tikv or resolved ts is not pushing for too long time",
 						zap.Uint64("regionID", rts.regionID),
 						zap.Stringer("span", state.getRegionSpan()),
 						zap.Duration("duration", sinceLastResolvedTs),
 						zap.Duration("lastEvent", sinceLastEvent),
 						zap.Uint64("resolvedTs", lastResolvedTs),
 					)
-					err = w.session.lockResolver.Resolve(ctx, rts.regionID, maxVersion)
-					if err != nil {
-						log.Warn("failed to resolve lock", zap.Uint64("regionID", rts.regionID), zap.Error(err))
-						continue
-					}
+					// Resolve locks for RawKV is not necessary. Add it back after we support TxnKV.
+					// err = w.session.lockResolver.Resolve(ctx, rts.regionID, maxVersion)
+					// if err != nil {
+					// 	log.Warn("failed to resolve lock", zap.Uint64("regionID", rts.regionID), zap.Error(err))
+					// 	continue
+					// }
 					rts.ts.penalty = 0
 				}
 				rts.ts.resolvedTs = lastResolvedTs

cdc/cdc/kv/store_op.go

@@ -14,20 +14,16 @@
 package kv
 
 import (
-	"fmt"
 	"sync"
 	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	tidbconfig "github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/kv"
-	tidbkv "github.com/pingcap/tidb/kv"
-	"github.com/pingcap/tidb/meta"
-	"github.com/pingcap/tidb/store"
-	"github.com/pingcap/tidb/store/driver"
+	tikvconfig "github.com/tikv/client-go/v2/config"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
+	"github.com/tikv/client-go/v2/txnkv"
 	"github.com/tikv/migration/cdc/cdc/model"
 	cerror "github.com/tikv/migration/cdc/pkg/errors"
 	"github.com/tikv/migration/cdc/pkg/flags"
@@ -38,7 +34,7 @@ import (
 // TiKVStorage is the tikv storage interface used by CDC.
 type TiKVStorage interface {
 	tikv.Storage
-	GetCachedCurrentVersion() (version tidbkv.Version, err error)
+	GetCachedCurrentVersion() (version uint64, err error)
 }
 
 const (
@@ -80,7 +76,7 @@ func newStorageWithCurVersionCache(storage tikv.Storage, cacheKey string) TiKVSt
 }
 
 // GetCachedCurrentVersion gets the cached version of currentVersion, and update the cache if necessary
-func (s *StorageWithCurVersionCache) GetCachedCurrentVersion() (version tidbkv.Version, err error) {
+func (s *StorageWithCurVersionCache) GetCachedCurrentVersion() (version uint64, err error) {
 	curVersionCacheMu.Lock()
 	entry, exists := curVersionCache[s.cacheKey]
 	curVersionCacheMu.Unlock()
@@ -104,37 +100,26 @@ func (s *StorageWithCurVersionCache) GetCachedCurrentVersion() (version tidbkv.V
 		entry.lastUpdated = time.Now()
 	}
 
-	version.Ver = entry.ts
+	version = entry.ts
 	return
 }
 
-// GetSnapshotMeta returns tidb meta information
-// TODO: Simplify the signature of this function
-func GetSnapshotMeta(tiStore tidbkv.Storage, ts uint64) (*meta.Meta, error) {
-	snapshot := tiStore.GetSnapshot(tidbkv.NewVersion(ts))
-	return meta.NewSnapshotMeta(snapshot), nil
-}
-
 // CreateTiStore creates a new tikv storage client
-func CreateTiStore(urls string, credential *security.Credential) (kv.Storage, error) {
+func CreateTiStore(urls string, credential *security.Credential) (tikv.Storage, error) {
 	urlv, err := flags.NewURLsValue(urls)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
-	// Ignore error if it is already registered.
-	_ = store.Register("tikv", driver.TiKVDriver{})
-
 	if credential.CAPath != "" {
-		conf := tidbconfig.GetGlobalConfig()
+		conf := tikvconfig.GetGlobalConfig()
 		conf.Security.ClusterSSLCA = credential.CAPath
 		conf.Security.ClusterSSLCert = credential.CertPath
 		conf.Security.ClusterSSLKey = credential.KeyPath
-		tidbconfig.StoreGlobalConfig(conf)
+		tikvconfig.StoreGlobalConfig(conf)
 	}
 
-	tiPath := fmt.Sprintf("tikv://%s?disableGC=true", urlv.HostString())
-	tiStore, err := store.New(tiPath)
+	tiStore, err := txnkv.NewClient(urlv.HostList())
 	if err != nil {
 		return nil, cerror.WrapError(cerror.ErrNewStore, err)
 	}

cdc/cdc/puller/puller.go

@@ -20,7 +20,6 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	tidbkv "github.com/pingcap/tidb/kv"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/migration/cdc/cdc/kv"
@@ -68,15 +67,11 @@ func NewPuller(
 	pdCli pd.Client,
 	grpcPool kv.GrpcPool,
 	regionCache *tikv.RegionCache,
-	kvStorage tidbkv.Storage,
+	tikvStorage tikv.Storage,
 	checkpointTs uint64,
 	spans []regionspan.Span,
 	enableOldValue bool,
 ) Puller {
-	tikvStorage, ok := kvStorage.(tikv.Storage)
-	if !ok {
-		log.Panic("can't create puller for non-tikv storage")
-	}
 	comparableSpans := make([]regionspan.ComparableSpan, len(spans))
 	for i := range spans {
 		comparableSpans[i] = regionspan.ToComparableSpan(spans[i])

cdc/cdc/puller/puller_test.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/pingcap/check"
 	"github.com/pingcap/errors"
-	tidbkv "github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/store/mockstore"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/migration/cdc/cdc/kv"
@@ -116,11 +115,17 @@ func (s *pullerSuite) newPullerForTest(
 	c *check.C,
 	spans []regionspan.Span,
 	checkpointTs uint64,
-) (*mockInjectedPuller, context.CancelFunc, *sync.WaitGroup, tidbkv.Storage) {
+) (*mockInjectedPuller, context.CancelFunc, *sync.WaitGroup, tikv.Storage) {
 	var wg sync.WaitGroup
 	ctx, cancel := context.WithCancel(context.Background())
 	store, err := mockstore.NewMockStore()
 	c.Assert(err, check.IsNil)
+
+	tikvStorage, ok := store.(tikv.Storage)
+	if !ok {
+		panic("can't create puller for non-tikv storage")
+	}
+
 	enableOldValue := true
 	backupNewCDCKVClient := kv.NewCDCKVClient
 	kv.NewCDCKVClient = newMockCDCKVClient
@@ -132,7 +137,7 @@ func (s *pullerSuite) newPullerForTest(
 	defer grpcPool.Close()
 	regionCache := tikv.NewRegionCache(pdCli)
 	defer regionCache.Close()
-	plr := NewPuller(ctx, pdCli, grpcPool, regionCache, store, checkpointTs, spans, enableOldValue)
+	plr := NewPuller(ctx, pdCli, grpcPool, regionCache, tikvStorage, checkpointTs, spans, enableOldValue)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
@@ -146,7 +151,7 @@ func (s *pullerSuite) newPullerForTest(
 		Puller: plr,
 		cli:    plr.(*pullerImpl).kvCli.(*mockCDCKVClient),
 	}
-	return mockPlr, cancel, &wg, store
+	return mockPlr, cancel, &wg, tikvStorage
 }
 
 func (s *pullerSuite) TestPullerResolvedForward(c *check.C) {

cdc/cdc/server.go

@@ -24,8 +24,8 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	tidbkv "github.com/pingcap/tidb/kv"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/tikv/client-go/v2/tikv"
 	pd "github.com/tikv/pd/client"
 	"go.etcd.io/etcd/client/pkg/v3/logutil"
 	clientv3 "go.etcd.io/etcd/client/v3"
@@ -63,7 +63,7 @@ type Server struct {
 	statusServer *http.Server
 	pdClient     pd.Client
 	etcdClient   *etcd.CDCEtcdClient
-	kvStorage    tidbkv.Storage
+	kvStorage    tikv.Storage
 	pdEndpoints  []string
 }
 

cdc/pkg/context/context.go

@@ -18,7 +18,6 @@ import (
 	"time"
 
 	"github.com/pingcap/log"
-	tidbkv "github.com/pingcap/tidb/kv"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/migration/cdc/cdc/kv"
@@ -37,7 +36,7 @@ import (
 // All field in Vars should be READ-ONLY and THREAD-SAFE
 type GlobalVars struct {
 	PDClient     pd.Client
-	KVStorage    tidbkv.Storage
+	KVStorage    tikv.Storage
 	CaptureInfo  *model.CaptureInfo
 	EtcdClient   *etcd.CDCEtcdClient
 	GrpcPool     kv.GrpcPool

cdc/pkg/flags/urls.go

@@ -45,6 +45,15 @@ func (us *URLsValue) HostString() string {
 	return strings.Join(all, ",")
 }
 
+// HostList return an list of host:port
+func (us *URLsValue) HostList() []string {
+	all := make([]string, len(*us))
+	for i, u := range *us {
+		all[i] = u.Host
+	}
+	return all
+}
+
 // NewURLsValue return a URLsValue from a string of URLs list
 func NewURLsValue(init string) (*URLsValue, error) {
 	v := &URLsValue{}

cdc/pkg/flags/urls_test.go

@@ -14,6 +14,7 @@
 package flags
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -39,6 +40,7 @@ func TestNewURLsValue(t *testing.T) {
 		require.Nil(t, err)
 		hs := urlsValue.HostString()
 		require.Equal(t, testCase.hostString, hs)
+		require.Equal(t, strings.Split(testCase.hostString, ","), urlsValue.HostList())
 	}
 }
 

cdc/pkg/util/ctx.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/kv"
+	"github.com/tikv/client-go/v2/tikv"
 	"go.uber.org/zap"
 )
 
@@ -55,7 +56,7 @@ func PutTimezoneInCtx(ctx context.Context, timezone *time.Location) context.Cont
 }
 
 // PutKVStorageInCtx returns a new child context with the given tikv store
-func PutKVStorageInCtx(ctx context.Context, store kv.Storage) context.Context {
+func PutKVStorageInCtx(ctx context.Context, store tikv.Storage) context.Context {
 	return context.WithValue(ctx, ctxKeyKVStorage, store)
 }
 

cdc/pkg/util/ctx_test.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/pingcap/check"
 	"github.com/pingcap/tidb/store/mockstore"
+	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/migration/cdc/pkg/util/testleak"
 	"go.uber.org/zap"
 )
@@ -104,8 +105,14 @@ func (s *ctxValueSuite) TestKeySpanInfoNotSet(c *check.C) {
 
 func (s *ctxValueSuite) TestShouldReturnKVStorage(c *check.C) {
 	defer testleak.AfterTest(c)()
-	kvStorage, _ := mockstore.NewMockStore()
-	defer kvStorage.Close()
+	store, _ := mockstore.NewMockStore()
+	defer store.Close()
+
+	kvStorage, ok := store.(tikv.Storage)
+	if !ok {
+		panic("can't create puller for non-tikv storage")
+	}
+
 	ctx := PutKVStorageInCtx(context.Background(), kvStorage)
 	kvStorage2, err := KVStorageFromCtx(ctx)
 	c.Assert(kvStorage2, check.DeepEquals, kvStorage)