br: no domain to run log command (#52127)

br/pkg/conn/BUILD.bazel

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

br/pkg/conn/conn.go

@@ -327,6 +327,28 @@ func (mgr *Mgr) GetMergeRegionSizeAndCount(ctx context.Context, client *http.Cli
 	return regionSplitSize, regionSplitKeys
 }
 
+// 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
+	type logbackup struct {
+		Enable bool `json:"enable"`
+	}
+	type config struct {
+		LogBackup logbackup `json:"log-backup"`
+	}
+	err := mgr.GetConfigFromTiKV(ctx, client, func(resp *http.Response) error {
+		c := &config{}
+		e := json.NewDecoder(resp.Body).Decode(c)
+		if e != nil {
+			log.Warn("Failed to parse log-backup enable from config", logutil.ShortError(e))
+			return e
+		}
+		logbackupEnable = logbackupEnable && c.LogBackup.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

@@ -436,6 +436,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

@@ -342,6 +342,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