From 63dfda3412268ce5c1aa222624252255b8644fed Mon Sep 17 00:00:00 2001 From: shr Date: Mon, 26 Oct 2026 16:08:14 +0800 Subject: [PATCH] fix 12074 --- weed/server/volume_server_handlers_read.go | 10 +- .../volume_server_handlers_read_test.go | 275 ++++++++++++++++++ 2 files changed, 284 insertions(+), 1 deletion(-) diff --git a/weed/server/volume_server_handlers_read.go b/weed/server/volume_server_handlers_read.go index 16ab5bf96a..c2b7c71ac0 100644 --- a/weed/server/volume_server_handlers_read.go +++ b/weed/server/volume_server_handlers_read.go @@ -65,6 +65,11 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re return } lookupResult, err := operation.LookupVolumeId(vs.GetMaster, vs.grpcDialOption, volumeId.String()) + if err != nil && vs.ReadMode == "proxy" { + glog.V(0).InfofCtx(r.Context(), "lookup error: %v %s", err, r.URL.Path) + InternalError(w, r, err) + return + } if err != nil || len(lookupResult.Locations) <= 0 { glog.V(0).Infoln("lookup error:", err, r.URL.Path) stats.VolumeServerFileReadInvalidNeedles.Inc() @@ -100,7 +105,7 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re query := r.URL.Query() query.Set(reqIsProxied, "true") r.URL.RawQuery = query.Encode() - request, err := http.NewRequest(http.MethodGet, r.URL.String(), nil) + request, err := http.NewRequestWithContext(r.Context(), r.Method, r.URL.String(), nil) if err != nil { glog.V(0).Infof("failed to instance http request of url %s: %v", redactedURL(r), err) InternalError(w, r, nil) @@ -114,6 +119,9 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re response, err := util_http.GetGlobalHttpClient().Do(request) if err != nil { + if r.Context().Err() != nil { + return + } stats.VolumeServerHandlerCounter.WithLabelValues(stats.FailedReadProxyReq).Inc() glog.V(0).Infof("request remote url %s: %v", redactedURL(r), err) InternalError(w, r, nil) diff --git a/weed/server/volume_server_handlers_read_test.go b/weed/server/volume_server_handlers_read_test.go index 1402a9954d..3f123abe7f 100644 --- a/weed/server/volume_server_handlers_read_test.go +++ b/weed/server/volume_server_handlers_read_test.go @@ -1,12 +1,25 @@ package weed_server import ( + "context" "errors" + "io" + "net" "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" + + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" ) func TestInternalErrorWritesTheReadError(t *testing.T) { @@ -96,3 +109,265 @@ func TestNotModifiedChecksIfNoneMatchFirst(t *testing.T) { }) } } + +type proxyTestMaster struct { + master_pb.UnimplementedSeaweedServer + lookup func(context.Context, *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) +} + +func (m *proxyTestMaster) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) { + return m.lookup(ctx, req) +} + +func newProxyTestHTTPServer(t *testing.T, handler http.Handler) *httptest.Server { + t.Helper() + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := &httptest.Server{Listener: listener, Config: &http.Server{Handler: handler}} + server.Start() + t.Cleanup(server.Close) + return server +} + +type proxyTestResponseWriter struct { + http.ResponseWriter + statusWritten chan struct{} +} + +func (w proxyTestResponseWriter) WriteHeader(status int) { + w.ResponseWriter.WriteHeader(status) + w.statusWritten <- struct{}{} +} + +func newProxyTestVolumeServer(t *testing.T, volumeID string, locations []*master_pb.Location, lookupErr error) *VolumeServer { + t.Helper() + operation.InvalidateVolumeIdLocationCache(volumeID) + + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + masterAddress := pb.NewServerAddress("127.0.0.1", 9333, listener.Addr().(*net.TCPAddr).Port) + masterServer := grpc.NewServer() + master_pb.RegisterSeaweedServer(masterServer, &proxyTestMaster{lookup: func(_ context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) { + if lookupErr != nil { + return nil, lookupErr + } + if len(req.VolumeOrFileIds) == 0 { + return nil, errors.New("lookup request contains no volume IDs") + } + return &master_pb.LookupVolumeResponse{VolumeIdLocations: []*master_pb.LookupVolumeResponse_VolumeIdLocation{{ + VolumeOrFileId: req.VolumeOrFileIds[0], + Locations: locations, + }}}, nil + }}) + go func() { + _ = masterServer.Serve(listener) + }() + + vs := &VolumeServer{ + store: &storage.Store{Ip: "192.0.2.10", Port: 8080}, + ReadMode: "proxy", + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + vs.setCurrentMaster(masterAddress) + t.Cleanup(func() { + operation.InvalidateVolumeIdLocationCache(volumeID) + pb.InvalidateGrpcConnection(masterAddress.ToGrpcAddress()) + masterServer.Stop() + }) + return vs +} + +func TestProxyReqToTargetServerPreservesMethodAndResponse(t *testing.T) { + const volumeID = "849300101" + type observedRequest struct { + method string + rangeHeader string + ifNoneMatch string + ifModifiedAt string + } + observed := make(chan observedRequest, 2) + peer := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + observed <- observedRequest{ + method: r.Method, + rangeHeader: r.Header.Get("Range"), + ifNoneMatch: r.Header.Get("If-None-Match"), + ifModifiedAt: r.Header.Get("If-Modified-Since"), + } + w.Header().Set("ETag", `"peer-etag"`) + w.Header().Set("Content-Range", "bytes 0-3/10") + w.Header().Set("X-Peer-Header", "preserved") + w.WriteHeader(http.StatusPartialContent) + _, _ = io.WriteString(w, "data") + })) + + remoteAddress := strings.TrimPrefix(peer.URL, "http://") + vs := newProxyTestVolumeServer(t, volumeID, []*master_pb.Location{{Url: remoteAddress}}, nil) + + for _, method := range []string{http.MethodGet, http.MethodHead} { + t.Run(method, func(t *testing.T) { + r := httptest.NewRequest(method, "/"+volumeID+",abcdef", nil) + r.Header.Set("Range", "bytes=0-3") + r.Header.Set("If-None-Match", `"client-etag"`) + r.Header.Set("If-Modified-Since", "Wed, 21 Oct 2015 07:28:00 GMT") + w := httptest.NewRecorder() + vs.proxyReqToTargetServer(w, r) + + got := <-observed + if got.method != method { + t.Errorf("peer received method %q, want %q", got.method, method) + } + if got.rangeHeader != "bytes=0-3" || got.ifNoneMatch != `"client-etag"` || got.ifModifiedAt != "Wed, 21 Oct 2015 07:28:00 GMT" { + t.Errorf("peer received conditional/range headers %+v", got) + } + if w.Code != http.StatusPartialContent { + t.Errorf("status %d, want %d", w.Code, http.StatusPartialContent) + } + if w.Header().Get("ETag") != `"peer-etag"` || w.Header().Get("Content-Range") != "bytes 0-3/10" || w.Header().Get("X-Peer-Header") != "preserved" { + t.Errorf("peer response headers were not preserved: %v", w.Header()) + } + wantBody := "data" + if method == http.MethodHead { + wantBody = "" + } + if gotBody := w.Body.String(); gotBody != wantBody { + t.Errorf("response body %q, want %q", gotBody, wantBody) + } + }) + } +} + +func TestProxyReqToTargetServerCancelsUpstreamWithInboundRequest(t *testing.T) { + const volumeID = "849300102" + peerStarted := make(chan struct{}) + peerCanceled := make(chan struct{}) + releasePeer := make(chan struct{}) + var releasePeerOnce sync.Once + peer := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(peerStarted) + select { + case <-r.Context().Done(): + close(peerCanceled) + case <-releasePeer: + } + })) + defer releasePeerOnce.Do(func() { close(releasePeer) }) + + remoteAddress := strings.TrimPrefix(peer.URL, "http://") + vs := newProxyTestVolumeServer(t, volumeID, []*master_pb.Location{{Url: remoteAddress}}, nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + r := httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil).WithContext(ctx) + w := httptest.NewRecorder() + proxyDone := make(chan struct{}) + go func() { + vs.proxyReqToTargetServer(w, r) + close(proxyDone) + }() + + select { + case <-peerStarted: + case <-time.After(2 * time.Second): + t.Fatal("peer did not receive the proxied request") + } + cancel() + select { + case <-peerCanceled: + case <-time.After(time.Second): + t.Error("canceling the inbound request did not cancel the peer request") + } + releasePeerOnce.Do(func() { close(releasePeer) }) + select { + case <-proxyDone: + case <-time.After(2 * time.Second): + t.Error("proxy handler did not finish after inbound cancellation") + } +} + +func TestProxyReqToTargetServerCancelsUpstreamBodyReadWithInboundRequest(t *testing.T) { + const volumeID = "849300106" + peerCanceled := make(chan struct{}) + releasePeer := make(chan struct{}) + var releasePeerOnce sync.Once + peer := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + select { + case <-r.Context().Done(): + close(peerCanceled) + case <-releasePeer: + } + })) + defer releasePeerOnce.Do(func() { close(releasePeer) }) + + remoteAddress := strings.TrimPrefix(peer.URL, "http://") + vs := newProxyTestVolumeServer(t, volumeID, []*master_pb.Location{{Url: remoteAddress}}, nil) + r := httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil) + ctx, cancel := context.WithCancel(r.Context()) + r = r.WithContext(ctx) + recorder := httptest.NewRecorder() + statusWritten := make(chan struct{}, 1) + w := proxyTestResponseWriter{ResponseWriter: recorder, statusWritten: statusWritten} + proxyDone := make(chan struct{}) + go func() { + vs.proxyReqToTargetServer(w, r) + close(proxyDone) + }() + + select { + case <-statusWritten: + case <-time.After(2 * time.Second): + cancel() + t.Fatal("proxy did not receive the peer response headers") + } + cancel() + select { + case <-peerCanceled: + case <-time.After(time.Second): + t.Error("canceling the inbound request did not cancel the peer response body") + } + releasePeerOnce.Do(func() { close(releasePeer) }) + select { + case <-proxyDone: + case <-time.After(2 * time.Second): + t.Error("proxy handler did not finish reading after inbound cancellation") + } +} + +func TestProxyReqToTargetServerDistinguishesLookupFailureFromNoLocation(t *testing.T) { + t.Run("master lookup failure is internal error", func(t *testing.T) { + const volumeID = "849300103" + vs := newProxyTestVolumeServer(t, volumeID, nil, status.Error(codes.Unavailable, "master unavailable")) + w := httptest.NewRecorder() + vs.proxyReqToTargetServer(w, httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil)) + if w.Code != http.StatusInternalServerError { + t.Errorf("status %d, want %d", w.Code, http.StatusInternalServerError) + } + }) + + for _, tc := range []struct { + name string + locations []*master_pb.Location + }{ + {name: "lookup returned no replica locations"}, + {name: "only local replica is not a usable proxy target", locations: []*master_pb.Location{{Url: "192.0.2.10:8080"}}}, + } { + t.Run(tc.name, func(t *testing.T) { + volumeID := "84930010" + if tc.name == "lookup returned no replica locations" { + volumeID += "4" + } else { + volumeID += "5" + } + vs := newProxyTestVolumeServer(t, volumeID, tc.locations, nil) + w := httptest.NewRecorder() + vs.proxyReqToTargetServer(w, httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil)) + if w.Code != http.StatusNotFound { + t.Errorf("status %d, want %d", w.Code, http.StatusNotFound) + } + }) + } +}