seaweedfs/weed/command/filer_remote_gateway_buckets_test.go

package command

import (
	"slices"
	"testing"

	"github.com/seaweedfs/seaweedfs/weed/filer"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
	"github.com/seaweedfs/seaweedfs/weed/replication/source"
)

type recordingRemoteMaker struct{ client *recordingRemote }

func (m recordingRemoteMaker) Make(*remote_pb.RemoteConf) (remote_storage.RemoteStorageClient, error) {
	return m.client, nil
}
func (m recordingRemoteMaker) HasBucket() bool { return true }

func gatewayTestOption(storageType string) (*RemoteGatewayOptions, *recordingRemote) {
	remote := &recordingRemote{}
	remote_storage.RemoteStorageClientMakers[storageType] = recordingRemoteMaker{client: remote}
	filerAddress := "localhost:1"
	return &RemoteGatewayOptions{
		filerAddress: &filerAddress,
		bucketsDir:   "/buckets",
		mappings: &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{
			"/buckets/gold": {Name: storageType, Bucket: "gold", Path: "/"},
		}},
		remoteConfs: map[string]*remote_pb.RemoteConf{storageType: {Name: storageType, Type: storageType}},
	}, remote
}

// TestGatewayDroppedLocalCopyIsNotUploaded: remote.uncache drops a synced
// entry's chunks and clears its sync stamp; shouldSendToRemote reads the
// cleared stamp in the remote mtime's second as dirty, so the update reaches
// the upload. The object at the key is the entry's only content and the
// chunkless upload would replace it with zero bytes, so it is skipped.
func TestGatewayDroppedLocalCopyIsNotUploaded(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-uncache")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-uncache") })
	process, err := option.makeBucketedEventProcessor(&source.FilerSource{})
	if err != nil {
		t.Fatal(err)
	}

	resp := droppedCopyUpdate(
		droppedCopyEntry(droppedCopySynced, chunk("3,01", "e1")),
		droppedCopyEntry(droppedCopyStamp()),
	)
	if !shouldSendToRemote(resp.EventNotification.NewEntry) {
		t.Fatal("precondition: the cleared stamp reads the entry as dirty")
	}
	if err := process(resp); err != nil {
		t.Fatal(err)
	}
	if len(remote.writes) != 0 {
		t.Errorf("writes = %+v, want none: the remote object is the entry's content and an upload would zero it", remote.writes)
	}
	if len(remote.deletes) != 0 || len(remote.metadataUpdates) != 0 {
		t.Errorf("deletes = %+v, metadata updates = %+v, want the remote object left alone", remote.deletes, remote.metadataUpdates)
	}
}

// TestGatewayUncachedRewriteIsStillUploaded: a rewrite of an uncached entry
// holds local data, so the dropped-copy skip does not apply and the rewrite
// uploads. The stamp cannot reach a filer in the test, so the call errors;
// the upload is what is asserted.
func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-rewrite")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rewrite") })
	process, err := option.makeBucketedEventProcessor(&source.FilerSource{})
	if err != nil {
		t.Fatal(err)
	}

	rewrite := entryWith(droppedCopyName, droppedCopyStamp(), chunk("4,02", "e2"))
	resp := droppedCopyUpdate(droppedCopyEntry(droppedCopyStamp()), rewrite)
	if err := process(resp); err != nil {
		t.Logf("stamp against the test filer fails as expected: %v", err)
	}
	if len(remote.writes) != 1 {
		t.Fatalf("writes = %+v, want the rewrite uploaded once", remote.writes)
	}
	if !slices.Equal(remote.writtenChunks[0], []string{"4,02"}) {
		t.Errorf("uploaded chunks %v, want the rewrite's [4,02]", remote.writtenChunks[0])
	}
}

