diff --git a/weed/filer/remote_mapping.go b/weed/filer/remote_mapping.go index 9c488d92e..359fc4463 100644 --- a/weed/filer/remote_mapping.go +++ b/weed/filer/remote_mapping.go @@ -3,6 +3,7 @@ package filer import ( "context" "fmt" + "strings" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -11,6 +12,21 @@ import ( "google.golang.org/protobuf/proto" ) +// FindMountedRemoteMapping returns the mount point covering dir and its remote +// location. Matches on path components so a sibling name sharing a prefix does +// not collide, and the longest mount wins when mounts nest. +func FindMountedRemoteMapping(mappings *remote_pb.RemoteStorageMapping, dir string) (localMountedDir string, remoteStorageMountedLocation *remote_pb.RemoteStorageLocation, err error) { + for k, loc := range mappings.Mappings { + if (dir == k || strings.HasPrefix(dir, strings.TrimSuffix(k, "/")+"/")) && len(k) > len(localMountedDir) { + localMountedDir, remoteStorageMountedLocation = k, loc + } + } + if localMountedDir == "" { + return "", nil, fmt.Errorf("%s is not mounted", dir) + } + return +} + func ReadMountMappings(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress) (mappings *remote_pb.RemoteStorageMapping, readErr error) { var oldContent []byte if readErr = pb.WithFilerClient(false, 0, filerAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { diff --git a/weed/filer/remote_mapping_test.go b/weed/filer/remote_mapping_test.go new file mode 100644 index 000000000..15fdce1ce --- /dev/null +++ b/weed/filer/remote_mapping_test.go @@ -0,0 +1,48 @@ +package filer + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFindMountedRemoteMapping(t *testing.T) { + mappings := &remote_pb.RemoteStorageMapping{ + Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/b": {Name: "outer"}, + "/buckets/b/sub": {Name: "inner"}, + "/buckets/foo": {Name: "foo"}, + }, + } + + tests := []struct { + dir string + wantMount string + wantName string + wantErr bool + }{ + {dir: "/buckets/b", wantMount: "/buckets/b", wantName: "outer"}, + {dir: "/buckets/b/other", wantMount: "/buckets/b", wantName: "outer"}, + {dir: "/buckets/b/sub", wantMount: "/buckets/b/sub", wantName: "inner"}, + {dir: "/buckets/b/sub/deep", wantMount: "/buckets/b/sub", wantName: "inner"}, + {dir: "/buckets/foo/x", wantMount: "/buckets/foo", wantName: "foo"}, + // a sibling sharing a name prefix is not under the mount + {dir: "/buckets/foobar", wantErr: true}, + {dir: "/buckets/other", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.dir, func(t *testing.T) { + mount, loc, err := FindMountedRemoteMapping(mappings, tt.dir) + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.wantMount, mount) + assert.Equal(t, tt.wantName, loc.Name) + }) + } +} diff --git a/weed/filer/remote_storage.go b/weed/filer/remote_storage.go index de33c8d20..48096bcd7 100644 --- a/weed/filer/remote_storage.go +++ b/weed/filer/remote_storage.go @@ -164,15 +164,9 @@ func DetectMountInfo(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddre return mappings, "", nil, nil, fmt.Errorf("need to specify '-dir' option") } - var localMountedDir string - var remoteStorageMountedLocation *remote_pb.RemoteStorageLocation - for k, loc := range mappings.Mappings { - if strings.HasPrefix(dir, k) { - localMountedDir, remoteStorageMountedLocation = k, loc - } - } - if localMountedDir == "" { - return mappings, localMountedDir, remoteStorageMountedLocation, nil, fmt.Errorf("%s is not mounted", dir) + localMountedDir, remoteStorageMountedLocation, findErr := FindMountedRemoteMapping(mappings, dir) + if findErr != nil { + return mappings, localMountedDir, remoteStorageMountedLocation, nil, findErr } // find remote storage configuration diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index 5e8454859..e23ffbedd 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -110,6 +110,20 @@ type RemoteStorageStreamReader interface { ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) } +// CacheWaitTimeout is how long a read of an uncached remote-only object waits +// for the local cache before serving another way: small files wait longer since +// their cache completes quickly, large files fail fast for better TTFB. +func CacheWaitTimeout(remoteSize int64) time.Duration { + switch { + case remoteSize > 500*1024*1024: + return 2 * time.Second + case remoteSize > 0 && remoteSize < 50*1024*1024: + return 10 * time.Second + default: + return 5 * time.Second + } +} + type RemoteStorageClientMaker interface { Make(remoteConf *remote_pb.RemoteConf) (RemoteStorageClient, error) HasBucket() bool diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index bfa2eb3ca..38b73111c 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -389,7 +389,7 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * entries, _, err := s3a.list(uploadDirectory, "", "", false, s3_constants.MaxS3MultipartParts+1) if err != nil { glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err) - if isFilerListNotFound(err) { + if isFilerNotFound(err) { stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc() return nil, nil, s3err.ErrNoSuchUpload } @@ -1041,7 +1041,7 @@ func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput if err != nil { // A missing .uploads folder normally lists as empty with no error; a // store that reports it as not-found still means an empty list. - if isFilerListNotFound(err) { + if isFilerNotFound(err) { return output, s3err.ErrNone } // surface a real store error; a masked empty 200 makes a resuming client treat the upload as gone @@ -1108,7 +1108,7 @@ func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListP if err != nil { // A store that reports the missing upload directory as not-found means // the upload is gone (completed or aborted), not a store error. - if isFilerListNotFound(err) { + if isFilerNotFound(err) { return nil, s3err.ErrNoSuchUpload } glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err) diff --git a/weed/s3api/s3_error_utils.go b/weed/s3api/s3_error_utils.go index bd38e50b7..13e499cee 100644 --- a/weed/s3api/s3_error_utils.go +++ b/weed/s3api/s3_error_utils.go @@ -7,14 +7,20 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) -// isFilerListNotFound reports whether a filer list error is a not-found. -// Unlike lookups (normalized in filer_pb.LookupEntry), list errors cross gRPC -// as raw status errors, so the sentinel survives only as text. Callers must -// pass errors from paths without user-controlled segments that could spoof it. -func isFilerListNotFound(err error) bool { - return errors.Is(err, filer_pb.ErrNotFound) || strings.Contains(err.Error(), filer_pb.ErrNotFound.Error()) +// isFilerNotFound reports whether a filer error is a not-found. +// Unlike lookups (normalized in filer_pb.LookupEntry), list and cache errors +// cross gRPC as raw status errors, so the sentinel survives as codes.NotFound +// or only as text. Callers must pass errors from paths without user-controlled +// segments that could spoof it. +func isFilerNotFound(err error) bool { + if err == nil { + return false + } + return errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound || strings.Contains(err.Error(), filer_pb.ErrNotFound.Error()) } // ErrorHandlers provide common error handling patterns for S3 API operations diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index dee0fb0cb..b56bca6c0 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -21,7 +21,10 @@ import ( "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/security" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" @@ -31,6 +34,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) // zeroBuf is a reusable buffer of zero bytes for padding operations @@ -1016,22 +1020,59 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R chunks = cachedEntry.GetChunks() entry = cachedEntry glog.V(1).Infof("streamFromVolumeServers: successfully cached remote object, got %d chunks", len(chunks)) - } else if cacheErr != nil && !errors.Is(cacheErr, context.DeadlineExceeded) && !errors.Is(cacheErr, context.Canceled) && status.Code(cacheErr) != codes.DeadlineExceeded && status.Code(cacheErr) != codes.Canceled { - // Permanent error (e.g. not found, permission denied) - return final status - glog.Errorf("streamFromVolumeServers: permanent cache error for %s/%s: %v", bucket, object, cacheErr) - if status.Code(cacheErr) == codes.NotFound { - s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) - } else { - s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) - } + } else if isFilerNotFound(cacheErr) { + // Authoritative: the entry vanished; the origin cannot resurrect it + glog.Errorf("streamFromVolumeServers: entry not found while caching %s/%s: %v", bucket, object, cacheErr) + s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) return newStreamErrorWithResponse(cacheErr) } else { - // Client disconnected during the cache wait: report cancellation, not 503, - // so we don't write to a closed connection. + // Client disconnected during the cache wait: report cancellation, not an + // error response, so we don't write to a closed connection. if ctxErr := r.Context().Err(); ctxErr != nil { return ctxErr } - // Cache not ready yet; return 503 with Retry-After for client backoff + // Cache not ready or failing locally (e.g. no assignable volume): serve + // straight from the origin while the detached cache keeps filling for + // later reads. An entry resolved to a specific version -- even on a + // latest-version read -- has no origin key, so it keeps the error + // paths below. + if cacheVersionId == "" || cacheVersionId == "null" { + if remoteReader, remoteErr := s3a.openRemoteStream(r.Context(), bucket, object, offset, size); remoteErr == nil { + defer remoteReader.Close() + s3a.setResponseHeaders(w, r, entry, totalSize) + w.Header().Set("Content-Length", strconv.FormatInt(size, 10)) + if isRangeRequest { + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize)) + w.WriteHeader(http.StatusPartialContent) + } else { + w.WriteHeader(http.StatusOK) + } + TimeToFirstByte(r.Method, t0, r) + cw := &countingWriter{w: w} + _, copyErr := io.CopyN(cw, remoteReader, size) + if copyErr == io.EOF { + // the origin returned fewer bytes than the entry's RemoteSize + copyErr = io.ErrUnexpectedEOF + } + if cw.written > 0 { + BucketTrafficSent(cw.written, r) + } + if copyErr != nil { + glog.V(2).Infof("streamFromVolumeServers: origin stream %s/%s ended after %d bytes: %v", bucket, object, cw.written, copyErr) + return newStreamErrorWithResponse(copyErr) + } + return nil + } else { + glog.Warningf("streamFromVolumeServers: origin stream %s/%s: %v", bucket, object, remoteErr) + } + } + // Origin unreadable. A permanent cache error is final; a transient one + // gets 503 so the client retries the still-filling cache. + if cacheErr != nil && !errors.Is(cacheErr, context.DeadlineExceeded) && !errors.Is(cacheErr, context.Canceled) && status.Code(cacheErr) != codes.DeadlineExceeded && status.Code(cacheErr) != codes.Canceled { + glog.Errorf("streamFromVolumeServers: permanent cache error for %s/%s: %v", bucket, object, cacheErr) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return newStreamErrorWithResponse(cacheErr) + } glog.V(1).Infof("streamFromVolumeServers: remote object %s/%s not cached yet, returning 503 for retry", bucket, object) w.Header().Set("Retry-After", "2") s3err.WriteErrorResponse(w, r, s3err.ErrServiceUnavailable) @@ -3096,6 +3137,52 @@ func (s3a *S3ApiServer) buildRemoteObjectPath(bucket, object string) (dir, name return dir, name } +// openRemoteStream opens a ranged read of a remote-only object straight from +// its mounted origin, resolving the mount and storage conf from the filer. +func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object string, offset, size int64) (io.ReadCloser, error) { + dir, name := s3a.buildRemoteObjectPath(bucket, object) + + var storageConf *remote_pb.RemoteConf + var localMountedDir string + var mountedLocation *remote_pb.RemoteStorageLocation + err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mappingContent, readErr := filer.ReadInsideFiler(ctx, client, filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE) + if readErr != nil { + return readErr + } + mappings, unmarshalErr := filer.UnmarshalRemoteStorageMappings(mappingContent) + if unmarshalErr != nil { + return unmarshalErr + } + var findErr error + localMountedDir, mountedLocation, findErr = filer.FindMountedRemoteMapping(mappings, dir) + if findErr != nil { + return findErr + } + confContent, readErr := filer.ReadInsideFiler(ctx, client, filer.DirectoryEtcRemote, mountedLocation.Name+filer.REMOTE_STORAGE_CONF_SUFFIX) + if readErr != nil { + return readErr + } + storageConf = &remote_pb.RemoteConf{} + return proto.Unmarshal(confContent, storageConf) + }) + if err != nil { + return nil, err + } + + client, err := remote_storage.GetRemoteStorage(storageConf) + if err != nil { + return nil, err + } + streamer, ok := client.(remote_storage.RemoteStorageStreamReader) + if !ok { + return nil, fmt.Errorf("remote storage type %s does not support streaming reads", storageConf.Type) + } + + loc := filer.MapFullPathToRemoteStorageLocation(util.FullPath(localMountedDir), mountedLocation, util.FullPath(dir).Child(name)) + return streamer.ReadFileAsStream(ctx, loc, offset, size) +} + // doCacheRemoteObject calls the filer's CacheRemoteObjectToLocalCluster gRPC endpoint. // This is the core caching function used by cacheRemoteObjectForStreaming. func (s3a *S3ApiServer) doCacheRemoteObject(ctx context.Context, dir, name string) (*filer_pb.Entry, error) { @@ -3145,18 +3232,7 @@ var remoteCacheStreamingTimeoutNS = int64(20 * time.Second) // between transient errors (timeout) and permanent errors (not found, permission denied). // The filer continues caching on detached context, so retry finds cached chunks. func (s3a *S3ApiServer) cacheRemoteObjectForStreamingWithShortTimeout(r *http.Request, entry *filer_pb.Entry, bucket, object, versionId string) (*filer_pb.Entry, error) { - // Adaptive timeout: smaller files can afford to wait longer since cache completes faster - pollTimeout := 5 * time.Second - if entry.RemoteEntry != nil && entry.RemoteEntry.RemoteSize > 0 { - // For very large files (>500MB), use shorter timeout to improve TTFB - // For smaller files, allow longer to increase cache hit rate - remoteSize := entry.RemoteEntry.RemoteSize - if remoteSize > 500*1024*1024 { - pollTimeout = 2 * time.Second - } else if remoteSize < 50*1024*1024 { - pollTimeout = 10 * time.Second - } - } + pollTimeout := remote_storage.CacheWaitTimeout(entry.GetRemoteEntry().GetRemoteSize()) cacheCtx, cancel := context.WithTimeout(r.Context(), pollTimeout) defer cancel() diff --git a/weed/s3api/s3api_remote_storage_test.go b/weed/s3api/s3api_remote_storage_test.go index 4ef982f9f..ccebd581a 100644 --- a/weed/s3api/s3api_remote_storage_test.go +++ b/weed/s3api/s3api_remote_storage_test.go @@ -1,11 +1,14 @@ package s3api import ( + "bytes" "context" "fmt" + "io" "net" "net/http" "net/http/httptest" + "strconv" "strings" "sync/atomic" "testing" @@ -14,11 +17,16 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb" "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/s3api/s3_constants" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) // TestIsInRemoteOnly tests the IsInRemoteOnly method on filer_pb.Entry @@ -498,16 +506,26 @@ func TestCopyObjectRemoteOnlySourceDetection(t *testing.T) { // fakeCacheFiler is a minimal SeaweedFiler gRPC server whose // CacheRemoteObjectToLocalCluster behavior is driven by a callback, so tests can -// model a slow (still-caching) filer or one that returns cached chunks. +// model a slow (still-caching) filer or one that returns cached chunks. The +// optional entries map serves LookupDirectoryEntry for inline-content paths +// such as the /etc/remote configuration. type fakeCacheFiler struct { filer_pb.UnimplementedSeaweedFilerServer - cache func(context.Context, *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) + cache func(context.Context, *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) + entries map[string][]byte } func (f *fakeCacheFiler) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { return f.cache(ctx, req) } +func (f *fakeCacheFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if content, ok := f.entries[req.Directory+"/"+req.Name]; ok { + return &filer_pb.LookupDirectoryEntryResponse{Entry: &filer_pb.Entry{Name: req.Name, Content: content}}, nil + } + return nil, filer_pb.ErrNotFound +} + // startFakeCacheFiler serves impl on a random localhost port and returns the // S3-style filer address whose ToGrpcAddress resolves back to that port. func startFakeCacheFiler(t *testing.T, impl *fakeCacheFiler) pb.ServerAddress { @@ -585,3 +603,180 @@ func TestCacheRemoteObjectForStreamingCached(t *testing.T) { require.NotNil(t, got) assert.Len(t, got.GetChunks(), 1) } + +// fakeStreamRemoteClient records the ReadFileAsStream call and serves from an +// in-memory byte slice. The embedded nil interface panics on any other call. +type fakeStreamRemoteClient struct { + remote_storage.RemoteStorageClient + data []byte + gotLoc *remote_pb.RemoteStorageLocation + gotOffset int64 + gotSize int64 +} + +func (c *fakeStreamRemoteClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (io.ReadCloser, error) { + c.gotLoc, c.gotOffset, c.gotSize = loc, offset, size + end := min(offset+size, int64(len(c.data))) + return io.NopCloser(bytes.NewReader(c.data[offset:end])), nil +} + +type fakeStreamRemoteMaker struct{ client *fakeStreamRemoteClient } + +func (m *fakeStreamRemoteMaker) Make(conf *remote_pb.RemoteConf) (remote_storage.RemoteStorageClient, error) { + return m.client, nil +} +func (m *fakeStreamRemoteMaker) HasBucket() bool { return true } + +// startStreamThroughFiler serves a filer whose cache RPC always fails with +// cacheErr and whose /etc/remote mounts /buckets/mybucket on a fake origin. +func startStreamThroughFiler(t *testing.T, remoteName string, cacheErr error) pb.ServerAddress { + mappingBytes, err := proto.Marshal(&remote_pb.RemoteStorageMapping{ + Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/mybucket": {Name: remoteName, Bucket: "origin-bucket", Path: "/data"}, + }, + }) + require.NoError(t, err) + confBytes, err := proto.Marshal(&remote_pb.RemoteConf{Name: remoteName, Type: "faketest"}) + require.NoError(t, err) + return startFakeCacheFiler(t, &fakeCacheFiler{ + cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + return nil, cacheErr + }, + entries: map[string][]byte{ + "/etc/remote/mount.mapping": mappingBytes, + "/etc/remote/" + remoteName + ".conf": confBytes, + }, + }) +} + +var stillCachingErr = status.Error(codes.DeadlineExceeded, "still caching") + +// TestS3ColdReadStreamsFromOrigin pins the stream-through path: when the filer +// reports the cache is still filling, the GET is served straight from the +// mounted origin instead of a 503 retry loop. +func TestS3ColdReadStreamsFromOrigin(t *testing.T) { + content := []byte("0123456789") + client := &fakeStreamRemoteClient{data: content} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + + entry := func() *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "obj.bin", + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(content))}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: int64(len(content))}, + } + } + + t.Run("full read", func(t *testing.T) { + s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-full", stillCachingErr)) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + + err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "") + + require.NoError(t, err) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, content, w.Body.Bytes()) + require.NotNil(t, client.gotLoc, "must read from the origin") + assert.Equal(t, "origin-bucket", client.gotLoc.Bucket) + assert.Equal(t, "/data/dir/obj.bin", client.gotLoc.Path) + assert.Equal(t, strconv.Itoa(len(content)), w.Header().Get("Content-Length")) + }) + + t.Run("range read", func(t *testing.T) { + s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-range", stillCachingErr)) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + r.Header.Set("Range", "bytes=2-5") + + err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "") + + require.NoError(t, err) + assert.Equal(t, http.StatusPartialContent, w.Code) + assert.Equal(t, content[2:6], w.Body.Bytes()) + assert.Equal(t, int64(2), client.gotOffset) + assert.Equal(t, int64(4), client.gotSize) + assert.Equal(t, fmt.Sprintf("bytes 2-5/%d", len(content)), w.Header().Get("Content-Range")) + }) + + t.Run("short origin read fails instead of silently truncating", func(t *testing.T) { + shortClient := &fakeStreamRemoteClient{data: content[:4]} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: shortClient} + defer func() { + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + }() + s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-short", stillCachingErr)) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + + err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "") + + require.Error(t, err) + assert.Contains(t, err.Error(), io.ErrUnexpectedEOF.Error()) + }) + + t.Run("version-specific read keeps 503 retry", func(t *testing.T) { + s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-versioned", stillCachingErr)) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin?versionId=v123", nil) + + err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "v123") + + require.Error(t, err) + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Equal(t, "2", w.Header().Get("Retry-After")) + }) + + t.Run("latest read of a versioned entry keeps 503 retry", func(t *testing.T) { + freshClient := &fakeStreamRemoteClient{data: content} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: freshClient} + defer func() { + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + }() + s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-latest-versioned", stillCachingErr)) + versionedEntry := entry() + versionedEntry.Extended = map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v456")} + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + + err := s3a.streamFromVolumeServers(w, r, versionedEntry, "", "mybucket", "dir/obj.bin", "") + + require.Error(t, err) + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Nil(t, freshClient.gotLoc, "the unversioned origin key must not be read for a versioned entry") + }) + + t.Run("local cache failure falls back to origin", func(t *testing.T) { + cacheErr := status.Error(codes.Internal, "assign: no free volumes") + s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-localfail", cacheErr)) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + + err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "") + + require.NoError(t, err) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, content, w.Body.Bytes()) + }) + + t.Run("entry not found stays 404", func(t *testing.T) { + notFoundForms := map[string]error{ + "canonical status": status.Error(codes.NotFound, "entry vanished"), + // the filer returns the raw sentinel, which crosses gRPC as + // codes.Unknown with only the message surviving + "raw sentinel": filer_pb.ErrNotFound, + } + for name, cacheErr := range notFoundForms { + t.Run(name, func(t *testing.T) { + s3a := newRemoteCacheTestServer(startStreamThroughFiler(t, "faketest-notfound-"+name[:3], cacheErr)) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + + err := s3a.streamFromVolumeServers(w, r, entry(), "", "mybucket", "dir/obj.bin", "") + + require.Error(t, err) + assert.Equal(t, http.StatusNotFound, w.Code) + }) + } + }) +} diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go index 930d5c455..4c0589ad9 100644 --- a/weed/server/filer_grpc_server_remote.go +++ b/weed/server/filer_grpc_server_remote.go @@ -2,9 +2,9 @@ package weed_server import ( "context" + "errors" "fmt" "sort" - "strings" "sync" "time" @@ -17,6 +17,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" ) @@ -48,6 +50,11 @@ func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req glog.V(2).Infof("CacheRemoteObjectToLocalCluster: shared result for %s", cacheKey) } if res.Err != nil { + // The sentinel would cross gRPC as codes.Unknown; make it canonical + // so remote callers can classify a vanished entry. + if errors.Is(res.Err, filer_pb.ErrNotFound) { + return nil, status.Error(codes.NotFound, res.Err.Error()) + } return nil, res.Err } if res.Val == nil { @@ -90,37 +97,10 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re glog.V(1).Infof("CacheRemoteObjectToLocalCluster: caching %s/%s (remote size: %d)", req.Directory, req.Name, entry.Remote.RemoteSize) - // load all mappings - mappingEntry, err := fs.filer.FindEntry(ctx, util.JoinPath(filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE)) + storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, req.Directory, req.Name) if err != nil { return nil, err } - mappings, err := filer.UnmarshalRemoteStorageMappings(mappingEntry.Content) - if err != nil { - return nil, err - } - - // find mapping - var remoteStorageMountedLocation *remote_pb.RemoteStorageLocation - var localMountedDir string - for k, loc := range mappings.Mappings { - if strings.HasPrefix(req.Directory, k) { - localMountedDir, remoteStorageMountedLocation = k, loc - } - } - if localMountedDir == "" { - return nil, fmt.Errorf("%s is not mounted", req.Directory) - } - - // find storage configuration - storageConfEntry, err := fs.filer.FindEntry(ctx, util.JoinPath(filer.DirectoryEtcRemote, remoteStorageMountedLocation.Name+filer.REMOTE_STORAGE_CONF_SUFFIX)) - if err != nil { - return nil, err - } - storageConf := &remote_pb.RemoteConf{} - if unMarshalErr := proto.Unmarshal(storageConfEntry.Content, storageConf); unMarshalErr != nil { - return nil, fmt.Errorf("unmarshal remote storage conf %s/%s: %v", filer.DirectoryEtcRemote, remoteStorageMountedLocation.Name+filer.REMOTE_STORAGE_CONF_SUFFIX, unMarshalErr) - } // detect storage option so, err := fs.detectStorageOption(ctx, req.Directory, "", "", 0, "", "", "", "") @@ -156,8 +136,6 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re altRequest.ExpectedDataSize = uint64(chunkSize) } - dest := util.FullPath(remoteStorageMountedLocation.Path).Child(string(util.FullPath(req.Directory).Child(req.Name))[len(localMountedDir):]) - var chunks []*filer_pb.FileChunk var chunksMu sync.Mutex var fetchAndWriteErr error @@ -239,14 +217,10 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re Auth: string(assignResult.Auth), DownloadConcurrency: downloadConcurrency, RemoteConf: storageConf, - RemoteLocation: &remote_pb.RemoteStorageLocation{ - Name: remoteStorageMountedLocation.Name, - Bucket: remoteStorageMountedLocation.Bucket, - Path: string(dest), - }, + RemoteLocation: remoteLocation, }) if fetchErr != nil { - return fmt.Errorf("volume server %s fetchAndWrite %s: %v", assignResult.Url, dest, fetchErr) + return fmt.Errorf("volume server %s fetchAndWrite %s: %v", assignResult.Url, remoteLocation.Path, fetchErr) } etag = resp.ETag return nil @@ -352,3 +326,33 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re return resp, nil } + +// resolveMountedRemote reads /etc/remote fresh (so conf changes need no restart) +// and maps dir/name to its remote storage conf and remote location. +func (fs *FilerServer) resolveMountedRemote(ctx context.Context, dir, name string) (*remote_pb.RemoteConf, *remote_pb.RemoteStorageLocation, error) { + mappingEntry, err := fs.filer.FindEntry(ctx, util.JoinPath(filer.DirectoryEtcRemote, filer.REMOTE_STORAGE_MOUNT_FILE)) + if err != nil { + return nil, nil, err + } + mappings, err := filer.UnmarshalRemoteStorageMappings(mappingEntry.Content) + if err != nil { + return nil, nil, err + } + + localMountedDir, remoteStorageMountedLocation, err := filer.FindMountedRemoteMapping(mappings, dir) + if err != nil { + return nil, nil, err + } + + storageConfEntry, err := fs.filer.FindEntry(ctx, util.JoinPath(filer.DirectoryEtcRemote, remoteStorageMountedLocation.Name+filer.REMOTE_STORAGE_CONF_SUFFIX)) + if err != nil { + return nil, nil, err + } + storageConf := &remote_pb.RemoteConf{} + if unMarshalErr := proto.Unmarshal(storageConfEntry.Content, storageConf); unMarshalErr != nil { + return nil, nil, fmt.Errorf("unmarshal remote storage conf %s/%s: %v", filer.DirectoryEtcRemote, remoteStorageMountedLocation.Name+filer.REMOTE_STORAGE_CONF_SUFFIX, unMarshalErr) + } + + remoteLocation := filer.MapFullPathToRemoteStorageLocation(util.FullPath(localMountedDir), remoteStorageMountedLocation, util.FullPath(dir).Child(name)) + return storageConf, remoteLocation, nil +} diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go index b2f9fe8b5..e12bf1445 100644 --- a/weed/server/filer_server_handlers_read.go +++ b/weed/server/filer_server_handlers_read.go @@ -16,10 +16,13 @@ import ( "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/remote_storage" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // Validates the preconditions. Returns true if GET/HEAD operation should not proceed. @@ -213,26 +216,43 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) chunks := entry.GetChunks() if entry.IsInRemoteOnly() { dir, name := entry.FullPath.DirAndName() - if resp, err := fs.CacheRemoteObjectToLocalCluster(ctx, &filer_pb.CacheRemoteObjectToLocalClusterRequest{ + // Bounded wait: a large download outlasts any client timeout, so + // serve straight from the origin once the wait expires while the + // detached cache keeps filling for later reads. + cacheCtx, cancelCache := context.WithTimeout(ctx, remote_storage.CacheWaitTimeout(entry.Remote.RemoteSize)) + resp, err := fs.CacheRemoteObjectToLocalCluster(cacheCtx, &filer_pb.CacheRemoteObjectToLocalClusterRequest{ Directory: dir, Name: name, - }); err != nil { + }) + cancelCache() + if err != nil { stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadCache).Inc() // Client disconnected: surface ctx error so caller stays silent. if ctxErr := ctx.Err(); ctxErr != nil { return nil, ctxErr } - // Entry vanished mid-cache: forward NotFound so caller maps to 404, - // not the 503 retry-loop. - if errors.Is(err, filer_pb.ErrNotFound) { - return nil, err + // Entry vanished mid-cache: forward the sentinel so caller maps to + // 404, not the 503 retry-loop. The cache RPC returns it as a + // canonical status, which errors.Is cannot see. + if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound { + return nil, filer_pb.ErrNotFound } - // Cache still filling: tag with sentinel so caller maps to 503 + Retry-After. + // A multipart Range prepares every part before writing any, which + // would hold one open origin connection per part and leak them + // when a later prepare fails; keep those on the 503 retry path. + if !strings.Contains(r.Header.Get("Range"), ",") { + glog.V(1).InfofCtx(ctx, "stream %s from remote while caching: %v", entry.FullPath, err) + if streamFn, remoteErr := fs.streamFromRemote(ctx, dir, name, offset, size); remoteErr == nil { + return streamFn, nil + } else { + glog.WarningfCtx(ctx, "stream %s from remote: %v", entry.FullPath, remoteErr) + } + } + // Origin unreadable: tag with sentinel so caller maps to 503 + Retry-After. glog.WarningfCtx(ctx, "CacheRemoteObjectToLocalCluster %s: %v", entry.FullPath, err) return nil, fmt.Errorf("cache %s: %w", entry.FullPath, ErrCacheNotReady) - } else { - chunks = resp.Entry.GetChunks() } + chunks = resp.Entry.GetChunks() } // Use a detached context for streaming so client disconnects/cancellations don't abort volume server operations, @@ -259,6 +279,53 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) }) } +// streamFromRemote serves a byte range of a remote-only entry straight from the +// mounted origin, so a first read is not blocked by the full local caching. +func (fs *FilerServer) streamFromRemote(ctx context.Context, dir, name string, offset, size int64) (filer.DoStreamContent, error) { + storageConf, remoteLocation, err := fs.resolveMountedRemote(ctx, dir, name) + if err != nil { + return nil, err + } + client, err := remote_storage.GetRemoteStorage(storageConf) + if err != nil { + return nil, err + } + streamer, ok := client.(remote_storage.RemoteStorageStreamReader) + if !ok { + return nil, fmt.Errorf("remote storage type %s does not support streaming reads", storageConf.Type) + } + reader, err := streamer.ReadFileAsStream(ctx, remoteLocation, offset, size) + if err != nil { + return nil, err + } + downloadThrottler := util.NewWriteThrottler(fs.option.DownloadMaxBytesPs) + return func(writer io.Writer) error { + defer reader.Close() + var written int64 + buf := make([]byte, 128*1024) + for { + n, readErr := reader.Read(buf) + if n > 0 { + if _, writeErr := writer.Write(buf[:n]); writeErr != nil { + return writeErr + } + written += int64(n) + downloadThrottler.MaybeSlowdown(int64(n)) + } + if readErr == io.EOF { + if written != size { + // the origin returned fewer bytes than the entry's RemoteSize + return fmt.Errorf("origin stream %s: %w after %d of %d bytes", remoteLocation.Path, io.ErrUnexpectedEOF, written, size) + } + return nil + } + if readErr != nil { + return readErr + } + } + }, nil +} + func (fs *FilerServer) maybeGetVolumeReadJwtAuthorizationToken(fileId string) string { // Only ever sign with the read key. A volume server enforces read JWTs // solely when jwt.signing.read.key is set, so falling back to the write key