master: honour -volume.fileSizeLimitMB on the master's /submit (#11176)

* fix(master): honour -volume.fileSizeLimitMB on the master's /submit - #6748

`weed server -volume.fileSizeLimitMB=2048` still refused anything over
256MB, and the reason is not the one the report assumes: the option does
reach the volume server. The master does not use it. Uploads through the
master's /submit are buffered by submitForClientHandler, which passed a
hardcoded 256MB to needle.ParseUpload, so the master rejected what the
volume server it started would have accepted.

The limit is now passed in. `weed master` gains its own -fileSizeLimitMB
with the same 256 default, so a standalone master behaves exactly as
before, and `weed server` and `weed mini` hand it the value their volume
server already got.

* master.follower: take the same upload limit, and say which flag to match

Review found the follower left behind. It serves /submit like the leader
and buffers uploads under the same limit, but kept the fixed 256MB, so a
cluster raised above that would accept an upload through the leader and
refuse the identical one through a follower.

Two smaller points from the same review: the master's flag description
named only the standalone volume server's spelling, and now names the
weed server and weed mini form too; and the under-limit test asserted on
the error message alone, so it would have passed had the limit rejected
that payload with different wording. It now requires the request to get
past parsing.

weed/command/master.go

@@ -54,6 +54,7 @@ type MasterOptions struct {
 	peers                      *string
 	mastersDeprecated          *string // deprecated, for backward compatibility in master.follower
 	volumeSizeLimitMB          *uint
+	fileSizeLimitMB            *int
 	volumePreallocate          *bool
 	maxParallelVacuumPerServer *int
 	// pulseSeconds       *int
@@ -89,6 +90,7 @@ func init() {
 	m.metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
 	m.peers = cmdMaster.Flag.String("peers", "", "all master nodes in comma separated ip:port list, example: 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095; use 'none' for single-master mode")
 	m.volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", util.DefaultVolumeSizeLimitMB, "Master stops directing writes to oversized volumes.")
+	m.fileSizeLimitMB = cmdMaster.Flag.Int("fileSizeLimitMB", 256, "limit the file size accepted by /submit, should match the volume servers' -fileSizeLimitMB (-volume.fileSizeLimitMB under weed server or weed mini, which set this for you)")
 	m.volumePreallocate = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.")
 	m.maxParallelVacuumPerServer = cmdMaster.Flag.Int("maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel per volume server")
 	// m.pulseSeconds = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
@@ -468,6 +470,7 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp
 		Master:                     masterAddress,
 		MetaFolder:                 *m.metaFolder,
 		VolumeSizeLimitMB:          uint32(*m.volumeSizeLimitMB),
+		FileSizeLimitMB:            *m.fileSizeLimitMB,
 		VolumePreallocate:          *m.volumePreallocate,
 		MaxParallelVacuumPerServer: *m.maxParallelVacuumPerServer,
 		// PulseSeconds:            *m.pulseSeconds,

weed/command/master_follower.go

@@ -29,6 +29,11 @@ func init() {
 	mf.ipBind = cmdMasterFollower.Flag.String("ip.bind", "", "ip address to bind to. Default to localhost.")
 	mf.peers = cmdMasterFollower.Flag.String("master", "localhost:9333", "all master nodes in comma separated ip:port list, example: 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095")
 	mf.mastersDeprecated = cmdMasterFollower.Flag.String("masters", "", "all master nodes in comma separated ip:port list (deprecated, use -master instead)")
+	// A follower serves /submit like the leader does, so it buffers uploads under
+	// the same limit and has to be told the same value. Left fixed at 256, a
+	// cluster raised above that would accept an upload through the leader and
+	// refuse the identical one through a follower.
+	mf.fileSizeLimitMB = cmdMasterFollower.Flag.Int("fileSizeLimitMB", 256, "limit the file size accepted by /submit, should match the leader's -fileSizeLimitMB")
 
 	mf.ip = aws.String(util.DetectedHostAddress())
 	mf.metaFolder = aws.String("")

weed/command/mini.go

@@ -1280,6 +1280,9 @@ func runMini(cmd *Command, args []string) bool {
 	miniOptions.v.rack = miniRack
 
 	miniMasterOptions.whiteList = miniWhiteListOption
+	// The master's /submit buffers the upload before handing it to the volume
+	// server started here, so it must not reject what that volume server accepts.
+	miniMasterOptions.fileSizeLimitMB = miniOptions.v.fileSizeLimitMB
 
 	miniFilerOptions.dataCenter = miniDataCenter
 	miniFilerOptions.rack = miniRack

weed/command/server.go

@@ -323,6 +323,9 @@ func runServer(cmd *Command, args []string) bool {
 	// masterOptions.pulseSeconds = pulseSeconds
 
 	masterOptions.whiteList = serverWhiteListOption
+	// The master's /submit buffers the upload before handing it to the volume
+	// server started here, so it must not reject what that volume server accepts.
+	masterOptions.fileSizeLimitMB = serverOptions.v.fileSizeLimitMB
 
 	filerOptions.dataCenter = serverDataCenter
 	filerOptions.rack = serverRack

weed/server/common.go

@@ -137,7 +137,7 @@ func debug(params ...interface{}) {
 	glog.V(4).Infoln(params...)
 }
 
-func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption) {
+func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, fileSizeLimitBytes int64) {
 	ctx := r.Context()
 	m := make(map[string]interface{})
 	if r.Method != http.MethodPost {
@@ -148,7 +148,7 @@ func submitForClientHandler(w http.ResponseWriter, r *http.Request, masterFn ope
 	debug("parsing upload file...")
 	bytesBuffer := bufPool.Get().(*bytes.Buffer)
 	defer bufPool.Put(bytesBuffer)
-	pu, pe := needle.ParseUpload(r, 256*1024*1024, bytesBuffer)
+	pu, pe := needle.ParseUpload(r, fileSizeLimitBytes, bytesBuffer)
 	if pe != nil {
 		writeJsonError(w, r, http.StatusBadRequest, pe)
 		return

weed/server/common_test.go

@@ -2,6 +2,7 @@ package weed_server
 
 import (
 	"bytes"
+	"context"
 	"io"
 	"mime/multipart"
 	"net/http"
@@ -9,7 +10,11 @@ import (
 	"strings"
 	"testing"
 
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 )
 
 func TestParseURL(t *testing.T) {
@@ -257,3 +262,61 @@ func TestProcessRangeRequestRanges(t *testing.T) {
 		}
 	}
 }
+
+// /submit must reject at the limit its master was given, not at a hardcoded
+// 256MB: weed server hands -volume.fileSizeLimitMB down to the master, and an
+// upload the volume server would store must not be turned away here (#6748).
+func TestSubmitForClientHandlerFileSizeLimit(t *testing.T) {
+	const fileSizeLimitBytes = int64(1 << 20)
+
+	submit := func(t *testing.T, dataSize int) *httptest.ResponseRecorder {
+		t.Helper()
+		var form bytes.Buffer
+		mw := multipart.NewWriter(&form)
+		part, err := mw.CreateFormFile("file", "test.bin")
+		if err != nil {
+			t.Fatalf("create form file: %v", err)
+		}
+		if _, err := part.Write(make([]byte, dataSize)); err != nil {
+			t.Fatalf("write form file: %v", err)
+		}
+		if err := mw.Close(); err != nil {
+			t.Fatalf("close multipart writer: %v", err)
+		}
+
+		// Assigning a file id is not under test, and there is no master to ask.
+		// A cancelled context fails that step at once for an accepted upload.
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		r := httptest.NewRequestWithContext(ctx, http.MethodPost, "/submit", bytes.NewReader(form.Bytes()))
+		r.Header.Set("Content-Type", mw.FormDataContentType())
+		w := httptest.NewRecorder()
+		masterFn := func(ctx context.Context) pb.ServerAddress { return pb.ServerAddress("localhost:9333") }
+		submitForClientHandler(w, r, masterFn, grpc.WithTransportCredentials(insecure.NewCredentials()), fileSizeLimitBytes)
+		return w
+	}
+
+	t.Run("over the limit", func(t *testing.T) {
+		w := submit(t, int(fileSizeLimitBytes)+1)
+		if w.Code != http.StatusBadRequest {
+			t.Fatalf("status: got %d want %d, body %q", w.Code, http.StatusBadRequest, w.Body.String())
+		}
+		if !strings.Contains(w.Body.String(), "over the limited") {
+			t.Errorf("body: got %q, want the file size limit error", w.Body.String())
+		}
+	})
+
+	t.Run("under the limit", func(t *testing.T) {
+		w := submit(t, int(fileSizeLimitBytes)-1)
+		if strings.Contains(w.Body.String(), "over the limited") {
+			t.Errorf("body: got %q, want no file size limit error", w.Body.String())
+		}
+		// Asserting on the message alone would still pass if the limit rejected
+		// this payload with different wording. Parser failures answer 400, and the
+		// cancelled assignment this request runs into answers 500, so a 400 here
+		// means the upload never got past parsing.
+		if w.Code == http.StatusBadRequest {
+			t.Errorf("status: got 400, want the request to reach assignment, body %q", w.Body.String())
+		}
+	})
+}

weed/server/master_server.go

@@ -48,6 +48,7 @@ type MasterOption struct {
 	Master                     pb.ServerAddress
 	MetaFolder                 string
 	VolumeSizeLimitMB          uint32
+	FileSizeLimitMB            int
 	VolumePreallocate          bool
 	MaxParallelVacuumPerServer int
 	// PulseSeconds            int

weed/server/master_server_handlers_admin.go

@@ -125,14 +125,15 @@ func (ms *MasterServer) redirectHandler(w http.ResponseWriter, r *http.Request)
 }
 
 func (ms *MasterServer) submitFromMasterServerHandler(w http.ResponseWriter, r *http.Request) {
+	fileSizeLimitBytes := int64(ms.option.FileSizeLimitMB) * 1024 * 1024
 	if ms.Topo.IsLeader() {
-		submitForClientHandler(w, r, func(ctx context.Context) pb.ServerAddress { return ms.option.Master }, ms.grpcDialOption)
+		submitForClientHandler(w, r, func(ctx context.Context) pb.ServerAddress { return ms.option.Master }, ms.grpcDialOption, fileSizeLimitBytes)
 	} else {
 		masterUrl, err := ms.Topo.Leader()
 		if err != nil {
 			writeJsonError(w, r, http.StatusInternalServerError, err)
 		} else {
-			submitForClientHandler(w, r, func(ctx context.Context) pb.ServerAddress { return masterUrl }, ms.grpcDialOption)
+			submitForClientHandler(w, r, func(ctx context.Context) pb.ServerAddress { return masterUrl }, ms.grpcDialOption, fileSizeLimitBytes)
 		}
 	}
 }