diff --git a/weed/command/filer_backup.go b/weed/command/filer_backup.go index e32c46eafa..38820e856d 100644 --- a/weed/command/filer_backup.go +++ b/weed/command/filer_backup.go @@ -43,6 +43,7 @@ type FilerBackupOptions struct { retentionDays *int initialSnapshot *bool concurrency *int + s3FlattenVersions *bool } var ( @@ -67,6 +68,7 @@ func init() { filerBackupOptions.ignore404Error = cmdFilerBackup.Flag.Bool("ignore404Error", true, "ignore 404 errors from filer") filerBackupOptions.initialSnapshot = cmdFilerBackup.Flag.Bool("initialSnapshot", false, "before subscribing to metadata updates, walk the live filer tree under -filerPath and seed the destination, then subscribe from the walk-start timestamp so concurrent changes are still captured. Runs on every start when -timeAgo is 0 and overwrites the saved checkpoint, so it also re-seeds a reinitialized destination. Remove it once the backup is caught up, otherwise it re-walks the whole tree on each restart.") filerBackupOptions.concurrency = cmdFilerBackup.Flag.Int("concurrency", DefaultConcurrencyLimit, "the maximum number of files that will be backed up concurrently.") + filerBackupOptions.s3FlattenVersions = cmdFilerBackup.Flag.Bool("s3.flattenVersions", false, "mirror the current S3 view of versioned buckets: under each .versions directory only the latest version is written to , a latest delete marker or the removal of the last version deletes when -doDeleteFiles is set, and version files are not mirrored individually") } var cmdFilerBackup = &Command{ @@ -141,7 +143,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti // get start time for the data sink startFrom := time.Unix(0, 0) - sinkId, legacySinkId := backupCheckpointIds(sourcePath, dataSink) + sinkId, legacySinkId := backupCheckpointIds(sourcePath, dataSink, *backupOption.s3FlattenVersions) runSnapshot := *backupOption.initialSnapshot && timeAgo == 0 if resumeTsNs > 0 { // The previous attempt pinned where it stopped — a failed event's @@ -184,12 +186,27 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti } dataSink.SetSourceFiler(filerSource) + var flattener *s3VersionFlattener + if *backupOption.s3FlattenVersions { + flattener = &s3VersionFlattener{ + filerSource: filerSource, + dataSink: dataSink, + sourcePath: sourcePath, + targetPath: targetPath, + doDeleteFiles: *backupOption.doDeleteFiles, + excludePaths: excludePaths, + reExcludeFileName: reExcludeFileName, + excludeFileNames: excludeFileNames, + excludePathPatterns: excludePathPatterns, + } + } + // Walk and seed the live tree, then subscribe from the walk-start watermark: // replaying the event log alone misses entries created-then-deleted before // the walk. The watermark overwrites any stale checkpoint, re-seeding a wiped // destination. if runSnapshot { - snapshotTsNs, err := runInitialSnapshot(sourceFiler.ToGrpcAddress(), filerSource, sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.ignore404Error) + snapshotTsNs, err := runInitialSnapshot(sourceFiler.ToGrpcAddress(), filerSource, sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.ignore404Error, flattener) if err != nil { return 0, fmt.Errorf("initial snapshot: %w", err) } @@ -206,11 +223,22 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti glog.V(0).Infof("initialSnapshot done; subscribing from %v", startFrom) } + generated := genProcessFunction(sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.doDeleteFiles, debug) + if flattener != nil { + inner := generated + generated = func(resp *filer_pb.SubscribeMetadataResponse) error { + defer flattener.lockFor(resp)() + if handled, err := flattener.process(resp); handled { + return err + } + return inner(resp) + } + } + var processEventFn func(*filer_pb.SubscribeMetadataResponse) error if *backupOption.ignore404Error { - processEventFnGenerated := genProcessFunction(sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.doDeleteFiles, debug) processEventFn = func(resp *filer_pb.SubscribeMetadataResponse) error { - err := processEventFnGenerated(resp) + err := generated(resp) if err == nil { return nil } @@ -225,7 +253,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti return err } } else { - processEventFn = genProcessFunction(sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.doDeleteFiles, debug) + processEventFn = generated } if *backupOption.disableErrorRetry { @@ -369,9 +397,16 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti // // NUL joins the fields because it cannot occur in a path or a configuration // value, so distinct field tuples cannot concatenate to the same hash input. -func backupCheckpointIds(sourcePath string, dataSink sink.ReplicationSink) (sinkId, legacySinkId int32) { - sinkId = int32(util.HashStringToLong(sourcePath + "\x00" + dataSink.GetName() + "\x00" + dataSink.GetDestinationIdentity())) - legacySinkId = int32(util.HashStringToLong(dataSink.GetName() + dataSink.GetSinkToDirectory())) +// The flattening mode is part of the identity too: resuming a checkpoint taken +// in the other mode would skip the flattened re-seed and leave mirrored +// version files behind, so a mode change must restart the snapshot. +func backupCheckpointIds(sourcePath string, dataSink sink.ReplicationSink, flattenVersions bool) (sinkId, legacySinkId int32) { + mode := "" + if flattenVersions { + mode = "\x00flatten" + } + sinkId = int32(util.HashStringToLong(sourcePath + "\x00" + dataSink.GetName() + "\x00" + dataSink.GetDestinationIdentity() + mode)) + legacySinkId = int32(util.HashStringToLong(dataSink.GetName() + dataSink.GetSinkToDirectory() + mode)) return } @@ -490,6 +525,7 @@ func runInitialSnapshot( excludePathPatterns []*wildcard.WildcardMatcher, dataSink sink.ReplicationSink, ignore404Error bool, + flattener *s3VersionFlattener, ) (int64, error) { // Metadata events are stamped server-side, so the backup host clock may be // ahead of the filer's. Take `now - 1min` as the subscription watermark so @@ -532,6 +568,28 @@ func runInitialSnapshot( return nil } + if flattener != nil { + if insideVersionsDir(parent) && strings.HasPrefix(entry.Name, "v_") { + return nil + } + if entry.IsDirectory && isVersionsDirName(entry.Name) { + // Seed only the version the pointer names; every other version + // file is skipped by the check above. + err := flattener.applyLatest(parentPath.Child(entry.Name), entry, true, nil) + if err != nil && ignore404Error { + if isIgnorable404(err) { + glog.V(0).Infof("initialSnapshot: versions dir %s moved on, ignore it: %s", sourceKey, err.Error()) + return nil + } + if isSourceLookupError(err) && filersink.SourceSupersedes(ctx, filerSource, sourceKey, filersink.EntryMtimeNs(entry)) { + glog.V(0).Infof("initialSnapshot: source superseded %s mid-walk, skip it: %s", sourceKey, err.Error()) + return nil + } + } + return err + } + } + targetKey := initialSnapshotTargetKey(dataSink, targetPath, sourcePath, sourceKey, entry) if err := dataSink.CreateEntry(targetKey, entry, nil); err != nil { // A file can be listed by TraverseBfs and then deleted before diff --git a/weed/command/filer_backup_offset_test.go b/weed/command/filer_backup_offset_test.go index e42efdaa2b..385612c719 100644 --- a/weed/command/filer_backup_offset_test.go +++ b/weed/command/filer_backup_offset_test.go @@ -81,7 +81,7 @@ func TestBackupOffset_HistoricalFallbackThenIndependence(t *testing.T) { filerAddr, dial := startKvFiler(t) s := &stubSink{name: "s3", dir: "/", destination: "endpoint-a|bucket-a|/"} - sinkId, legacySinkId := backupCheckpointIds("/", s) + sinkId, legacySinkId := backupCheckpointIds("/", s, false) // pre-upgrade state: only the shared historical checkpoint exists if err := setOffset(dial, filerAddr, BackupKeyPrefix, legacySinkId, 111); err != nil { @@ -117,8 +117,8 @@ func TestBackupOffset_HistoricalFallbackThenIndependence(t *testing.T) { func TestBackupOffset_DistinctDestinationsDoNotShare(t *testing.T) { filerAddr, dial := startKvFiler(t) - idA, legacyA := backupCheckpointIds("/", &stubSink{name: "s3", dir: "/", destination: "s3.us-west-004.backblazeb2.com|seaweed-backup-a|/"}) - idB, legacyB := backupCheckpointIds("/", &stubSink{name: "s3", dir: "/", destination: "s3.us-east-005.backblazeb2.com|seaweed-backup-b|/"}) + idA, legacyA := backupCheckpointIds("/", &stubSink{name: "s3", dir: "/", destination: "s3.us-west-004.backblazeb2.com|seaweed-backup-a|/"}, false) + idB, legacyB := backupCheckpointIds("/", &stubSink{name: "s3", dir: "/", destination: "s3.us-east-005.backblazeb2.com|seaweed-backup-b|/"}, false) if err := setOffset(dial, filerAddr, BackupKeyPrefix, idA, 1000); err != nil { t.Fatalf("write A: %v", err) diff --git a/weed/command/filer_backup_test.go b/weed/command/filer_backup_test.go index 640a678bbe..504386bf92 100644 --- a/weed/command/filer_backup_test.go +++ b/weed/command/filer_backup_test.go @@ -287,8 +287,8 @@ func TestBackupCheckpointIds_DistinctDestinations(t *testing.T) { backupA := &stubSink{name: "s3", dir: "/", destination: "s3.us-west-004.backblazeb2.com|seaweed-backup-a|/"} backupB := &stubSink{name: "s3", dir: "/", destination: "s3.us-east-005.backblazeb2.com|seaweed-backup-b|/"} - idA, legacyA := backupCheckpointIds("/", backupA) - idB, legacyB := backupCheckpointIds("/", backupB) + idA, legacyA := backupCheckpointIds("/", backupA, false) + idB, legacyB := backupCheckpointIds("/", backupB, false) if idA == idB { t.Errorf("backups to different destinations share checkpoint id %d", idA) @@ -305,8 +305,8 @@ func TestBackupCheckpointIds_DistinctDestinations(t *testing.T) { // shared offset lets the faster one push the slower one past unseen events. func TestBackupCheckpointIds_DistinctSourcePaths(t *testing.T) { s := &stubSink{name: "s3", dir: "/", destination: "endpoint|bucket|/"} - idA, _ := backupCheckpointIds("/buckets/a", s) - idB, _ := backupCheckpointIds("/buckets/b", s) + idA, _ := backupCheckpointIds("/buckets/a", s, false) + idB, _ := backupCheckpointIds("/buckets/b", s, false) if idA == idB { t.Errorf("backups of different source paths share checkpoint id %d", idA) } @@ -317,7 +317,7 @@ func TestBackupCheckpointIds_DistinctSourcePaths(t *testing.T) { // backups lose their checkpoint on upgrade and replay from zero. func TestBackupCheckpointIds_LegacyFormulaUnchanged(t *testing.T) { s := &stubSink{name: "s3", dir: "/data", destination: "endpoint|bucket|/data"} - _, legacy := backupCheckpointIds("/", s) + _, legacy := backupCheckpointIds("/", s, false) if want := int32(util.HashStringToLong("s3" + "/data")); legacy != want { t.Errorf("legacy id = %d, want historical formula value %d", legacy, want) } @@ -326,19 +326,30 @@ func TestBackupCheckpointIds_LegacyFormulaUnchanged(t *testing.T) { // The NUL joins keep the hash input injective: field values spelling out // other fields' content must not concatenate to the same input. func TestBackupCheckpointIds_NoAliasing(t *testing.T) { - idA, _ := backupCheckpointIds("/src", &stubSink{name: "s3", dir: "/", destination: "/d=>s3|/other"}) - idB, _ := backupCheckpointIds("/src=>s3|/d", &stubSink{name: "s3", dir: "/", destination: "/other"}) + idA, _ := backupCheckpointIds("/src", &stubSink{name: "s3", dir: "/", destination: "/d=>s3|/other"}, false) + idB, _ := backupCheckpointIds("/src=>s3|/d", &stubSink{name: "s3", dir: "/", destination: "/other"}, false) if idA == idB { t.Errorf("field content spelling a separator aliases checkpoint id %d", idA) } } +// Toggling -s3.flattenVersions must not resume the other mode's checkpoint: +// switching modes needs a fresh snapshot, not a mid-stream resume. +func TestBackupCheckpointIds_FlattenModeDistinct(t *testing.T) { + s := &stubSink{name: "s3", dir: "/", destination: "endpoint|bucket|/"} + idFlat, legacyFlat := backupCheckpointIds("/buckets/a", s, true) + idPlain, legacyPlain := backupCheckpointIds("/buckets/a", s, false) + if idFlat == idPlain || legacyFlat == legacyPlain { + t.Errorf("flatten mode shares checkpoint ids: (%d,%d) vs (%d,%d)", idFlat, legacyFlat, idPlain, legacyPlain) + } +} + // Restarting the same configuration must derive the same key, or every // restart would orphan its checkpoint. func TestBackupCheckpointIds_Stable(t *testing.T) { s := &stubSink{name: "s3", dir: "/", destination: "endpoint|bucket|/"} - id1, legacy1 := backupCheckpointIds("/buckets/a", s) - id2, legacy2 := backupCheckpointIds("/buckets/a", s) + id1, legacy1 := backupCheckpointIds("/buckets/a", s, false) + id2, legacy2 := backupCheckpointIds("/buckets/a", s, false) if id1 != id2 || legacy1 != legacy2 { t.Errorf("ids not stable: (%d,%d) vs (%d,%d)", id1, legacy1, id2, legacy2) } diff --git a/weed/command/filer_backup_versions.go b/weed/command/filer_backup_versions.go new file mode 100644 index 0000000000..57f4dab7f7 --- /dev/null +++ b/weed/command/filer_backup_versions.go @@ -0,0 +1,355 @@ +package command + +import ( + "context" + "errors" + "path" + "regexp" + "strings" + "sync" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/replication/sink" + "github.com/seaweedfs/seaweedfs/weed/s3api" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" +) + +// s3VersionFlattener mirrors the current S3 view of versioned buckets into the +// sink: under each ".versions" directory only the latest version is +// written to "", so the destination always holds a plain restorable copy. +// Version files never reach the sink themselves — the latest-version pointer +// on the .versions directory decides what the flattened key holds. +type s3VersionFlattener struct { + filerSource filer_pb.FilerClient + dataSink sink.ReplicationSink + sourcePath string + targetPath string + doDeleteFiles bool + excludePaths []string + reExcludeFileName *regexp.Regexp + excludeFileNames []*wildcard.WildcardMatcher + excludePathPatterns []*wildcard.WildcardMatcher + flatLocks [256]sync.Mutex +} + +func isVersionsDirName(name string) bool { + return strings.HasSuffix(name, s3_constants.VersionsFolder) +} + +func insideVersionsDir(dir string) bool { + return isVersionsDirName(path.Base(dir)) +} + +// lockFor serializes the handling of one event by the key the flattened mirror +// writes to: versions-related events lock their object key, everything else its +// own path, so a pointer update and a regular-path write to the same object +// cannot overlap. Both sides of the event are locked, since a rename or delete +// can retire one flattened key while writing another. Non-barrier dir updates +// run concurrently in the metadata processor, which would otherwise let an +// older flattened upload finish last. +func (f *s3VersionFlattener) lockFor(resp *filer_pb.SubscribeMetadataResponse) func() { + oldKey, newKey := eventFlatKeys(resp) + if oldKey == "" { + oldKey = newKey + } + if newKey == "" { + newKey = oldKey + } + if newKey == "" { + oldKey, newKey = resp.Directory, resp.Directory + } + a := util.HashStringToLong(oldKey) & int64(len(f.flatLocks)-1) + b := util.HashStringToLong(newKey) & int64(len(f.flatLocks)-1) + if a == b { + f.flatLocks[a].Lock() + return f.flatLocks[a].Unlock + } + if b < a { + a, b = b, a + } + f.flatLocks[a].Lock() + f.flatLocks[b].Lock() + return func() { + f.flatLocks[b].Unlock() + f.flatLocks[a].Unlock() + } +} + +func eventFlatKeys(resp *filer_pb.SubscribeMetadataResponse) (string, string) { + message := resp.EventNotification + return flatKeyOf(resp.Directory, message.OldEntry), + flatKeyOf(filer_pb.MetadataEventTargetDirectory(resp), message.NewEntry) +} + +func flatKeyOf(dir string, entry *filer_pb.Entry) string { + if entry == nil { + return "" + } + if insideVersionsDir(dir) && strings.HasPrefix(entry.Name, "v_") { + return string(objectKeyOf(util.FullPath(dir))) + } + if entry.IsDirectory && isVersionsDirName(entry.Name) { + return string(objectKeyOf(util.FullPath(dir).Child(entry.Name))) + } + return string(util.FullPath(dir).Child(entry.Name)) +} + +// process handles the versions-related part of one metadata event and reports +// whether the event was consumed. Version files under a .versions directory +// are dropped outright; create/update/delete of the .versions directory +// itself is mapped onto the flattened object key through its latest-version +// pointer. Anything else is left for the generic event path. +func (f *s3VersionFlattener) process(resp *filer_pb.SubscribeMetadataResponse) (bool, error) { + message := resp.EventNotification + if message == nil { + return false, nil + } + targetDir := filer_pb.MetadataEventTargetDirectory(resp) + + name := "" + if message.NewEntry != nil { + name = message.NewEntry.Name + } else if message.OldEntry != nil { + name = message.OldEntry.Name + } + if strings.HasPrefix(name, "v_") && (insideVersionsDir(resp.Directory) || insideVersionsDir(targetDir)) { + // A version file changed; only a change to the version the pointer + // still names can move the flattened key (e.g. a tag update), so + // re-resolve the pointer and refresh just for that child. + versionsDir := resp.Directory + if !insideVersionsDir(versionsDir) { + versionsDir = targetDir + } + dirEntry, _, _, err := filer_pb.GetEntry(context.Background(), f.filerSource, util.FullPath(versionsDir)) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return true, nil + } + return true, err + } + if name != string(dirEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]) { + return true, nil + } + return true, f.applyLatest(util.FullPath(versionsDir), dirEntry, false, message.Signatures) + } + + oldVersionsDir := "" + newVersionsDir := "" + if message.OldEntry != nil && message.OldEntry.IsDirectory && isVersionsDirName(message.OldEntry.Name) { + oldVersionsDir = string(util.FullPath(resp.Directory).Child(message.OldEntry.Name)) + } + if message.NewEntry != nil && message.NewEntry.IsDirectory && isVersionsDirName(message.NewEntry.Name) { + newVersionsDir = string(util.FullPath(targetDir).Child(message.NewEntry.Name)) + } + if oldVersionsDir == "" && newVersionsDir == "" { + return f.processRegularPath(resp, targetDir, message) + } + + if newVersionsDir != "" { + if err := f.applyLatest(util.FullPath(newVersionsDir), message.NewEntry, message.OldEntry == nil, message.Signatures); err != nil { + return true, err + } + } + // The versions directory is gone (deleted outright or renamed away): + // whatever the flattened key held is no longer current. + if oldVersionsDir != "" && oldVersionsDir != newVersionsDir { + objectKey := objectKeyOf(util.FullPath(oldVersionsDir)) + if f.isExcludedObject(objectKey, util.FullPath(oldVersionsDir), message.OldEntry) { + return true, nil + } + if err := f.deleteObject(objectKey, message.Signatures); err != nil { + return true, err + } + } + return true, nil +} + +func objectKeyOf(versionsDir util.FullPath) util.FullPath { + dir, name := versionsDir.DirAndName() + return util.FullPath(dir).Child(strings.TrimSuffix(name, s3_constants.VersionsFolder)) +} + +// processRegularPath guards the generic event path for keys that carry a +// .versions directory: a regular-path entry (the null object, or its delete) +// can coexist with a pointer that names a live version, and only the pointer +// decides the flattened key — e.g. deleting the null object must not remove a +// flattened key whose current version is live. A rename touches two keys, so +// each side is resolved on its own; when exactly one side is versioned the +// other is handed to the generic path as a lone delete or create. Keys with +// no .versions sibling are left to the generic path entirely. +func (f *s3VersionFlattener) processRegularPath(resp *filer_pb.SubscribeMetadataResponse, targetDir string, message *filer_pb.EventNotification) (bool, error) { + oldHandled, err := f.refreshIfVersioned(util.FullPath(resp.Directory), message.OldEntry, message.Signatures) + if err != nil { + return true, err + } + newHandled := oldHandled + if !filer_pb.IsUpdate(resp) { + newHandled, err = f.refreshIfVersioned(util.FullPath(targetDir), message.NewEntry, message.Signatures) + if err != nil { + return true, err + } + } + if !filer_pb.IsRename(resp) || oldHandled == newHandled { + return oldHandled || newHandled, nil + } + if oldHandled { + // The flattened old key keeps whatever its pointer still names; the + // generic path only needs to create the moved entry. + message.OldEntry = nil + } else { + // The flattened old key held the moved file; the generic path only + // needs to delete it. + message.NewEntry = nil + } + return false, nil +} + +// refreshIfVersioned re-resolves the flattened key of dir/entry.Name when a +// .versions directory sits next to it, reporting whether that sibling exists. +// The pointer decides what the flattened key holds, so a regular-path event +// only refreshes through it and never writes the flattened key directly. +func (f *s3VersionFlattener) refreshIfVersioned(dir util.FullPath, entry *filer_pb.Entry, signatures []int32) (bool, error) { + if entry == nil || entry.IsDirectory { + return false, nil + } + versionsDir := dir.Child(entry.Name + s3_constants.VersionsFolder) + dirEntry, _, _, err := filer_pb.GetEntry(context.Background(), f.filerSource, versionsDir) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return false, nil + } + return true, err + } + return true, f.applyLatest(versionsDir, dirEntry, false, signatures) +} + +// applyLatest writes the object a .versions pointer currently names to the +// flattened key: the latest version file, a delete marker (flat delete), the +// null object at the regular path, or — once every version is gone and the +// pointer is cleared — nothing. +func (f *s3VersionFlattener) applyLatest(versionsDir util.FullPath, dirEntry *filer_pb.Entry, isCreate bool, signatures []int32) error { + objectKey := objectKeyOf(versionsDir) + if f.isExcludedObject(objectKey, versionsDir, dirEntry) { + return nil + } + + extended := dirEntry.Extended + switch { + case string(extended[s3_constants.ExtLatestVersionIsDeleteMarker]) == "true": + return f.deleteObject(objectKey, signatures) + case string(extended[s3_constants.ExtNullVersionIsLatestKey]) == "true": + return f.uploadNullLatest(objectKey, versionsDir, signatures) + default: + if latestName := string(extended[s3_constants.ExtLatestVersionFileNameKey]); latestName != "" { + return f.uploadLatest(objectKey, versionsDir.Child(latestName), signatures) + } + // A create event lands before the pointer is written; an update with + // no pointer means the last version was removed. + if isCreate { + return nil + } + return f.deleteObject(objectKey, signatures) + } +} + +// uploadNullLatest flattens the regular-path null object while the pointer +// names it latest. A ?versionId=null delete removes that object without +// touching the pointer, so on a miss the newest remaining version wins (or a +// flat delete when none survives) — the same rescan the S3 read path falls +// back to. +func (f *s3VersionFlattener) uploadNullLatest(objectKey, versionsDir util.FullPath, signatures []int32) error { + if _, _, _, err := filer_pb.GetEntry(context.Background(), f.filerSource, objectKey); err == nil { + return f.uploadLatest(objectKey, objectKey, signatures) + } else if !errors.Is(err, filer_pb.ErrNotFound) { + return err + } + latestName, err := f.latestVersionFileName(versionsDir) + if err != nil { + return err + } + if latestName == "" { + return f.deleteObject(objectKey, signatures) + } + return f.uploadLatest(objectKey, versionsDir.Child(latestName), signatures) +} + +// latestVersionFileName returns the name of the newest version file under +// versionsDir by version-id order — delete markers included, the upload +// resolves them to a flat delete — or "" when no version entry remains. It +// paginates: the filer caps one request at DirListingLimit, and old-format +// version names sort oldest-first, so a single page can miss the true latest. +func (f *s3VersionFlattener) latestVersionFileName(versionsDir util.FullPath) (string, error) { + var latestName, latestId string + err := f.filerSource.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + startFrom := "" + for { + var lastName string + var pageCount uint32 + if err := filer_pb.SeaweedList(context.Background(), client, string(versionsDir), "", func(entry *filer_pb.Entry, _ bool) error { + pageCount++ + if entry == nil { + return nil + } + lastName = entry.Name + versionId := s3api.VersionIdFromEntry(entry) + if versionId != "" && (latestId == "" || s3api.CompareVersionIds(versionId, latestId) < 0) { + latestId, latestName = versionId, entry.Name + } + return nil + }, startFrom, false, filer.PaginationSize); err != nil { + return err + } + if pageCount < filer.PaginationSize || lastName == "" { + return nil + } + startFrom = lastName + } + }) + return latestName, err +} + +func (f *s3VersionFlattener) uploadLatest(objectKey, sourceKey util.FullPath, signatures []int32) error { + entry, _, _, err := filer_pb.GetEntry(context.Background(), f.filerSource, sourceKey) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + // The pointed-at version (or null object) is not readable yet or is + // already gone again; its own event or a later pointer update + // converges the flattened key. + glog.V(1).Infof("flatten %s: source %s not found: %v", objectKey, sourceKey, err) + return nil + } + return err + } + if string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" { + return f.deleteObject(objectKey, signatures) + } + var mTime int64 + if entry.Attributes != nil { + mTime = entry.Attributes.Mtime + } + return sink.CreateEntryLatest(f.dataSink, destKey(f.dataSink, f.targetPath, f.sourcePath, objectKey, mTime), entry, signatures) +} + +// isExcludedObject applies the source scope and exclusion rules to the +// flattened object key — including its own name, so a file-name pattern that +// excludes "k" also excludes the flattened "k" even though the event names +// "k.versions". +func (f *s3VersionFlattener) isExcludedObject(objectKey, versionsDir util.FullPath, dirEntry *filer_pb.Entry) bool { + dir, name := objectKey.DirAndName() + return !util.IsEqualOrUnder(string(objectKey), f.sourcePath) || + matchesExcludePath(string(objectKey), f.excludePaths) || + matchesExcludePath(string(versionsDir), f.excludePaths) || + isEntryExcluded(dir, dirEntry, f.reExcludeFileName, f.excludeFileNames, f.excludePathPatterns) || + isEntryExcluded(dir, &filer_pb.Entry{Name: name}, f.reExcludeFileName, f.excludeFileNames, f.excludePathPatterns) +} + +func (f *s3VersionFlattener) deleteObject(objectKey util.FullPath, signatures []int32) error { + if !f.doDeleteFiles || f.dataSink.IsIncremental() { + return nil + } + return sink.DeleteEntry(f.dataSink, destKey(f.dataSink, f.targetPath, f.sourcePath, objectKey, 0), false, true, false, signatures) +} diff --git a/weed/command/filer_backup_versions_test.go b/weed/command/filer_backup_versions_test.go new file mode 100644 index 0000000000..1910f828ae --- /dev/null +++ b/weed/command/filer_backup_versions_test.go @@ -0,0 +1,492 @@ +package command + +import ( + "context" + "fmt" + "sort" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/replication/source" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" +) + +type lookupEntryFilerClient struct { + entries map[string]*filer_pb.Entry +} + +func (c *lookupEntryFilerClient) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { + return fn(&lookupEntrySeaweedClient{entries: c.entries}) +} +func (c *lookupEntryFilerClient) AdjustedUrl(location *filer_pb.Location) string { + return location.Url +} +func (c *lookupEntryFilerClient) GetDataCenter() string { return "" } + +type lookupEntrySeaweedClient struct { + filer_pb.SeaweedFilerClient + entries map[string]*filer_pb.Entry +} + +func (c *lookupEntrySeaweedClient) LookupDirectoryEntry(ctx context.Context, in *filer_pb.LookupDirectoryEntryRequest, opts ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + entry, found := c.entries[in.Directory+"/"+in.Name] + if !found { + return nil, filer_pb.ErrNotFound + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil +} + +func (c *lookupEntrySeaweedClient) ListEntries(_ context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + var entries []*filer_pb.Entry + for fullPath, entry := range c.entries { + if dir, _ := util.FullPath(fullPath).DirAndName(); dir == in.Directory { + entries = append(entries, entry) + } + } + sort.Slice(entries, func(i, j int) bool { return entries[i].Name < entries[j].Name }) + if in.StartFromFileName != "" { + idx := sort.Search(len(entries), func(i int) bool { return entries[i].Name >= in.StartFromFileName }) + if !in.InclusiveStartFrom { + for idx < len(entries) && entries[idx].Name == in.StartFromFileName { + idx++ + } + } + entries = entries[idx:] + } + limit := in.Limit + if limit == 0 { + limit = filer.PaginationSize // the filer substitutes DirListingLimit for 0 + } + if uint64(len(entries)) > uint64(limit) { + entries = entries[:limit] + } + return &verifyTestStream{entries: entries}, nil +} + +type recordSink struct { + created map[string]*filer_pb.Entry + deleted []string +} + +func (s *recordSink) GetName() string { return "record" } +func (s *recordSink) Initialize(configuration util.Configuration, prefix string) error { + return nil +} +func (s *recordSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error { + s.deleted = append(s.deleted, key) + return nil +} +func (s *recordSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error { + s.created[key] = entry + return nil +} +func (s *recordSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (bool, error) { + return true, nil +} +func (s *recordSink) GetSinkToDirectory() string { return "/" } +func (s *recordSink) GetDestinationIdentity() string { return "record" } +func (s *recordSink) SetSourceFiler(fs *source.FilerSource) {} +func (s *recordSink) IsIncremental() bool { return false } + +func newFlattener(entries map[string]*filer_pb.Entry) (*s3VersionFlattener, *recordSink) { + rs := &recordSink{created: make(map[string]*filer_pb.Entry)} + f := &s3VersionFlattener{ + filerSource: &lookupEntryFilerClient{entries: entries}, + dataSink: rs, + sourcePath: "/buckets", + targetPath: "/", + doDeleteFiles: true, + } + return f, rs +} + +func event(dir, parent string, oldEntry, newEntry *filer_pb.Entry) *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{ + Directory: dir, + EventNotification: &filer_pb.EventNotification{ + OldEntry: oldEntry, + NewEntry: newEntry, + NewParentPath: parent, + }, + } +} + +func TestFlattenDropsVersionFileEvents(t *testing.T) { + f, rs := newFlattener(map[string]*filer_pb.Entry{}) + v := &filer_pb.Entry{Name: "v_abc", Chunks: []*filer_pb.FileChunk{{FileId: "1,01"}}} + + for _, resp := range []*filer_pb.SubscribeMetadataResponse{ + event("/buckets/b/k.versions", "", nil, v), + event("/buckets/b/k.versions", "", v, nil), + event("/buckets/b/k.versions", "/buckets/b/k.versions", v, v), + } { + handled, err := f.process(resp) + require.NoError(t, err) + require.True(t, handled, "version file events should be consumed") + } + require.Empty(t, rs.created) + require.Empty(t, rs.deleted) +} + +func TestFlattenPassesThroughNormalEvents(t *testing.T) { + f, rs := newFlattener(nil) + + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, &filer_pb.Entry{Name: "k"})) + require.NoError(t, err) + require.False(t, handled) + + // a regular file literally named v_foo outside a .versions dir is not a + // version file + handled, err = f.process(event("/buckets/b", "/buckets/b", nil, &filer_pb.Entry{Name: "v_notes"})) + require.NoError(t, err) + require.False(t, handled) + + // a plain file whose name ends with .versions is not a versions directory + handled, err = f.process(event("/buckets/b", "/buckets/b", nil, &filer_pb.Entry{Name: "x.versions"})) + require.NoError(t, err) + require.False(t, handled) + require.Empty(t, rs.created) +} + +func TestFlattenUploadsLatestVersion(t *testing.T) { + version := &filer_pb.Entry{ + Name: "v_2", + Attributes: &filer_pb.FuseAttributes{Mtime: 7}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,02"}}, + } + f, rs := newFlattener(map[string]*filer_pb.Entry{"/buckets/b/k.versions/v_2": version}) + + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_2")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Same(t, version, rs.created["/b/k"]) + require.Empty(t, rs.deleted) +} + +func TestFlattenDeletesOnLatestDeleteMarker(t *testing.T) { + f, rs := newFlattener(nil) + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionIsDeleteMarker: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", dir, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Equal(t, []string{"/b/k"}, rs.deleted) + require.Empty(t, rs.created) +} + +func TestFlattenDeleteHonorsDoDeleteFiles(t *testing.T) { + f, rs := newFlattener(nil) + f.doDeleteFiles = false + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionIsDeleteMarker: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", dir, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Empty(t, rs.deleted) +} + +func TestFlattenDeleteOnClearedPointerAndRemovedDir(t *testing.T) { + f, rs := newFlattener(nil) + withPointer := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_1")}, + } + cleared := &filer_pb.Entry{Name: "k.versions", IsDirectory: true, Extended: map[string][]byte{}} + + handled, err := f.process(event("/buckets/b", "/buckets/b", withPointer, cleared)) + require.NoError(t, err) + require.True(t, handled) + require.Equal(t, []string{"/b/k"}, rs.deleted) + + rs.deleted = nil + handled, err = f.process(event("/buckets/b", "", cleared, nil)) + require.NoError(t, err) + require.True(t, handled) + require.Equal(t, []string{"/b/k"}, rs.deleted) +} + +func TestFlattenFreshVersionsDirIsSkipped(t *testing.T) { + f, rs := newFlattener(nil) + // a .versions dir create carries no pointer yet — the update that sets it + // follows in the same write + dir := &filer_pb.Entry{Name: "k.versions", IsDirectory: true} + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Empty(t, rs.deleted) + require.Empty(t, rs.created) +} + +func TestFlattenNullVersionIsLatest(t *testing.T) { + nullObject := &filer_pb.Entry{ + Name: "k", + Attributes: &filer_pb.FuseAttributes{Mtime: 9}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,09"}}, + } + f, rs := newFlattener(map[string]*filer_pb.Entry{"/buckets/b/k": nullObject}) + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtNullVersionIsLatestKey: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Same(t, nullObject, rs.created["/b/k"]) +} + +func TestFlattenNullVersionDeletedRescansNewestVersion(t *testing.T) { + // deleting ?versionId=null removes the regular object without touching the + // null-is-latest pointer: the flattened key must fall back to the newest + // remaining version, the same rescan the S3 read path does + older := &filer_pb.Entry{Name: "v_00000000000000010000000000000000", Chunks: []*filer_pb.FileChunk{{FileId: "1,01"}}} + newer := &filer_pb.Entry{Name: "v_00000000000000020000000000000000", Chunks: []*filer_pb.FileChunk{{FileId: "1,02"}}} + f, rs := newFlattener(map[string]*filer_pb.Entry{ + "/buckets/b/k.versions/" + older.Name: older, + "/buckets/b/k.versions/" + newer.Name: newer, + }) + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtNullVersionIsLatestKey: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Same(t, newer, rs.created["/b/k"]) + require.Empty(t, rs.deleted) +} + +func TestFlattenNullVersionDeletedNewestIsDeleteMarker(t *testing.T) { + marker := &filer_pb.Entry{ + Name: "v_00000000000000020000000000000000", + Extended: map[string][]byte{s3_constants.ExtDeleteMarkerKey: []byte("true")}, + } + older := &filer_pb.Entry{Name: "v_00000000000000010000000000000000", Chunks: []*filer_pb.FileChunk{{FileId: "1,01"}}} + f, rs := newFlattener(map[string]*filer_pb.Entry{ + "/buckets/b/k.versions/" + older.Name: older, + "/buckets/b/k.versions/" + marker.Name: marker, + }) + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtNullVersionIsLatestKey: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Equal(t, []string{"/b/k"}, rs.deleted) + require.Empty(t, rs.created) +} + +func TestFlattenNullVersionDeletedNoVersionsLeftDeletes(t *testing.T) { + f, rs := newFlattener(map[string]*filer_pb.Entry{}) + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtNullVersionIsLatestKey: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Equal(t, []string{"/b/k"}, rs.deleted) +} + +func TestFlattenNullVersionDeletedScansPastOnePage(t *testing.T) { + // old-format version names sort oldest-first and the filer caps one + // listing at DirListingLimit, so with a long history the newest version + // sits beyond the first page — the rescan must read every page + entries := map[string]*filer_pb.Entry{} + var newest *filer_pb.Entry + for i := 1; i <= filer.PaginationSize+1; i++ { + v := &filer_pb.Entry{ + Name: fmt.Sprintf("v_%032x", i), + Chunks: []*filer_pb.FileChunk{{FileId: fmt.Sprintf("1,%02x", i)}}, + } + entries["/buckets/b/k.versions/"+v.Name] = v + newest = v + } + f, rs := newFlattener(entries) + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtNullVersionIsLatestKey: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Same(t, newest, rs.created["/b/k"]) +} + +func TestFlattenVersionFileEventRefreshesCurrentVersion(t *testing.T) { + // a tag-only update touches the version file, not the pointer: re-resolving + // the pointer must still upload the current version to the flat key + version := &filer_pb.Entry{Name: "v_2", Chunks: []*filer_pb.FileChunk{{FileId: "1,02"}}} + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_2")}, + } + f, rs := newFlattener(map[string]*filer_pb.Entry{ + "/buckets/b/k.versions": dir, + "/buckets/b/k.versions/v_2": version, + }) + + tagUpdate := &filer_pb.Entry{Name: "v_2", Extended: map[string][]byte{"X-Amz-Tagging-env": []byte("prod")}} + handled, err := f.process(event("/buckets/b/k.versions", "/buckets/b/k.versions", tagUpdate, tagUpdate)) + require.NoError(t, err) + require.True(t, handled) + require.Same(t, version, rs.created["/b/k"]) +} + +func TestFlattenExcludedObjectNameIsNotCopiedOrDeleted(t *testing.T) { + f, rs := newFlattener(nil) + f.excludeFileNames = wildcard.CompileWildcardMatchers("*.tmp") + + dir := &filer_pb.Entry{ + Name: "notes.tmp.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionIsDeleteMarker: []byte("true")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", dir, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Empty(t, rs.deleted, "delete on the excluded flat key must be skipped") + + handled, err = f.process(event("/buckets/b", "", dir, nil)) + require.NoError(t, err) + require.True(t, handled) + require.Empty(t, rs.deleted) +} + +func TestFlattenRegularPathFollowsVersionsPointer(t *testing.T) { + // deleting the null object must not remove the flattened key while the + // .versions pointer names a live version + version := &filer_pb.Entry{Name: "v_2", Chunks: []*filer_pb.FileChunk{{FileId: "1,02"}}} + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_2")}, + } + f, rs := newFlattener(map[string]*filer_pb.Entry{ + "/buckets/b/k.versions": dir, + "/buckets/b/k.versions/v_2": version, + }) + + handled, err := f.process(event("/buckets/b", "", &filer_pb.Entry{Name: "k"}, nil)) + require.NoError(t, err) + require.True(t, handled) + require.Same(t, version, rs.created["/b/k"]) + require.Empty(t, rs.deleted) +} + +func TestFlattenRenameLeavesDeleteForGenericPath(t *testing.T) { + // rename a -> b where only b carries a .versions directory: the flattener + // refreshes b through the pointer but the old flattened key held the moved + // file, so the event must reach the generic path as a lone delete of a + version := &filer_pb.Entry{Name: "v_9", Chunks: []*filer_pb.FileChunk{{FileId: "1,09"}}} + dir := &filer_pb.Entry{ + Name: "b.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_9")}, + } + f, rs := newFlattener(map[string]*filer_pb.Entry{ + "/buckets/b/b.versions": dir, + "/buckets/b/b.versions/v_9": version, + }) + + resp := event("/buckets/b", "/buckets/b", &filer_pb.Entry{Name: "a"}, &filer_pb.Entry{Name: "b"}) + handled, err := f.process(resp) + require.NoError(t, err) + require.False(t, handled, "old side has no .versions sibling — generic path must delete a") + require.NotNil(t, resp.EventNotification.OldEntry) + require.Nil(t, resp.EventNotification.NewEntry) + require.Same(t, version, rs.created["/b/b"]) +} + +func TestFlattenRenameLeavesCreateForGenericPath(t *testing.T) { + // rename a -> b where only a carries a .versions directory: the pointer + // still decides flattened a, so the generic path must see a lone create + // of b and never delete the flattened old key + version := &filer_pb.Entry{Name: "v_2", Chunks: []*filer_pb.FileChunk{{FileId: "1,02"}}} + dir := &filer_pb.Entry{ + Name: "a.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_2")}, + } + f, rs := newFlattener(map[string]*filer_pb.Entry{ + "/buckets/b/a.versions": dir, + "/buckets/b/a.versions/v_2": version, + }) + + resp := event("/buckets/b", "/buckets/b", &filer_pb.Entry{Name: "a"}, &filer_pb.Entry{Name: "b"}) + handled, err := f.process(resp) + require.NoError(t, err) + require.False(t, handled) + require.Nil(t, resp.EventNotification.OldEntry) + require.NotNil(t, resp.EventNotification.NewEntry) + require.Same(t, version, rs.created["/b/a"]) + require.Empty(t, rs.deleted) +} + +func TestFlattenRenameBothSidesVersioned(t *testing.T) { + versionA := &filer_pb.Entry{Name: "v_2", Chunks: []*filer_pb.FileChunk{{FileId: "1,02"}}} + versionB := &filer_pb.Entry{Name: "v_9", Chunks: []*filer_pb.FileChunk{{FileId: "1,09"}}} + f, rs := newFlattener(map[string]*filer_pb.Entry{ + "/buckets/b/a.versions": { + Name: "a.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_2")}, + }, + "/buckets/b/a.versions/v_2": versionA, + "/buckets/b/b.versions": { + Name: "b.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_9")}, + }, + "/buckets/b/b.versions/v_9": versionB, + }) + + resp := event("/buckets/b", "/buckets/b", &filer_pb.Entry{Name: "a"}, &filer_pb.Entry{Name: "b"}) + handled, err := f.process(resp) + require.NoError(t, err) + require.True(t, handled) + require.NotNil(t, resp.EventNotification.OldEntry) + require.NotNil(t, resp.EventNotification.NewEntry) + require.Same(t, versionA, rs.created["/b/a"]) + require.Same(t, versionB, rs.created["/b/b"]) +} + +func TestFlattenDeleteMarkerVersionEntry(t *testing.T) { + marker := &filer_pb.Entry{ + Name: "v_3", + Extended: map[string][]byte{s3_constants.ExtDeleteMarkerKey: []byte("true")}, + } + f, rs := newFlattener(map[string]*filer_pb.Entry{"/buckets/b/k.versions/v_3": marker}) + dir := &filer_pb.Entry{ + Name: "k.versions", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v_3")}, + } + handled, err := f.process(event("/buckets/b", "/buckets/b", nil, dir)) + require.NoError(t, err) + require.True(t, handled) + require.Equal(t, []string{"/b/k"}, rs.deleted) +} diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 67657acb62..4a4d7ea1c4 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -1,6 +1,7 @@ package filersink import ( + "bytes" "context" "errors" "fmt" @@ -15,6 +16,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/wdclient" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" "github.com/seaweedfs/seaweedfs/weed/security" @@ -282,6 +284,17 @@ func (fs *FilerSink) entryMissing(key string) (bool, error) { } func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error { + return fs.createEntry(key, entry, signatures, true) +} + +// CreateEntryLatest writes an entry the caller already resolved as the current +// source state, so the stale-replay check against the key's own source path +// does not apply. +func (fs *FilerSink) CreateEntryLatest(key string, entry *filer_pb.Entry, signatures []int32) error { + return fs.createEntry(key, entry, signatures, false) +} + +func (fs *FilerSink) createEntry(key string, entry *filer_pb.Entry, signatures []int32, checkSourceVersion bool) error { return fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { @@ -294,7 +307,14 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [ } // glog.V(1).Infof("lookup: %v", lookupRequest) if resp, err := filer_pb.LookupEntry(context.Background(), client, lookupRequest); err == nil { - if destinationSatisfiesSource(resp.Entry, entry) { + // A resolved-latest write carries the caller's pick of the current + // version, which can be older than the destination copy it replaces + // (the deleted latest) — only provably identical content skips it. + skip := destinationHoldsContent(resp.Entry, entry) + if checkSourceVersion { + skip = destinationSatisfiesSource(resp.Entry, entry) + } + if skip { glog.V(3).Infof("skip overwriting %s", key) return nil } @@ -307,13 +327,15 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [ // arrival — skip before any read. When the source holds a newer live // version, replicate it directly: the superseding event may never be // delivered to this sink. - if live, movedOn := fs.sourceVersion(key, entry); movedOn { - if live == nil { - glog.Warningf("skip stale entry %s, source superseded before replicate", key) - return nil + if checkSourceVersion { + if live, movedOn := fs.sourceVersion(key, entry); movedOn { + if live == nil { + glog.Warningf("skip stale entry %s, source superseded before replicate", key) + return nil + } + glog.Warningf("entry %s superseded before replicate; syncing live version", key) + entry = live } - glog.Warningf("entry %s superseded before replicate; syncing live version", key) - entry = live } replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry), "") @@ -646,6 +668,77 @@ func chooseUpdateAction(existing, oldEntry, incoming *filer_pb.Entry) updateActi return updateNormal } +// destinationHoldsContent reports whether the destination already holds exactly +// the resolved entry for a CreateEntryLatest write, where a newer destination +// mtime proves nothing: the resolved version may be older than the copy it +// replaces. Skip only when the writable metadata already matches and the content +// is identical — chunk provenance, or an equal ETag over full byte coverage. +func destinationHoldsContent(existing, reference *filer_pb.Entry) bool { + if existing == nil { + return false + } + // A skip keeps the destination's stored metadata, so identical bytes are not + // enough: the resolved version's writable attributes and extended entries + // must already match what the write would install, or a stale tag/version id + // survives under equal content. + if !sameWritableMetadata(existing, reference) { + return false + } + if destinationMatchesReference(existing, reference) { + return true + } + return sameInlineContent(existing, reference) && + coversReference(existing, reference) && + filer.ETag(existing) == filer.ETag(reference) +} + +// sameWritableMetadata reports whether the destination already stores the metadata +// the resolved write would install. Inode and creation time are filer-assigned or +// preserved across rewrites — comparing them would re-replicate forever on fields +// a write can never change — so only the writable fields decide. +func sameWritableMetadata(existing, reference *filer_pb.Entry) bool { + exAttr, refAttr := normalizedAttributes(existing.GetAttributes()), normalizedAttributes(reference.GetAttributes()) + if exAttr != nil && refAttr.GetTtlSec() == 0 { + // a TTL-less write inherits the destination's storage-rule TTL, so the + // stored value is one a rewrite reproduces, not one it would change + exAttr.TtlSec = 0 + } + if !proto.Equal(exAttr, refAttr) { + return false + } + ex, ref := existing.GetExtended(), reference.GetExtended() + if len(ex) != len(ref) { + return false + } + for k, v := range ex { + refValue, found := ref[k] + if !found || !bytes.Equal(v, refValue) { + return false + } + } + return true +} + +// normalizedAttributes masks the fields the destination filer assigns or preserves +// on write and applies the store's zero-value defaults (access/change time fall +// back to mtime), so the comparison covers only fields a rewrite controls. +func normalizedAttributes(attr *filer_pb.FuseAttributes) *filer_pb.FuseAttributes { + if attr == nil { + return nil + } + a := proto.Clone(attr).(*filer_pb.FuseAttributes) + a.Inode = 0 + a.Crtime = 0 + a.CrtimeNs = 0 + if a.Atime == 0 && a.AtimeNs == 0 { + a.Atime, a.AtimeNs = a.Mtime, a.MtimeNs + } + if a.Ctime == 0 { + a.Ctime, a.CtimeNs = a.Mtime, a.MtimeNs + } + return a +} + // destinationSatisfiesSource reports whether an existing destination entry already // satisfies the incoming source entry, so CreateEntry can skip the write. Coverage // is checked first: a chunk-truncated destination (covering fewer bytes than the @@ -662,6 +755,9 @@ func destinationSatisfiesSource(existing, incoming *filer_pb.Entry) bool { return true // destination strictly newer: incoming is older → keep destination } if !destinationMatchesReference(existing, incoming) { + if !sameInlineContent(existing, incoming) { + return false // equal chunk lists hide nothing: the inline bytes differ + } if !coversReference(existing, incoming) { return false // missing a byte range the source covers (truncated/gapped) } @@ -701,7 +797,27 @@ func destinationMatchesReference(existing, reference *filer_pb.Entry) bool { return false } } - return true + // Identical chunk lists say nothing about content stored outside them: + // a chunk-less entry keeps its bytes in Content, a remote-only entry in the + // object RemoteEntry points at. Both must compare equal to claim a match. + return sameInlineContent(existing, reference) +} + +// sameInlineContent reports whether two entries carry the same out-of-chunk +// content — inline bytes, or the same remote object — so a destination can be +// trusted to hold the reference's bytes when the chunk comparison is done. +func sameInlineContent(a, b *filer_pb.Entry) bool { + ra, rb := a.GetRemoteEntry(), b.GetRemoteEntry() + if (ra == nil) != (rb == nil) { + return false + } + if ra != nil && (ra.StorageName != rb.StorageName || + ra.RemoteETag != rb.RemoteETag || + ra.RemoteMtime != rb.RemoteMtime || + ra.RemoteSize != rb.RemoteSize) { + return false + } + return bytes.Equal(a.GetContent(), b.GetContent()) } // hasStaleBackupChunk reports whether existing holds a chunk that this backup diff --git a/weed/replication/sink/filersink/filer_sink_test.go b/weed/replication/sink/filersink/filer_sink_test.go index eeab374ccf..c276fad0b1 100644 --- a/weed/replication/sink/filersink/filer_sink_test.go +++ b/weed/replication/sink/filersink/filer_sink_test.go @@ -61,6 +61,17 @@ func dstEntry(mtime int64, srcFids ...string) *filer_pb.Entry { return e } +// inlineEntry builds a chunk-less entry at mtime holding its bytes in Content — +// the small-file layout, where chunk comparisons alone cannot tell versions apart. +func inlineEntry(mtime int64, content string) *filer_pb.Entry { + return &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: mtime}, Content: []byte(content)} +} + +func extEntry(e *filer_pb.Entry, k, v string) *filer_pb.Entry { + e.Extended = map[string][]byte{k: []byte(v)} + return e +} + // ivEntry builds an entry at mtime with chunks at the given [offset,end) byte // ranges — for coverage/containment tests where exact ranges matter. func ivEntry(mtime int64, ivs ...[2]int64) *filer_pb.Entry { @@ -243,6 +254,11 @@ func TestDestinationSatisfiesSource(t *testing.T) { // out-of-order: newer (smaller) destination must not be rolled back by an // older, larger source replay. {"newer smaller dest, older larger source -> skip", ivEntry(300, [2]int64{0, 5}), ivEntry(100, [2]int64{0, 10}), true}, + // chunk-less inline files: identical chunk lists must not hide changed bytes + {"inline same -> skip", inlineEntry(200, "x"), inlineEntry(200, "x"), true}, + {"inline different, dest older -> replicate", inlineEntry(200, "x"), inlineEntry(300, "y"), false}, + {"inline different, equal mtime -> replicate", inlineEntry(200, "x"), inlineEntry(200, "y"), false}, + {"inline different, dest newer -> skip", inlineEntry(300, "x"), inlineEntry(200, "y"), true}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -253,6 +269,67 @@ func TestDestinationSatisfiesSource(t *testing.T) { } } +// destinationHoldsContent gates CreateEntryLatest's skip: the resolved entry is +// the picked current version, so a newer destination mtime must not keep the +// replaced copy — only provably identical content skips. +func TestDestinationHoldsContent(t *testing.T) { + md5A := []byte("0123456789abcdef") + md5B := []byte("fedcba9876543210") + cases := []struct { + name string + existing, incoming *filer_pb.Entry + want bool + }{ + {"nil existing", nil, entryChunks(200, 4), false}, + {"identical -> skip", entryChunks(200, 4), entryChunks(200, 4), true}, + {"in-sync (fast path) -> skip", dstEntry(200, "A", "B"), srcEntry(200, "A", "B"), true}, + {"md5 equal and complete -> skip", entryMd5(200, 4, md5A), entryMd5(200, 4, md5A), true}, + {"truncated (no md5) -> write", entryChunks(200, 2), entryChunks(200, 4), false}, + {"md5-backed truncated -> write", entryMd5(200, 2, md5A), entryMd5(200, 4, md5A), false}, + // the resolved write installs an older version over the deleted latest's + // newer copy: different content must be replaced regardless of mtime + {"complete, dest newer, different -> write", entryMd5(300, 4, md5A), entryMd5(200, 4, md5B), false}, + {"newer smaller dest -> write", ivEntry(300, [2]int64{0, 5}), ivEntry(100, [2]int64{0, 10}), false}, + {"equal-mtime different -> write", entryMd5(200, 4, md5A), entryMd5(200, 4, md5B), false}, + // chunk-less inline files: matching on the empty chunk lists would keep + // the previous version's bytes — only identical inline content may skip + {"inline same -> skip", inlineEntry(200, "x"), inlineEntry(200, "x"), true}, + {"inline different -> write", inlineEntry(200, "x"), inlineEntry(200, "y"), false}, + {"inline different, dest newer -> write", inlineEntry(300, "x"), inlineEntry(200, "y"), false}, + {"inline vs empty -> write", inlineEntry(200, "x"), entry(200, 0), false}, + // identical content is not enough: a skip keeps the stored metadata, so a + // stale writable attribute or extended entry must force the rewrite + {"same chunks, different mtime -> write", dstEntry(300, "A", "B"), srcEntry(200, "A", "B"), false}, + {"same chunks, stale tag -> write", extEntry(dstEntry(200, "A", "B"), "k", "v1"), extEntry(srcEntry(200, "A", "B"), "k", "v2"), false}, + {"same chunks, same tag -> skip", extEntry(dstEntry(200, "A", "B"), "k", "v"), extEntry(srcEntry(200, "A", "B"), "k", "v"), true}, + {"same chunks, renamed empty tag -> write", extEntry(dstEntry(200, "A", "B"), "a", ""), extEntry(srcEntry(200, "A", "B"), "b", ""), false}, + {"same chunks, same empty tag -> skip", extEntry(dstEntry(200, "A", "B"), "a", ""), extEntry(srcEntry(200, "A", "B"), "a", ""), true}, + {"same chunks, stale uid -> write", func() *filer_pb.Entry { e := dstEntry(200, "A", "B"); e.Attributes.Uid = 1; return e }(), srcEntry(200, "A", "B"), false}, + // a TTL-less write inherits the destination's storage-rule TTL, so a + // stored rule TTL is one a rewrite reproduces — only an explicit TTL + // mismatch forces it + {"same chunks, rule ttl -> skip", func() *filer_pb.Entry { e := dstEntry(200, "A", "B"); e.Attributes.TtlSec = 60; return e }(), srcEntry(200, "A", "B"), true}, + {"same chunks, ttl mismatch -> write", func() *filer_pb.Entry { e := dstEntry(200, "A", "B"); e.Attributes.TtlSec = 60; return e }(), func() *filer_pb.Entry { e := srcEntry(200, "A", "B"); e.Attributes.TtlSec = 30; return e }(), false}, + // fields a rewrite cannot change must not block the skip: inode/crtime are + // filer-assigned or preserved, and a stored atime/ctime defaults to mtime + {"assigned fields differ -> skip", func() *filer_pb.Entry { + e := dstEntry(200, "A", "B") + e.Attributes.Inode = 42 + e.Attributes.Crtime = 999 + e.Attributes.Atime = 200 + e.Attributes.Ctime = 200 + return e + }(), srcEntry(200, "A", "B"), true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := destinationHoldsContent(tc.existing, tc.incoming); got != tc.want { + t.Fatalf("destinationHoldsContent = %v, want %v", got, tc.want) + } + }) + } +} + // destinationMatchesReference is the allocation-free fast path: true only when the // destination holds exactly the reference's chunks in order (by SourceFileId). It // must be conservative — never a false positive — so reorders, count mismatches, @@ -280,6 +357,10 @@ func TestDestinationMatchesReference(t *testing.T) { &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: 0}, Chunks: []*filer_pb.FileChunk{ {FileId: "dst", SourceFileId: "A", Offset: 0, Size: 2}, }}, srcEntry(0, "A"), false}, + // equal chunk lists are not enough when content lives outside chunks + {"inline same", inlineEntry(0, "x"), inlineEntry(0, "x"), true}, + {"inline different", inlineEntry(0, "x"), inlineEntry(0, "y"), false}, + {"inline vs none", inlineEntry(0, "x"), srcEntry(0), false}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go index a3352e14ba..dd4c7bac07 100644 --- a/weed/replication/sink/replication_sink.go +++ b/weed/replication/sink/replication_sink.go @@ -40,6 +40,22 @@ type MetadataOnlyDeleter interface { DeleteEntryKeepingRemoteObject(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error } +// LatestVersionWriter is an optional capability for sinks whose CreateEntry +// guards a replayed write against the live source at the key's own path. A +// caller that already resolved the current source state — a flattened latest +// version lives under a different source path than the key it lands on — must +// not be skipped as stale by that check. +type LatestVersionWriter interface { + CreateEntryLatest(key string, entry *filer_pb.Entry, signatures []int32) error +} + +func CreateEntryLatest(s ReplicationSink, key string, entry *filer_pb.Entry, signatures []int32) error { + if w, ok := s.(LatestVersionWriter); ok { + return w.CreateEntryLatest(key, entry, signatures) + } + return s.CreateEntry(key, entry, signatures) +} + func DeleteEntry(s ReplicationSink, key string, isDirectory, deleteIncludeChunks, keepRemoteObject bool, signatures []int32) error { if deleter, ok := s.(MetadataOnlyDeleter); ok && keepRemoteObject { return deleter.DeleteEntryKeepingRemoteObject(key, isDirectory, deleteIncludeChunks, signatures) diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index f7ec672d29..ae6c3b98fd 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -2121,6 +2121,13 @@ func (s3a *S3ApiServer) recoverLatestVersionWithoutPointer(bucket, normalizedObj return nil, fmt.Errorf("no version metadata in .versions directory and no regular object found for %s/%s: %w", bucket, normalizedObject, filer_pb.ErrNotFound) } +// VersionIdFromEntry exposes versionIdFromEntry to callers outside this +// package that resolve version files the same way (e.g. filer.backup's +// flattener rescanning .versions). +func VersionIdFromEntry(entry *filer_pb.Entry) string { + return versionIdFromEntry(entry) +} + // versionIdFromEntry returns a .versions child entry's version id, preferring // the Seaweed-X-Amz-Version-Id attribute and falling back to the v_ // file name so entries written outside the normal PUT path still self-heal. diff --git a/weed/s3api/s3api_version_id.go b/weed/s3api/s3api_version_id.go index c481f54f7f..568ecfde84 100644 --- a/weed/s3api/s3api_version_id.go +++ b/weed/s3api/s3api_version_id.go @@ -99,6 +99,12 @@ func getVersionTimestamp(versionId string) int64 { return int64(timestampPart) } +// CompareVersionIds exposes compareVersionIds to callers outside this package +// that must pick the newest version the same way (e.g. filer.backup's flattener). +func CompareVersionIds(a, b string) int { + return compareVersionIds(a, b) +} + // compareVersionIds compares two version IDs for sorting (newest first) // Returns: negative if a is newer, positive if b is newer, 0 if equal // Handles both old and new format version IDs