From b763a5f6bf84fe8352089d0b3b8556fc01041f73 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 18 Jun 2026 17:22:32 -0700 Subject: [PATCH] s3: improve TTFB for large remote objects (#10010) * s3: add streaming reader interface for remote storage Add RemoteStorageStreamReader optional interface to support efficient streaming of large remote objects without buffering entire file in memory. This enables future stream-through caching where data can be served to clients while simultaneously writing to volume servers. Implement ReadFileAsStream() for S3, GCS, and Azure backends using their native streaming APIs. This provides the foundation for improving TTFB on large remote file access by serving data directly from remote storage while background cache operation populates local chunks. The streaming interface allows remote storage backends to return io.ReadCloser, enabling efficient memory usage for multi-GB objects compared to the current ReadFile() approach which buffers entire ranges in memory. * s3: adaptive timeout for remote object caching to improve TTFB Use size-aware cache polling timeout to balance cache-hit rate against time-to-first-byte: - Small files (<50MB): 10s timeout - more likely to complete caching before timeout, improving subsequent request performance - Medium files (50-500MB): 5s timeout - default balance - Large files (>500MB): 2s timeout - fail-fast to improve initial TTFB for very large downloads This reduces waiting time for large remote files while maintaining high cache-hit rate for smaller files that cache quickly. * s3: address code review feedback for stream-through cache - Move startBackgroundRemoteCache call after policy recheck to avoid cache side effects for denied requests (authorization first) - Make startBackgroundRemoteCache version-aware by accepting versionId parameter and using buildVersionedRemoteObjectPath - Add timeout (5 minutes) to background cache context to prevent goroutine pile-up if RPC stalls under load - Update cacheRemoteObjectForStreamingWithShortTimeout to return both entry and error, allowing callers to distinguish transient errors (timeout/cancellation) from permanent errors (not found, denied) - Update streamFromVolumeServers to handle permanent cache errors with appropriate HTTP status codes (404 for not found, 503 for transient) --- .../azure/azure_storage_client.go | 16 +++ weed/remote_storage/gcs/gcs_storage_client.go | 5 + weed/remote_storage/remote_storage.go | 6 + weed/remote_storage/s3/s3_storage_client.go | 15 +++ weed/s3api/s3api_object_handlers.go | 123 +++++++++++++++--- 5 files changed, 144 insertions(+), 21 deletions(-) diff --git a/weed/remote_storage/azure/azure_storage_client.go b/weed/remote_storage/azure/azure_storage_client.go index 38970e850..2ed10cf3d 100644 --- a/weed/remote_storage/azure/azure_storage_client.go +++ b/weed/remote_storage/azure/azure_storage_client.go @@ -344,6 +344,22 @@ func (az *azureRemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.Remot return data, nil } +func (az *azureRemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) { + key := loc.Path[1:] + blobClient := az.client.ServiceClient().NewContainerClient(loc.Bucket).NewBlockBlobClient(key) + + downloadResponse, err := blobClient.DownloadStream(ctx, &blob.DownloadStreamOptions{ + Range: blob.HTTPRange{ + Offset: offset, + Count: size, + }, + }) + if err != nil { + return nil, fmt.Errorf("failed to open stream for %s%s: %v", loc.Bucket, loc.Path, err) + } + return downloadResponse.Body, nil +} + func (az *azureRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) { return nil } diff --git a/weed/remote_storage/gcs/gcs_storage_client.go b/weed/remote_storage/gcs/gcs_storage_client.go index 69831aee5..7c812fdad 100644 --- a/weed/remote_storage/gcs/gcs_storage_client.go +++ b/weed/remote_storage/gcs/gcs_storage_client.go @@ -212,6 +212,11 @@ func (gcs *gcsRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation return } +func (gcs *gcsRemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) { + key := loc.Path[1:] + return gcs.client.Bucket(loc.Bucket).Object(key).NewRangeReader(ctx, offset, size) +} + func (gcs *gcsRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) { return nil } diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index 0a6a63e1d..117be244a 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -95,6 +95,12 @@ type RemoteStorageConcurrentReader interface { ReadFileWithConcurrency(loc *remote_pb.RemoteStorageLocation, offset int64, size int64, concurrency int) (data []byte, err error) } +// RemoteStorageStreamReader is an optional interface for remote storage clients +// that support streaming reads with io.Reader for efficient memory usage. +type RemoteStorageStreamReader interface { + ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) +} + type RemoteStorageClientMaker interface { Make(remoteConf *remote_pb.RemoteConf) (RemoteStorageClient, error) HasBucket() bool diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index 9c7755019..e2a706c66 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -261,6 +261,21 @@ func (s *s3RemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.RemoteSto return writerAt.Bytes(), nil } +func (s *s3RemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) { + output, err := s.conn.GetObjectWithContext(ctx, &s3.GetObjectInput{ + Bucket: aws.String(loc.Bucket), + Key: aws.String(loc.Path[1:]), + Range: aws.String(fmt.Sprintf("bytes=%d-%d", offset, offset+size-1)), + }) + if err != nil { + if aerr, ok := err.(awserr.Error); ok && aerr.Code() == s3.ErrCodeNoSuchKey { + return nil, remote_storage.ErrRemoteObjectNotFound + } + return nil, fmt.Errorf("failed to open stream for %s%s: %v", loc.Bucket, loc.Path, err) + } + return output.Body, nil +} + func (s *s3RemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) { return nil } diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 869dc32ac..58830b0bc 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -776,19 +776,24 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) return } - // Handle remote storage objects: cache to local cluster if object is remote-only - // This uses singleflight to deduplicate concurrent caching requests for the same object - // On cache error, gracefully falls back to streaming from remote - if objectEntryForSSE.IsInRemoteOnly() { - objectEntryForSSE = s3a.cacheRemoteObjectWithDedup(r.Context(), bucket, object, objectEntryForSSE) - } - // Re-check bucket policy with object entry for tag-based conditions (e.g., s3:ExistingObjectTag) if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, string(s3_constants.ACTION_READ), objectEntryForSSE.Extended, "GetObjectHandler"); errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) return } + // Handle remote storage objects: initiate background caching without blocking + // This implements stream-through caching: serve first byte immediately while + // caching happens in the background, improving TTFB for large files + if objectEntryForSSE.IsInRemoteOnly() { + // Start background cache without waiting (non-blocking) + // Only after authorization passes to avoid cache side effects for denied requests + versionId := r.URL.Query().Get("versionId") + cacheVersionId := resolvedSourceVersionId(versionId, objectEntryForSSE) + s3a.startBackgroundRemoteCache(bucket, object, cacheVersionId, objectEntryForSSE) + // Continue with streaming immediately - will serve from remote or cached chunks + } + // Check if PartNumber query parameter is present (for multipart GET requests) partNumberStr := r.URL.Query().Get("partNumber") if partNumberStr == "" { @@ -978,28 +983,31 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R len(chunks), totalSize, isRangeRequest, offset, size) if len(chunks) == 0 { - // Check if this is a remote-only entry that needs caching - // This handles the case where initial caching attempt timed out or failed + // Check if this is a remote-only entry if entry.IsInRemoteOnly() { - glog.V(1).Infof("streamFromVolumeServers: entry is remote-only, attempting to cache before streaming") - // Latest-version reads carry an empty query versionId even when the - // entry lives at .versions/v_; resolve from the entry itself. + glog.V(1).Infof("streamFromVolumeServers: entry is remote-only, attempting stream-through cache") cacheVersionId := resolvedSourceVersionId(versionId, entry) - cachedEntry := s3a.cacheRemoteObjectForStreaming(r, entry, bucket, object, cacheVersionId) - if cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 { + cachedEntry, cacheErr := s3a.cacheRemoteObjectForStreamingWithShortTimeout(r, entry, bucket, object, cacheVersionId) + if cacheErr == nil && cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 { + // Cache completed, use cached chunks chunks = cachedEntry.GetChunks() entry = cachedEntry glog.V(1).Infof("streamFromVolumeServers: successfully cached remote object, got %d chunks", len(chunks)) - } else { - // Client disconnected: report cancellation, not 503. - if ctxErr := r.Context().Err(); ctxErr != nil { - return ctxErr + } 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) } - // Cache still filling: 503 + Retry-After so SDKs back off and retry. + return newStreamErrorWithResponse(cacheErr) + } else { + // Cache not ready yet; return 503 with Retry-After for client backoff glog.V(1).Infof("streamFromVolumeServers: remote object %s/%s not cached yet, returning 503 for retry", bucket, object) - w.Header().Set("Retry-After", "5") + w.Header().Set("Retry-After", "2") s3err.WriteErrorResponse(w, r, s3err.ErrServiceUnavailable) - return newStreamErrorWithResponse(fmt.Errorf("remote object not cached yet")) + return newStreamErrorWithResponse(fmt.Errorf("remote object not cached yet, will be available on retry")) } } else if totalSize > 0 && len(entry.Content) == 0 { // Not a remote entry but has size without content - this is a data integrity issue @@ -3114,6 +3122,51 @@ func cachedEntryHasLocalData(entry *filer_pb.Entry) bool { // Atomic so tests can shorten it without racing the read path. var remoteCacheStreamingTimeoutNS = int64(20 * time.Second) +// cacheRemoteObjectForStreamingWithShortTimeout polls for cache completion with an adaptive timeout. +// Timeout is based on file size: small files wait longer to maximize cache hits, large files +// fail-fast to improve TTFB. Returns the cached entry and error to allow callers to distinguish +// 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 + } + } + + cacheCtx, cancel := context.WithTimeout(r.Context(), pollTimeout) + defer cancel() + + dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId) + + glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: polling cache status for %s/%s (timeout=%v)", dir, name, pollTimeout) + + cachedEntry, err := s3a.doCacheRemoteObject(cacheCtx, dir, name) + if err != nil { + // Distinguish transient errors (timeout/cancellation) from permanent errors + if cacheCtx.Err() != nil { + glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: %s/%s not cached within %v", dir, name, pollTimeout) + return nil, cacheCtx.Err() + } + glog.V(2).Infof("cacheRemoteObjectForStreamingWithShortTimeout: cache error for %s/%s: %v", dir, name, err) + return nil, err + } + + if cachedEntry != nil && len(cachedEntry.GetChunks()) > 0 { + glog.V(1).Infof("cacheRemoteObjectForStreamingWithShortTimeout: successfully cached %s/%s (%d chunks)", dir, name, len(cachedEntry.GetChunks())) + return cachedEntry, nil + } + + return nil, nil +} + // cacheRemoteObjectForStreaming caches a remote-only object to the local cluster for streaming. // 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 @@ -3179,6 +3232,34 @@ func (s3a *S3ApiServer) cacheRemoteObjectForCopy(ctx context.Context, bucket, ob return nil } +// startBackgroundRemoteCache initiates caching without blocking the request. +// This enables fast TTFB: the client's request returns immediately (via 503 if not +// cached), while a background task fills the cache. Subsequent requests will find +// cached chunks via singleflight deduplication in the filer's CacheRemoteObjectToLocalCluster. +// Uses detached context with reasonable timeout to prevent goroutine pile-up. +func (s3a *S3ApiServer) startBackgroundRemoteCache(bucket, object, versionId string, entry *filer_pb.Entry) { + if !entry.IsInRemoteOnly() { + return + } + + dir, name := s3a.buildVersionedRemoteObjectPath(bucket, object, versionId) + + // Start background cache without blocking. The filer's CacheRemoteObjectToLocalCluster + // uses singleflight internally, so concurrent requests will all benefit from the same + // cache operation. + go func() { + // Use timeout to bound goroutine and prevent pile-up if RPC stalls under load + bgCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + _, err := s3a.doCacheRemoteObject(bgCtx, dir, name) + if err != nil { + glog.V(2).Infof("startBackgroundRemoteCache: cache failed for %s/%s: %v", bucket, object, err) + } else { + glog.V(2).Infof("startBackgroundRemoteCache: cached %s/%s in background", bucket, object) + } + }() +} + // resolvedSourceVersionId falls back to the version recorded on the entry // when the request didn't carry one — necessary for latest-version reads // in versioning-enabled buckets, where the entry lives at .versions/v_.