Merge branch 'master' into out2in

br/pkg/config/kv.go

@@ -56,3 +56,19 @@ func ParseMergeRegionSizeFromConfig(resp []byte) (uint64, uint64, error) {
 	urs := uint64(rs)
 	return urs, c.Cop.RegionSplitKeys, nil
 }
+
+func ParseLogBackupEnableFromConfig(resp []byte) (bool, error) {
+	type logbackup struct {
+		Enable bool `json:"enable"`
+	}
+
+	type config struct {
+		LogBackup logbackup `json:"log-backup"`
+	}
+	var c config
+	e := json.Unmarshal(resp, &c)
+	if e != nil {
+		return false, e
+	}
+	return c.LogBackup.Enable, nil
+}

br/pkg/conn/BUILD.bazel

@@ -46,7 +46,7 @@ go_test(
     ],
     embed = [":conn"],
     flaky = True,
-    shard_count = 7,
+    shard_count = 8,
     deps = [
         "//br/pkg/config",
         "//br/pkg/conn/util",

br/pkg/conn/conn.go

@@ -352,6 +352,25 @@ func (mgr *Mgr) ProcessTiKVConfigs(ctx context.Context, cfg *kvconfig.KVConfig,
 	}
 }
 
+// IsLogBackupEnabled is used for br to check whether tikv has enabled log backup.
+func (mgr *Mgr) IsLogBackupEnabled(ctx context.Context, client *http.Client) (bool, error) {
+	logbackupEnable := true
+	err := mgr.GetConfigFromTiKV(ctx, client, func(resp *http.Response) error {
+		respBytes, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+		enable, err := kvconfig.ParseLogBackupEnableFromConfig(respBytes)
+		if err != nil {
+			log.Warn("Failed to parse log-backup enable from config", logutil.ShortError(err))
+			return err
+		}
+		logbackupEnable = logbackupEnable && enable
+		return nil
+	})
+	return logbackupEnable, errors.Trace(err)
+}
+
 // GetConfigFromTiKV get configs from all alive tikv stores.
 func (mgr *Mgr) GetConfigFromTiKV(ctx context.Context, cli *http.Client, fn func(*http.Response) error) error {
 	allStores, err := GetAllTiKVStoresWithRetry(ctx, mgr.GetPDClient(), util.SkipTiFlash)

br/pkg/conn/conn_test.go

@@ -455,6 +455,169 @@ func TestGetMergeRegionSizeAndCount(t *testing.T) {
 	}
 }
 
