diff --git a/.github/workflows/master-cold-start-tests.yml b/.github/workflows/master-cold-start-tests.yml new file mode 100644 index 000000000..ba98f59e0 --- /dev/null +++ b/.github/workflows/master-cold-start-tests.yml @@ -0,0 +1,79 @@ +name: "Master Cold Start Tests" + +on: + push: + branches: [ master ] + paths: + - 'weed/server/master_*.go' + - 'weed/topology/**' + - 'weed/operation/**' + - 'test/master_cold_start/**' + - 'test/testutil/**' + - '.github/workflows/master-cold-start-tests.yml' + pull_request: + branches: [ master ] + paths: + - 'weed/server/master_*.go' + - 'weed/topology/**' + - 'weed/operation/**' + - 'test/master_cold_start/**' + - 'test/testutil/**' + - '.github/workflows/master-cold-start-tests.yml' + +concurrency: + group: ${{ github.head_ref || github.ref }}/master-cold-start-tests + cancel-in-progress: true + +permissions: + contents: read + +jobs: + master-cold-start-tests: + name: Master Cold Start Tests + runs-on: ubuntu-22.04 + timeout-minutes: 10 + steps: + - name: Check out code + uses: actions/checkout@v7 + with: + persist-credentials: false + + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version-file: 'go.mod' + + - name: Build weed binary + run: | + cd weed && go install -buildvcs=false + + - name: Run master cold start tests + # test/master_cold_start boots a fresh master plus empty volume + # servers and requires the very first assign (HTTP and gRPC, no + # client retries) to complete a write: the assign that triggers + # volume growth must wait for it instead of failing with + # "volume growth in progress". + run: | + export WEED_BINARY=$(go env GOPATH)/bin/weed + go test -v -timeout=8m ./test/master_cold_start/... + + - name: Collect server logs on failure + if: failure() + run: | + # test/master_cold_start/cluster.go keeps failing-test dirs created + # via os.MkdirTemp("", "seaweedfs_master_cold_start_it_") with each + # process log under /logs/. + mkdir -p /tmp/master-cold-start-logs + find /tmp -maxdepth 1 -type d -name "seaweedfs_master_cold_start_it_*" 2>/dev/null | while read dir; do + echo "Found test directory: $dir" + cp -r "$dir" /tmp/master-cold-start-logs/ 2>/dev/null || true + done + find /tmp/master-cold-start-logs -type f -name "*.log" -print -exec tail -n 100 {} \; 2>/dev/null || echo "No logs found" + + - name: Archive logs + if: failure() + uses: actions/upload-artifact@v7 + with: + name: master-cold-start-test-logs + path: /tmp/master-cold-start-logs/ + retention-days: 7 diff --git a/test/master_cold_start/assign_test.go b/test/master_cold_start/assign_test.go new file mode 100644 index 000000000..6a5686a59 --- /dev/null +++ b/test/master_cold_start/assign_test.go @@ -0,0 +1,119 @@ +package master_cold_start + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "mime/multipart" + "net/http" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" +) + +// The very first write against a fresh cluster must succeed in one shot: the +// assign that triggers volume growth waits for it instead of shedding itself +// with "volume growth in progress". No client retries — a single request per +// assign, like a plain HTTP client. +func TestColdStartFirstWrite(t *testing.T) { + c := StartCluster(t, 3) + + // Each subtest uses its own replication so its volume layout starts cold. + t.Run("http_assign_and_write", func(t *testing.T) { + client := &http.Client{Timeout: 15 * time.Second} + resp, err := client.Get(c.MasterURL() + "/dir/assign?replication=002") + if err != nil { + t.Fatalf("dir/assign: %v", err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("dir/assign returned %d: %s", resp.StatusCode, body) + } + var assign struct { + Fid string `json:"fid"` + Url string `json:"url"` + Error string `json:"error"` + } + if err := json.Unmarshal(body, &assign); err != nil { + t.Fatalf("parse assign response %q: %v", body, err) + } + if assign.Error != "" || assign.Fid == "" { + t.Fatalf("assign failed: %s", body) + } + + content := []byte("cold start write") + if err := uploadAndReadBack(client, assign.Url, assign.Fid, content); err != nil { + t.Fatal(err) + } + }) + + t.Run("grpc_assign", func(t *testing.T) { + grpcDialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + err := pb.WithMasterClient(context.Background(), false, pb.ServerAddress(c.MasterAddress()), grpcDialOption, false, + func(client master_pb.SeaweedClient) error { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + resp, err := client.Assign(ctx, &master_pb.AssignRequest{Count: 1, Replication: "001"}) + if err != nil { + return err + } + if resp.Error != "" || resp.Fid == "" { + return fmt.Errorf("assign failed: %+v", resp) + } + return nil + }) + if err != nil { + t.Fatalf("single grpc assign on cold cluster: %v", err) + } + }) +} + +// uploadAndReadBack writes content to the assigned fid and reads it back. +func uploadAndReadBack(client *http.Client, volumeUrl, fid string, content []byte) error { + var buf bytes.Buffer + writer := multipart.NewWriter(&buf) + part, err := writer.CreateFormFile("file", "cold_start.txt") + if err != nil { + return err + } + part.Write(content) + writer.Close() + + target := fmt.Sprintf("http://%s/%s", volumeUrl, fid) + req, err := http.NewRequest(http.MethodPost, target, &buf) + if err != nil { + return err + } + req.Header.Set("Content-Type", writer.FormDataContentType()) + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("upload to %s: %w", target, err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK { + return fmt.Errorf("upload to %s returned %d: %s", target, resp.StatusCode, body) + } + + resp, err = client.Get(target) + if err != nil { + return fmt.Errorf("read back %s: %w", target, err) + } + got, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("read back %s returned %d", target, resp.StatusCode) + } + if !bytes.Equal(got, content) { + return fmt.Errorf("read back %s: got %q, want %q", target, got, content) + } + return nil +} diff --git a/test/master_cold_start/cluster.go b/test/master_cold_start/cluster.go new file mode 100644 index 000000000..abfb4b414 --- /dev/null +++ b/test/master_cold_start/cluster.go @@ -0,0 +1,297 @@ +// Package master_cold_start boots a fresh master plus empty volume servers +// and exercises the very first write against the cluster. The first assign +// triggers volume growth server-side; it must succeed within a single +// request, without client retries. +package master_cold_start + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/testutil" +) + +const ( + waitTimeout = 30 * time.Second + waitTick = 200 * time.Millisecond +) + +type process struct { + name string + cmd *exec.Cmd + logFile string +} + +// Cluster is one fresh master plus n empty volume servers. +type Cluster struct { + t testing.TB + weedBinary string + baseDir string + + masterPort int + masterGrpcPort int + procs []*process +} + +// StartCluster boots a master and volumeServerCount empty volume servers, +// then waits for all of them to register with the master. +func StartCluster(t testing.TB, volumeServerCount int) *Cluster { + t.Helper() + + weedBinary, err := findOrBuildWeedBinary() + if err != nil { + t.Fatalf("resolve weed binary: %v", err) + } + + baseDir, err := os.MkdirTemp("", "seaweedfs_master_cold_start_it_") + if err != nil { + t.Fatalf("create temp dir: %v", err) + } + logsDir := filepath.Join(baseDir, "logs") + os.MkdirAll(logsDir, 0o755) + + ports, err := testutil.AllocateMiniPorts(1 + volumeServerCount) + if err != nil { + t.Fatalf("allocate ports: %v", err) + } + + c := &Cluster{ + t: t, + weedBinary: weedBinary, + baseDir: baseDir, + masterPort: ports[0], + masterGrpcPort: ports[0] + testutil.GrpcPortOffset, + } + t.Cleanup(func() { + c.StopAll() + if t.Failed() { + c.DumpLogs() + t.Logf("cold-start logs kept at %s", baseDir) + } else { + os.RemoveAll(baseDir) + } + }) + + masterDir := filepath.Join(baseDir, "master") + os.MkdirAll(masterDir, 0o755) + c.startProcess("master", filepath.Join(logsDir, "master.log"), + "master", + "-ip=127.0.0.1", + "-port="+strconv.Itoa(c.masterPort), + "-port.grpc="+strconv.Itoa(c.masterGrpcPort), + "-mdir="+masterDir, + "-volumeSizeLimitMB=32", + "-defaultReplication=000", + ) + if err := c.waitForMaster(waitTimeout); err != nil { + t.Fatalf("master not ready: %v", err) + } + + for i := 0; i < volumeServerCount; i++ { + volDir := filepath.Join(baseDir, fmt.Sprintf("vol%d", i)) + os.MkdirAll(volDir, 0o755) + name := fmt.Sprintf("volume%d", i) + c.startProcess(name, filepath.Join(logsDir, name+".log"), + "volume", + "-ip=127.0.0.1", + "-port="+strconv.Itoa(ports[1+i]), + "-port.grpc="+strconv.Itoa(ports[1+i]+testutil.GrpcPortOffset), + "-dir="+volDir, + "-max=10", + "-mserver=127.0.0.1:"+strconv.Itoa(c.masterPort), + ) + } + if err := c.waitForVolumeServers(volumeServerCount, waitTimeout); err != nil { + t.Fatalf("volume servers not registered: %v", err) + } + return c +} + +func (c *Cluster) startProcess(name, logFile string, args ...string) { + c.t.Helper() + f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + c.t.Fatalf("create log for %s: %v", name, err) + } + defer f.Close() // the child owns its own descriptor after Start + cmd := exec.Command(c.weedBinary, args...) + cmd.Dir = c.baseDir + cmd.Stdout = f + cmd.Stderr = f + if err := cmd.Start(); err != nil { + c.t.Fatalf("start %s: %v", name, err) + } + c.procs = append(c.procs, &process{name: name, cmd: cmd, logFile: logFile}) +} + +// StopAll interrupts every process, volume servers first so the master does +// not log reconnect noise. +func (c *Cluster) StopAll() { + for i := len(c.procs) - 1; i >= 0; i-- { + p := c.procs[i] + if p.cmd.Process == nil { + continue + } + _ = p.cmd.Process.Signal(os.Interrupt) + done := make(chan error, 1) + go func() { done <- p.cmd.Wait() }() + select { + case <-time.After(10 * time.Second): + _ = p.cmd.Process.Kill() + <-done + case <-done: + } + } + c.procs = nil +} + +// MasterURL returns the master HTTP endpoint. +func (c *Cluster) MasterURL() string { + return fmt.Sprintf("http://127.0.0.1:%d", c.masterPort) +} + +// MasterAddress returns "127.0.0.1:port" for the master HTTP port. +func (c *Cluster) MasterAddress() string { + return fmt.Sprintf("127.0.0.1:%d", c.masterPort) +} + +func (c *Cluster) waitForMaster(timeout time.Duration) error { + client := &http.Client{Timeout: 1 * time.Second} + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + resp, err := client.Get(c.MasterURL() + "/cluster/status") + if err == nil { + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + var cs struct { + IsLeader bool `json:"IsLeader"` + } + if json.Unmarshal(body, &cs) == nil && cs.IsLeader { + return nil + } + } + time.Sleep(waitTick) + } + return fmt.Errorf("master did not become leader within %v", timeout) +} + +// waitForVolumeServers polls /dir/status until want data nodes are registered +// with free volume slots. +func (c *Cluster) waitForVolumeServers(want int, timeout time.Duration) error { + client := &http.Client{Timeout: 2 * time.Second} + deadline := time.Now().Add(timeout) + var last string + for time.Now().Before(deadline) { + resp, err := client.Get(c.MasterURL() + "/dir/status") + if err == nil { + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + var status struct { + Topology struct { + Max int64 `json:"Max"` + DataCenters []struct { + Racks []struct { + DataNodes []struct { + Url string `json:"Url"` + } `json:"DataNodes"` + } `json:"Racks"` + } `json:"DataCenters"` + } `json:"Topology"` + } + if json.Unmarshal(body, &status) == nil { + nodes := 0 + for _, dc := range status.Topology.DataCenters { + for _, rack := range dc.Racks { + nodes += len(rack.DataNodes) + } + } + if nodes >= want && status.Topology.Max > 0 { + return nil + } + last = fmt.Sprintf("%d/%d nodes, %d slots", nodes, want, status.Topology.Max) + } + } + time.Sleep(waitTick) + } + return fmt.Errorf("volume servers not registered within %v (%s)", timeout, last) +} + +// DumpLogs prints the tail of every process log. +func (c *Cluster) DumpLogs() { + logsDir := filepath.Join(c.baseDir, "logs") + entries, _ := os.ReadDir(logsDir) + for _, e := range entries { + c.t.Logf("=== %s tail ===\n%s", e.Name(), tailFile(filepath.Join(logsDir, e.Name()))) + } +} + +func tailFile(path string) string { + f, err := os.Open(path) + if err != nil { + return "(no log)" + } + defer f.Close() + scanner := bufio.NewScanner(f) + lines := make([]string, 0, 50) + for scanner.Scan() { + lines = append(lines, scanner.Text()) + if len(lines) > 50 { + lines = lines[1:] + } + } + return strings.Join(lines, "\n") +} + +func findOrBuildWeedBinary() (string, error) { + if fromEnv := os.Getenv("WEED_BINARY"); fromEnv != "" { + if isExecutableFile(fromEnv) { + return fromEnv, nil + } + return "", fmt.Errorf("WEED_BINARY not executable: %s", fromEnv) + } + + repoRoot := "" + if _, file, _, ok := runtime.Caller(0); ok { + repoRoot = filepath.Clean(filepath.Join(filepath.Dir(file), "..", "..")) + } + if repoRoot == "" { + return "", fmt.Errorf("unable to detect repository root") + } + + binDir := filepath.Join(os.TempDir(), "seaweedfs_master_cold_start_it_bin") + os.MkdirAll(binDir, 0o755) + binPath := filepath.Join(binDir, "weed") + if isExecutableFile(binPath) { + return binPath, nil + } + + cmd := exec.Command("go", "build", "-o", binPath, ".") + cmd.Dir = filepath.Join(repoRoot, "weed") + var out bytes.Buffer + cmd.Stdout = &out + cmd.Stderr = &out + if err := cmd.Run(); err != nil { + return "", fmt.Errorf("build weed binary: %w\n%s", err, out.String()) + } + return binPath, nil +} + +func isExecutableFile(path string) bool { + info, err := os.Stat(path) + if err != nil || info.IsDir() { + return false + } + return info.Mode().Perm()&0o111 != 0 +} diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go index 6626bf0bc..d6db78a4b 100644 --- a/weed/server/master_grpc_server_assign.go +++ b/weed/server/master_grpc_server_assign.go @@ -28,7 +28,7 @@ func (ms *MasterServer) StreamAssign(server master_pb.Seaweed_StreamAssignServer glog.Errorf("StreamAssign failed to receive: %v", err) return err } - resp, err := ms.Assign(context.Background(), req) + resp, err := ms.Assign(server.Context(), req) if err != nil { // Return transient errors (warmup, growth-in-progress shed) as in-band // error responses instead of killing the stream, so pooled connections @@ -106,18 +106,19 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest vl.SetLastGrowCount(req.WritableVolumeCount) var ( - lastErr error - maxTimeout = time.Second * 10 - startTime = time.Now() + lastErr error + maxTimeout = time.Second * 10 + startTime = time.Now() + initiatedGrow bool ) for time.Now().Sub(startTime) < maxTimeout { fid, count, dnList, shouldGrow, err := ms.Topo.PickForWrite(req.Count, option, vl, req.ExpectedDataSize) - if shouldGrow && !vl.HasGrowRequest() && !ms.option.VolumeGrowthDisabled { + if shouldGrow && !initiatedGrow && !ms.option.VolumeGrowthDisabled && vl.AddGrowRequestIfAbsent() { + initiatedGrow = true if err != nil && ms.Topo.AvailableSpaceFor(option) <= 0 { err = fmt.Errorf("%s and no free volumes left for %s", err.Error(), option.String()) } - vl.AddGrowRequest() ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{ Option: option, Count: req.WritableVolumeCount, @@ -129,20 +130,29 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest stats.MasterPickForWriteErrorCounter.Inc() lastErr = err if (req.DataCenter != "" || req.Rack != "") && strings.Contains(err.Error(), topology.NoWritableVolumes) { - break + glog.V(0).Infof("assign %v %v: %v", req, option.String(), err) + return nil, err } - // Growth is the remedy and already in flight; don't pin a goroutine - // spinning out the timeout under an assign herd. - if shouldGrow && vl.HasGrowRequest() { + if shouldGrow { if ms.Topo.AvailableSpaceFor(option) <= 0 { break // out of space: surface the real error, not a retryable shed } - // ResourceExhausted, not Unavailable: clients retry it (assign_file_id.go) - // but the gRPC layer doesn't treat it as a dead channel, so the shed - // doesn't tear down the shared master connection mid-herd. - return nil, status.Errorf(codes.ResourceExhausted, "no writable volumes for %s, volume growth in progress", option.String()) + // Only the initiator waits, and only while the growth it triggered + // is still pending: followers shed fast so a herd doesn't pin a + // goroutine each, and an initiator whose growth concluded without + // yielding a writable volume sheds so client retries re-trigger + // growth instead of looping it here. ResourceExhausted, not + // Unavailable: clients retry it (assign_file_id.go) without + // invalidating the shared master connection. + if initiatedGrow != vl.HasGrowRequest() { + return nil, status.Errorf(codes.ResourceExhausted, "no writable volumes for %s, volume growth in progress", option.String()) + } + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(200 * time.Millisecond): } - time.Sleep(200 * time.Millisecond) continue } dn := dnList.Head() @@ -171,6 +181,11 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest Replicas: replicas, }, nil } + // The initiator timed out with its growth still pending: shed retryably + // rather than failing the write that growth is about to satisfy. + if initiatedGrow && vl.HasGrowRequest() && ms.Topo.AvailableSpaceFor(option) > 0 { + return nil, status.Errorf(codes.ResourceExhausted, "no writable volumes for %s, volume growth in progress", option.String()) + } if lastErr != nil { glog.V(0).Infof("assign %v %v: %v", req, option.String(), lastErr) } diff --git a/weed/server/master_grpc_server_assign_test.go b/weed/server/master_grpc_server_assign_test.go index 3d2b45732..7b65002f8 100644 --- a/weed/server/master_grpc_server_assign_test.go +++ b/weed/server/master_grpc_server_assign_test.go @@ -12,7 +12,9 @@ import ( "google.golang.org/grpc/status" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/types" @@ -33,6 +35,7 @@ func newLeaderMaster() *MasterServer { return &MasterServer{ Topo: topo, option: &MasterOption{}, + guard: security.NewGuard(nil, "", 0, "", 0), volumeGrowthRequestChan: make(chan *topology.VolumeGrowRequest, 1<<6), } } @@ -45,7 +48,7 @@ func markGrowthInFlight(t *testing.T, topo *topology.Topology, req *master_pb.As require.NoError(t, err) ttl, err := needle.ReadTTL(req.Ttl) require.NoError(t, err) - topo.GetVolumeLayout(req.Collection, rp, ttl, types.ToDiskType(req.DiskType)).AddGrowRequest() + topo.GetVolumeLayout(req.Collection, rp, ttl, types.ToDiskType(req.DiskType)).AddGrowRequestIfAbsent() } // With free space but no writable volume and growth already in flight, Assign @@ -76,6 +79,99 @@ func TestAssignShedsLoadWhenGrowthInFlight(t *testing.T) { assert.Len(t, ms.volumeGrowthRequestChan, 0) } +// The assign that triggers growth must not shed itself: on a cold cluster there +// is no other request to pick up the volume, so it waits for the growth it +// started and returns the fresh volume once it lands. +func TestAssignInitiatorWaitsForItsOwnGrowth(t *testing.T) { + ms := newLeaderMaster() + dn := ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100}) + + req := &master_pb.AssignRequest{Count: 1, Replication: "000"} + rp, err := super_block.NewReplicaPlacementFromString(req.Replication) + require.NoError(t, err) + + // Simulate growth landing shortly after it is requested by registering a + // writable volume, then draining the request the initiator enqueued. + go func() { + req := <-ms.volumeGrowthRequestChan + v := storage.VolumeInfo{ + Id: needle.VolumeId(1), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: rp, + Ttl: needle.EMPTY_TTL, + } + dn.UpdateVolumes([]storage.VolumeInfo{v}) + ms.Topo.RegisterVolumeLayout(v, dn) + ms.Topo.GetVolumeLayout(req.Option.Collection, rp, needle.EMPTY_TTL, types.ToDiskType("")).DoneGrowRequest() + }() + + start := time.Now() + resp, err := ms.Assign(context.Background(), req) + elapsed := time.Since(start) + + require.NoError(t, err) + require.NotNil(t, resp) + assert.NotEmpty(t, resp.Fid) + // It waited for growth rather than shedding, but well inside the retry budget. + assert.Less(t, elapsed, 5*time.Second) +} + +// An initiator whose growth concludes without yielding a writable volume +// (failed or discarded growth) sheds retryably instead of re-triggering +// growth for the rest of its budget. +func TestAssignInitiatorShedsWhenGrowthConcludesUnfulfilled(t *testing.T) { + ms := newLeaderMaster() + ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100}) + + req := &master_pb.AssignRequest{Count: 1, Replication: "000"} + rp, err := super_block.NewReplicaPlacementFromString(req.Replication) + require.NoError(t, err) + vl := ms.Topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.ToDiskType("")) + + // Growth consumer that fails: clears the flag without registering volumes. + go func() { + <-ms.volumeGrowthRequestChan + vl.DoneGrowRequest() + }() + + start := time.Now() + resp, err := ms.Assign(context.Background(), req) + elapsed := time.Since(start) + + require.Error(t, err) + require.Nil(t, resp) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.ResourceExhausted, st.Code()) + assert.Less(t, elapsed, 2*time.Second) + // Growth was triggered exactly once, not re-enqueued after the failure. + assert.Len(t, ms.volumeGrowthRequestChan, 0) +} + +// A cancelled request stops waiting instead of sleeping out the 10s budget. +func TestAssignAbortsOnCancel(t *testing.T) { + ms := newLeaderMaster() + ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100}) + + // Initiator with its growth never concluding: nobody drains the chan. + req := &master_pb.AssignRequest{Count: 1, Replication: "000"} + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(300 * time.Millisecond) + cancel() + }() + + start := time.Now() + _, err := ms.Assign(ctx, req) + elapsed := time.Since(start) + + require.ErrorIs(t, err, context.Canceled) + assert.Less(t, elapsed, 2*time.Second) +} + // Out of space, Assign fails fast with the real error rather than masking it as // a retryable "growth in progress". func TestAssignFailsFastWhenOutOfSpace(t *testing.T) { diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go index 6d5eb6c28..0a8d30b34 100644 --- a/weed/server/master_server_handlers.go +++ b/weed/server/master_server_handlers.go @@ -158,9 +158,10 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType) var ( - lastErr error - maxTimeout = time.Second * 10 - startTime = time.Now() + lastErr error + maxTimeout = time.Second * 10 + startTime = time.Now() + initiatedGrow bool ) if !ms.Topo.DataCenterExists(option.DataCenter) { @@ -172,12 +173,12 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) for time.Since(startTime) < maxTimeout { fid, count, dnList, shouldGrow, err := ms.Topo.PickForWrite(requestedCount, option, vl, expectedDataSize) - if shouldGrow && !vl.HasGrowRequest() && !ms.option.VolumeGrowthDisabled { + if shouldGrow && !initiatedGrow && !ms.option.VolumeGrowthDisabled && vl.AddGrowRequestIfAbsent() { + initiatedGrow = true glog.V(0).Infof("dirAssign volume growth %v from %v", option.String(), r.RemoteAddr) if err != nil && ms.Topo.AvailableSpaceFor(option) <= 0 { err = fmt.Errorf("%s and no free volumes left for %s", err.Error(), option.String()) } - vl.AddGrowRequest() ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{ Option: option, Count: uint32(writableVolumeCount), @@ -187,18 +188,25 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) if err != nil { stats.MasterPickForWriteErrorCounter.Inc() lastErr = err - // See Assign: shed instead of spinning when growth is already in flight. - if shouldGrow && vl.HasGrowRequest() { + if shouldGrow { if ms.Topo.AvailableSpaceFor(option) <= 0 { break // out of space: surface the real error (406 below) } - w.Header().Set("Retry-After", "1") - writeJsonQuiet(w, r, http.StatusServiceUnavailable, operation.AssignResult{ - Error: fmt.Sprintf("no writable volumes for %s, volume growth in progress", option.String()), - }) - return + // See Assign: only the initiator waits, and only while the + // growth it triggered is still pending. + if initiatedGrow != vl.HasGrowRequest() { + w.Header().Set("Retry-After", "1") + writeJsonQuiet(w, r, http.StatusServiceUnavailable, operation.AssignResult{ + Error: fmt.Sprintf("no writable volumes for %s, volume growth in progress", option.String()), + }) + return + } + } + select { + case <-r.Context().Done(): + return // client gone + case <-time.After(200 * time.Millisecond): } - time.Sleep(200 * time.Millisecond) continue } else { ms.maybeAddJwtAuthorization(w, fid, true) @@ -211,6 +219,14 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) } } + // See Assign: initiator that timed out with growth still pending stays retryable. + if initiatedGrow && vl.HasGrowRequest() && ms.Topo.AvailableSpaceFor(option) > 0 { + w.Header().Set("Retry-After", "1") + writeJsonQuiet(w, r, http.StatusServiceUnavailable, operation.AssignResult{ + Error: fmt.Sprintf("no writable volumes for %s, volume growth in progress", option.String()), + }) + return + } if lastErr != nil { writeJsonQuiet(w, r, http.StatusNotAcceptable, operation.AssignResult{Error: lastErr.Error()}) } else { diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 8f389c6e6..3f47c0110 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -642,8 +642,12 @@ func (vl *VolumeLayout) remainingSize(vid needle.VolumeId) uint64 { func (vl *VolumeLayout) HasGrowRequest() bool { return vl.growRequest.Load() } -func (vl *VolumeLayout) AddGrowRequest() { - vl.growRequest.Store(true) + +// AddGrowRequestIfAbsent atomically claims the pending-growth flag. It returns +// true for the one caller that transitions it from unset to set (the growth +// initiator); concurrent callers get false and are followers of that growth. +func (vl *VolumeLayout) AddGrowRequestIfAbsent() bool { + return vl.growRequest.CompareAndSwap(false, true) } func (vl *VolumeLayout) DoneGrowRequest() { vl.growRequest.Store(false)