volume: report the needle read error with a failed read (#11934)

* volume: report the needle read error with a failed read

A read that fails on the volume server, for example on a CRC mismatch,
returned a bare 500, so mount and filer logs only showed the status.
The volume server now writes the error into the response body, and the
read helpers add it to the error they return.

Co-authored-by: Bruce <82158894+half-life666@users.noreply.github.com>

* volume: keep the 500 bare on HEAD requests

writeJsonError's body write fails with ErrBodyNotAllowed under net/http
when the request method is HEAD, adding an error-writing log line on top
of the failed read. Send only the status, matching the no-error path.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* util/http: decode gzip-encoded error bodies in statusError

Full-chunk reads set Accept-Encoding: gzip themselves, so the
transport does not transparently decompress; a gzip-encoded error
body reached statusError as compressed bytes and produced an
unreadable message. statusError now takes the response and decodes
Content-Encoding before extracting the bounded message.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: log the read error when the 500 stays bare

On HEAD requests the body write is skipped, which also skipped the
V(0) response log inside writeJson — a failed HEAD read left no
cause in normal volume-server logs. Log the error in the same
format before the bare 500.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* util/http: check the status before decoding gzip in GetAuthenticated

The gzip reader eagerly consumes the header when it is created, so
statusError saw a drained or corrupt body on a gzipped error response
and the server message was lost. Match the other read helpers and check
the status first.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* util/http: export StatusError

Other volume read paths outside this file need the same error body.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: redact jwt query param when logging failed-read URLs

The new read-failure logs print the request URL, and volume reads accept
?jwt= as a credential, so a failed authenticated read would write the
bearer token to the volume server log.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* util/http: surface the error body on the remaining chunk-read paths

* volume: redact duplicate jwt params in failed-read logs

* operation: pass the lookup jwt on the chunk needle read

---------

Co-authored-by: Bruce <82158894+half-life666@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

weed/operation/chunked_file.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
@@ -101,6 +102,9 @@ func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (wri
 	if offset > 0 {
 		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
 	}
+	if jwt != "" {
+		req.Header.Set("Authorization", security.BearerPrefix+jwt)
+	}
 
 	resp, err := util_http.Do(req)
 	if err != nil {
@@ -118,7 +122,7 @@ func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (wri
 	case http.StatusPartialContent:
 		break
 	default:
-		return written, fmt.Errorf("Read chunk needle error: [%d] %s", resp.StatusCode, fileUrl)
+		return written, util_http.StatusError(fileUrl, resp)
 
 	}
 	return io.Copy(w, resp.Body)

weed/s3api/s3api_object_handlers.go

@@ -2251,8 +2251,9 @@ func (s3a *S3ApiServer) readChunkRange(ctx context.Context, fileId string, offse
 	}
 
 	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
+		err = util_http.StatusError(chunkUrl, resp)
 		resp.Body.Close()
-		return nil, fmt.Errorf("unexpected status code %d for chunk %s", resp.StatusCode, fileId)
+		return nil, err
 	}
 
 	return resp.Body, nil

weed/server/common.go

@@ -109,7 +109,7 @@ func writeJson(w http.ResponseWriter, r *http.Request, httpStatus int, obj inter
 
 	if httpStatus >= 400 {
 		glog.V(0).Infof("response method:%s URL:%s with httpStatus:%d and JSON:%s",
-			r.Method, r.URL.String(), httpStatus, string(bytes))
+			r.Method, redactedURL(r), httpStatus, string(bytes))
 	}
 
 	w.Header().Set("Content-Type", "application/json")
@@ -119,10 +119,21 @@ func writeJson(w http.ResponseWriter, r *http.Request, httpStatus int, obj inter
 	return
 }
 
+func redactedURL(r *http.Request) string {
+	if _, ok := r.URL.Query()["jwt"]; !ok {
+		return r.URL.String()
+	}
+	u := *r.URL
+	q := u.Query()
+	q.Set("jwt", "[redacted]")
+	u.RawQuery = q.Encode()
+	return u.String()
+}
+
 // wrapper for writeJson - just logs errors
 func writeJsonQuiet(w http.ResponseWriter, r *http.Request, httpStatus int, obj interface{}) {
 	if err := writeJson(w, r, httpStatus, obj); err != nil {
-		glog.V(0).Infof("error writing JSON status %s %d: %v", r.URL, httpStatus, err)
+		glog.V(0).Infof("error writing JSON status %s %d: %v", redactedURL(r), httpStatus, err)
 		glog.V(1).Infof("JSON content: %+v", obj)
 	}
 }

weed/server/common_test.go

@@ -400,3 +400,18 @@ func TestSubmitForClientHandlerFileSizeLimit(t *testing.T) {
 		}
 	})
 }
