log-backup: calculate global checkpoint-ts from storage

br/pkg/stream/stream_mgr.go

@@ -36,13 +36,19 @@ import (
 const (
 	streamBackupMetaPrefix = "v1/backupmeta"
 
+	streamBackupGlobalCheckpointPrefix = "v1/global_checkpoint"
+
 	metaDataWorkerPoolSize = 128
 )
 
 func GetStreamBackupMetaPrefix() string {
 	return streamBackupMetaPrefix
 }
 
+func GetStreamBackupGlobalCheckpointPrefix() string {
+	return streamBackupGlobalCheckpointPrefix
+}
+
 // appendTableObserveRanges builds key ranges corresponding to `tblIDS`.
 func appendTableObserveRanges(tblIDs []int64) []kv.KeyRange {
 	krs := make([]kv.KeyRange, 0, len(tblIDs))

br/pkg/task/stream.go

@@ -17,6 +17,7 @@ package task
 import (
 	"bytes"
 	"context"
+	"encoding/binary"
 	"fmt"
 	"net/http"
 	"strings"
@@ -1264,7 +1265,7 @@ func getLogRange(
 	logMinTS := mathutil.Max(logStartTS, truncateTS)
 
 	// get max global resolved ts from metas.
-	logMaxTS, err := getGlobalResolvedTS(ctx, s)
+	logMaxTS, err := getGlobalCheckpointFromStorage(ctx, s)
 	if err != nil {
 		return 0, 0, errors.Trace(err)
 	}
@@ -1273,6 +1274,25 @@ func getLogRange(
 	return logMinTS, logMaxTS, nil
 }
 
+func getGlobalCheckpointFromStorage(ctx context.Context, s storage.ExternalStorage) (uint64, error) {
+	var globalCheckPointTS uint64 = 0
+	opt := storage.WalkOption{SubDir: stream.GetStreamBackupGlobalCheckpointPrefix()}
+	err := s.WalkDir(ctx, &opt, func(path string, size int64) error {
+		if !strings.HasSuffix(path, ".ts") {
+			return nil
+		}
+
+		buff, err := s.ReadFile(ctx, path)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		ts := binary.LittleEndian.Uint64(buff)
+		globalCheckPointTS = mathutil.Max(ts, globalCheckPointTS)
+		return nil
+	})
+	return globalCheckPointTS, errors.Trace(err)
+}
+
 // getFullBackupTS gets the snapshot-ts of full bakcup
 func getFullBackupTS(
 	ctx context.Context,

br/pkg/task/stream_test.go

@@ -16,6 +16,7 @@ package task
 
 import (
 	"context"
+	"encoding/binary"
 	"fmt"
 	"path/filepath"
 	"testing"
@@ -194,3 +195,67 @@ func TestGetGlobalResolvedTS2(t *testing.T) {
 	require.Nil(t, err)
 	require.Equal(t, uint64(99), globalResolvedTS)
 }
+
+func fakeCheckpointFiles(
+	ctx context.Context,
+	tmpDir string,
+	infos []fakeGlobalCheckPoint,
+) error {
+	cpDir := filepath.Join(tmpDir, stream.GetStreamBackupGlobalCheckpointPrefix())
+	s, err := storage.NewLocalStorage(cpDir)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	// create normal files belong to global-checkpoint files
+	for _, info := range infos {
+		filename := fmt.Sprintf("%v.ts", info.storeID)
+		buff := make([]byte, 8)
+		binary.LittleEndian.PutUint64(buff, info.global_checkpoint)
+		if _, err := s.Create(ctx, filename); err != nil {
+			return errors.Trace(err)
+		}
+		if err := s.WriteFile(ctx, filename, buff); err != nil {
+			return errors.Trace(err)
+		}
+	}
+
+	// create a file not belonging to global-checkpoint-ts files
+	filename := fmt.Sprintf("%v.tst", 1)
+	err = s.WriteFile(ctx, filename, []byte("ping"))
+	return errors.AddStack(err)
+}
+
+type fakeGlobalCheckPoint struct {
+	storeID           int64
+	global_checkpoint uint64
+}
+
+func TestGetGlobalCheckpointFromStorage(t *testing.T) {
+	ctx := context.Background()
+	tmpdir := t.TempDir()
+	s, err := storage.NewLocalStorage(tmpdir)
+	require.Nil(t, err)
+
+	infos := []fakeGlobalCheckPoint{
+		{
+			storeID:           1,
+			global_checkpoint: 98,
+		},
+		{
+			storeID:           2,
+			global_checkpoint: 90,
+		},
+		{
+			storeID:           2,
+			global_checkpoint: 99,
+		},
+	}
+
+	err = fakeCheckpointFiles(ctx, tmpdir, infos)
+	require.Nil(t, err)
+
+	ts, err := getGlobalCheckpointFromStorage(ctx, s)
+	require.Nil(t, err)
+	require.Equal(t, ts, uint64(99))
+}