+func TestIsLogBackupEnabled(t *testing.T) {
+	cases := []struct {
+		stores  []*metapb.Store
+		content []string
+		enable  bool
+		err     bool
+	}{
+		{
+			stores: []*metapb.Store{
+				{
+					Id:    1,
+					State: metapb.StoreState_Up,
+					Labels: []*metapb.StoreLabel{
+						{
+							Key:   "engine",
+							Value: "tiflash",
+						},
+					},
+				},
+			},
+			content: []string{""},
+			enable:  true,
+			err:     false,
+		},
+		{
+			stores: []*metapb.Store{
+				{
+					Id:    1,
+					State: metapb.StoreState_Up,
+					Labels: []*metapb.StoreLabel{
+						{
+							Key:   "engine",
+							Value: "tiflash",
+						},
+					},
+				},
+				{
+					Id:    2,
+					State: metapb.StoreState_Up,
+					Labels: []*metapb.StoreLabel{
+						{
+							Key:   "engine",
+							Value: "tikv",
+						},
+					},
+				},
+			},
+			content: []string{
+				"",
+				// Assuming the TiKV has failed due to some reason.
+				"",
+			},
+			enable: false,
+			err:    true,
+		},
+		{
+			stores: []*metapb.Store{
+				{
+					Id:    1,
+					State: metapb.StoreState_Up,
+					Labels: []*metapb.StoreLabel{
+						{
+							Key:   "engine",
+							Value: "tikv",
+						},
+					},
+				},
+			},
+			content: []string{
+				"{\"log-level\": \"debug\", \"log-backup\": {\"enable\": true}}",
+			},
+			enable: true,
+			err:    false,
+		},
+		{
+			stores: []*metapb.Store{
+				{
+					Id:    1,
+					State: metapb.StoreState_Up,
+					Labels: []*metapb.StoreLabel{
+						{
+							Key:   "engine",
+							Value: "tikv",
+						},
+					},
+				},
+			},
+			content: []string{
+				"{\"log-level\": \"debug\", \"log-backup\": {\"enable\": false}}",
+			},
+			enable: false,
+			err:    false,
+		},
+		{
+			stores: []*metapb.Store{
+				{
+					Id:    1,
+					State: metapb.StoreState_Up,
+					Labels: []*metapb.StoreLabel{
+						{
+							Key:   "engine",
+							Value: "tikv",
+						},
+					},
+				},
+				{
+					Id:    2,
+					State: metapb.StoreState_Up,
+					Labels: []*metapb.StoreLabel{
+						{
+							Key:   "engine",
+							Value: "tikv",
+						},
+					},
+				},
+			},
+			content: []string{
+				"{\"log-level\": \"debug\", \"log-backup\": {\"enable\": true}}",
+				"{\"log-level\": \"debug\", \"log-backup\": {\"enable\": false}}",
+			},
+			enable: false,
+			err:    false,
+		},
+	}
+
+	pctx := context.Background()
+	for _, ca := range cases {
+		ctx, cancel := context.WithCancel(pctx)
+		pdCli := utils.FakePDClient{Stores: ca.stores}
+		require.Equal(t, len(ca.content), len(ca.stores))
+		count := 0
+		mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch strings.TrimSpace(r.URL.Path) {
+			case "/config":
+				if len(ca.content[count]) == 0 {
+					cancel()
+				}
+				_, _ = fmt.Fprint(w, ca.content[count])
+			default:
+				http.NotFoundHandler().ServeHTTP(w, r)
+			}
+			count++
+		}))
+
+		for _, s := range ca.stores {
+			s.Address = mockServer.URL
+			s.StatusAddress = mockServer.URL
+		}
+
+		httpCli := mockServer.Client()
+		mgr := &Mgr{PdController: &pdutil.PdController{}}
+		mgr.PdController.SetPDClient(pdCli)
+		enable, err := mgr.IsLogBackupEnabled(ctx, httpCli)
+		if ca.err {
+			require.Error(t, err)
+		} else {
+			require.NoError(t, err)
+			require.Equal(t, ca.enable, enable)
+		}
+		mockServer.Close()
+	}
+}
+
 func TestHandleTiKVAddress(t *testing.T) {
 	cases := []struct {
 		store      *metapb.Store

br/pkg/stream/stream_status.go

@@ -344,6 +344,15 @@ func NewStatusController(meta *MetaDataClient, mgr PDInfoProvider, view TaskPrin
 	}
 }
 
+func (ctl *StatusController) Close() error {
+	if ctl.meta != nil {
+		if err := ctl.meta.Close(); err != nil {
+			return errors.Trace(err)
+		}
+	}
+	return nil
+}
+
 // fillTask queries and fills the extra information for a raw task.
 func (ctl *StatusController) fillTask(ctx context.Context, task Task) (TaskStatus, error) {
 	var err error

br/pkg/streamhelper/advancer.go

@@ -444,7 +444,7 @@ func (c *CheckpointAdvancer) onTaskEvent(ctx context.Context, e TaskEvent) error
 		if err := c.env.ClearV3GlobalCheckpointForTask(ctx, e.Name); err != nil {
 			log.Warn("failed to clear global checkpoint", logutil.ShortError(err))
 		}
-		if _, err := c.env.BlockGCUntil(ctx, 0); err != nil {
+		if err := c.env.UnblockGC(ctx); err != nil {
 			log.Warn("failed to remove service GC safepoint", logutil.ShortError(err))
 		}
 		metrics.LastCheckpoint.DeleteLabelValues(e.Name)

br/pkg/streamhelper/advancer_env.go

@@ -4,8 +4,10 @@ package streamhelper
 
 import (
 	"context"
+	"math"
 	"time"
 
+	"github.com/pingcap/errors"
 	logbackup "github.com/pingcap/kvproto/pkg/logbackuppb"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/pkg/config"
@@ -46,7 +48,21 @@ type PDRegionScanner struct {
 // Returns the minimal service GC safe point across all services.
 // If the arguments is `0`, this would remove the service safe point.
 func (c PDRegionScanner) BlockGCUntil(ctx context.Context, at uint64) (uint64, error) {
-	return c.UpdateServiceGCSafePoint(ctx, logBackupServiceID, int64(logBackupSafePointTTL.Seconds()), at)
+	minimalSafePoint, err := c.UpdateServiceGCSafePoint(
+		ctx, logBackupServiceID, int64(logBackupSafePointTTL.Seconds()), at)
+	if err != nil {
+		return 0, errors.Annotate(err, "failed to block gc until")
+	}
+	if minimalSafePoint > at {
+		return 0, errors.Errorf("minimal safe point %d is greater than the target %d", minimalSafePoint, at)
+	}
+	return at, nil
+}
+
+func (c PDRegionScanner) UnblockGC(ctx context.Context) error {
+	// set ttl to 0, means remove the safe point.
+	_, err := c.UpdateServiceGCSafePoint(ctx, logBackupServiceID, 0, math.MaxUint64)
+	return err
 }
 
 // TODO: It should be able to synchoronize the current TS with the PD.

br/pkg/streamhelper/advancer_test.go

@@ -208,7 +208,7 @@ func TestGCServiceSafePoint(t *testing.T) {
 	req.Eventually(func() bool {
 		env.fakeCluster.mu.Lock()
 		defer env.fakeCluster.mu.Unlock()
-		return env.serviceGCSafePoint == 0
+		return env.serviceGCSafePoint != 0 && env.serviceGCSafePointDeleted
 	}, 3*time.Second, 100*time.Millisecond)
 }
 

br/pkg/streamhelper/basic_lib_for_test.go

@@ -100,10 +100,11 @@ type fakeCluster struct {
 	regions   []*region
 	testCtx   *testing.T
 
-	onGetClient        func(uint64) error
-	onClearCache       func(uint64) error
-	serviceGCSafePoint uint64
-	currentTS          uint64
+	onGetClient               func(uint64) error
+	onClearCache              func(uint64) error
+	serviceGCSafePoint        uint64
+	serviceGCSafePointDeleted bool
+	currentTS                 uint64
 }
 
 func (r *region) splitAt(newID uint64, k string) *region {
@@ -264,17 +265,20 @@ func (f *fakeStore) GetLastFlushTSOfRegion(ctx context.Context, in *logbackup.Ge
 func (f *fakeCluster) BlockGCUntil(ctx context.Context, at uint64) (uint64, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	if at == 0 {
-		f.serviceGCSafePoint = at
-		return at, nil
-	}
 	if f.serviceGCSafePoint > at {
 		return f.serviceGCSafePoint, nil
 	}
 	f.serviceGCSafePoint = at
 	return at, nil
 }
 
+func (f *fakeCluster) UnblockGC(ctx context.Context) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.serviceGCSafePointDeleted = true
+	return nil
+}
+
 func (f *fakeCluster) FetchCurrentTS(ctx context.Context) (uint64, error) {
 	return f.currentTS, nil
 }

br/pkg/streamhelper/regioniter.go

@@ -43,6 +43,9 @@ type TiKVClusterMeta interface {
 	// For now, all tasks (exactly one task in fact) use the same checkpoint.
 	BlockGCUntil(ctx context.Context, at uint64) (uint64, error)
 
+	// UnblockGC used to remove the service GC safe point in PD.
+	UnblockGC(ctx context.Context) error
+
 	FetchCurrentTS(ctx context.Context) (uint64, error)
 }
 

br/pkg/streamhelper/regioniter_test.go

@@ -83,6 +83,10 @@ func (c constantRegions) BlockGCUntil(ctx context.Context, at uint64) (uint64, e
 	return 0, status.Error(codes.Unimplemented, "Unsupported operation")
 }
 
+func (c constantRegions) UnblockGC(ctx context.Context) error {
+	return status.Error(codes.Unimplemented, "Unsupported operation")
+}
+
 // TODO: It should be able to synchoronize the current TS with the PD.
 func (c constantRegions) FetchCurrentTS(ctx context.Context) (uint64, error) {
 	return oracle.ComposeTS(time.Now().UnixMilli(), 0), nil