+
+func TestRedactedURLDropsTheJwt(t *testing.T) {
+	r := httptest.NewRequest(http.MethodGet, "/1,abc?jwt=secret-token&size=5", nil)
+	if got := redactedURL(r); strings.Contains(got, "secret-token") || !strings.Contains(got, "size=5") {
+		t.Errorf("got %q, want jwt redacted and other params kept", got)
+	}
+	r = httptest.NewRequest(http.MethodGet, "/1,abc?jwt=&jwt=secret-token&size=5", nil)
+	if got := redactedURL(r); strings.Contains(got, "secret-token") {
+		t.Errorf("got %q, want every jwt value redacted", got)
+	}
+	r = httptest.NewRequest(http.MethodGet, "/1,abc?size=5", nil)
+	if got := redactedURL(r); got != "/1,abc?size=5" {
+		t.Errorf("got %q, want the URL unchanged without a jwt", got)
+	}
+}

weed/server/volume_server_handlers_read.go

@@ -40,10 +40,20 @@ func NotFound(w http.ResponseWriter) {
 	w.WriteHeader(http.StatusNotFound)
 }
 
-func InternalError(w http.ResponseWriter) {
+func InternalError(w http.ResponseWriter, r *http.Request, err error) {
 	stats.VolumeServerFileReadFailures.Inc()
 	stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorGetInternal).Inc()
-	w.WriteHeader(http.StatusInternalServerError)
+	if err == nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	if r.Method == http.MethodHead {
+		glog.V(0).Infof("response method:%s URL:%s with httpStatus:%d and error:%v",
+			r.Method, redactedURL(r), http.StatusInternalServerError, err)
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	writeJsonError(w, r, http.StatusInternalServerError, err)
 }
 
 func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Request) {
@@ -92,8 +102,8 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re
 		r.URL.RawQuery = query.Encode()
 		request, err := http.NewRequest(http.MethodGet, r.URL.String(), nil)
 		if err != nil {
-			glog.V(0).Infof("failed to instance http request of url %s: %v", r.URL.String(), err)
-			InternalError(w)
+			glog.V(0).Infof("failed to instance http request of url %s: %v", redactedURL(r), err)
+			InternalError(w, r, nil)
 			return
 		}
 		for k, vv := range r.Header {
@@ -105,8 +115,8 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re
 		response, err := util_http.GetGlobalHttpClient().Do(request)
 		if err != nil {
 			stats.VolumeServerHandlerCounter.WithLabelValues(stats.FailedReadProxyReq).Inc()
-			glog.V(0).Infof("request remote url %s: %v", r.URL.String(), err)
-			InternalError(w)
+			glog.V(0).Infof("request remote url %s: %v", redactedURL(r), err)
+			InternalError(w, r, nil)
 			return
 		}
 		defer util_http.CloseResponse(response)
@@ -217,7 +227,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		if invalid_needle || err == storage.ErrorDeleted {
 			NotFound(w)
 		} else {
-			InternalError(w)
+			InternalError(w, r, err)
 		}
 		return
 	}

weed/server/volume_server_handlers_read_test.go

@@ -1,11 +1,37 @@
 package weed_server
 
 import (
+	"errors"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 )
 
+func TestInternalErrorWritesTheReadError(t *testing.T) {
+	readErr := errors.New("invalid CRC for needle 2a (got 0a1b2c3d, want 4e5f6071), data on disk corrupted: needle data corrupted")
+	w := httptest.NewRecorder()
+	InternalError(w, httptest.NewRequest(http.MethodGet, "/1,abc", nil), readErr)
+	if w.Code != http.StatusInternalServerError {
+		t.Fatalf("status %d, want 500", w.Code)
+	}
+	if body := strings.TrimSpace(w.Body.String()); body != `{"error":"`+readErr.Error()+`"}` {
+		t.Fatalf("body %q", body)
+	}
+
+	w = httptest.NewRecorder()
+	InternalError(w, httptest.NewRequest(http.MethodGet, "/1,abc", nil), nil)
+	if w.Code != http.StatusInternalServerError || w.Body.Len() != 0 {
+		t.Fatalf("without an error: status %d body %q, want a bare 500", w.Code, w.Body.String())
+	}
+
+	w = httptest.NewRecorder()
+	InternalError(w, httptest.NewRequest(http.MethodHead, "/1,abc", nil), readErr)
+	if w.Code != http.StatusInternalServerError || w.Body.Len() != 0 {
+		t.Fatalf("on a HEAD request: status %d body %q, want a bare 500", w.Code, w.Body.String())
+	}
+}
+
 // A HEAD only attempts a metadata-only read, like a GET: a needle up to
 // PagedReadLimit is read whole. Demanding metadata only would turn that one
 // read into a header read followed by a trailer read.

weed/util/http/http_global_client_util.go

@@ -119,6 +119,11 @@ func GetAuthenticated(ctx context.Context, url, jwt string) ([]byte, bool, error
 	recordReachable(request.URL.Host)
 	defer CloseResponse(response)
 
+	if response.StatusCode >= 400 {
+		retryable := response.StatusCode >= 500
+		return nil, retryable, StatusError(url, response)
+	}
+
 	var reader io.ReadCloser
 	switch response.Header.Get("Content-Encoding") {
 	case "gzip":
@@ -132,10 +137,6 @@ func GetAuthenticated(ctx context.Context, url, jwt string) ([]byte, bool, error
 	}
 
 	b, err := io.ReadAll(reader)
-	if response.StatusCode >= 400 {
-		retryable := response.StatusCode >= 500
-		return nil, retryable, fmt.Errorf("%s: %s", url, response.Status)
-	}
 	if err != nil {
 		if ctxErr := ctx.Err(); ctxErr != nil {
 			return nil, false, ctxErr
@@ -334,7 +335,7 @@ func ReadUrl(ctx context.Context, fileUrl string, cipherKey []byte, isContentCom
 	defer CloseResponse(r)
 
 	if r.StatusCode >= 400 {
-		return 0, fmt.Errorf("%s: %s", fileUrl, r.Status)
+		return 0, StatusError(fileUrl, r)
 	}
 
 	var reader io.ReadCloser
@@ -414,7 +415,7 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte,
 			return false, fmt.Errorf("%s: %s: %w", fileUrl, r.Status, ErrTooManyRequests)
 		}
 		retryable = r.StatusCode >= 499
-		return retryable, fmt.Errorf("%s: %s", fileUrl, r.Status)
+		return retryable, StatusError(fileUrl, r)
 	}
 
 	var reader io.ReadCloser
@@ -504,8 +505,9 @@ func ReadUrlAsReaderCloser(fileUrl string, jwt string, rangeHeader string) (*htt
 		return nil, nil, err
 	}
 	if r.StatusCode >= 400 {
+		err = StatusError(fileUrl, r)
 		CloseResponse(r)
-		return nil, nil, fmt.Errorf("%s: %s", fileUrl, r.Status)
+		return nil, nil, err
 	}
 
 	var reader io.ReadCloser
@@ -523,6 +525,29 @@ func ReadUrlAsReaderCloser(fileUrl string, jwt string, rangeHeader string) (*htt
 	return r, reader, nil
 }
 
+// StatusError adds the message a server writes into the body of a failed
+// request, such as a volume server CRC mismatch, to the status line.
+func StatusError(fileUrl string, r *http.Response) error {
+	var reader io.Reader = r.Body
+	if r.Header.Get("Content-Encoding") == "gzip" {
+		if gz, err := gzip.NewReader(r.Body); err == nil {
+			reader = gz
+		}
+	}
+	body, _ := io.ReadAll(io.LimitReader(reader, 1024))
+	msg := strings.TrimSpace(string(body))
+	var m struct {
+		Error string `json:"error"`
+	}
+	if json.Unmarshal(body, &m) == nil && m.Error != "" {
+		msg = m.Error
+	}
+	if msg == "" {
+		return fmt.Errorf("%s: %s", fileUrl, r.Status)
+	}
+	return fmt.Errorf("%s: %s: %s", fileUrl, r.Status, msg)
+}
+
 func CloseResponse(resp *http.Response) {
 	if resp == nil || resp.Body == nil {
 		return
@@ -782,7 +807,7 @@ func readUrlDirectToBuffer(ctx context.Context, fileUrl, jwt string, buffer []by
 			return 0, false, fmt.Errorf("%s: %s: %w", fileUrl, r.Status, ErrTooManyRequests)
 		}
 		retryable = r.StatusCode >= 499
-		return 0, retryable, fmt.Errorf("%s: %s", fileUrl, r.Status)
+		return 0, retryable, StatusError(fileUrl, r)
 	}
 
 	// Read directly into the buffer without intermediate copying

weed/util/http/http_global_client_util_test.go

@@ -1,6 +1,8 @@
 package http
 
 import (
+	"bytes"
+	"compress/gzip"
 	"context"
 	"net/http"
 	"net/http/httptest"
@@ -92,6 +94,76 @@ func TestReadUrlAsStreamReturnsGzipReaderError(t *testing.T) {
 	}
 }
 
+func TestReadErrorsCarryTheServerMessage(t *testing.T) {
+	InitGlobalHttpClient()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = w.Write([]byte(`{"error":"invalid CRC for needle 2a (got 0a1b2c3d, want 4e5f6071), data on disk corrupted: needle data corrupted"}`))
+	}))
+	defer server.Close()
+
+	want := server.URL + ": 500 Internal Server Error: invalid CRC for needle 2a (got 0a1b2c3d, want 4e5f6071), data on disk corrupted: needle data corrupted"
+
+	if _, err := ReadUrl(context.Background(), server.URL, nil, false, true, 0, 0, make([]byte, 16)); err == nil || err.Error() != want {
+		t.Errorf("ReadUrl: got %v, want %q", err, want)
+	}
+	if _, err := ReadUrlAsStream(context.Background(), server.URL, "", nil, false, true, 0, 0, func(data []byte) {}); err == nil || err.Error() != want {
+		t.Errorf("ReadUrlAsStream: got %v, want %q", err, want)
+	}
+	if _, _, err := ReadUrlAsReaderCloser(server.URL, "", ""); err == nil || err.Error() != want {
+		t.Errorf("ReadUrlAsReaderCloser: got %v, want %q", err, want)
+	}
+	if _, _, err := readUrlDirectToBuffer(context.Background(), server.URL, "", make([]byte, 16)); err == nil || err.Error() != want {
+		t.Errorf("readUrlDirectToBuffer: got %v, want %q", err, want)
+	}
+	if _, _, err := GetAuthenticated(context.Background(), server.URL, ""); err == nil || err.Error() != want {
+		t.Errorf("GetAuthenticated: got %v, want %q", err, want)
+	}
+}
+
+func TestReadErrorsDecodeGzipBody(t *testing.T) {
+	InitGlobalHttpClient()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var buf bytes.Buffer
+		gz := gzip.NewWriter(&buf)
+		_, _ = gz.Write([]byte(`{"error":"invalid CRC for needle 2a"}`))
+		_ = gz.Close()
+		w.Header().Set("Content-Type", "application/json")
+		w.Header().Set("Content-Encoding", "gzip")
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = w.Write(buf.Bytes())
+	}))
+	defer server.Close()
+
+	want := server.URL + ": 500 Internal Server Error: invalid CRC for needle 2a"
+	if _, err := ReadUrl(context.Background(), server.URL, nil, false, true, 0, 0, make([]byte, 16)); err == nil || err.Error() != want {
+		t.Errorf("ReadUrl: got %v, want %q", err, want)
+	}
+	if _, err := ReadUrlAsStream(context.Background(), server.URL, "", nil, false, true, 0, 0, func(data []byte) {}); err == nil || err.Error() != want {
+		t.Errorf("ReadUrlAsStream: got %v, want %q", err, want)
+	}
+	if _, _, err := GetAuthenticated(context.Background(), server.URL, ""); err == nil || err.Error() != want {
+		t.Errorf("GetAuthenticated: got %v, want %q", err, want)
+	}
+}
+
+func TestReadErrorWithoutBodyIsTheStatus(t *testing.T) {
+	InitGlobalHttpClient()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	want := server.URL + ": 500 Internal Server Error"
+	if _, err := ReadUrlAsStream(context.Background(), server.URL, "", nil, false, true, 0, 0, func(data []byte) {}); err == nil || err.Error() != want {
+		t.Errorf("got %v, want %q", err, want)
+	}
+}
+
 func TestDeleteReturnsInvalidRequestErrorBeforeAddingAuth(t *testing.T) {
 	defer func() {
 		if r := recover(); r != nil {