// TestGatewayRemoteOnlyRenameCopiesTheObject: a remote-only entry holds its
// content only as the object the renamed-from path maps to. The rename copies
// that object to the destination key, stamps the entry so reads stop going
// through the pending-source marker, and removes the source. Uploading the
// chunkless entry would put a zero-byte object over it, and deleting the
// source first would lose the only copy.
func TestGatewayRemoteOnlyRenameCopiesTheObject(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-rename")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rename") })

	oldPath := "/buckets/gold/dir/old.txt"
	stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-rename", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100}
	oldEntry := &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp}
	current := &filer_pb.Entry{
		Name:        "new.txt",
		Attributes:  &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5},
		RemoteEntry: stamp,
		Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(oldPath)},
	}
	resp := &filer_pb.SubscribeMetadataResponse{
		Directory: "/buckets/gold/dir",
		EventNotification: &filer_pb.EventNotification{
			OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: current,
		},
	}
	filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/gold/dir/new.txt": current}}

	remote.objects = map[string][]byte{"/dir/old.txt": []byte("hello")}
	remote.statETag = `"abc"`
	remote.statMtime = 100

	if err := gatewayRemoteOnlyRename(filerClient, option, resp, current); err != nil {
		t.Fatal(err)
	}
	if len(remote.writes) != 1 || remote.writes[0].Path != "/dir/new.txt" {
		t.Fatalf("writes = %+v, want the source object copied to /dir/new.txt", remote.writes)
	}
	if string(remote.written[0]) != "hello" {
		t.Errorf("copied %q, want the source object's content", remote.written[0])
	}
	if filerClient.updates != 1 {
		t.Errorf("stamped the entry %d times, want 1 to clear the pending-source marker", filerClient.updates)
	}
	if len(remote.deletes) != 1 || remote.deletes[0].Path != "/dir/old.txt" {
		t.Errorf("deletes = %+v, want the source object removed", remote.deletes)
	}
}

// TestGatewayRemoteOnlyRenameAlreadyComplete: when the destination already
// holds what the stamp describes (an earlier run copied it), the rename only
// refreshes the stamp and removes the source.
func TestGatewayRemoteOnlyRenameAlreadyComplete(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-rename-complete")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rename-complete") })

	oldPath := "/buckets/gold/dir/old.txt"
	stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-rename-complete", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100}
	oldEntry := &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp}
	current := &filer_pb.Entry{
		Name:        "new.txt",
		Attributes:  &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5},
		RemoteEntry: stamp,
		Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(oldPath)},
	}
	resp := &filer_pb.SubscribeMetadataResponse{
		Directory: "/buckets/gold/dir",
		EventNotification: &filer_pb.EventNotification{
			OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: current,
		},
	}
	filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/gold/dir/new.txt": current}}

	remote.objects = map[string][]byte{"/dir/old.txt": []byte("hello"), "/dir/new.txt": []byte("hello")}
	remote.statETag = `"abc"`
	remote.statMtime = 100

	if err := gatewayRemoteOnlyRename(filerClient, option, resp, current); err != nil {
		t.Fatal(err)
	}
	if len(remote.writes) != 0 {
		t.Errorf("writes = %+v, want none: the destination already holds the content", remote.writes)
	}
	if len(remote.deletes) != 1 || remote.deletes[0].Path != "/dir/old.txt" {
		t.Errorf("deletes = %+v, want the source object removed", remote.deletes)
	}
}

// TestGatewayRenameReleasesPendingSource: a rename whose old entry read a
// pending source (an earlier rename's deferred copy had not finished) leaves
// that source behind once the renamed entry no longer reads through a
// marker. The gateway deletes it, guarded by the old entry's stamp.
func TestGatewayRenameReleasesPendingSource(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-release")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-release") })

	stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-release", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100}
	oldEntry := &filer_pb.Entry{
		Name:        "mid.txt",
		Attributes:  &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5},
		RemoteEntry: stamp,
		Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/gold/dir/orig.txt")},
	}
	newEntry := &filer_pb.Entry{Name: "new.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, RemoteEntry: stamp}
	message := &filer_pb.EventNotification{OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: newEntry}
	filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{
		"/buckets/gold/dir/new.txt": newEntry,
	}}
	remote.objects = map[string][]byte{"/dir/orig.txt": []byte("hello")}
	remote.statETag = `"abc"`
	remote.statMtime = 100

	if err := releaseGatewayRenamedSource(filerClient, option, message); err != nil {
		t.Fatal(err)
	}
	if len(remote.deletes) != 1 || remote.deletes[0].Path != "/dir/orig.txt" {
		t.Errorf("deletes = %+v, want the pending source released", remote.deletes)
	}
}

// TestGatewayRenameKeepsHeldPendingSource: the pending source stays while a
// remote-only entry still holds its path.
func TestGatewayRenameKeepsHeldPendingSource(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-release")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-release") })

	stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-release", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100}
	oldEntry := &filer_pb.Entry{
		Name:        "mid.txt",
		Attributes:  &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5},
		RemoteEntry: stamp,
		Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/gold/dir/orig.txt")},
	}
	newEntry := &filer_pb.Entry{Name: "new.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, RemoteEntry: stamp}
	heldSource := &filer_pb.Entry{Name: "orig.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 300, FileSize: 5}, RemoteEntry: stamp}
	message := &filer_pb.EventNotification{OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: newEntry}
	filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{
		"/buckets/gold/dir/new.txt":  newEntry,
		"/buckets/gold/dir/orig.txt": heldSource,
	}}
	remote.objects = map[string][]byte{"/dir/orig.txt": []byte("hello")}
	remote.statETag = `"abc"`
	remote.statMtime = 100

	if err := releaseGatewayRenamedSource(filerClient, option, message); err != nil {
		t.Fatal(err)
	}
	if len(remote.deletes) != 0 {
		t.Errorf("deletes = %+v, want the pending source kept while an entry holds its path", remote.deletes)
	}
}

