Repository: juicedata/juicefs
Pull request #3083 (2022-12-14 07:20:22)

meta/tikv: add pid info in tikv log
<img width="2527" alt="image" src="https://user-images.githubusercontent.com/31313340/207530959-19206345-064e-4014-a8b6-7105b87ab268.png">



Patch:
diff --git a/cmd/debug.go b/cmd/debug.go
index 4a0757b7..5dbd2b58 100644
--- a/cmd/debug.go
+++ b/cmd/debug.go
@@ -121,7 +121,7 @@ func copyFile(srcPath, destPath string, requireRootPrivileges bool) error {
 }
 
 func getCmdMount(mp string) (uid, pid, cmd string, err error) {
-	psArgs := []string{"/bin/sh", "-c", fmt.Sprintf("ps -ef | grep -v grep | grep -E \"%s mount | /sbin/mount.juicefs\" | grep %s", os.Args[0], mp)}
+	psArgs := []string{"/bin/sh", "-c", fmt.Sprintf("ps -ef | grep -v grep | grep mount | grep %s", mp)}
 	ret, err := exec.Command(psArgs[0], psArgs[1:]...).CombinedOutput()
 	if err != nil {
 		return "", "", "", fmt.Errorf("failed to execute command `%s`: %v", strings.Join(psArgs, " "), err)
@@ -218,7 +218,7 @@ func getPprofPort(pid, amp string, requireRootPrivileges bool) (int, error) {
 	if err := json.Unmarshal(content, &cfg); err != nil {
 		logger.Warnf("failed to unmarshal config file: %v", err)
 	}
-	if cfg.Port.DebugAgent != "" {
+	if cfg.Port != nil {
 		if len(strings.Split(cfg.Port.DebugAgent, ":")) >= 2 {
 			if port, err := strconv.Atoi(strings.Split(cfg.Port.DebugAgent, ":")[1]); err != nil {
 				logger.Warnf("failed to parse debug agent port: %v", err)
@@ -455,6 +455,14 @@ func collectLog(ctx *cli.Context, cmd string, requireRootPrivileges bool, currDi
 		logger.Warnf("Collecting log currently only support Linux/macOS")
 		return nil
 	}
+	matched, err := regexp.MatchString(`.*\s-d\s.*`, cmd)
+	if err != nil {
+		return fmt.Errorf("failed to match cmd: %v", err)
+	}
+	if !matched {
+		logger.Warnf("The juicefs mount by foreground, the log will not be collected")
+		return nil
+	}
 	logPath, err := getLogPath(cmd, requireRootPrivileges)
 	if err != nil {
 		return fmt.Errorf("failed to get log path: %v", err)
@@ -474,9 +482,7 @@ func collectSysInfo(ctx *cli.Context, currDir string) error {
 
 	result := fmt.Sprintf(`Platform: 
 %s %s
-%s
-JuiceFS Version:
-%s`, runtime.GOOS, runtime.GOARCH, sysInfo, ctx.App.Version)
+%s`, runtime.GOOS, runtime.GOARCH, sysInfo)
 
 	sysPath := filepath.Join(currDir, "system-info.log")
 	sysFile, err := os.Create(sysPath)
@@ -543,7 +549,7 @@ func debug(ctx *cli.Context) error {
 	}
 
 	if err := collectSysInfo(ctx, currDir); err != nil {
-		return err
+		logger.Errorf("Failed to collect system info: %v", err)
 	}
 
 	uid, pid, cmd, err := getCmdMount(amp)
@@ -560,15 +566,15 @@ func debug(ctx *cli.Context) error {
 
 	var wg sync.WaitGroup
 	if err := collectSpecialFile(ctx, amp, currDir, requireRootPrivileges, &wg); err != nil {
-		return err
+		logger.Errorf("Failed to collect special file: %v", err)
 	}
 
 	if err := collectLog(ctx, cmd, requireRootPrivileges, currDir); err != nil {
-		return err
+		logger.Errorf("Failed to collect log: %v", err)
 	}
 
 	if err := collectPprof(ctx, cmd, pid, amp, requireRootPrivileges, currDir, &wg); err != nil {
-		return err
+		logger.Errorf("Failed to collect pprof: %v", err)
 	}
 
 	wg.Wait()
diff --git a/cmd/mount.go b/cmd/mount.go
index a4b6d410..b3bf24d6 100644
--- a/cmd/mount.go
+++ b/cmd/mount.go
@@ -615,6 +615,9 @@ func mount(c *cli.Context) error {
 	if c.Bool("background") && os.Getenv("JFS_FOREGROUND") == "" {
 		daemonRun(c, addr, vfsConf, metaCli)
 	} else {
+		if c.IsSet("log") {
+			logger.Warnf("--log flag is ignored in foreground mode, the log output will be Stderr")
+		}
 		go checkMountpoint(vfsConf.Format.Name, mp, c.String("log"), false)
 	}
 
@@ -628,5 +631,7 @@ func mount(c *cli.Context) error {
 	v := vfs.NewVFS(vfsConf, metaCli, store, registerer, registry)
 	initBackgroundTasks(c, vfsConf, metaConf, metaCli, blob, registerer, registry)
 	mount_main(v, c)
-	return metaCli.CloseSession()
+	err = metaCli.CloseSession()
+	logger.Infof("The juicefs mount process exit successfully, mountpoint: %s", metaConf.MountPoint)
+	return err
 }
diff --git a/cmd/status.go b/cmd/status.go
index 515f293f..b82ed00d 100644
--- a/cmd/status.go
+++ b/cmd/status.go
@@ -64,14 +64,14 @@ type sections struct {
 }
 
 type statistic struct {
-	UsedSpace         uint64
-	AvailableSpace    uint64
-	UsedInodes        uint64
-	AvailableInodes   uint64
-	DeletedSliceCount int64 `json:",omitempty"`
-	DeletedSliceSize  int64 `json:",omitempty"`
-	DeletedFileCount  int64 `json:",omitempty"`
-	DeletedFileSize   int64 `json:",omitempty"`
+	UsedSpace               uint64
+	AvailableSpace          uint64
+	UsedInodes              uint64
+	AvailableInodes         uint64
+	TrashSliceCount         int64 `json:",omitempty"`
+	TrashSliceSize          int64 `json:",omitempty"`
+	PendingDeletedFileCount int64 `json:",omitempty"`
+	PendingDeletedFileSize  int64 `json:",omitempty"`
 }
 
 func printJson(v interface{}) {
@@ -115,9 +115,9 @@ func status(ctx *cli.Context) error {
 
 	if ctx.Bool("more") {
 		progress := utils.NewProgress(false, false)
-		slicesSpinner := progress.AddDoubleSpinner("Delayed Slices")
+		slicesSpinner := progress.AddDoubleSpinner("Trash Slices")
 		defer slicesSpinner.Done()
-		fileSpinner := progress.AddDoubleSpinner("Delayed Files")
+		fileSpinner := progress.AddDoubleSpinner("Pending Deleted Files")
 		defer fileSpinner.Done()
 
 		err = m.ScanDeletedObject(
@@ -137,8 +137,8 @@ func status(ctx *cli.Context) error {
 			logger.Fatalf("statistic: %s", err)
 			return err
 		}
-		stat.DeletedSliceCount, stat.DeletedSliceSize = slicesSpinner.Current()
-		stat.DeletedFileCount, stat.DeletedFileSize = fileSpinner.Current()
+		stat.TrashSliceCount, stat.TrashSliceSize = slicesSpinner.Current()
+		stat.PendingDeletedFileCount, stat.PendingDeletedFileSize = fileSpinner.Current()
 	}
 
 	printJson(&sections{format, sessions, stat})
diff --git a/pkg/fuse/fuse.go b/pkg/fuse/fuse.go
index 86423adb..c3f32cb9 100644
--- a/pkg/fuse/fuse.go
+++ b/pkg/fuse/fuse.go
@@ -19,6 +19,7 @@ package fuse
 import (
 	"fmt"
 	"os"
+	"os/exec"
 	"runtime"
 	"strings"
 	"syscall"
@@ -470,6 +471,13 @@ func Serve(v *vfs.VFS, options string, xattrs, ioctl bool) error {
 	}
 	fssrv, err := fuse.NewServer(imp, conf.Meta.MountPoint, &opt)
 	if err != nil {
+		if execErr, ok := err.(*exec.Error); ok {
+			if pathErr, ok := execErr.Unwrap().(*os.PathError); ok &&
+				strings.Contains(pathErr.Path, "fusermount") &&
+				pathErr.Unwrap() == syscall.ENOENT {
+				return fmt.Errorf("fuse is not installed. Please install it first")
+			}
+		}
 		return fmt.Errorf("fuse: %s", err)
 	}
 	if runtime.GOOS == "linux" {
diff --git a/pkg/meta/base.go b/pkg/meta/base.go
index 81797ecb..7586ac91 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -64,7 +64,7 @@ type engine interface {
 	doFindDeletedFiles(ts int64, limit int) (map[Ino]uint64, error) // limit < 0 means all
 	doDeleteFileData(inode Ino, length uint64)
 	doCleanupSlices()
-	doCleanupDelayedSlices(edge int64, limit int) (int, error)
+	doCleanupDelayedSlices(edge int64) (int, error)
 	doDeleteSlice(id uint64, size uint32) error
 
 	doGetAttr(ctx Context, inode Ino, attr *Attr) syscall.Errno
@@ -1446,15 +1446,10 @@ func (m *baseMeta) cleanupDelayedSlices() {
 	now := time.Now()
 	edge := now.Unix() - int64(m.fmt.TrashDays)*24*3600
 	logger.Debugf("Cleanup delayed slices: started with edge %d", edge)
-	if count, err := m.en.doCleanupDelayedSlices(edge, 3e5); err == nil {
-		msg := fmt.Sprintf("Cleanup delayed slices: deleted %d slices in %v", count, time.Since(now))
-		if count >= 3e5 {
-			logger.Warnf("%s (reached max limit, stop)", msg)
-		} else {
-			logger.Debugf(msg)
-		}
-	} else {
+	if count, err := m.en.doCleanupDelayedSlices(edge); err != nil {
 		logger.Warnf("Cleanup delayed slices: deleted %d slices in %v, but got error: %s", count, time.Since(now), err)
+	} else if count > 0 {
+		logger.Infof("Cleanup delayed slices: deleted %d slices in %v", count, time.Since(now))
 	}
 }
 
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index 765421d8..3c69d9ee 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -2437,8 +2437,9 @@ func (m *redisMeta) doDeleteFileData_(inode Ino, length uint64, tracking string)
 	_ = m.rdb.ZRem(ctx, m.delfiles(), tracking)
 }
 
-func (r *redisMeta) doCleanupDelayedSlices(edge int64, limit int) (int, error) {
+func (r *redisMeta) doCleanupDelayedSlices(edge int64) (int, error) {
 	ctx := Background
+	start := time.Now()
 	stop := fmt.Errorf("reach limit")
 	var count int
 	var ss []Slice
@@ -2489,9 +2490,9 @@ func (r *redisMeta) doCleanupDelayedSlices(edge int64, limit int) (int, error) {
 					count++
 				}
 			}
-			if count >= limit {
-				return stop
-			}
+		}
+		if time.Since(start) > 50*time.Minute {
+			return stop
 		}
 		return nil
 	})
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index cccc075f..970a7a72 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -2328,54 +2328,58 @@ func (m *dbMeta) doDeleteFileData(inode Ino, length uint64) {
 	})
 }
 
-func (m *dbMeta) doCleanupDelayedSlices(edge int64, limit int) (int, error) {
-	var result []delslices
-	_ = m.roTxn(func(s *xorm.Session) error {
-		result = nil
-		return s.Where("deleted < ?", edge).Limit(limit, 0).Find(&result)
-	})
-
+func (m *dbMeta) doCleanupDelayedSlices(edge int64) (int, error) {
+	start := time.Now()
 	var count int
 	var ss []Slice
-	for _, ds := range result {
-		if err := m.txn(func(ses *xorm.Session) error {
-			ss = ss[:0]
-			ds := delslices{Id: ds.Id}
-			if ok, e := ses.ForUpdate().Get(&ds); e != nil {
-				return e
-			} else if !ok {
-				return nil
-			}
-			m.decodeDelayedSlices(ds.Slices, &ss)
-			if len(ss) == 0 {
-				return fmt.Errorf("invalid value for delayed slices %d: %v", ds.Id, ds.Slices)
-			}
-			for _, s := range ss {
-				if _, e := ses.Exec("update jfs_chunk_ref set refs=refs-1 where chunkid=? and size=?", s.Id, s.Size); e != nil {
+	var result []delslices
+	var batch int = 1e6
+	for {
+		_ = m.roTxn(func(s *xorm.Session) error {
+			result = result[:0]
+			return s.Where("deleted < ?", edge).Limit(batch, 0).Find(&result)
+		})
+
+		for _, ds := range result {
+			if err := m.txn(func(ses *xorm.Session) error {
+				ss = ss[:0]
+				ds := delslices{Id: ds.Id}
+				if ok, e := ses.ForUpdate().Get(&ds); e != nil {
 					return e
+				} else if !ok {
+					return nil
+				}
+				m.decodeDelayedSlices(ds.Slices, &ss)
+				if len(ss) == 0 {
+					return fmt.Errorf("invalid value for delayed slices %d: %v", ds.Id, ds.Slices)
+				}
+				for _, s := range ss {
+					if _, e := ses.Exec("update jfs_chunk_ref set refs=refs-1 where chunkid=? and size=?", s.Id, s.Size); e != nil {
+						return e
+					}
 				}
+				_, e := ses.Delete(&delslices{Id: ds.Id})
+				return e
+			}); err != nil {
+				logger.Warnf("Cleanup delayed slices %d: %s", ds.Id, err)
+				continue
 			}
-			_, e := ses.Delete(&delslices{Id: ds.Id})
-			return e
-		}); err != nil {
-			logger.Warnf("Cleanup delayed slices %d: %s", ds.Id, err)
-			continue
-		}
-		for _, s := range ss {
-			var ref = sliceRef{Id: s.Id}
-			err := m.roTxn(func(s *xorm.Session) error {
-				ok, err := s.Get(&ref)
-				if err == nil && !ok {
-					err = errors.New("not found")
+			for _, s := range ss {
+				var ref = sliceRef{Id: s.Id}
+				err := m.roTxn(func(s *xorm.Session) error {
+					ok, err := s.Get(&ref)
+					if err == nil && !ok {
+						err = errors.New("not found")
+					}
+					return err
+				})
+				if err == nil && ref.Refs <= 0 {
+					m.deleteSlice(s.Id, s.Size)
+					count++
 				}
-				return err
-			})
-			if err == nil && ref.Refs <= 0 {
-				m.deleteSlice(s.Id, s.Size)
-				count++
 			}
 		}
-		if count >= limit {
+		if len(result) < batch || time.Since(start) > 50*time.Minute {
 			break
 		}
 	}
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index fbfd0416..bb3766b1 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -43,6 +43,7 @@ type kvTxn interface {
 	gets(keys ...[]byte) [][]byte
 	scanRange(begin, end []byte) map[string][]byte
 	scanKeys(prefix []byte) [][]byte
+	scanKeysRange(begin, end []byte, limit int, filter func(k []byte) bool) [][]byte
 	scanValues(prefix []byte, limit int, filter func(k, v []byte) bool) map[string][]byte
 	exist(prefix []byte) bool
 	set(key, value []byte)
@@ -1969,52 +1970,52 @@ func (m *kvMeta) doDeleteFileData(inode Ino, length uint64) {
 	_ = m.deleteKeys(m.delfileKey(inode, length))
 }
 
-func (m *kvMeta) doCleanupDelayedSlices(edge int64, limit int) (int, error) {
-	var keys [][]byte
-	if err := m.client.txn(func(tx kvTxn) error {
-		vals := tx.scanRange(m.delSliceKey(0, 0), m.delSliceKey(edge, 0))
-		for k := range vals {
-			if len(k) != 1+8+8 { // delayed slices: Lttttttttcccccccc
-				continue
-			}
-			keys = append(keys, []byte(k))
-		}
-		return nil
-	}); err != nil {
-		logger.Warnf("Scan delayed slices: %s", err)
-		return 0, err
-	}
-
+func (m *kvMeta) doCleanupDelayedSlices(edge int64) (int, error) {
+	start := time.Now()
 	var count int
 	var ss []Slice
 	var rs []int64
-	for _, key := range keys {
-		if err := m.txn(func(tx kvTxn) error {
-			ss, rs = ss[:0], rs[:0]
-			buf := tx.get(key)
-			if len(buf) == 0 {
-				return nil
-			}
-			m.decodeDelayedSlices(buf, &ss)
-			if len(ss) == 0 {
-				return fmt.Errorf("invalid value for delayed slices %s: %v", key, buf)
-			}
-			for _, s := range ss {
-				rs = append(rs, tx.incrBy(m.sliceKey(s.Id, s.Size), -1))
-			}
-			tx.dels(key)
+	var keys [][]byte
+	var batch int = 1e5
+	for {
+		if err := m.client.txn(func(tx kvTxn) error {
+			keys = tx.scanKeysRange(m.delSliceKey(0, 0), m.delSliceKey(edge, 0), batch, func(k []byte) bool {
+				return len(k) == 1+8+8 // delayed slices: Lttttttttcccccccc
+			})
 			return nil
 		}); err != nil {
-			logger.Warnf("Cleanup delayed slices %s: %s", key, err)
-			continue
+			logger.Warnf("Scan delayed slices: %s", err)
+			return count, err
 		}
-		for i, s := range ss {
-			if rs[i] < 0 {
-				m.deleteSlice(s.Id, s.Size)
-				count++
+
+		for _, key := range keys {
+			if err := m.txn(func(tx kvTxn) error {
+				ss, rs = ss[:0], rs[:0]
+				buf := tx.get(key)
+				if len(buf) == 0 {
+					return nil
+				}
+				m.decodeDelayedSlices(buf, &ss)
+				if len(ss) == 0 {
+					return fmt.Errorf("invalid value for delayed slices %s: %v", key, buf)
+				}
+				for _, s := range ss {
+					rs = append(rs, tx.incrBy(m.sliceKey(s.Id, s.Size), -1))
+				}
+				tx.dels(key)
+				return nil
+			}); err != nil {
+				logger.Warnf("Cleanup delayed slices %s: %s", key, err)
+				continue
+			}
+			for i, s := range ss {
+				if rs[i] < 0 {
+					m.deleteSlice(s.Id, s.Size)
+					count++
+				}
 			}
 		}
-		if count >= limit {
+		if len(keys) < batch || time.Since(start) > 50*time.Minute {
 			break
 		}
 	}
diff --git a/pkg/meta/tkv_badger.go b/pkg/meta/tkv_badger.go
index 9a020576..79222f50 100644
--- a/pkg/meta/tkv_badger.go
+++ b/pkg/meta/tkv_badger.go
@@ -92,6 +92,30 @@ func (tx *badgerTxn) scanKeys(prefix []byte) [][]byte {
 	return ret
 }
 
+func (tx *badgerTxn) scanKeysRange(begin, end []byte, limit int, filter func(k []byte) bool) [][]byte {
+	if limit == 0 {
+		return nil
+	}
+	it := tx.t.NewIterator(badger.IteratorOptions{
+		PrefetchValues: true,
+		PrefetchSize:   1024,
+	})
+	defer it.Close()
+	var ret [][]byte
+	for it.Rewind(); it.Valid(); it.Next() {
+		key := it.Item().KeyCopy(nil)
+		if filter == nil || filter(key) {
+			ret = append(ret, key)
+			if limit > 0 {
+				if limit--; limit == 0 {
+					break
+				}
+			}
+		}
+	}
+	return ret
+}
+
 func (tx *badgerTxn) scanValues(prefix []byte, limit int, filter func(k, v []byte) bool) map[string][]byte {
 	if limit == 0 {
 		return nil
diff --git a/pkg/meta/tkv_etcd.go b/pkg/meta/tkv_etcd.go
index 383961d8..0a24b655 100644
--- a/pkg/meta/tkv_etcd.go
+++ b/pkg/meta/tkv_etcd.go
@@ -141,6 +141,28 @@ func (tx *etcdTxn) scanKeys(prefix []byte) [][]byte {
 	return keys
 }
 
+func (tx *etcdTxn) scanKeysRange(begin_, end_ []byte, limit int, filter func(k []byte) bool) [][]byte {
+	if limit == 0 {
+		return nil
+	}
+	resp, err := tx.kv.Get(tx.ctx, string(begin_), etcd.WithRange(string(end_)), etcd.WithKeysOnly())
+	if err != nil {
+		panic(fmt.Errorf("get range [%v-%v): %s", string(begin_), string(end_), err))
+	}
+	var keys [][]byte
+	for _, kv := range resp.Kvs {
+		if filter == nil || filter(kv.Key) {
+			k := string(kv.Key)
+			tx.observed[k] = kv.ModRevision
+			keys = append(keys, kv.Key)
+			if limit > 0 && len(keys) >= limit {
+				break
+			}
+		}
+	}
+	return keys
+}
+
 func (tx *etcdTxn) scanValues(prefix []byte, limit int, filter func(k, v []byte) bool) map[string][]byte {
 	if limit == 0 {
 		return nil
diff --git a/pkg/meta/tkv_fdb.go b/pkg/meta/tkv_fdb.go
index e81eb580..4ad05609 100644
--- a/pkg/meta/tkv_fdb.go
+++ b/pkg/meta/tkv_fdb.go
@@ -160,6 +160,23 @@ func (tx *fdbTxn) scanKeys(prefix []byte) [][]byte {
 	return ret
 }
 
+func (tx *fdbTxn) scanKeysRange(begin, end []byte, limit int, filter func(k []byte) bool) [][]byte {
+	ret := make([][]byte, 0)
+	iter := tx.range0(begin, end)
+	for iter.Advance() {
+		r := iter.MustGet()
+		if filter == nil || filter(r.Key) {
+			ret = append(ret, r.Key)
+			if limit > 0 {
+				if limit--; limit == 0 {
+					break
+				}
+			}
+		}
+	}
+	return ret
+}
+
 func (tx *fdbTxn) scanValues(prefix []byte, limit int, filter func(k, v []byte) bool) map[string][]byte {
 	ret := make(map[string][]byte)
 	iter := tx.range0(prefix, nextKey(prefix))
diff --git a/pkg/meta/tkv_mem.go b/pkg/meta/tkv_mem.go
index 6afcf611..3d08238d 100644
--- a/pkg/meta/tkv_mem.go
+++ b/pkg/meta/tkv_mem.go
@@ -132,6 +132,34 @@ func (tx *memTxn) scanKeys(prefix_ []byte) [][]byte {
 	return keys
 }
 
+func (tx *memTxn) scanKeysRange(begin_, end_ []byte, limit int, filter func(k []byte) bool) [][]byte {
+	if limit == 0 {
+		return nil
+	}
+	tx.store.Lock()
+	defer tx.store.Unlock()
+	begin := string(begin_)
+	end := string(end_)
+	var keys [][]byte
+	tx.store.items.AscendGreaterOrEqual(&kvItem{key: begin}, func(i btree.Item) bool {
+		it := i.(*kvItem)
+		if end == "" || it.key < end {
+			if filter == nil || filter([]byte(it.key)) {
+				tx.observed[it.key] = it.ver
+				keys = append(keys, []byte(it.key))
+				if limit > 0 {
+					if limit--; limit == 0 {
+						return false
+					}
+				}
+			}
+			return true
+		}
+		return false
+	})
+	return keys
+}
+
 func (tx *memTxn) scanValues(prefix []byte, limit int, filter func(k, v []byte) bool) map[string][]byte {
 	if limit == 0 {
 		return nil
diff --git a/pkg/meta/tkv_prefix.go b/pkg/meta/tkv_prefix.go
index 9ee06398..ee0560b2 100644
--- a/pkg/meta/tkv_prefix.go
+++ b/pkg/meta/tkv_prefix.go
@@ -62,6 +62,19 @@ func (tx *prefixTxn) scanKeys(prefix []byte) [][]byte {
 	return keys
 }
 
+func (tx *prefixTxn) scanKeysRange(begin, end []byte, limit int, filter func(k []byte) bool) [][]byte {
+	keys := tx.kvTxn.scanKeysRange(tx.realKey(begin), tx.realKey(end), limit, func(k []byte) bool {
+		if filter == nil {
+			return true
+		}
+		return filter(tx.origKey(k))
+	})
+	for i, k := range keys {
+		keys[i] = tx.origKey(k)
+	}
+	return keys
+}
+
 func (tx *prefixTxn) scanValues(prefix []byte, limit int, filter func(k, v []byte) bool) map[string][]byte {
 	r := tx.kvTxn.scanValues(tx.realKey(prefix), limit, func(k, v []byte) bool {
 		if filter == nil {
diff --git a/pkg/meta/tkv_tikv.go b/pkg/meta/tkv_tikv.go
index ea9cd56c..a453e487 100644
--- a/pkg/meta/tkv_tikv.go
+++ b/pkg/meta/tkv_tikv.go
@@ -22,6 +22,7 @@ package meta
 import (
 	"context"
 	"net/url"
+	"os"
 	"strings"
 
 	plog "github.com/pingcap/log"
@@ -32,6 +33,7 @@ import (
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/client-go/v2/txnkv"
 	"github.com/tikv/client-go/v2/txnkv/txnutil"
+	"go.uber.org/zap"
 )
 
 func init() {
@@ -54,7 +56,7 @@ func newTikvClient(addr string) (tkvClient, error) {
 	default:
 		plvl = "dpanic"
 	}
-	l, prop, _ := plog.InitLogger(&plog.Config{Level: plvl})
+	l, prop, _ := plog.InitLogger(&plog.Config{Level: plvl}, zap.Fields(zap.String("component", "tikv"), zap.Int("pid", os.Getpid())))
 	plog.ReplaceGlobals(l, prop)
 
 	tUrl, err := url.Parse("tikv://" + addr)
@@ -153,6 +155,33 @@ func (tx *tikvTxn) scanKeys(prefix []byte) [][]byte {
 	return ret
 }
 
+func (tx *tikvTxn) scanKeysRange(begin, end []byte, limit int, filter func(k []byte) bool) [][]byte {
+	if limit == 0 {
+		return nil
+	}
+	it, err := tx.Iter(begin, end)
+	if err != nil {
+		panic(err)
+	}
+	defer it.Close()
+	var ret [][]byte
+	for it.Valid() {
+		key := it.Key()
+		if filter == nil || filter(key) {
+			ret = append(ret, it.Key())
+			if limit > 0 {
+				if limit--; limit == 0 {
+					break
+				}
+			}
+		}
+		if err = it.Next(); err != nil {
+			panic(err)
+		}
+	}
+	return ret
+}
+
 func (tx *tikvTxn) scanValues(prefix []byte, limit int, filter func(k, v []byte) bool) map[string][]byte {
 	return tx.scanRange0(prefix, nextKey(prefix), limit, filter)
 }


Test patch:
diff --git a/pkg/meta/base_test.go b/pkg/meta/base_test.go
index 8e2ab054..b98d0f04 100644
--- a/pkg/meta/base_test.go
+++ b/pkg/meta/base_test.go
@@ -999,7 +999,7 @@ func testCompaction(t *testing.T, m Meta, trash bool) {
 		if len(sliceMap[1]) < 200 {
 			t.Fatalf("list delayed slices %d is less than 200", len(sliceMap[1]))
 		}
-		m.(engine).doCleanupDelayedSlices(time.Now().Unix()+1, 1000)
+		m.(engine).doCleanupDelayedSlices(time.Now().Unix() + 1)
 		l.Lock()
 		deletes = len(deleted)
 		l.Unlock()
diff --git a/pkg/meta/tkv_test.go b/pkg/meta/tkv_test.go
index 6d18a844..125f6036 100644
--- a/pkg/meta/tkv_test.go
+++ b/pkg/meta/tkv_test.go
@@ -104,6 +104,12 @@ func testTKV(t *testing.T, c tkvClient) {
 	if len(keys) != 2 || string(keys[0]) != "k" || string(keys[1]) != "k2" {
 		t.Fatalf("keys: %+v", keys)
 	}
+	txn(func(kt kvTxn) {
+		keys = kt.scanKeysRange([]byte("a"), []byte("z"), -1, func(k []byte) bool { return len(k) == 1 })
+	})
+	if len(keys) != 2 || string(keys[0]) != "k" || string(keys[1]) != "v" {
+		t.Fatalf("keys: %+v", keys)
+	}
 	txn(func(kt kvTxn) { keys = kt.scanKeys(nil) })
 	if len(keys) != 3 || string(keys[0]) != "k" || string(keys[1]) != "k2" || string(keys[2]) != "v" {
 		t.Fatalf("keys: %+v", keys)