meta: support changelog replay

cmd/changelog.go

@@ -17,19 +17,24 @@
 package cmd
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"os/signal"
+	"syscall"
 
 	"github.com/juicedata/juicefs/pkg/meta"
 	"github.com/urfave/cli/v2"
 )
 
 func cmdChangelog() *cli.Command {
 	return &cli.Command{
-		Name:      "changelog",
-		Action:    changelog,
-		Category:  "INSPECTOR",
-		Usage:     "Tail the changelog of a volume",
-		ArgsUsage: "META-URL",
+		Name:            "changelog",
+		Action:          changelog,
+		Category:        "INSPECTOR",
+		Usage:           "Tail the changelog of a volume",
+		ArgsUsage:       "META-URL",
+		HideHelpCommand: true,
 		Description: `
 Show the changelog of metadata operations on the volume. This requires the changelog feature
 to be enabled via "juicefs config META-URL --changelog".
@@ -38,12 +43,48 @@ Examples:
 $ juicefs changelog redis://localhost
 
 # Start tailing from a specific version
-$ juicefs changelog redis://localhost --from 100`,
+$ juicefs changelog redis://localhost --from 100
+
+# Validate that every entry can be parsed, without writing to the destination
+$ juicefs changelog replay redis://src redis://dst --dry-run`,
+		Subcommands: []*cli.Command{
+			{
+				Name:      "replay",
+				Usage:     "Replay the changelog of a volume on another volume",
+				ArgsUsage: "SRC-META-URL DST-META-URL",
+				Action:    changelogReplay,
+				Description: `
+Replay the changelog of SRC-META-URL on DST-META-URL.
+
+Examples:
+# Start from the backup used to create the destination
+$ juicefs changelog replay redis://src redis://dst --backup /tmp/meta.bin
+
+# Start from a specific version
+$ juicefs changelog replay redis://src redis://dst --from 100
+
+# Only parse and validate entries
+$ juicefs changelog replay redis://src redis://dst --dry-run --from 100`,
+			},
+		},
 		Flags: []cli.Flag{
+			&cli.StringFlag{
+				Name:  "backup",
+				Usage: "path to the baseline JSON or binary backup (overrides --from)",
+			},
 			&cli.Int64Flag{
 				Name:  "from",
 				Usage: "show changelog from this version (0 means from the latest)",
 			},
+			&cli.BoolFlag{
+				Name:  "follow",
+				Value: true,
+				Usage: "keep waiting for new entries after catching up",
+			},
+			&cli.BoolFlag{
+				Name:  "dry-run",
+				Usage: "only parse and validate the changelog, do not write to the destination",
+			},
 		},
 	}
 }
@@ -60,9 +101,98 @@ func changelog(ctx *cli.Context) error {
 		return fmt.Errorf("changelog is not enabled, use `juicefs config %s --changelog` to enable it", metaUri)
 	}
 
