BackupShard: support incremental backup

go/cmd/vtctldclient/command/backups.go

@@ -44,7 +44,7 @@ var (
 	}
 	// BackupShard makes a BackupShard gRPC call to a vtctld.
 	BackupShard = &cobra.Command{
-		Use:   "BackupShard [--concurrency <concurrency>] [--allow-primary] [--upgrade-safe] <keyspace/shard>",
+		Use:   "BackupShard [--concurrency <concurrency>] [--allow-primary] [--incremental-from-pos=<pos>|auto] [--upgrade-safe] <keyspace/shard>",
 		Short: "Finds the most up-to-date REPLICA, RDONLY, or SPARE tablet in the given shard and uses the BackupStorage service on that tablet to create and store a new backup.",
 		Long: `Finds the most up-to-date REPLICA, RDONLY, or SPARE tablet in the given shard and uses the BackupStorage service on that tablet to create and store a new backup.
 
@@ -119,9 +119,10 @@ func commandBackup(cmd *cobra.Command, args []string) error {
 }
 
 var backupShardOptions = struct {
-	AllowPrimary bool
-	Concurrency  uint64
-	UpgradeSafe  bool
+	AllowPrimary       bool
+	Concurrency        uint64
+	IncrementalFromPos string
+	UpgradeSafe        bool
 }{}
 
 func commandBackupShard(cmd *cobra.Command, args []string) error {
@@ -133,11 +134,12 @@ func commandBackupShard(cmd *cobra.Command, args []string) error {
 	cli.FinishedParsing(cmd)
 
 	stream, err := client.BackupShard(commandCtx, &vtctldatapb.BackupShardRequest{
-		Keyspace:     keyspace,
-		Shard:        shard,
-		AllowPrimary: backupOptions.AllowPrimary,
-		Concurrency:  backupOptions.Concurrency,
-		UpgradeSafe:  backupOptions.UpgradeSafe,
+		Keyspace:           keyspace,
+		Shard:              shard,
+		AllowPrimary:       backupOptions.AllowPrimary,
+		Concurrency:        backupOptions.Concurrency,
+		IncrementalFromPos: backupShardOptions.IncrementalFromPos,
+		UpgradeSafe:        backupOptions.UpgradeSafe,
 	})
 	if err != nil {
 		return err
@@ -287,6 +289,7 @@ func init() {
 
 	BackupShard.Flags().BoolVar(&backupShardOptions.AllowPrimary, "allow-primary", false, "Allow the primary of a shard to be used for the backup. WARNING: If using the builtin backup engine, this will shutdown mysqld on the primary and stop writes for the duration of the backup.")
 	BackupShard.Flags().Uint64Var(&backupShardOptions.Concurrency, "concurrency", 4, "Specifies the number of compression/checksum jobs to run simultaneously.")
+	BackupShard.Flags().StringVar(&backupShardOptions.IncrementalFromPos, "incremental-from-pos", "", "Position of previous backup. Default: empty. If given, then this backup becomes an incremental backup from given position. If value is 'auto', backup taken from last successful backup position")
 	BackupShard.Flags().BoolVar(&backupOptions.UpgradeSafe, "upgrade-safe", false, "Whether to use innodb_fast_shutdown=0 for the backup so it is safe to use for MySQL upgrades.")
 	Root.AddCommand(BackupShard)
 

go/vt/vtctl/backup.go

@@ -114,6 +114,7 @@ func (b *backupEventStreamLogger) Send(resp *vtctldatapb.BackupResponse) error {
 func commandBackupShard(ctx context.Context, wr *wrangler.Wrangler, subFlags *pflag.FlagSet, args []string) error {
 	concurrency := subFlags.Int("concurrency", 4, "Specifies the number of compression/checksum jobs to run simultaneously")
 	allowPrimary := subFlags.Bool("allow_primary", false, "Whether to use primary tablet for backup. Warning!! If you are using the builtin backup engine, this will shutdown your primary mysql for as long as it takes to create a backup.")
+	incrementalFromPos := subFlags.String("incremental_from_pos", "", "Position of previous backup. Default: empty. If given, then this backup becomes an incremental backup from given position. If value is 'auto', backup taken from last successful backup position")
 	upgradeSafe := subFlags.Bool("upgrade-safe", false, "Whether to use innodb_fast_shutdown=0 for the backup so it is safe to use for MySQL upgrades.")
 
 	if err := subFlags.Parse(args); err != nil {
@@ -129,11 +130,12 @@ func commandBackupShard(ctx context.Context, wr *wrangler.Wrangler, subFlags *pf
 	}
 
 	return wr.VtctldServer().BackupShard(&vtctldatapb.BackupShardRequest{
-		Keyspace:     keyspace,
-		Shard:        shard,
-		Concurrency:  uint64(*concurrency),
-		AllowPrimary: *allowPrimary,
-		UpgradeSafe:  *upgradeSafe,
+		Keyspace:           keyspace,
+		Shard:              shard,
+		Concurrency:        uint64(*concurrency),
+		AllowPrimary:       *allowPrimary,
+		IncrementalFromPos: *incrementalFromPos,
+		UpgradeSafe:        *upgradeSafe,
 	}, &backupEventStreamLogger{logger: wr.Logger(), ctx: ctx})
 }
 

go/vt/vtctl/grpcvtctldserver/server.go

@@ -417,6 +417,7 @@ func (s *VtctldServer) BackupShard(req *vtctldatapb.BackupShardRequest, stream v
 	span.Annotate("shard", req.Shard)
 	span.Annotate("allow_primary", req.AllowPrimary)
 	span.Annotate("concurrency", req.Concurrency)
+	span.Annotate("incremental_from_pos", req.IncrementalFromPos)
 
 	tablets, stats, err := reparentutil.ShardReplicationStatuses(ctx, s.ts, s.tmc, req.Keyspace, req.Shard)
 	if err != nil {
@@ -459,7 +460,7 @@ func (s *VtctldServer) BackupShard(req *vtctldatapb.BackupShardRequest, stream v
 
 	span.Annotate("tablet_alias", topoproto.TabletAliasString(backupTablet.Alias))
 
-	r := &vtctldatapb.BackupRequest{Concurrency: req.Concurrency, AllowPrimary: req.AllowPrimary, UpgradeSafe: req.UpgradeSafe}
+	r := &vtctldatapb.BackupRequest{Concurrency: req.Concurrency, AllowPrimary: req.AllowPrimary, UpgradeSafe: req.UpgradeSafe, IncrementalFromPos: req.IncrementalFromPos}
 	err = s.backupTablet(ctx, backupTablet, r, stream)
 	return err
 }

go/vt/vtctl/grpcvtctldserver/server_test.go

@@ -920,6 +920,72 @@ func TestBackupShard(t *testing.T) {
 				assert.Equal(t, 3, len(responses), "expected 3 messages from backupclient stream")
 			},
 		},
+		{
+			name: "incremental-from-pos",
+			ts:   memorytopo.NewServer("zone1"),
+			tmc: &testutil.TabletManagerClient{
+				Backups: map[string]struct {
+					Events        []*logutilpb.Event
+					EventInterval time.Duration
+					EventJitter   time.Duration
+					ErrorAfter    time.Duration
+				}{
+					"zone1-0000000100": {
+						Events: []*logutilpb.Event{{}, {}, {}},
+					},
+				},
+				PrimaryPositionResults: map[string]struct {
+					Position string
+					Error    error
+				}{
+					"zone1-0000000200": {
+						Position: "some-position",
+					},
+				},
+				ReplicationStatusResults: map[string]struct {
+					Position *replicationdatapb.Status
+					Error    error
+				}{
+					"zone1-0000000100": {
+						Position: &replicationdatapb.Status{
+							ReplicationLagSeconds: 0,
+						},
+					},
+				},
+				SetReplicationSourceResults: map[string]error{
+					"zone1-0000000100": nil,
+				},
+			},
+			tablets: []*topodatapb.Tablet{
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  100,
+					},
+					Keyspace: "ks",
+					Shard:    "-",
+					Type:     topodatapb.TabletType_REPLICA,
+				},
+				{
+					Alias: &topodatapb.TabletAlias{
+						Cell: "zone1",
+						Uid:  200,
+					},
+					Keyspace: "ks",
+					Shard:    "-",
+					Type:     topodatapb.TabletType_PRIMARY,
+				},
+			},
+			req: &vtctldatapb.BackupShardRequest{
+				Keyspace:           "ks",
+				Shard:              "-",
+				IncrementalFromPos: "auto",
+			},
+			assertion: func(t *testing.T, responses []*vtctldatapb.BackupResponse, err error) {
+				assert.ErrorIs(t, err, io.EOF, "expected Recv loop to end with io.EOF")
+				assert.Equal(t, 3, len(responses), "expected 3 messages from backupclient stream")
+			},
+		},
 		{
 			name: "no available tablet",
 			ts:   memorytopo.NewServer("zone1"),

proto/vtctldata.proto

@@ -304,6 +304,9 @@ message BackupShardRequest {
   // UpgradeSafe indicates if the backup should be taken with innodb_fast_shutdown=0
   // so that it's a backup that can be used for an upgrade.
   bool upgrade_safe = 5;
+  // IncrementalFromPos indicates a position of a previous backup. When this value is non-empty
+  // then the backup becomes incremental and applies as of given position.
+  string incremental_from_pos = 6;
 }
 
 message ChangeTabletTypeRequest {