diff --git a/weed/filer/reader_at.go b/weed/filer/reader_at.go index d87198f54..55231a193 100644 --- a/weed/filer/reader_at.go +++ b/weed/filer/reader_at.go @@ -288,6 +288,15 @@ func (c *ChunkReadAt) doReadAt(ctx context.Context, p []byte, offset int64) (n i } if err != nil { + // a failed chunk leaves its window untouched while the tasks after it + // may well have filled theirs, so only the prefix up to the hole is + // data the caller may use + for _, task := range tasks { + if int64(task.bytesRead) != task.bufferEnd-task.bufferStart { + n = int(task.bufferStart) + task.bytesRead + break + } + } return n, ts, err } } diff --git a/weed/filer/reader_at_test.go b/weed/filer/reader_at_test.go index 782e4b342..fde3fc0e3 100644 --- a/weed/filer/reader_at_test.go +++ b/weed/filer/reader_at_test.go @@ -3,10 +3,14 @@ package filer import ( "bytes" "context" + "errors" "io" "math" "strconv" "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) type mockChunkCache struct { @@ -202,3 +206,54 @@ func TestReaderAtSparseFileDoesNotLeak(t *testing.T) { testReadAt(t, readerAt, 0, 3, 3, io.EOF, []byte{2, 2, 2}, []byte{0, 0, 0}) testReadAt(t, readerAt, 1, 2, 2, io.EOF, []byte{2, 2}, []byte{0, 0}) } + +// holeChunkCache serves every chunk but one, so that chunk falls through to a +// lookup that fails. +type holeChunkCache struct { + mockChunkCache + missingFileId string +} + +func (c *holeChunkCache) ReadChunkAt(data []byte, fileId string, offset uint64) (int, error) { + if fileId == c.missingFileId { + return 0, nil + } + return c.mockChunkCache.ReadChunkAt(data, fileId, offset) +} + +func (c *holeChunkCache) GetMaxFilePartSizeInCache() uint64 { + return math.MaxUint64 +} + +// The parallel reads place their bytes directly in the output buffer, so a +// failing middle chunk leaves a hole with valid data after it. Reporting that +// length would hand the caller bytes it never read. +func TestReaderAtParallelFailureReturnsContiguousPrefix(t *testing.T) { + visibles := NewIntervalList[*VisibleInterval]() + for i, fileId := range []string{"1", "2", "3"} { + addVisibleInterval(visibles, &VisibleInterval{ + start: int64(i) * 4, + stop: int64(i)*4 + 4, + fileId: fileId, + chunkSize: 4, + }) + } + + readerAt := &ChunkReadAt{ + ctx: context.Background(), + chunkViews: ViewFromVisibleIntervals(visibles, 0, math.MaxInt64), + fileSize: 12, + readerCache: NewReaderCache(3, &holeChunkCache{missingFileId: "2"}, func(ctx context.Context, fileId string) ([]string, error) { + return nil, errors.New("volume down") + }, nil), + readerPattern: NewReaderPattern(), + prefetchCount: 3, + } + + buf := make([]byte, 12) + n, err := readerAt.ReadAt(buf, 0) + + require.Error(t, err) + assert.Equal(t, 4, n, "only the bytes before the failed chunk are readable") + assert.Equal(t, []byte{1, 1, 1, 1}, buf[:n]) +} diff --git a/weed/s3api/s3api_local_read_fallback_test.go b/weed/s3api/s3api_local_read_fallback_test.go new file mode 100644 index 000000000..c692ff336 --- /dev/null +++ b/weed/s3api/s3api_local_read_fallback_test.go @@ -0,0 +1,318 @@ +package s3api + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "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/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/util/chunk_cache" + "github.com/seaweedfs/seaweedfs/weed/wdclient" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubReaderAt stands in for ChunkReadAt in probe unit tests. +type stubReaderAt struct { + n int + err error +} + +func (s stubReaderAt) ReadAtWithTime(ctx context.Context, p []byte, offset int64) (int, int64, error) { + return s.n, 0, s.err +} + +// blockingReaderAt never returns until the context is cancelled. +type blockingReaderAt struct{} + +func (blockingReaderAt) ReadAtWithTime(ctx context.Context, p []byte, offset int64) (int, int64, error) { + <-ctx.Done() + return 0, 0, ctx.Err() +} + +func TestProbeReadable(t *testing.T) { + t.Run("readable local copy", func(t *testing.T) { + assert.NoError(t, probeReadable(context.Background(), stubReaderAt{n: 1}, 0, time.Second)) + }) + + t.Run("final byte with trailing EOF counts as readable", func(t *testing.T) { + assert.NoError(t, probeReadable(context.Background(), stubReaderAt{n: 1, err: io.EOF}, 0, time.Second)) + }) + + t.Run("zero-byte EOF is unreadable", func(t *testing.T) { + assert.Error(t, probeReadable(context.Background(), stubReaderAt{n: 0, err: io.EOF}, 0, time.Second)) + }) + + t.Run("zero-byte read without error is unreadable", func(t *testing.T) { + assert.ErrorIs(t, probeReadable(context.Background(), stubReaderAt{n: 0, err: nil}, 0, time.Second), io.ErrUnexpectedEOF) + }) + + t.Run("read error surfaces", func(t *testing.T) { + boom := errors.New("volume: connection refused") + assert.ErrorIs(t, probeReadable(context.Background(), stubReaderAt{err: boom}, 0, time.Second), boom) + }) + + t.Run("stuck volume trips the timeout instead of blocking", func(t *testing.T) { + start := time.Now() + err := probeReadable(context.Background(), blockingReaderAt{}, 0, 50*time.Millisecond) + assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.Less(t, time.Since(start), time.Second, "must not block past the probe timeout") + }) +} + +func TestShouldFallBackToRemote(t *testing.T) { + const size = int64(100) + remoteEntry := func() *filer_pb.Entry { + return &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(size)}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: size}, + } + } + server := func() *S3ApiServer { + return &S3ApiServer{option: &S3ApiServerOption{}} + } + + t.Run("matching remote", func(t *testing.T) { + assert.True(t, server().shouldFallBackToRemote(remoteEntry(), size, "")) + }) + + t.Run("null versionId is treated as unversioned", func(t *testing.T) { + assert.True(t, server().shouldFallBackToRemote(remoteEntry(), size, "null")) + }) + + t.Run("versioned read cannot use the unversioned remote key", func(t *testing.T) { + assert.False(t, server().shouldFallBackToRemote(remoteEntry(), size, "v123")) + }) + + t.Run("latest read that resolves to a version cannot use the unversioned key", func(t *testing.T) { + versioned := remoteEntry() + versioned.Extended = map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v9")} + assert.False(t, server().shouldFallBackToRemote(versioned, size, "")) + }) + + t.Run("no remote entry", func(t *testing.T) { + local := &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{FileSize: uint64(size)}} + assert.False(t, server().shouldFallBackToRemote(local, size, "")) + }) + + t.Run("size mismatch is not served as identical bytes", func(t *testing.T) { + assert.False(t, server().shouldFallBackToRemote(remoteEntry(), size+1, "")) + }) +} + +// The fake filer does not implement LookupVolume, so every chunk read through +// this ReaderCache fails, standing in for an unreachable volume server. +func newLocalReadFallbackServer(t *testing.T, filerAddr pb.ServerAddress) *S3ApiServer { + t.Helper() + s3a := newRemoteCacheTestServer(filerAddr) + fc := wdclient.NewFilerClient(s3a.option.Filers, s3a.option.GrpcDialOption, s3a.option.DataCenter) + s3a.filerClient = fc + s3a.readerCache = filer.NewReaderCache(8, (*chunk_cache.TieredChunkCache)(nil), fc.GetLookupFileIdFunction(), fc) + return s3a +} + +// a remote-mounted object whose cached copy is in the metadata but unreadable +func cachedEntry(content []byte) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "obj.bin", + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(content))}, + Chunks: []*filer_pb.FileChunk{{FileId: "1,0123456789ab", Size: uint64(len(content)), Offset: 0}}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: int64(len(content))}, + } +} + +func TestS3CachedReadFallsBackToRemote(t *testing.T) { + content := []byte("0123456789") + + t.Run("unreadable local copy is served from the mounted remote", func(t *testing.T) { + client := &fakeStreamRemoteClient{data: content} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-cachedfail", nil)) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + + err := s3a.streamFromVolumeServers(w, r, cachedEntry(content), "", "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 mounted remote") + assert.Equal(t, "/data/dir/obj.bin", client.gotLoc.Path) + }) + + t.Run("range fallback serves the requested window from the remote", func(t *testing.T) { + client := &fakeStreamRemoteClient{data: content} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-cachedrange", nil)) + 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, cachedEntry(content), "", "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) + }) + + t.Run("a local-only object keeps the existing error", func(t *testing.T) { + client := &fakeStreamRemoteClient{data: content} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-nofallback", nil)) + local := cachedEntry(content) + local.RemoteEntry = nil + w := httptest.NewRecorder() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil).WithContext(ctx) + + err := s3a.streamFromVolumeServers(w, r, local, "", "mybucket", "dir/obj.bin", "") + + require.Error(t, err) + assert.Nil(t, client.gotLoc, "an object that is not remote-mounted has no remote to fall back to") + }) + + t.Run("versioned read does not fall back to the unversioned remote key", func(t *testing.T) { + client := &fakeStreamRemoteClient{data: content} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-cachedversioned", nil)) + versioned := cachedEntry(content) + versioned.Extended = map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v456")} + w := httptest.NewRecorder() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin?versionId=v456", nil).WithContext(ctx) + + err := s3a.streamFromVolumeServers(w, r, versioned, "", "mybucket", "dir/obj.bin", "v456") + + require.Error(t, err) + assert.Nil(t, client.gotLoc, "a versioned read must not serve the unversioned remote object") + }) +} + +// a multi-chunk object whose later chunk's volume server is unreachable +type truncatedReaderAt struct { + data []byte + breakAt int64 + err error +} + +func (t truncatedReaderAt) ReadAt(p []byte, offset int64) (int, error) { + if offset >= t.breakAt { + return 0, t.err + } + return copy(p, t.data[offset:t.breakAt]), nil +} + +// a client that disconnected mid-response +type failingWriter struct{ err error } + +func (f failingWriter) Write(p []byte) (int, error) { return 0, f.err } + +// the window the probe cannot cover: the byte at offset reads fine, the response +// is committed, and only then does a later chunk turn out to be unreadable +func TestStreamRangeToClientFinishesFromRemote(t *testing.T) { + content := []byte("0123456789") + size := int64(len(content)) + newServer := func(t *testing.T, name string) (*S3ApiServer, *fakeStreamRemoteClient) { + client := &fakeStreamRemoteClient{data: content} + remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client} + return newLocalReadFallbackServer(t, startStreamThroughFiler(t, name, nil)), client + } + + t.Run("later chunk failure is finished from the remote", func(t *testing.T) { + s3a, client := newServer(t, "faketest-midstream") + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + local := truncatedReaderAt{data: content, breakAt: 4, err: errors.New("volume: connection refused")} + + written, err := s3a.streamRangeToClient(w, r, local, cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "") + + require.NoError(t, err) + assert.Equal(t, size, written) + assert.Equal(t, content, w.Body.Bytes()) + assert.Equal(t, int64(4), client.gotOffset, "the remote must resume where the local copy stopped") + assert.Equal(t, int64(6), client.gotSize) + }) + + t.Run("a short local read reported as clean EOF is not served truncated", func(t *testing.T) { + s3a, client := newServer(t, "faketest-shortread") + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + local := truncatedReaderAt{data: content, breakAt: 7, err: io.EOF} + + written, err := s3a.streamRangeToClient(w, r, local, cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "") + + require.NoError(t, err) + assert.Equal(t, size, written) + assert.Equal(t, content, w.Body.Bytes()) + assert.Equal(t, int64(7), client.gotOffset) + }) + + t.Run("range read resumes at the absolute offset", func(t *testing.T) { + s3a, client := newServer(t, "faketest-midstreamrange") + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + local := truncatedReaderAt{data: content, breakAt: 5, err: errors.New("volume: connection refused")} + + written, err := s3a.streamRangeToClient(w, r, local, cachedEntry(content), "mybucket", "dir/obj.bin", 2, 6, size, "") + + require.NoError(t, err) + assert.Equal(t, int64(6), written) + assert.Equal(t, content[2:8], w.Body.Bytes()) + assert.Equal(t, int64(5), client.gotOffset) + assert.Equal(t, int64(3), client.gotSize) + }) + + t.Run("a remote overwritten since caching is not spliced onto the local prefix", func(t *testing.T) { + s3a, client := newServer(t, "faketest-midstreamchanged") + client.stat = &filer_pb.RemoteEntry{RemoteSize: size, RemoteETag: "reuploaded"} + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + boom := errors.New("volume: connection refused") + + written, err := s3a.streamRangeToClient(w, r, truncatedReaderAt{data: content, breakAt: 4, err: boom}, cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "") + + require.Error(t, err) + assert.Equal(t, int64(4), written, "a truncated body beats one mixing two generations") + assert.Nil(t, client.gotLoc, "must not read a remote that no longer matches what was cached") + }) + + t.Run("a local-only object keeps the mid-stream error", func(t *testing.T) { + s3a, client := newServer(t, "faketest-midstreamlocal") + local := cachedEntry(content) + local.RemoteEntry = nil + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + boom := errors.New("volume: connection refused") + + written, err := s3a.streamRangeToClient(w, r, truncatedReaderAt{data: content, breakAt: 4, err: boom}, local, "mybucket", "dir/obj.bin", 0, size, size, "") + + assert.ErrorIs(t, err, boom) + assert.Equal(t, int64(4), written) + assert.Nil(t, client.gotLoc, "an object that is not remote-mounted has no remote to fall back to") + }) + + t.Run("a failed write to the client is not refetched from the remote", func(t *testing.T) { + s3a, client := newServer(t, "faketest-midstreamwrite") + r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil) + broken := errors.New("write: broken pipe") + + written, err := s3a.streamRangeToClient(failingWriter{err: broken}, r, bytes.NewReader(content), cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "") + + assert.ErrorIs(t, err, broken) + assert.Zero(t, written) + assert.Nil(t, client.gotLoc, "the client is gone; refetching from the remote is wasted work") + }) +} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index b56bca6c0..d9340ac29 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -43,13 +43,17 @@ var zeroBuf = make([]byte, 32*1024) // countingWriter wraps an io.Writer to count bytes written type countingWriter struct { - w io.Writer - written int64 + w io.Writer + written int64 + writeErr error } func (cw *countingWriter) Write(p []byte) (int, error) { n, err := cw.w.Write(p) cw.written += int64(n) + if err != nil { + cw.writeErr = err + } return n, err } @@ -1037,33 +1041,11 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R // 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) + if served, streamErr := s3a.serveObjectFromRemoteMount(w, r, entry, bucket, object, offset, size, isRangeRequest, totalSize, t0); served { + if streamErr != nil { + return newStreamErrorWithResponse(streamErr) } 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 @@ -1119,6 +1101,15 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R } else { glog.Errorf("streamFromVolumeServers: failed to resolve chunks: %v", err) } + if s3a.shouldFallBackToRemote(entry, totalSize, versionId) { + glog.V(1).Infof("streamFromVolumeServers: resolving chunks for %s/%s failed, falling back to remote mount", bucket, object) + if served, streamErr := s3a.serveObjectFromRemoteMount(w, r, entry, bucket, object, offset, size, isRangeRequest, totalSize, t0); served { + if streamErr != nil { + return newStreamErrorWithResponse(streamErr) + } + return nil + } + } // Write S3-compliant XML error response s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) return newStreamErrorWithResponse(fmt.Errorf("failed to resolve chunks: %v", err)) @@ -1142,6 +1133,28 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R reader := filer.NewChunkReaderAtFromClient(ctx, s3a.readerCache, chunkViews, totalSize, filer.DefaultPrefetchCount) streamPrepTime = time.Since(tStreamPrep) + // A cached chunk whose volume server is down, or whose needle was evicted and + // now 404s under retry-backoff, would stall and then 500 after the headers are + // already sent -- serve the authoritative remote instead. + if s3a.shouldFallBackToRemote(entry, totalSize, versionId) { + if probeErr := probeReadable(ctx, reader, offset, localReadProbeTimeout); probeErr != nil { + if isCanceledStreamingError(probeErr) { + glog.V(3).Infof("streamFromVolumeServers: request canceled while probing %s/%s: %v", bucket, object, probeErr) + return probeErr + } + glog.V(1).Infof("streamFromVolumeServers: local read of %s/%s failed (%v), falling back to remote mount", bucket, object, probeErr) + if served, streamErr := s3a.serveObjectFromRemoteMount(w, r, entry, bucket, object, offset, size, isRangeRequest, totalSize, t0); served { + if streamErr != nil { + return newStreamErrorWithResponse(streamErr) + } + return nil + } + glog.Errorf("streamFromVolumeServers: local read of %s/%s failed and remote mount unavailable: %v", bucket, object, probeErr) + s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) + return newStreamErrorWithResponse(probeErr) + } + } + // All validation and preparation successful - NOW set headers and write status tHeaderSet := time.Now() s3a.setResponseHeaders(w, r, entry, totalSize) @@ -1170,40 +1183,147 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R // io.CopyBuffer drains them as fast memcpys. tStreamExec := time.Now() glog.V(4).Infof("streamFromVolumeServers: starting chunk reader, offset=%d, size=%d", offset, size) - cw := &countingWriter{w: w} - // Cap the copy buffer to the response size so small-object GETs (common - // for thumbnails, config files, etc.) don't allocate a 256 KiB scratch - // buffer per request. - const maxCopyBuf = 256 * 1024 - copyBufSize := int64(maxCopyBuf) - if size > 0 && size < copyBufSize { - copyBufSize = size - } - copyBuf := make([]byte, copyBufSize) - _, err = io.CopyBuffer(cw, io.NewSectionReader(reader, offset, size), copyBuf) + written, err := s3a.streamRangeToClient(w, r, reader, entry, bucket, object, offset, size, totalSize, versionId) streamExecTime = time.Since(tStreamExec) // Track traffic even on partial writes for accurate egress accounting - if cw.written > 0 { - BucketTrafficSent(cw.written, r) + if written > 0 { + BucketTrafficSent(written, r) } if err != nil { switch { case isCanceledStreamingError(err): // Client disconnected mid-stream (e.g. Nginx upstream timeout, browser cancel) - expected - glog.V(3).Infof("streamFromVolumeServers: client disconnected after writing %d bytes: %v", cw.written, err) + glog.V(3).Infof("streamFromVolumeServers: client disconnected after writing %d bytes: %v", written, err) case errors.Is(err, context.DeadlineExceeded): // Server-side deadline exceeded - unexpected, warrants operator attention - glog.Warningf("streamFromVolumeServers: server-side deadline exceeded after writing %d bytes: %v", cw.written, err) + glog.Warningf("streamFromVolumeServers: server-side deadline exceeded after writing %d bytes: %v", written, err) default: - glog.Errorf("streamFromVolumeServers: streamFn failed after writing %d bytes: %v", cw.written, err) + glog.Errorf("streamFromVolumeServers: streamFn failed after writing %d bytes: %v", written, err) } // Streaming error after WriteHeader was called - response already partially written return newStreamErrorWithResponse(err) } - glog.V(4).Infof("streamFromVolumeServers: streamFn completed successfully, wrote %d bytes", cw.written) + glog.V(4).Infof("streamFromVolumeServers: streamFn completed successfully, wrote %d bytes", written) return nil } +// short, so a down volume server or an evicted needle under retry-backoff falls +// back promptly instead of stalling the request +const localReadProbeTimeout = 2 * time.Second + +// shouldFallBackToRemote reports whether a failed local read may be served from +// the mounted remote instead: openRemoteStream only reads the unversioned key, +// and the sizes must match for the bytes to be identical under the +// already-computed Content-Length. +func (s3a *S3ApiServer) shouldFallBackToRemote(entry *filer_pb.Entry, totalSize int64, versionId string) bool { + if v := resolvedSourceVersionId(versionId, entry); v != "" && v != "null" { + return false + } + remote := entry.GetRemoteEntry() + return totalSize > 0 && remote != nil && remote.GetRemoteSize() == totalSize +} + +// the context-honoring read of filer.ChunkReadAt, so probeReadable can bound it +type ctxReaderAt interface { + ReadAtWithTime(ctx context.Context, p []byte, offset int64) (int, int64, error) +} + +// probeReadable decides whether the local copy is usable by reading one byte at +// offset. Only a returned byte counts -- a zero-byte read, EOF or not, would +// otherwise wave through a stream that then truncates. +func probeReadable(ctx context.Context, reader ctxReaderAt, offset int64, timeout time.Duration) error { + probeCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + var b [1]byte + n, _, err := reader.ReadAtWithTime(probeCtx, b[:], offset) + if n == len(b) { + return nil + } + if err == nil { + err = io.ErrUnexpectedEOF + } + return err +} + +// serveObjectFromRemoteMount serves [offset, offset+size) straight from the +// mounted remote. served=false means nothing was written and the caller still +// owns the error path; once served is true the response is committed. +func (s3a *S3ApiServer) serveObjectFromRemoteMount(w http.ResponseWriter, r *http.Request, entry *filer_pb.Entry, bucket, object string, offset, size int64, isRangeRequest bool, totalSize int64, t0 time.Time) (served bool, err error) { + remoteReader, remoteErr := s3a.openRemoteStream(r.Context(), bucket, object, offset, size, nil) + if remoteErr != nil { + glog.Warningf("streamFromVolumeServers: origin stream %s/%s: %v", bucket, object, remoteErr) + return false, 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 true, copyErr + } + return true, nil +} + +// streamRangeToClient copies the local copy into the already-committed response. +// The probe only proved the byte at offset readable, so a later chunk can still +// be gone: finish such a body from the remote -- if it is still the generation +// that was cached -- instead of truncating it under the Content-Length already +// sent. +func (s3a *S3ApiServer) streamRangeToClient(w io.Writer, r *http.Request, reader io.ReaderAt, entry *filer_pb.Entry, bucket, object string, offset, size, totalSize int64, versionId string) (int64, error) { + // small-object GETs shouldn't allocate a 256 KiB scratch buffer per request + const maxCopyBuf = 256 * 1024 + copyBufSize := int64(maxCopyBuf) + if size > 0 && size < copyBufSize { + copyBufSize = size + } + cw := &countingWriter{w: w} + _, err := io.CopyBuffer(cw, io.NewSectionReader(reader, offset, size), make([]byte, copyBufSize)) + if err == nil && cw.written < size { + // a short local read surfaces as a clean EOF + err = io.ErrUnexpectedEOF + } + if err == nil || cw.writeErr != nil || isCanceledStreamingError(err) || !s3a.shouldFallBackToRemote(entry, totalSize, versionId) { + return cw.written, err + } + + remaining := size - cw.written + glog.V(1).Infof("streamFromVolumeServers: local read of %s/%s failed after %d bytes (%v), finishing from the remote mount", bucket, object, cw.written, err) + // the bytes already sent are the cached generation, so splice on the remote + // only while it still is that generation + remoteReader, remoteErr := s3a.openRemoteStream(r.Context(), bucket, object, offset+cw.written, remaining, entry.GetRemoteEntry()) + if remoteErr != nil { + glog.Warningf("streamFromVolumeServers: origin stream %s/%s: %v", bucket, object, remoteErr) + return cw.written, err + } + defer remoteReader.Close() + if _, copyErr := io.CopyN(cw, remoteReader, remaining); copyErr != nil { + if copyErr == io.EOF { + copyErr = io.ErrUnexpectedEOF + } + glog.V(2).Infof("streamFromVolumeServers: origin stream %s/%s ended after %d bytes: %v", bucket, object, cw.written, copyErr) + return cw.written, copyErr + } + return cw.written, nil +} + // Shared HTTP client for volume server requests (connection pooling) var volumeServerHTTPClient = &http.Client{ Timeout: 5 * time.Minute, @@ -3139,7 +3259,9 @@ func (s3a *S3ApiServer) buildRemoteObjectPath(bucket, object string) (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) { +// cached, when set, is the remote generation the local copy was made from: the +// stream is refused unless the remote still matches it. +func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object string, offset, size int64, cached *filer_pb.RemoteEntry) (io.ReadCloser, error) { dir, name := s3a.buildRemoteObjectPath(bucket, object) var storageConf *remote_pb.RemoteConf @@ -3180,6 +3302,15 @@ func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object str } loc := filer.MapFullPathToRemoteStorageLocation(util.FullPath(localMountedDir), mountedLocation, util.FullPath(dir).Child(name)) + if cached != nil { + current, statErr := client.StatFile(loc) + if statErr != nil { + return nil, statErr + } + if current.GetRemoteSize() != cached.GetRemoteSize() || current.GetRemoteETag() != cached.GetRemoteETag() || current.GetRemoteMtime() != cached.GetRemoteMtime() { + return nil, fmt.Errorf("remote %s/%s changed since it was cached", bucket, object) + } + } return streamer.ReadFileAsStream(ctx, loc, offset, size) } diff --git a/weed/s3api/s3api_remote_storage_test.go b/weed/s3api/s3api_remote_storage_test.go index ccebd581a..92c910e0c 100644 --- a/weed/s3api/s3api_remote_storage_test.go +++ b/weed/s3api/s3api_remote_storage_test.go @@ -609,11 +609,19 @@ func TestCacheRemoteObjectForStreamingCached(t *testing.T) { type fakeStreamRemoteClient struct { remote_storage.RemoteStorageClient data []byte + stat *filer_pb.RemoteEntry // what the remote reports now; defaults to matching data gotLoc *remote_pb.RemoteStorageLocation gotOffset int64 gotSize int64 } +func (c *fakeStreamRemoteClient) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) { + if c.stat != nil { + return c.stat, nil + } + return &filer_pb.RemoteEntry{RemoteSize: int64(len(c.data))}, nil +} + 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)))