-	last := ctx.Int64("from")
-	return m.ScanChangelog(meta.Background(), last, func(ver int64, entry string) error {
+	opt := &meta.ChangelogScanOption{From: ctx.Int64("from"), Follow: ctx.Bool("follow")}
+	return m.ScanChangelog(meta.WrapContext(ctx.Context), opt, func(ver int64, entry string) error {
 		fmt.Printf("%d: %s\n", ver, entry)
 		return nil
 	})
 }
+
+func changelogReplay(ctx *cli.Context) error {
+	setup(ctx, 2)
+	srcUri, dstUri := ctx.Args().Get(0), ctx.Args().Get(1)
+	removePassword(srcUri)
+	removePassword(dstUri)
+
+	if srcUri == dstUri {
+		return fmt.Errorf("source and destination metadata URLs must be different")
+	}
+	src := meta.NewClient(srcUri, nil)
+	srcFormat, err := src.Load(true)
+	if err != nil {
+		return err
+	}
+	if !srcFormat.ChangeLog {
+		return fmt.Errorf("changelog is not enabled on the source, use `juicefs config %s --changelog` to enable it", srcUri)
+	}
+	dryRun := ctx.Bool("dry-run")
+	replayer, err := meta.NewReplayer(dstUri, srcFormat, dryRun)
+	if err != nil {
+		return err
+	}
+	defer replayer.Close()
+	opt := &meta.ChangelogScanOption{From: ctx.Int64("from"), Follow: ctx.Bool("follow")}
+	if path := ctx.String("backup"); path != "" {
+		format, err := func() (*meta.Format, error) {
+			r, err := open(path, "", "")
+			if err != nil {
+				return nil, err
+			}
+			defer r.Close()
+			format, from, seen, err := meta.LoadChangelogBaseline(r)
+			if err != nil {
+				return nil, err
+			}
+			opt.From, opt.Seen = from, seen
+			return format, nil
+		}()
+		if err != nil {
+			return fmt.Errorf("read changelog baseline from backup: %w", err)
+		}
+		if format.UUID != srcFormat.UUID {
+			return fmt.Errorf("backup volume UUID %q does not match the source %q", format.UUID, srcFormat.UUID)
+		}
+		logger.Infof("Start from backup lastChangelog %d with %d rewind entries", opt.From, len(opt.Seen))
+	}
+
+	signalCtx, stop := signal.NotifyContext(ctx.Context, syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	logger.Infof("Start changelog replay: from=%d follow=%t dry-run=%t", opt.From, opt.Follow, dryRun)
+	replayCtx := meta.WrapContext(signalCtx)
+	lastVersion := opt.From
+	err = src.ScanChangelog(replayCtx, opt, func(ver int64, entry string) error {
+		if err := replayCtx.Err(); err != nil {
+			return err
+		}
+		logger.Debugf("Changelog entry: version=%d entry=%q", ver, entry)
+		e, err := meta.ParseChangeEntry(ver, entry)
+		op := "unknown"
+		if err == nil {
+			op = e.Op
+			err = e.Validate()
+		}
+		if err == nil && !dryRun {
+			err = replayer.Replay(replayCtx, e)
+		}
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				return err
+			}
+			logger.Errorf("Failed to replay changelog: version=%d operation=%s error=%s", ver, op, err)
+			return err
+		} else if dryRun {
+			logger.Infof("Validated changelog: version=%d operation=%s dry-run=true", ver, op)
+		}
+		lastVersion = ver
+		return nil
+	})
+	if errors.Is(err, context.Canceled) {
+		logger.Infof("Changelog replay stopped: last processed version=%d", lastVersion)
+		return nil
+	}
+	if err == nil {
+		logger.Infof("Changelog replay finished: last processed version=%d", lastVersion)
+	}
+	return err
+}

pkg/meta/backup.go

