diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 6ce81b249..869dc32ac 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -16,6 +16,7 @@ import ( "sort" "strconv" "strings" + "sync/atomic" "time" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -3105,18 +3106,37 @@ func cachedEntryHasLocalData(entry *filer_pb.Entry) bool { return entry != nil && (len(entry.GetChunks()) > 0 || len(entry.Content) > 0) } +// remoteCacheStreamingTimeoutNS bounds (nanoseconds) how long the streaming read +// path waits for the filer to finish caching a remote-only object. Without a +// bound a multi-GB download outlasts the client's read timeout, so the request +// is canceled before the caller can emit 503 + Retry-After. Kept under common S3 +// client read timeouts (~60s) given the 30s dedup attempt that precedes it. +// Atomic so tests can shorten it without racing the read path. +var remoteCacheStreamingTimeoutNS = int64(20 * time.Second) + // cacheRemoteObjectForStreaming caches a remote-only object to the local cluster for streaming. -// Uses the request context (no artificial timeout) so the caching can complete. -// Returns the cached entry only when chunks are present; the streaming caller -// is not wired to read inline Content from a cache result here. +// Bounded so a slow large-file download returns to the caller (which emits 503 + +// Retry-After) before the client gives up; the filer keeps caching on a detached +// context, so a retry streams from the cached chunks. Returns the cached entry +// only when chunks are present; the caller cannot read inline Content here. func (s3a *S3ApiServer) cacheRemoteObjectForStreaming(r *http.Request, entry *filer_pb.Entry, bucket, object, versionId string) *filer_pb.Entry { + timeout := time.Duration(atomic.LoadInt64(&remoteCacheStreamingTimeoutNS)) + cacheCtx, cancel := context.WithTimeout(r.Context(), timeout) + defer cancel() + dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId) glog.V(1).Infof("cacheRemoteObjectForStreaming: caching %s/%s (remote size: %d, versionId: %s)", dir, name, entry.RemoteEntry.RemoteSize, versionId) - cachedEntry, err := s3a.doCacheRemoteObject(r.Context(), dir, name) + cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name) if err != nil { - glog.Errorf("cacheRemoteObjectForStreaming: failed to cache %s/%s: %v", dir, name, err) + // A bounded-wait timeout (or client disconnect) is not a cache failure: + // the filer keeps downloading and the caller maps a nil return to 503. + if cacheCtx.Err() != nil { + glog.V(1).Infof("cacheRemoteObjectForStreaming: %s/%s not ready within %v (will retry)", dir, name, timeout) + } else { + glog.Errorf("cacheRemoteObjectForStreaming: failed to cache %s/%s: %v", dir, name, err) + } return nil } diff --git a/weed/s3api/s3api_remote_storage_test.go b/weed/s3api/s3api_remote_storage_test.go index bda3f938d..4ef982f9f 100644 --- a/weed/s3api/s3api_remote_storage_test.go +++ b/weed/s3api/s3api_remote_storage_test.go @@ -1,13 +1,24 @@ package s3api import ( + "context" + "fmt" + "net" + "net/http" + "net/http/httptest" "strings" + "sync/atomic" "testing" + "time" "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/s3api/s3_constants" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" ) // TestIsInRemoteOnly tests the IsInRemoteOnly method on filer_pb.Entry @@ -484,3 +495,93 @@ 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. +type fakeCacheFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + cache func(context.Context, *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) +} + +func (f *fakeCacheFiler) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + return f.cache(ctx, req) +} + +// 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 { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + srv := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(srv, impl) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + port := lis.Addr().(*net.TCPAddr).Port + return pb.ServerAddress(fmt.Sprintf("127.0.0.1:1.%d", port)) +} + +func newRemoteCacheTestServer(filerAddr pb.ServerAddress) *S3ApiServer { + return &S3ApiServer{ + option: &S3ApiServerOption{ + Filers: []pb.ServerAddress{filerAddr}, + BucketsPath: "/buckets", + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }, + } +} + +func remoteOnlyEntry(name string, size int64) *filer_pb.Entry { + return &filer_pb.Entry{Name: name, RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: size}} +} + +// TestCacheRemoteObjectForStreamingTimeout pins the fix for large-file cold +// GetObject: when the filer is still caching, the streaming path returns nil +// within its bound (so the handler emits 503 + Retry-After) instead of blocking +// on the raw request context until the client gives up. +func TestCacheRemoteObjectForStreamingTimeout(t *testing.T) { + filerAddr := startFakeCacheFiler(t, &fakeCacheFiler{ + cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + <-ctx.Done() // never finishes within the bound + return nil, ctx.Err() + }, + }) + s3a := newRemoteCacheTestServer(filerAddr) + + const shortTimeout = 200 * time.Millisecond + defer func(prev int64) { atomic.StoreInt64(&remoteCacheStreamingTimeoutNS, prev) }(atomic.LoadInt64(&remoteCacheStreamingTimeoutNS)) + atomic.StoreInt64(&remoteCacheStreamingTimeoutNS, int64(shortTimeout)) + + r := httptest.NewRequest(http.MethodGet, "/mybucket/large.bin", nil) + start := time.Now() + got := s3a.cacheRemoteObjectForStreaming(r, remoteOnlyEntry("large.bin", 1<<30), "mybucket", "large.bin", "") + elapsed := time.Since(start) + + assert.Nil(t, got, "uncached object must return nil so the caller maps to 503") + assert.GreaterOrEqual(t, elapsed, shortTimeout/2, "must wait on the bounded timeout, not return early on a setup error") + assert.Less(t, elapsed, 5*time.Second, "must not block on the full download") + assert.NoError(t, r.Context().Err(), "request context stays alive so the caller chooses 503, not cancellation") +} + +// TestCacheRemoteObjectForStreamingCached confirms that once the filer reports +// local chunks, the streaming path returns the cached entry to stream from. +func TestCacheRemoteObjectForStreamingCached(t *testing.T) { + filerAddr := startFakeCacheFiler(t, &fakeCacheFiler{ + cache: func(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { + return &filer_pb.CacheRemoteObjectToLocalClusterResponse{ + Entry: &filer_pb.Entry{ + Name: req.Name, + Chunks: []*filer_pb.FileChunk{{FileId: "1,abc", Size: 100, Offset: 0}}, + }, + }, nil + }, + }) + s3a := newRemoteCacheTestServer(filerAddr) + + r := httptest.NewRequest(http.MethodGet, "/mybucket/obj.bin", nil) + got := s3a.cacheRemoteObjectForStreaming(r, remoteOnlyEntry("obj.bin", 100), "mybucket", "obj.bin", "") + + require.NotNil(t, got) + assert.Len(t, got.GetChunks(), 1) +}