diff --git a/weed/operation/chunked_file.go b/weed/operation/chunked_file.go index a5746c8462..5df1647bc1 100644 --- a/weed/operation/chunked_file.go +++ b/weed/operation/chunked_file.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "sort" + "strings" "sync" "google.golang.org/grpc" @@ -94,7 +95,7 @@ func (cm *ChunkManifest) DeleteChunks(masterFn GetMasterFn, usePublicUrl bool, g return nil } -func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (written int64, e error) { +func readChunkNeedle(fileUrl string, w io.Writer, offset int64, size int64, jwt string) (written int64, e error) { req, err := http.NewRequest(http.MethodGet, fileUrl, nil) if err != nil { return written, err @@ -125,7 +126,11 @@ func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (wri return written, util_http.StatusError(fileUrl, resp) } - return io.Copy(w, resp.Body) + written, e = io.CopyN(w, resp.Body, size) + if e == io.EOF { + e = fmt.Errorf("%s: %d of %d bytes: %w", fileUrl, written, size, io.ErrUnexpectedEOF) + } + return written, e } func NewChunkedFileReader(chunkList []*ChunkInfo, master pb.ServerAddress, grpcDialOption grpc.DialOption) *ChunkedFileReader { @@ -175,26 +180,69 @@ func (cf *ChunkedFileReader) WriteTo(w io.Writer) (n int64, err error) { return n, ErrInvalidRange } for ; chunkIndex < len(cf.chunkList); chunkIndex++ { - ci := cf.chunkList[chunkIndex] - // if we need read date from local volume server first? - fileUrl, jwt, lookupError := LookupFileId(func(_ context.Context) pb.ServerAddress { - return cf.master - }, cf.grpcDialOption, ci.Fid) - if lookupError != nil { - return n, lookupError - } - if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset, jwt); e != nil { + wn, e := cf.readChunk(cf.chunkList[chunkIndex], w, chunkStartOffset) + n += wn + cf.pos += wn + if e != nil { return n, e - } else { - n += wn - cf.pos += wn } - chunkStartOffset = 0 } return n, nil } +// readChunk reads the chunk from the first replica that serves it. A failure +// drops the volume's cached locations, and once every cached replica has +// failed the master is asked again. A replica that already sent part of the +// chunk is not continued elsewhere: a volume server streams a needle before +// its checksum is verified, so those bytes may be the corrupt ones. +func (cf *ChunkedFileReader) readChunk(ci *ChunkInfo, w io.Writer, offset int64) (written int64, err error) { + masterFn := func(_ context.Context) pb.ServerAddress { + return cf.master + } + vid, _, _ := strings.Cut(ci.Fid, ",") + dst := &writeErrorRecorder{Writer: w} + tried := make(map[string]bool) + for range 2 { + fileUrls, jwt, lookupError := lookupFileIdUrls(masterFn, cf.grpcDialOption, ci.Fid) + if lookupError != nil { + if err == nil { + err = lookupError + } + return 0, err + } + for _, fileUrl := range fileUrls { + if tried[fileUrl] { + continue + } + tried[fileUrl] = true + written, err = readChunkNeedle(fileUrl, dst, offset, ci.Size-offset, jwt) + if err == nil || dst.err != nil { + return written, err + } + glog.V(0).Infof("read chunk %s from %s: %v", ci.Fid, fileUrl, err) + InvalidateVolumeIdLocationCache(vid) + if written > 0 { + return written, err + } + } + } + return 0, err +} + +type writeErrorRecorder struct { + io.Writer + err error +} + +func (r *writeErrorRecorder) Write(p []byte) (int, error) { + n, err := r.Writer.Write(p) + if err != nil { + r.err = err + } + return n, err +} + func (cf *ChunkedFileReader) ReadAt(p []byte, off int64) (n int, err error) { cf.Seek(off, 0) return cf.Read(p) diff --git a/weed/operation/chunked_file_test.go b/weed/operation/chunked_file_test.go new file mode 100644 index 0000000000..589f77f16d --- /dev/null +++ b/weed/operation/chunked_file_test.go @@ -0,0 +1,257 @@ +package operation + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +type fakeVolumeLocationServer struct { + master_pb.UnimplementedSeaweedServer + mu sync.Mutex + calls int + urls []string +} + +func (s *fakeVolumeLocationServer) LookupVolume(_ context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.calls++ + resp := &master_pb.LookupVolumeResponse{} + for _, vid := range req.VolumeOrFileIds { + var locs []*master_pb.Location + for _, u := range s.urls { + locs = append(locs, &master_pb.Location{Url: u}) + } + resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{ + VolumeOrFileId: vid, + Locations: locs, + }) + } + return resp, nil +} + +func (s *fakeVolumeLocationServer) lookups() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.calls +} + +type chunkReplica struct { + *httptest.Server + mu sync.Mutex + ranges []string +} + +func (c *chunkReplica) host() string { + return strings.TrimPrefix(c.URL, "http://") +} + +func (c *chunkReplica) requests() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.ranges...) +} + +// newChunkReplica serves data; with cutAfter >= 0 it announces the full length +// but drops the connection after cutAfter bytes. +func newChunkReplica(t *testing.T, data []byte, cutAfter int) *chunkReplica { + c := &chunkReplica{} + c.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c.mu.Lock() + c.ranges = append(c.ranges, r.Header.Get("Range")) + c.mu.Unlock() + if cutAfter < 0 { + http.ServeContent(w, r, "", time.Time{}, bytes.NewReader(data)) + return + } + w.Header().Set("Content-Length", strconv.Itoa(len(data))) + w.WriteHeader(http.StatusOK) + w.Write(data[:cutAfter]) + w.(http.Flusher).Flush() + conn, _, err := w.(http.Hijacker).Hijack() + if err == nil { + conn.Close() + } + })) + t.Cleanup(c.Close) + return c +} + +func deadAddress(t *testing.T) string { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := lis.Addr().String() + lis.Close() + return addr +} + +func testChunkData(size int) []byte { + data := make([]byte, size) + for i := range data { + data[i] = byte(i * 7) + } + return data +} + +func TestChunkedFileReaderTriesTheNextReplica(t *testing.T) { + const vid = "91201" + data := testChunkData(4096) + healthy := newChunkReplica(t, data, -1) + fake := &fakeVolumeLocationServer{urls: []string{healthy.host()}} + master := startFakeMasterServer(t, fake) + dial := grpc.WithTransportCredentials(insecure.NewCredentials()) + + vc.Set(vid, []Location{{Url: deadAddress(t)}, {Url: healthy.host(), DataInRemote: true}}, time.Minute) + t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) }) + + chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}} + reader := NewChunkedFileReader(chunks, master, dial) + reader.Seek(1000, io.SeekStart) + got, err := io.ReadAll(reader) + reader.Close() + if err != nil { + t.Fatalf("read: %v", err) + } + if !bytes.Equal(got, data[1000:]) { + t.Fatalf("read %d bytes, want the last %d bytes of the chunk", len(got), len(data)-1000) + } + if r := healthy.requests(); len(r) != 1 || r[0] != "bytes=1000-" { + t.Fatalf("healthy replica requests = %q, want one from byte 1000", r) + } + if _, err := vc.Get(vid); err == nil { + t.Fatalf("cached locations kept after a replica failed") + } + if fake.lookups() != 0 { + t.Fatalf("master lookups = %d, want 0", fake.lookups()) + } +} + +func TestChunkedFileReaderDoesNotContinueAPartlySentChunk(t *testing.T) { + const vid = "91205" + data := testChunkData(4096) + broken := newChunkReplica(t, data, 1000) + healthy := newChunkReplica(t, data, -1) + fake := &fakeVolumeLocationServer{urls: []string{healthy.host()}} + master := startFakeMasterServer(t, fake) + dial := grpc.WithTransportCredentials(insecure.NewCredentials()) + + vc.Set(vid, []Location{{Url: broken.host()}, {Url: healthy.host(), DataInRemote: true}}, time.Minute) + t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) }) + + chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}} + reader := NewChunkedFileReader(chunks, master, dial) + _, err := io.ReadAll(reader) + reader.Close() + if err == nil { + t.Fatalf("read of a chunk cut off mid-body succeeded") + } + if r := healthy.requests(); len(r) != 0 { + t.Fatalf("healthy replica requests = %q, want none", r) + } + + reader = NewChunkedFileReader(chunks, master, dial) + got, err := io.ReadAll(reader) + reader.Close() + if err != nil || !bytes.Equal(got, data) { + t.Fatalf("second read: %d bytes, err %v", len(got), err) + } + if fake.lookups() != 1 || len(broken.requests()) != 1 { + t.Fatalf("second read: %d master lookups, %d requests to the broken replica; want 1 and 1", fake.lookups(), len(broken.requests())) + } +} + +func TestChunkedFileReaderAsksMasterWhenCachedReplicasFail(t *testing.T) { + const vid = "91202" + first, second := testChunkData(3000), testChunkData(2000) + healthy := newChunkReplica(t, second, -1) + fake := &fakeVolumeLocationServer{urls: []string{healthy.host()}} + master := startFakeMasterServer(t, fake) + dial := grpc.WithTransportCredentials(insecure.NewCredentials()) + + firstReplica := newChunkReplica(t, first, -1) + const otherVid = "91203" + vc.Set(otherVid, []Location{{Url: firstReplica.host()}}, time.Minute) + vc.Set(vid, []Location{{Url: deadAddress(t)}}, time.Minute) + t.Cleanup(func() { + InvalidateVolumeIdLocationCache(vid) + InvalidateVolumeIdLocationCache(otherVid) + }) + + chunks := []*ChunkInfo{ + {Fid: otherVid + ",01637037d6", Offset: 0, Size: int64(len(first))}, + {Fid: vid + ",02637037d6", Offset: int64(len(first)), Size: int64(len(second))}, + } + reader := NewChunkedFileReader(chunks, master, dial) + defer reader.Close() + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("read: %v", err) + } + if want := append(append([]byte(nil), first...), second...); !bytes.Equal(got, want) { + t.Fatalf("read %d bytes, want %d", len(got), len(want)) + } + if fake.lookups() != 1 { + t.Fatalf("master lookups = %d, want 1", fake.lookups()) + } +} + +func TestChunkedFileReaderTreatsAShortBodyAsFailed(t *testing.T) { + const vid = "91206" + data := testChunkData(2048) + empty := newChunkReplica(t, nil, -1) + healthy := newChunkReplica(t, data, -1) + vc.Set(vid, []Location{{Url: empty.host()}, {Url: healthy.host(), DataInRemote: true}}, time.Minute) + t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) }) + + chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}} + reader := NewChunkedFileReader(chunks, pb.ServerAddress("127.0.0.1:1"), grpc.WithTransportCredentials(insecure.NewCredentials())) + defer reader.Close() + got, err := io.ReadAll(reader) + if err != nil || !bytes.Equal(got, data) { + t.Fatalf("read: %d bytes, err %v; want %d bytes", len(got), err, len(data)) + } + if len(empty.requests()) != 1 || len(healthy.requests()) != 1 { + t.Fatalf("requests: empty %d, healthy %d; want 1 and 1", len(empty.requests()), len(healthy.requests())) + } +} + +type failingWriter struct{} + +func (failingWriter) Write([]byte) (int, error) { return 0, errors.New("client gone") } + +func TestChunkedFileReaderStopsWhenTheClientGoesAway(t *testing.T) { + const vid = "91204" + data := testChunkData(1024) + a, b := newChunkReplica(t, data, -1), newChunkReplica(t, data, -1) + vc.Set(vid, []Location{{Url: a.host()}, {Url: b.host()}}, time.Minute) + t.Cleanup(func() { InvalidateVolumeIdLocationCache(vid) }) + + chunks := []*ChunkInfo{{Fid: vid + ",01637037d6", Offset: 0, Size: int64(len(data))}} + reader := NewChunkedFileReader(chunks, pb.ServerAddress("127.0.0.1:1"), grpc.WithTransportCredentials(insecure.NewCredentials())) + if _, err := reader.WriteTo(failingWriter{}); err == nil || err.Error() != "client gone" { + t.Fatalf("WriteTo error = %v, want the writer's error", err) + } + if n := len(a.requests()) + len(b.requests()); n != 1 { + t.Fatalf("replica requests = %d, want 1", n) + } + if _, err := vc.Get(vid); err != nil { + t.Fatalf("a client write error dropped the cached locations: %v", err) + } +} diff --git a/weed/operation/lookup.go b/weed/operation/lookup.go index a82504277a..77326a10ac 100644 --- a/weed/operation/lookup.go +++ b/weed/operation/lookup.go @@ -51,31 +51,41 @@ var ( // authorization the master stamped on the volume; pass it through to the // volume server on the read request. func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrl string, jwt string, err error) { + fullUrls, jwt, err := lookupFileIdUrls(masterFn, grpcDialOption, fileId) + if err != nil { + return "", jwt, err + } + return fullUrls[0], jwt, nil +} + +// lookupFileIdUrls returns the read URLs of every replica of fileId, local +// replicas first, each group in random order. +func lookupFileIdUrls(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId string) (fullUrls []string, jwt string, err error) { parts := strings.Split(fileId, ",") if len(parts) != 2 { - return "", jwt, errors.New("Invalid fileId " + fileId) + return nil, jwt, errors.New("Invalid fileId " + fileId) } lookup, lookupError := LookupVolumeId(masterFn, grpcDialOption, parts[0]) if lookupError != nil { - return "", jwt, lookupError + return nil, jwt, lookupError } if len(lookup.Locations) == 0 { - return "", jwt, errors.New("File Not Found") + return nil, jwt, errors.New("File Not Found") } - localUrls := make([]string, 0, len(lookup.Locations)) + var localUrls, remoteUrls []string for _, loc := range lookup.Locations { - if !loc.DataInRemote { - localUrls = append(localUrls, loc.Url) - } - } - if len(localUrls) == 0 { - for _, loc := range lookup.Locations { - localUrls = append(localUrls, loc.Url) + fullUrl := "http://" + loc.Url + "/" + fileId + if loc.DataInRemote { + remoteUrls = append(remoteUrls, fullUrl) + } else { + localUrls = append(localUrls, fullUrl) } } + rand.Shuffle(len(localUrls), func(i, j int) { localUrls[i], localUrls[j] = localUrls[j], localUrls[i] }) + rand.Shuffle(len(remoteUrls), func(i, j int) { remoteUrls[i], remoteUrls[j] = remoteUrls[j], remoteUrls[i] }) - return "http://" + localUrls[rand.IntN(len(localUrls))] + "/" + fileId, lookup.Jwt, nil + return append(localUrls, remoteUrls...), lookup.Jwt, nil } func LookupVolumeId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid string) (*LookupResult, error) {