diff --git a/weed/command/master.go b/weed/command/master.go index 87a3cd68a..61bbea730 100644 --- a/weed/command/master.go +++ b/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, diff --git a/weed/command/master_follower.go b/weed/command/master_follower.go index 60d6e65ee..249f51dcf 100644 --- a/weed/command/master_follower.go +++ b/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("") diff --git a/weed/command/mini.go b/weed/command/mini.go index 146ad5ee5..ec34f6e55 100644 --- a/weed/command/mini.go +++ b/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 diff --git a/weed/command/server.go b/weed/command/server.go index 72b2e7a2d..6a53c4467 100644 --- a/weed/command/server.go +++ b/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 diff --git a/weed/server/common.go b/weed/server/common.go index 8399d927c..f719d6881 100644 --- a/weed/server/common.go +++ b/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 diff --git a/weed/server/common_test.go b/weed/server/common_test.go index 79bcad9b9..8042f78d7 100644 --- a/weed/server/common_test.go +++ b/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()) + } + }) +} diff --git a/weed/server/master_server.go b/weed/server/master_server.go index d921203c3..f31cbb4de 100644 --- a/weed/server/master_server.go +++ b/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 diff --git a/weed/server/master_server_handlers_admin.go b/weed/server/master_server_handlers_admin.go index 03f485a09..54802d93d 100644 --- a/weed/server/master_server_handlers_admin.go +++ b/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) } } }