log-backup: implement subscription of flush

br/pkg/streamhelper/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "advancer_env.go",
         "client.go",
         "collector.go",
+        "flush_subscriber.go",
         "models.go",
         "prefix_scanner.go",
         "regioniter.go",
@@ -26,6 +27,8 @@ go_library(
         "//kv",
         "//metrics",
         "//owner",
+        "//util/codec",
+        "//util/engine",
         "//util/mathutil",
         "@com_github_gogo_protobuf//proto",
         "@com_github_golang_protobuf//proto",
@@ -40,8 +43,11 @@ go_library(
         "@com_github_tikv_pd_client//:client",
         "@io_etcd_go_etcd_client_v3//:client",
         "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//keepalive",
+        "@org_golang_google_grpc//status",
         "@org_golang_x_sync//errgroup",
+        "@org_uber_go_multierr//:multierr",
         "@org_uber_go_zap//:zap",
     ],
 )
@@ -54,6 +60,7 @@ go_test(
         "basic_lib_for_test.go",
         "integration_test.go",
         "regioniter_test.go",
+        "subscription_test.go",
     ],
     flaky = True,
     race = "on",
@@ -64,11 +71,11 @@ go_test(
         "//br/pkg/logutil",
         "//br/pkg/redact",
         "//br/pkg/storage",
-        "//br/pkg/streamhelper/config",
         "//br/pkg/streamhelper/spans",
         "//br/pkg/utils",
         "//kv",
         "//tablecodec",
+        "//util/codec",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_kvproto//pkg/brpb",
         "@com_github_pingcap_kvproto//pkg/errorpb",
@@ -82,6 +89,7 @@ go_test(
         "@io_etcd_go_etcd_server_v3//mvcc",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//codes",
+        "@org_golang_google_grpc//metadata",
         "@org_golang_google_grpc//status",
         "@org_uber_go_zap//:zap",
         "@org_uber_go_zap//zapcore",

br/pkg/streamhelper/advancer.go

@@ -63,6 +63,9 @@ type CheckpointAdvancer struct {
 
 	checkpoints   *spans.ValueSortedFull
 	checkpointsMu sync.Mutex
+
+	subscriber   *FlushSubscriber
+	subscriberMu sync.Mutex
 }
 
 // NewCheckpointAdvancer creates a checkpoint advancer with the env.
@@ -105,7 +108,7 @@ func (c *CheckpointAdvancer) GetCheckpointInRange(ctx context.Context, start, en
 		}
 		log.Debug("scan region", zap.Int("len", len(rs)))
 		for _, r := range rs {
-			err := collector.collectRegion(r)
+			err := collector.CollectRegion(r)
 			if err != nil {
 				log.Warn("meet error during getting checkpoint", logutil.ShortError(err))
 				return err
@@ -135,7 +138,7 @@ func (c *CheckpointAdvancer) tryAdvance(ctx context.Context, length int, getRang
 	workers := utils.NewWorkerPool(uint(config.DefaultMaxConcurrencyAdvance)*4, "sub ranges")
 	eg, cx := errgroup.WithContext(ctx)
 	collector := NewClusterCollector(ctx, c.env)
-	collector.setOnSuccessHook(func(u uint64, kr kv.KeyRange) {
+	collector.SetOnSuccessHook(func(u uint64, kr kv.KeyRange) {
 		c.checkpointsMu.Lock()
 		defer c.checkpointsMu.Unlock()
 		c.checkpoints.Merge(spans.Valued{Key: kr, Value: u})
@@ -166,15 +169,16 @@ func tsoBefore(n time.Duration) uint64 {
 	return oracle.ComposeTS(now.UnixMilli()-n.Milliseconds(), 0)
 }
 
-// CalculateGlobalCheckpointLight tries to advance the global checkpoint by the cache.
-func (c *CheckpointAdvancer) CalculateGlobalCheckpointLight(ctx context.Context) (uint64, error) {
+func (c *CheckpointAdvancer) CalculateGlobalCheckpointLight(ctx context.Context, threshold time.Duration) (uint64, error) {
 	var targets []spans.Valued
-	c.checkpoints.TraverseValuesLessThan(tsoBefore(config.DefaultTryAdvanceThreshold), func(v spans.Valued) bool {
+	c.checkpoints.TraverseValuesLessThan(tsoBefore(threshold), func(v spans.Valued) bool {
 		targets = append(targets, v)
 		return true
 	})
 	if len(targets) == 0 {
-		return 0, nil
+		c.checkpointsMu.Lock()
+		defer c.checkpointsMu.Unlock()
+		return c.checkpoints.MinValue(), nil
 	}
 	samples := targets
 	if len(targets) > 3 {
@@ -188,7 +192,9 @@ func (c *CheckpointAdvancer) CalculateGlobalCheckpointLight(ctx context.Context)
 	if err != nil {
 		return 0, err
 	}
+	c.checkpointsMu.Lock()
 	ts := c.checkpoints.MinValue()
+	c.checkpointsMu.Unlock()
 	return ts, nil
 }
 
@@ -285,6 +291,8 @@ func (c *CheckpointAdvancer) onTaskEvent(ctx context.Context, e TaskEvent) error
 		c.task = nil
 		c.taskRange = nil
 		c.checkpoints = nil
+		// This would be synced by `taskMu`, perhaps we'd better rename that to `tickMu`.
+		c.subscriber.Clear()
 		if err := c.env.ClearV3GlobalCheckpointForTask(ctx, e.Name); err != nil {
 			log.Warn("failed to clear global checkpoint", logutil.ShortError(err))
 		}
@@ -323,14 +331,67 @@ func (c *CheckpointAdvancer) advanceCheckpointBy(ctx context.Context, getCheckpo
 	return nil
 }
 
+func (c *CheckpointAdvancer) stopSubscriber() {
+	c.subscriberMu.Lock()
+	defer c.subscriberMu.Unlock()
+	c.subscriber.Drop()
+	c.subscriber = nil
+}
+
+func (c *CheckpointAdvancer) spawnSubscriptionHandler(ctx context.Context) {
+	c.subscriberMu.Lock()
+	defer c.subscriberMu.Unlock()
+	c.subscriber = NewSubscriber(c.env, c.env, WithMasterContext(ctx))
+	es := c.subscriber.Events()
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case event, ok := <-es:
+				if !ok {
+					return
+				}
+				c.checkpointsMu.Lock()
+				log.Debug("Accepting region flush event.",
+					zap.Stringer("range", logutil.StringifyRange(event.Key)),
+					zap.Uint64("checkpoint", event.Value))
+				c.checkpoints.Merge(event)
+				c.checkpointsMu.Unlock()
+			}
+		}
+	}()
+}
+
+func (c *CheckpointAdvancer) subscribeTick(ctx context.Context) error {
+	if c.subscriber == nil {
+		return nil
+	}
+	if err := c.subscriber.UpdateStoreTopology(ctx); err != nil {
+		log.Warn("[log backup advancer] Error when updating store topology.", logutil.ShortError(err))
+	}
+	c.subscriber.HandleErrors(ctx)
+	return c.subscriber.PendingErrors()
+}
+
 func (c *CheckpointAdvancer) tick(ctx context.Context) error {
 	c.taskMu.Lock()
 	defer c.taskMu.Unlock()
 	if c.task == nil {
 		log.Debug("No tasks yet, skipping advancing.")
 		return nil
 	}
-	err := c.advanceCheckpointBy(ctx, c.CalculateGlobalCheckpointLight)
+
+	threshold := c.Config().GetDefaultStartPollThreshold()
+	if err := c.subscribeTick(ctx); err != nil {
+		log.Warn("[log backup advancer] Subscriber meet error, would polling the checkpoint.", logutil.ShortError(err))
+		threshold = c.Config().GetSubscriberErrorStartPollThreshold()
+	}
+
+	err := c.advanceCheckpointBy(ctx, func(ctx context.Context) (uint64, error) {
+		return c.CalculateGlobalCheckpointLight(ctx, threshold)
+	})
 	if err != nil {
 		return err
 	}

br/pkg/streamhelper/advancer_daemon.go

@@ -30,6 +30,7 @@ func (c *CheckpointAdvancer) OnTick(ctx context.Context) (err error) {
 func (c *CheckpointAdvancer) OnStart(ctx context.Context) {
 	metrics.AdvancerOwner.Set(1.0)
 	c.StartTaskListener(ctx)
+	c.spawnSubscriptionHandler(ctx)
 	go func() {
 		<-ctx.Done()
 		c.onStop()
@@ -43,6 +44,7 @@ func (c *CheckpointAdvancer) Name() string {
 
 func (c *CheckpointAdvancer) onStop() {
 	metrics.AdvancerOwner.Set(0.0)
+	c.stopSubscriber()
 }
 
 func OwnerManagerForLogBackup(ctx context.Context, etcdCli *clientv3.Client) owner.Manager {

br/pkg/streamhelper/advancer_env.go

@@ -9,6 +9,7 @@ import (
 	logbackup "github.com/pingcap/kvproto/pkg/logbackuppb"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/config"
+	"github.com/pingcap/tidb/util/engine"
 	pd "github.com/tikv/pd/client"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"google.golang.org/grpc"
@@ -18,7 +19,7 @@ import (
 // Env is the interface required by the advancer.
 type Env interface {
 	// The region scanner provides the region information.
-	RegionScanner
+	TiKVClusterMeta
 	// LogBackupService connects to the TiKV, so we can collect the region checkpoints.
 	LogBackupService
 	// StreamMeta connects to the metadata service (normally PD).
@@ -48,6 +49,23 @@ func (c PDRegionScanner) RegionScan(ctx context.Context, key []byte, endKey []by
 	return rls, nil
 }
 
+func (c PDRegionScanner) Stores(ctx context.Context) ([]Store, error) {
+	res, err := c.Client.GetAllStores(ctx, pd.WithExcludeTombstone())
+	if err != nil {
+		return nil, err
+	}
+	r := make([]Store, 0, len(res))
+	for _, re := range res {
+		if !engine.IsTiFlash(re) {
+			r = append(r, Store{
+				BootAt: uint64(re.StartTimestamp),
+				ID:     re.GetId(),
+			})
+		}
+	}
+	return r, nil
+}
+
 // clusterEnv is the environment for running in the real cluster.
 type clusterEnv struct {
 	clis *utils.StoreManager

br/pkg/streamhelper/advancer_test.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/br/pkg/streamhelper"
-	"github.com/pingcap/tidb/br/pkg/streamhelper/config"
 	"github.com/pingcap/tidb/kv"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zapcore"
@@ -51,9 +50,6 @@ func TestTick(t *testing.T) {
 	env := &testEnv{fakeCluster: c, testCtx: t}
 	adv := streamhelper.NewCheckpointAdvancer(env)
 	adv.StartTaskListener(ctx)
-	adv.UpdateConfigWith(func(cac *config.Config) {
-		cac.FullScanTick = 0
-	})
 	require.NoError(t, adv.OnTick(ctx))
 	for i := 0; i < 5; i++ {
 		cp := c.advanceCheckpoints()
@@ -76,9 +72,6 @@ func TestWithFailure(t *testing.T) {
 	env := &testEnv{fakeCluster: c, testCtx: t}
 	adv := streamhelper.NewCheckpointAdvancer(env)
 	adv.StartTaskListener(ctx)
-	adv.UpdateConfigWith(func(cac *config.Config) {
-		cac.FullScanTick = 0
-	})
 	require.NoError(t, adv.OnTick(ctx))
 
 	cp := c.advanceCheckpoints()