// TestGatewayRemoteOnlyRenameSourceOutOfCoverage: a rename source outside the
// gateway's buckets is left untouched; when the destination already holds
// the content the rename only refreshes the stamp instead of failing.
func TestGatewayRemoteOnlyRenameSourceOutOfCoverage(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-outsrc")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-outsrc") })

	stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-outsrc", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100}
	oldEntry := &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp}
	current := &filer_pb.Entry{
		Name:        "new.txt",
		Attributes:  &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5},
		RemoteEntry: stamp,
		Extended:    map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/unmanaged/dir/old.txt")},
	}
	resp := &filer_pb.SubscribeMetadataResponse{
		Directory: "/buckets/gold/dir",
		EventNotification: &filer_pb.EventNotification{
			OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: current,
		},
	}
	filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/gold/dir/new.txt": current}}

	remote.objects = map[string][]byte{"/dir/new.txt": []byte("hello")}
	remote.statETag = `"abc"`
	remote.statMtime = 100

	if err := gatewayRemoteOnlyRename(filerClient, option, resp, current); err != nil {
		t.Fatalf("the completed destination should finalize without touching the out-of-coverage source: %v", err)
	}
	if filerClient.updates != 1 {
		t.Errorf("stamped the entry %d times, want 1 to clear the pending-source marker", filerClient.updates)
	}
	if len(remote.deletes) != 0 || len(remote.writes) != 0 {
		t.Errorf("deletes = %+v, writes = %+v, want the unmanaged source and destination left alone", remote.deletes, remote.writes)
	}
}

// TestGatewayRenameOutKeepsRemoteOnlyObject: a remote-only entry renamed out
// of the gateway's coverage keeps the remote object: it is the entry's only
// content.
func TestGatewayRenameOutKeepsRemoteOnlyObject(t *testing.T) {
	option, remote := gatewayTestOption("gatewaytest-out")
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-out") })
	process, err := option.makeBucketedEventProcessor(&source.FilerSource{})
	if err != nil {
		t.Fatal(err)
	}

	stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-out", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100}
	resp := &filer_pb.SubscribeMetadataResponse{
		Directory: "/buckets/gold/dir",
		EventNotification: &filer_pb.EventNotification{
			OldEntry:      &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp},
			NewParentPath: "/outside/dir",
			NewEntry:      &filer_pb.Entry{Name: "new.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, RemoteEntry: stamp},
		},
	}
	if err := process(resp); err != nil {
		t.Fatal(err)
	}
	if len(remote.deletes) != 0 {
		t.Errorf("deletes = %+v, want the remote-only object kept", remote.deletes)
	}
}

func TestGatewayMetadataOnlyDeleteKeepsTheRemoteObject(t *testing.T) {
	remote := &recordingRemote{}
	remote_storage.RemoteStorageClientMakers["gatewaytest"] = recordingRemoteMaker{client: remote}
	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest") })
	option := &RemoteGatewayOptions{
		bucketsDir:  "/buckets",
		mappings:    &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{"/buckets/b": {Name: "gatewaytest", Bucket: "b", Path: "/"}}},
		remoteConfs: map[string]*remote_pb.RemoteConf{"gatewaytest": {Name: "gatewaytest", Type: "gatewaytest"}},
	}
	process, err := option.makeBucketedEventProcessor(nil)
	if err != nil {
		t.Fatal(err)
	}
	deleteEvent := func(extended map[string][]byte) *filer_pb.SubscribeMetadataResponse {
		return &filer_pb.SubscribeMetadataResponse{
			Directory: "/buckets/b/dir",
			EventNotification: &filer_pb.EventNotification{
				OldEntry: &filer_pb.Entry{Name: "obj.bin", Attributes: &filer_pb.FuseAttributes{}, Extended: extended},
			},
		}
	}

	if err := process(deleteEvent(map[string][]byte{filer.ExtKeepRemoteObjectKey: []byte("true")})); err != nil {
		t.Fatal(err)
	}
	if len(remote.deletes) != 0 {
		t.Fatalf("deletes = %+v, want none for a metadata-only delete", remote.deletes)
	}
	if err := process(deleteEvent(nil)); err != nil {
		t.Fatal(err)
	}
	if len(remote.deletes) != 1 {
		t.Fatalf("deletes = %+v, want the ordinary delete", remote.deletes)
	}
}