diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 0021c2ec52..feb37fbd7f 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -698,6 +698,7 @@ message SubscribeMetadataResponse { repeated SubscribeMetadataResponse events = 4; // batch of additional events (backlog catch-up) repeated LogFileChunkRef log_file_refs = 5; // log file chunk refs for client direct-read int64 flushed_ts_ns = 6; // local log-buffer flush watermark: everything at or below it is on disk + int64 mutation_ts_ns = 7; // metadata mutation commit stamp; ts_ns carries the delivery stream position instead } message ListMetadataSubscribersRequest { repeated string client_types = 1; // optional filter by client type, e.g. "mount"; empty = all @@ -822,6 +823,7 @@ message CacheRemoteObjectToLocalClusterRequest { string name = 2; int32 chunk_concurrency = 3; // parallel chunk downloads per file, 0 = default (8) int32 download_concurrency = 4; // multipart download concurrency per chunk (if supported by remote storage), 0 = default (5 for S3) + int64 max_size = 5; // reject remote objects larger than this before downloading, 0 = no limit } message CacheRemoteObjectToLocalClusterResponse { Entry entry = 1; diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index eaea6a9f0f..feb37fbd7f 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -823,6 +823,7 @@ message CacheRemoteObjectToLocalClusterRequest { string name = 2; int32 chunk_concurrency = 3; // parallel chunk downloads per file, 0 = default (8) int32 download_concurrency = 4; // multipart download concurrency per chunk (if supported by remote storage), 0 = default (5 for S3) + int64 max_size = 5; // reject remote objects larger than this before downloading, 0 = no limit } message CacheRemoteObjectToLocalClusterResponse { Entry entry = 1; diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 8ba6fcbfd3..cb5f44b2b5 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -5447,6 +5447,7 @@ type CacheRemoteObjectToLocalClusterRequest struct { Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` ChunkConcurrency int32 `protobuf:"varint,3,opt,name=chunk_concurrency,json=chunkConcurrency,proto3" json:"chunk_concurrency,omitempty"` // parallel chunk downloads per file, 0 = default (8) DownloadConcurrency int32 `protobuf:"varint,4,opt,name=download_concurrency,json=downloadConcurrency,proto3" json:"download_concurrency,omitempty"` // multipart download concurrency per chunk (if supported by remote storage), 0 = default (5 for S3) + MaxSize int64 `protobuf:"varint,5,opt,name=max_size,json=maxSize,proto3" json:"max_size,omitempty"` // reject remote objects larger than this before downloading, 0 = no limit unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -5509,6 +5510,13 @@ func (x *CacheRemoteObjectToLocalClusterRequest) GetDownloadConcurrency() int32 return 0 } +func (x *CacheRemoteObjectToLocalClusterRequest) GetMaxSize() int64 { + if x != nil { + return x.MaxSize + } + return 0 +} + type CacheRemoteObjectToLocalClusterResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Entry *Entry `protobuf:"bytes,1,opt,name=entry,proto3" json:"entry,omitempty"` @@ -7872,12 +7880,13 @@ const file_filer_proto_rawDesc = "" + "\x19worm_grace_period_seconds\x18\x0f \x01(\x04R\x16wormGracePeriodSeconds\x12=\n" + "\x1bworm_retention_time_seconds\x18\x10 \x01(\x04R\x18wormRetentionTimeSecondsB\b\n" + "\x06_fsyncB\a\n" + - "\x05_worm\"\xba\x01\n" + + "\x05_worm\"\xd5\x01\n" + "&CacheRemoteObjectToLocalClusterRequest\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x12\n" + "\x04name\x18\x02 \x01(\tR\x04name\x12+\n" + "\x11chunk_concurrency\x18\x03 \x01(\x05R\x10chunkConcurrency\x121\n" + - "\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\"\xdd\x01\n" + + "\x14download_concurrency\x18\x04 \x01(\x05R\x13downloadConcurrency\x12\x19\n" + + "\bmax_size\x18\x05 \x01(\x03R\amaxSize\"\xdd\x01\n" + "'CacheRemoteObjectToLocalClusterResponse\x12%\n" + "\x05entry\x18\x01 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12J\n" + "\x0emetadata_event\x18\x02 \x01(\v2#.filer_pb.SubscribeMetadataResponseR\rmetadataEvent\x12\x1a\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 29d11c7e15..8cd7734a72 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -5173,6 +5173,11 @@ func (m *CacheRemoteObjectToLocalClusterRequest) MarshalToSizedBufferVT(dAtA []b i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.MaxSize != 0 { + i = protohelpers.EncodeVarint(dAtA, i, uint64(m.MaxSize)) + i-- + dAtA[i] = 0x28 + } if m.DownloadConcurrency != 0 { i = protohelpers.EncodeVarint(dAtA, i, uint64(m.DownloadConcurrency)) i-- @@ -8683,6 +8688,9 @@ func (m *CacheRemoteObjectToLocalClusterRequest) SizeVT() (n int) { if m.DownloadConcurrency != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.DownloadConcurrency)) } + if m.MaxSize != 0 { + n += 1 + protohelpers.SizeOfVarint(uint64(m.MaxSize)) + } n += len(m.unknownFields) return n } @@ -23123,6 +23131,25 @@ func (m *CacheRemoteObjectToLocalClusterRequest) UnmarshalVT(dAtA []byte) error break } } + case 5: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field MaxSize", wireType) + } + m.MaxSize = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.MaxSize |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go index 18be095c59..78f5e1f610 100644 --- a/weed/server/filer_grpc_server_remote.go +++ b/weed/server/filer_grpc_server_remote.go @@ -31,42 +31,74 @@ func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req // This benefits all clients: S3 API, filer HTTP, Hadoop, etc. cacheKey := req.Directory + "/" + req.Name + // A caller's size bound is enforced on the entry it resolves, before it + // joins the shared download: the rejection stays with the bounded caller + // while allowed downloads still dedupe by path across all callers. + if req.MaxSize > 0 { + if entry, _, err := fs.fencedFindEntry(ctx, util.JoinPath(req.Directory, req.Name)); err == nil && + entry.IsInRemoteOnly() && entry.Remote.RemoteSize > req.MaxSize { + return nil, &remoteObjectTooLargeError{path: cacheKey, size: entry.Remote.RemoteSize, limit: req.MaxSize} + } + } + // Detach from caller ctx: on failure the error path deletes every chunk // already written, so cancelling mid-download loses all progress. For // blobs large enough that the download outlasts the caller's timeout // the retry loop never converges. bgCtx := context.WithoutCancel(ctx) - // DoChan (vs Do) so the caller can bail out on ctx.Done() while the - // singleflight goroutine keeps caching on bgCtx; otherwise this handler - // goroutine stays blocked for the full download after the client is gone. - ch := fs.remoteCacheGroup.DoChan(cacheKey, func() (interface{}, error) { - return fs.doCacheRemoteObjectToLocalCluster(bgCtx, req) - }) + for { + // DoChan (vs Do) so the caller can bail out on ctx.Done() while the + // singleflight goroutine keeps caching on bgCtx; otherwise this handler + // goroutine stays blocked for the full download after the client is gone. + ch := fs.remoteCacheGroup.DoChan(cacheKey, func() (interface{}, error) { + return fs.doCacheRemoteObjectToLocalCluster(bgCtx, req) + }) - select { - case <-ctx.Done(): - // Caller gave up; the detached cache keeps running and a later - // request will find the entry cached (or join the same singleflight). - return nil, ctx.Err() - case res := <-ch: - if res.Shared { - glog.V(2).Infof("CacheRemoteObjectToLocalCluster: shared result for %s", cacheKey) + select { + case <-ctx.Done(): + // Caller gave up; the detached cache keeps running and a later + // request will find the entry cached (or join the same singleflight). + return nil, ctx.Err() + case res := <-ch: + if res.Shared { + glog.V(2).Infof("CacheRemoteObjectToLocalCluster: shared result for %s", cacheKey) + } + if res.Err != nil { + var tooLarge *remoteObjectTooLargeError + if errors.As(res.Err, &tooLarge) && (req.MaxSize == 0 || tooLarge.size <= req.MaxSize) { + // The shared flight ran under a co-tenant's tighter bound; this + // caller's own bound allows the object, so it starts a fresh + // flight rather than share that rejection. + continue + } + return nil, cacheRemoteObjectError(res.Err) + } + if res.Val == nil { + return nil, fmt.Errorf("unexpected nil result from singleflight") + } + resp, ok := res.Val.(*filer_pb.CacheRemoteObjectToLocalClusterResponse) + if !ok { + return nil, fmt.Errorf("unexpected result type from singleflight") + } + return resp, nil } - if res.Err != nil { - return nil, cacheRemoteObjectError(res.Err) - } - if res.Val == nil { - return nil, fmt.Errorf("unexpected nil result from singleflight") - } - resp, ok := res.Val.(*filer_pb.CacheRemoteObjectToLocalClusterResponse) - if !ok { - return nil, fmt.Errorf("unexpected result type from singleflight") - } - return resp, nil } } +// remoteObjectTooLargeError reports that the entry a shared download resolved +// exceeds a bound. It travels through the singleflight result, so callers +// whose own bound permits the size retry instead of inheriting the rejection. +type remoteObjectTooLargeError struct { + path string + size int64 + limit int64 +} + +func (e *remoteObjectTooLargeError) Error() string { + return fmt.Sprintf("remote object %s is %d bytes, over the %d byte limit", e.path, e.size, e.limit) +} + // doCacheRemoteObjectToLocalCluster performs the actual caching operation. // This is called from singleflight, so only one instance runs per object. func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { @@ -94,6 +126,12 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re return resp, nil } + // Enforce the bound on the entry this flight downloads: the caller's + // pre-flight check read an entry a remote metadata refresh can replace. + if req.MaxSize > 0 && entry.Remote.RemoteSize > req.MaxSize { + return nil, &remoteObjectTooLargeError{path: string(lockPath), size: entry.Remote.RemoteSize, limit: req.MaxSize} + } + glog.V(1).Infof("CacheRemoteObjectToLocalCluster: caching %s/%s (remote size: %d)", req.Directory, req.Name, entry.Remote.RemoteSize) readDir, readName := fs.filer.RemoteReadPath(entry, req.Directory, req.Name) diff --git a/weed/server/filer_grpc_server_remote_test.go b/weed/server/filer_grpc_server_remote_test.go index 7c5f165b3a..73ce5358bc 100644 --- a/weed/server/filer_grpc_server_remote_test.go +++ b/weed/server/filer_grpc_server_remote_test.go @@ -227,3 +227,69 @@ func TestPruneOnANonOwnerLeavesTheDeleteToTheOwner(t *testing.T) { assert.False(t, fs.pruneEntryMissingFromRemote(context.Background(), entry.FullPath, entry, remote_storage.ErrRemoteObjectNotFound)) assert.Contains(t, store.entries, string(entry.FullPath), "only the write owner may delete the entry") } + +func pollFindCalls(store *renameTestStore, path string, n int) bool { + for i := 0; i < 4000; i++ { + store.mu.Lock() + calls := store.findCalls[path] + store.mu.Unlock() + if calls >= n { + return true + } + time.Sleep(time.Millisecond) + } + return false +} + +func awaitFindCalls(t *testing.T, store *renameTestStore, path string, n int) { + t.Helper() + require.True(t, pollFindCalls(store, path, n), "entry %s saw fewer than %d finds", path, n) +} + +// A remote metadata refresh can swap the entry between a caller's pre-flight +// check and the lookup the shared download resolves: the bound must apply to +// the latter, and a co-tenant sharing the flight must not inherit a rejection +// its own bound does not call for. +func TestCacheBoundAppliesToTheDownloadedEntry(t *testing.T) { + fs, store, _ := newRemoteMountTestServer(t) + store.findDelay = 500 * time.Millisecond + + small := remoteOnlyFileEntry("/other/obj.bin") + small.Remote.RemoteSize = 4 + store.entries[string(small.FullPath)] = small + + big := small.ShallowClone() + big.Remote = proto.Clone(small.Remote).(*filer_pb.RemoteEntry) + big.Remote.RemoteSize = 100 + go func() { + if !pollFindCalls(store, string(small.FullPath), 1) { + return + } + store.mu.Lock() + store.entries[string(small.FullPath)] = big + store.mu.Unlock() + }() + + bounded := make(chan error, 1) + go func() { + _, err := fs.CacheRemoteObjectToLocalCluster(context.Background(), &filer_pb.CacheRemoteObjectToLocalClusterRequest{ + Directory: "/other", Name: "obj.bin", MaxSize: 5, + }) + bounded <- err + }() + + awaitFindCalls(t, store, string(small.FullPath), 1) + time.Sleep(80 * time.Millisecond) // let the bounded flight reach its own find + _, joinErr := fs.CacheRemoteObjectToLocalCluster(context.Background(), &filer_pb.CacheRemoteObjectToLocalClusterRequest{ + Directory: "/other", Name: "obj.bin", + }) + + err := <-bounded + var tooLarge *remoteObjectTooLargeError + require.ErrorAs(t, err, &tooLarge) + assert.Equal(t, int64(100), tooLarge.size) + assert.Equal(t, int64(5), tooLarge.limit) + + require.Error(t, joinErr) + assert.False(t, errors.As(joinErr, &tooLarge), "unbounded joiner retries rather than sharing the rejection") +} diff --git a/weed/worker/tasks/iceberg/compact_order_test.go b/weed/worker/tasks/iceberg/compact_order_test.go index 1a36f7618f..49367a5a6d 100644 --- a/weed/worker/tasks/iceberg/compact_order_test.go +++ b/weed/worker/tasks/iceberg/compact_order_test.go @@ -363,6 +363,7 @@ func TestResolveCompactionOrderSkipsUncoveredSortField(t *testing.T) { t.Fatalf("uncovered sort field: got %+v, want nil", order) } } + // A full-bin repack that produces nothing must not discard the ordered runs // it was meant to replace — they are still valid compaction work. func TestCompactDataFilesKeepsRunsWhenRepackFails(t *testing.T) { diff --git a/weed/worker/tasks/iceberg/filer_io.go b/weed/worker/tasks/iceberg/filer_io.go index b2a447eeb7..2531e1ecb5 100644 --- a/weed/worker/tasks/iceberg/filer_io.go +++ b/weed/worker/tasks/iceberg/filer_io.go @@ -181,14 +181,24 @@ func (s *tableState) loadMetadata(ctx context.Context, client filer_pb.SeaweedFi if s.Metadata != nil { return nil } - entry, err := lookupFileByIcebergPath(ctx, client, bucketName, s.DataPath, s.MetadataLocation) + entry, fullPath, err := lookupResolvedEntry(ctx, client, bucketName, s.DataPath, s.MetadataLocation) if err != nil { return fmt.Errorf("load metadata file %s: %w", s.MetadataLocation, err) } - if len(entry.Content) == 0 && len(entry.Chunks) > 0 { - if size := int64(filer.FileSize(entry)); size > metadataFileMaxBytes { - return fmt.Errorf("metadata file %s too large: %d bytes", s.MetadataLocation, size) - } + // Bound what the location would pull before it is pulled: a remote + // entry's object is downloaded whole by the cache call, a chunked + // file's read is bounded by FileSize. + if size := metadataEntrySize(entry); size > metadataFileMaxBytes { + return fmt.Errorf("metadata file %s too large: %d bytes", s.MetadataLocation, size) + } + entry, err = cacheRemoteOnlyEntry(ctx, client, fullPath, entry, metadataFileMaxBytes) + if err != nil { + return fmt.Errorf("load metadata file %s: %w", s.MetadataLocation, err) + } + // The cache call returns the entry the filer then held; a replacement + // written during the download keeps the name but not the checked size. + if size := metadataEntrySize(entry); size > metadataFileMaxBytes { + return fmt.Errorf("metadata file %s too large: %d bytes", s.MetadataLocation, size) } content, err := readEntryContent(ctx, client, entry, s.MetadataLocation) if err != nil { @@ -202,6 +212,23 @@ func (s *tableState) loadMetadata(ctx context.Context, client filer_pb.SeaweedFi return nil } +// metadataEntrySize is the bytes an entry's content would pull into memory: +// a remote object's size or a chunked file's FileSize. Inline content is +// exempt — the REST load path accepts it unbounded and the lookup already +// holds it. +func metadataEntrySize(entry *filer_pb.Entry) int64 { + if len(entry.Content) > 0 { + return 0 + } + if isRemoteOnly(entry) { + return entry.GetRemoteEntry().GetRemoteSize() + } + if len(entry.Chunks) > 0 { + return int64(filer.FileSize(entry)) + } + return 0 +} + func isFilerNotFound(err error) bool { return errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound } @@ -258,10 +285,57 @@ func addFilerPathSpellings(set map[string]struct{}, bucketName, dataPath, iceber } // lookupFileByIcebergPath returns the filer entry holding an Iceberg-style -// path. dataPath is the bucket-relative directory the table's files live in. +// path, ready to read. dataPath is the bucket-relative directory the table's +// files live in. A file of a remote-mounted bucket that is not cached locally +// is cached first, as the S3 gateway does on a read. func lookupFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, dataPath, icebergPath string) (*filer_pb.Entry, error) { - entry, _, err := lookupResolvedEntry(ctx, client, bucketName, dataPath, icebergPath) - return entry, err + entry, fullPath, err := lookupResolvedEntry(ctx, client, bucketName, dataPath, icebergPath) + if err != nil { + return nil, err + } + return cacheRemoteOnlyEntry(ctx, client, fullPath, entry, 0) +} + +// cacheRemoteOnlyEntry downloads a remote-only entry's object to the local +// cluster and returns the cached entry. maxSize bounds what the filer will +// download: the entry it resolves at cache time can differ from the one +// checked here, and the download would run to its full remote size. One +// retry covers a cache call racing an uncache of the same entry; an entry +// still remote-only after caching is an error. +func cacheRemoteOnlyEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, fullPath string, entry *filer_pb.Entry, maxSize int64) (*filer_pb.Entry, error) { + if !isRemoteOnly(entry) { + return entry, nil + } + dir, name := path.Dir(fullPath), path.Base(fullPath) + var err error + for attempt := 0; attempt < 2 && isRemoteOnly(entry); attempt++ { + var resp *filer_pb.CacheRemoteObjectToLocalClusterResponse + resp, err = client.CacheRemoteObjectToLocalCluster(ctx, &filer_pb.CacheRemoteObjectToLocalClusterRequest{ + Directory: dir, + Name: name, + MaxSize: maxSize, + }) + switch { + case err != nil: + case resp == nil || resp.Entry == nil: + err = errors.New("no entry returned") + default: + entry = resp.Entry + } + } + if err != nil { + return nil, fmt.Errorf("cache remote file %s: %w", fullPath, err) + } + if isRemoteOnly(entry) { + return nil, fmt.Errorf("cache remote file %s: still remote-only after caching", fullPath) + } + return entry, nil +} + +// isRemoteOnly reports whether an entry of a remote-mounted bucket holds no +// bytes locally: no chunks and no inline content, only its RemoteEntry. +func isRemoteOnly(entry *filer_pb.Entry) bool { + return entry.IsInRemoteOnly() && len(entry.Content) == 0 } // loadFileByIcebergPath loads a file from the filer given an Iceberg-style path. @@ -276,9 +350,14 @@ func loadFileByIcebergPath(ctx context.Context, client filer_pb.SeaweedFilerClie // readEntryContent returns an entry's bytes: its inline content, or a chunked // file read through the volume servers. func readEntryContent(ctx context.Context, client filer_pb.SeaweedFilerClient, entry *filer_pb.Entry, name string) ([]byte, error) { - if len(entry.Content) > 0 || len(entry.Chunks) == 0 { + if len(entry.Content) > 0 { return entry.Content, nil } + if len(entry.Chunks) == 0 { + // Everything read here is Iceberg metadata, a manifest or a data + // file, none of which can be empty. + return nil, fmt.Errorf("file %s is empty (%d bytes in its attributes, no content or chunks)", name, entry.GetAttributes().GetFileSize()) + } initGlobalHTTPClientOnce.Do(util_http.InitGlobalHttpClient) reader := filer.NewFileReader(singleFilerClient{client: client}, entry) diff --git a/weed/worker/tasks/iceberg/filer_io_save_test.go b/weed/worker/tasks/iceberg/filer_io_save_test.go index 2040f11e5f..45bf79f84a 100644 --- a/weed/worker/tasks/iceberg/filer_io_save_test.go +++ b/weed/worker/tasks/iceberg/filer_io_save_test.go @@ -1,6 +1,7 @@ package iceberg import ( + "bytes" "context" "encoding/json" "io" @@ -10,12 +11,14 @@ import ( "strings" "sync" "testing" + "time" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) // fakeVolumeServer accepts chunk uploads and records what it received, so a -// test can check the bytes actually left the filer entry. +// test can check the bytes actually left the filer entry. It also serves +// chunk reads of what it holds. type fakeVolumeServer struct { mu sync.Mutex server *httptest.Server @@ -29,6 +32,18 @@ func startFakeVolumeServer(t *testing.T) *fakeVolumeServer { v := &fakeVolumeServer{parts: make(map[string][]byte)} v.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fid := strings.TrimPrefix(r.URL.Path, "/") + if r.Method == http.MethodGet || r.Method == http.MethodHead { + // A chunk read, answered as a volume server does (ranges included). + v.mu.Lock() + data, ok := v.parts[fid] + v.mu.Unlock() + if !ok { + http.NotFound(w, r) + return + } + http.ServeContent(w, r, "", time.Time{}, bytes.NewReader(data)) + return + } // The uploader sends the payload as a single form-data part named // "file" with an empty filename, which FormFile will not match. reader, err := r.MultipartReader() @@ -58,6 +73,12 @@ func startFakeVolumeServer(t *testing.T) *fakeVolumeServer { return v } +func (v *fakeVolumeServer) put(fid string, data []byte) { + v.mu.Lock() + defer v.mu.Unlock() + v.parts[fid] = data +} + func (v *fakeVolumeServer) hostPort(t *testing.T) string { t.Helper() u, err := url.Parse(v.server.URL) diff --git a/weed/worker/tasks/iceberg/remote_only_test.go b/weed/worker/tasks/iceberg/remote_only_test.go new file mode 100644 index 0000000000..7421303c2d --- /dev/null +++ b/weed/worker/tasks/iceberg/remote_only_test.go @@ -0,0 +1,304 @@ +package iceberg + +import ( + "bytes" + "context" + "fmt" + "net" + "path" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// remoteCachingFiler is the fake filer plus what a remote mount needs, shaped +// as the real filer answers: a file of a remote-mounted bucket that is not +// cached locally has only its RemoteEntry; caching it writes its bytes to a +// volume and returns the entry with chunks and no inline content, so a read +// goes through the chunk reader, the volume lookup and the volume server. +type remoteCachingFiler struct { + *fakeFilerServer + volume *fakeVolumeServer + volumeHost string + + mu sync.Mutex + remote map[string][]byte // dir/name -> the remote object's bytes + cacheCalls int + nextFid int + stayRemote bool // answer without caching, as on a racing uncache + failFirst int // fail the first N cache calls + // cacheEntry swaps the stored entry at cache time, as a replacement + // written during the download would. + cacheEntry func(dir, name string, entry *filer_pb.Entry) *filer_pb.Entry +} + +func (f *remoteCachingFiler) CacheRemoteObjectToLocalCluster(_ context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + f.mu.Lock() + body, ok := f.remote[req.Directory+"/"+req.Name] + f.cacheCalls++ + f.nextFid++ + fid := fmt.Sprintf("7,%08x", f.nextFid) + stay := f.stayRemote + fail := f.failFirst > 0 + if fail { + f.failFirst-- + } + f.mu.Unlock() + if fail { + return nil, fmt.Errorf("remote read timed out") + } + entry := f.getEntry(req.Directory, req.Name) + if !ok || entry == nil { + return nil, fmt.Errorf("no remote object %s/%s", req.Directory, req.Name) + } + if req.MaxSize > 0 && entry.GetRemoteEntry().GetRemoteSize() > req.MaxSize { + return nil, fmt.Errorf("remote object %s/%s is %d bytes, over the %d byte limit", + req.Directory, req.Name, entry.GetRemoteEntry().GetRemoteSize(), req.MaxSize) + } + if stay { + return &filer_pb.CacheRemoteObjectToLocalClusterResponse{Entry: entry}, nil + } + if f.cacheEntry != nil { + entry = f.cacheEntry(req.Directory, req.Name, entry) + } + f.volume.put(fid, body) + cached := &filer_pb.Entry{Name: entry.Name, Attributes: entry.Attributes, RemoteEntry: entry.RemoteEntry, + Chunks: []*filer_pb.FileChunk{{FileId: fid, Offset: 0, Size: uint64(len(body)), ModifiedTsNs: time.Now().UnixNano()}}} + f.putEntry(req.Directory, req.Name, cached) + return &filer_pb.CacheRemoteObjectToLocalClusterResponse{Entry: cached}, nil +} + +func (f *remoteCachingFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { + out := &filer_pb.LookupVolumeResponse{LocationsMap: map[string]*filer_pb.Locations{}} + for _, vid := range req.VolumeIds { + out.LocationsMap[vid] = &filer_pb.Locations{Locations: []*filer_pb.Location{{Url: f.volumeHost, PublicUrl: f.volumeHost}}} + } + return out, nil +} + +func startRemoteCachingFiler(t *testing.T) (*remoteCachingFiler, filer_pb.SeaweedFilerClient) { + t.Helper() + vol := startFakeVolumeServer(t) + srv := &remoteCachingFiler{fakeFilerServer: newFakeFilerServer(), volume: vol, volumeHost: vol.hostPort(t), remote: map[string][]byte{}} + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + server := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(server, srv) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.GracefulStop) + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { conn.Close() }) + return srv, filer_pb.NewSeaweedFilerClient(conn) +} + +// putRemoteOnly records body as the remote object behind an Iceberg path and +// puts its remote-only entry: no chunks, no content, only the RemoteEntry. +func putRemoteOnly(t *testing.T, srv *remoteCachingFiler, bucket, dataPath, file string, body []byte) { + t.Helper() + full, err := icebergFilerPath(bucket, dataPath, file) + if err != nil { + t.Fatalf("path: %v", err) + } + dir, name := path.Dir(full), path.Base(full) + srv.remote[dir+"/"+name] = body + srv.putEntry(dir, name, &filer_pb.Entry{Name: name, + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(body)), Mtime: time.Now().Unix()}, + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "r", RemoteSize: int64(len(body)), RemoteMtime: time.Now().Unix()}}) +} + +func TestRemoteOnlyFileIsCachedThenReadThroughItsChunks(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + const bucket, dataPath, file = "b", "ns/t", "data/d1-deletes.parquet" + putRemoteOnly(t, srv, bucket, dataPath, file, positionDeleteParquet(t, []positionDeleteTestRow{ + {"s3://b/ns/t/data/f1.parquet", 3}, {"s3://b/ns/t/data/f1.parquet", 9}, + })) + + var got map[string][]int64 + var err error + noPanic(t, func() { + got, err = readPositionDeleteFile(context.Background(), client, bucket, dataPath, file) + }) + if err != nil { + t.Fatalf("read: %v", err) + } + if srv.cacheCalls != 1 { + t.Fatalf("CacheRemoteObjectToLocalCluster called %d times, want 1", srv.cacheCalls) + } + if n := len(got["s3://b/ns/t/data/f1.parquet"]); n != 2 { + t.Fatalf("got %v, want the two positions of the remote object", got) + } +} + +func TestRemoteOnlyFileSchemaIsReadThroughItsChunks(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + const bucket, dataPath, file = "b", "ns/t", "data/d2-deletes.parquet" + putRemoteOnly(t, srv, bucket, dataPath, file, positionDeleteParquet(t, []positionDeleteTestRow{{"s3://b/ns/t/data/f1.parquet", 1}})) + schema, err := parquetSchemaByIcebergPath(context.Background(), client, bucket, dataPath, file) + if err != nil { + t.Fatalf("schema: %v", err) + } + if _, ok := schema.Lookup("pos"); !ok { + t.Fatalf("schema %v has no pos column", schema) + } +} + +func TestACachedFileIsNotCachedAgain(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + const bucket, dataPath, file = "b", "ns/t", "data/d3-deletes.parquet" + full, _ := icebergFilerPath(bucket, dataPath, file) + body := positionDeleteParquet(t, []positionDeleteTestRow{{"s3://b/ns/t/data/f1.parquet", 1}}) + srv.volume.put("7,000000ff", body) + srv.putEntry(path.Dir(full), path.Base(full), &filer_pb.Entry{Name: path.Base(full), + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(body))}, + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "r", RemoteSize: int64(len(body))}, + Chunks: []*filer_pb.FileChunk{{FileId: "7,000000ff", Offset: 0, Size: uint64(len(body)), ModifiedTsNs: 1}}}) + if _, err := readPositionDeleteFile(context.Background(), client, bucket, dataPath, file); err != nil { + t.Fatalf("read: %v", err) + } + if srv.cacheCalls != 0 { + t.Fatalf("CacheRemoteObjectToLocalCluster called %d times for a cached file", srv.cacheCalls) + } +} + +func TestAFileStillRemoteOnlyAfterCachingIsACacheError(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + srv.stayRemote = true + const bucket, dataPath, file = "b", "ns/t", "data/d4-deletes.parquet" + putRemoteOnly(t, srv, bucket, dataPath, file, positionDeleteParquet(t, []positionDeleteTestRow{{"s3://b/ns/t/data/f1.parquet", 1}})) + var err error + noPanic(t, func() { + _, err = readPositionDeleteFile(context.Background(), client, bucket, dataPath, file) + }) + if err == nil || !strings.Contains(err.Error(), "still remote-only") { + t.Fatalf("err = %v, want a cache error", err) + } +} + +func TestAFailedCacheCallIsRetriedOnce(t *testing.T) { + const bucket, dataPath, file = "b", "ns/t", "data/d5-deletes.parquet" + body := positionDeleteParquet(t, []positionDeleteTestRow{{"s3://b/ns/t/data/f1.parquet", 1}}) + + srv, client := startRemoteCachingFiler(t) + putRemoteOnly(t, srv, bucket, dataPath, file, body) + srv.failFirst = 1 + if _, err := readPositionDeleteFile(context.Background(), client, bucket, dataPath, file); err != nil { + t.Fatalf("one failed cache call, then a good one: %v", err) + } + if srv.cacheCalls != 2 { + t.Fatalf("%d cache calls, want 2", srv.cacheCalls) + } + + srv, client = startRemoteCachingFiler(t) + putRemoteOnly(t, srv, bucket, dataPath, file, body) + srv.failFirst = 2 + if _, err := readPositionDeleteFile(context.Background(), client, bucket, dataPath, file); err == nil || !strings.Contains(err.Error(), "remote read timed out") { + t.Fatalf("two failed cache calls: err = %v, want the cache error", err) + } +} + +// A location-only record pointing at a remote object over the metadata bound +// is rejected on the remote entry's size, without downloading the object: the +// bound exists because the catalog pointer may name any file in the bucket. +func TestARemoteOnlyMetadataFileOverTheBoundIsNotCached(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + original := metadataFileMaxBytes + metadataFileMaxBytes = 8 + defer func() { metadataFileMaxBytes = original }() + + setup := tableSetup{BucketName: "b", Namespace: "ns", TableName: "t"} + putRemoteOnly(t, srv, setup.BucketName, setup.tablePath(), "metadata/big.metadata.json", + bytes.Repeat([]byte{1}, int(metadataFileMaxBytes)+1)) + state := &tableState{DataPath: setup.tablePath(), MetadataLocation: fmt.Sprintf( + "s3://%s/%s/metadata/big.metadata.json", setup.BucketName, setup.tablePath())} + if err := state.loadMetadata(context.Background(), client, setup.BucketName); err == nil || !strings.Contains(err.Error(), "too large") { + t.Fatalf("err = %v, want a too-large rejection", err) + } + if srv.cacheCalls != 0 { + t.Fatalf("%d cache calls, want 0", srv.cacheCalls) + } +} + +// A remote object that grew past the bound between the worker's check and the +// filer's cache resolve is rejected before its bytes are downloaded: the +// bound rides the cache request for the entry the filer resolves, not only +// the one the worker saw. +func TestAMetadataFileThatGrewPastTheBoundIsNotDownloaded(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + original := metadataFileMaxBytes + metadataFileMaxBytes = 8 + defer func() { metadataFileMaxBytes = original }() + + setup := tableSetup{BucketName: "b", Namespace: "ns", TableName: "t"} + putRemoteOnly(t, srv, setup.BucketName, setup.tablePath(), "metadata/grew.metadata.json", []byte{1}) + full, err := icebergFilerPath(setup.BucketName, setup.tablePath(), "metadata/grew.metadata.json") + if err != nil { + t.Fatalf("path: %v", err) + } + dir, name := path.Dir(full), path.Base(full) + client = swapEntryOnCacheClient{SeaweedFilerClient: client, swap: func() { + srv.putEntry(dir, name, &filer_pb.Entry{Name: name, + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(metadataFileMaxBytes) + 1, Mtime: time.Now().Unix()}, + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "r", RemoteSize: metadataFileMaxBytes + 1, RemoteMtime: time.Now().Unix()}}) + }} + state := &tableState{DataPath: setup.tablePath(), MetadataLocation: fmt.Sprintf( + "s3://%s/%s/metadata/grew.metadata.json", setup.BucketName, setup.tablePath())} + if err := state.loadMetadata(context.Background(), client, setup.BucketName); err == nil || !strings.Contains(err.Error(), "over the") { + t.Fatalf("err = %v, want a bound rejection", err) + } +} + +// swapEntryOnCacheClient runs swap inside the cache call, as a replacement +// write landing between the worker's lookup and the filer's resolve would. +type swapEntryOnCacheClient struct { + filer_pb.SeaweedFilerClient + swap func() +} + +func (c swapEntryOnCacheClient) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest, opts ...grpc.CallOption) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + c.swap() + return c.SeaweedFilerClient.CacheRemoteObjectToLocalCluster(ctx, req, opts...) +} + +// The bound also applies to the entry the cache call returns: a replacement +// written during the download can be larger than the remote entry that was +// checked. +func TestAReplacedMetadataFileOverTheBoundIsNotRead(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + original := metadataFileMaxBytes + metadataFileMaxBytes = 8 + defer func() { metadataFileMaxBytes = original }() + + setup := tableSetup{BucketName: "b", Namespace: "ns", TableName: "t"} + putRemoteOnly(t, srv, setup.BucketName, setup.tablePath(), "metadata/swapped.metadata.json", + bytes.Repeat([]byte{1}, 4)) + srv.cacheEntry = func(_, _ string, entry *filer_pb.Entry) *filer_pb.Entry { + return &filer_pb.Entry{Name: entry.Name, RemoteEntry: entry.RemoteEntry, + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(metadataFileMaxBytes) + 1, Mtime: time.Now().Unix()}} + } + state := &tableState{DataPath: setup.tablePath(), MetadataLocation: fmt.Sprintf( + "s3://%s/%s/metadata/swapped.metadata.json", setup.BucketName, setup.tablePath())} + if err := state.loadMetadata(context.Background(), client, setup.BucketName); err == nil || !strings.Contains(err.Error(), "too large") { + t.Fatalf("err = %v, want a too-large rejection", err) + } +} + +func TestAFileWithNoContentOrChunksIsAnEmptyFileError(t *testing.T) { + srv, client := startRemoteCachingFiler(t) + const bucket, dataPath, file = "b", "ns/t", "snap-1.avro" + full, _ := icebergFilerPath(bucket, dataPath, file) + srv.putEntry(path.Dir(full), path.Base(full), &filer_pb.Entry{Name: path.Base(full), + Attributes: &filer_pb.FuseAttributes{FileSize: 2057}}) // its size says bytes, it holds none + if _, err := loadFileByIcebergPath(context.Background(), client, bucket, dataPath, file); err == nil || !strings.Contains(err.Error(), "empty") { + t.Fatalf("err = %v, want an empty-file error", err) + } +}