@@ -17,11 +17,13 @@
 package meta
 
 import (
+	"bufio"
 	"context"
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"
 
 	"github.com/juicedata/juicefs/pkg/meta/pb"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -189,6 +191,167 @@ func readBackupSource(r io.Reader) (pb.Footer_Engine, error) {
 	return footer.Msg.Source, nil
 }
 
+// LoadChangelogBaseline reads the volume format, lastChangelog and the changelog
+// versions included in a JSON or binary metadata backup. A binary backup that is
+// not seekable is copied to a temporary file.
+func LoadChangelogBaseline(r io.Reader) (*Format, int64, map[uint64]struct{}, error) {
+	reader := bufio.NewReader(r)
+	var first byte
+
+	for {
+		prefix, err := reader.Peek(1)
+		if err != nil {
+			return nil, 0, nil, err
+		}
+		first = prefix[0]
+		if first != ' ' && first != '\t' && first != '\r' && first != '\n' {
+			break
+		}
+		_, _ = reader.Discard(1)
+	}
+
+	var format Format
+	var from int64
+	var err error
+	seen := make(map[uint64]struct{})
+	if first == '{' {
+		from, err = readJSONBaseline(reader, &format, seen)
+	} else {
+		rs, ok := r.(io.ReadSeeker)
+		if !ok {
+			fp, err := os.CreateTemp("", "juicefs-backup-*")
+			if err != nil {
+				return nil, 0, nil, err
+			}
+			defer os.Remove(fp.Name())
+			defer fp.Close()
+			if _, err := io.Copy(fp, reader); err != nil {
+				return nil, 0, nil, fmt.Errorf("copy backup: %w", err)
+			}
+			rs = fp
+		}
+		from, err = readBinaryBaseline(rs, &format, seen)
+	}
+	if err != nil {
+		return nil, 0, nil, err
+	}
+	if format.UUID == "" {
+		return nil, 0, nil, fmt.Errorf("backup does not contain a volume UUID")
+	}
+	if from <= 0 {
+		return nil, 0, nil, fmt.Errorf("backup does not contain a positive lastChangelog")
+	}
+	for id := range seen {
+		if id == 0 || id > uint64(from) {
+			return nil, 0, nil, fmt.Errorf("backup changelog %d is outside the baseline ending at %d", int64(id), from)
+		}
+	}
+	return &format, from, seen, nil
+}
+
+func readJSONBaseline(r io.Reader, format *Format, seen map[uint64]struct{}) (int64, error) {
+	dec := json.NewDecoder(r)
+	dec.UseNumber()
+	if _, err := dec.Token(); err != nil {
+		return 0, err
+	}
+	var from int64
+	for dec.More() {
+		name, err := dec.Token()
+		if err != nil {
+			return 0, err
+		}
+		switch name {
+		case "Setting":
+			err = dec.Decode(format)
+		case "Counters":
+			var counters DumpedCounters
+			err = dec.Decode(&counters)
+			from = counters.LastChangelog
+		case "ChangeLog":
+			var logs []struct{ Version int64 }
+			err = dec.Decode(&logs)
+			for _, log := range logs {
+				seen[uint64(log.Version)] = struct{}{}
+			}
+		case "FSTree", "Trash":
+			// Dump writes the file trees after all other fields.
+			return from, nil
+		default:
+			err = skipJSONValue(dec)
+		}
+		if err != nil {
+			return 0, fmt.Errorf("read backup field %s: %w", name, err)
+		}
+	}
+	_, err := dec.Token()
+	return from, err
+}
+
+// skipJSONValue skips the next value without retaining it in memory.
+func skipJSONValue(dec *json.Decoder) error {
+	depth := 0
+	for {
+		token, err := dec.Token()
+		if err != nil {
+			return err
+		}
+		switch token {
+		case json.Delim('{'), json.Delim('['):
+			depth++
+		case json.Delim('}'), json.Delim(']'):
+			depth--
+		}
+		if depth == 0 {
+			return nil
+		}
+	}
+}
+
+func readBinaryBaseline(r io.ReadSeeker, format *Format, seen map[uint64]struct{}) (int64, error) {
+	bak := &BakFormat{}
+	footer, err := bak.ReadFooter(r)
+	if err != nil {
+		return 0, err
+	}
+	var from int64
+	for _, name := range []string{"format", "counter", "changeLog"} {
+		info := footer.Msg.Infos[name]
+		if info == nil {
+			continue
+		}
+		for _, offset := range info.Offset {
+			if _, err := r.Seek(int64(offset), io.SeekStart); err != nil {
+				return 0, err
+			}
+			seg, err := bak.ReadSegment(r)
+			if err != nil {
+				return 0, err
+			}
+			if seg.Name() != name {
+				return 0, fmt.Errorf("expected backup segment %s, got %s", name, seg.Name())
+			}
+			switch seg.typ {
+			case segTypeFormat:
+				if err := json.Unmarshal(seg.val.(*pb.Format).Data, format); err != nil {
+					return 0, err
+				}
+			case segTypeCounter:
+				for _, counter := range seg.val.(*pb.Batch).Counters {
+					if counter.Key == "lastChangelog" {
+						from = counter.Value
+					}
+				}
+			case segTypeChangeLog:
+				for _, log := range seg.val.(*pb.Batch).Changelogs {
+					seen[uint64(log.Version)] = struct{}{}
+				}
+			}
+		}
+	}
+	return from, nil
+}
+
 type BakFooter struct {
 	Msg *pb.Footer
 	Len uint64