diff --git a/weed/command/master.go b/weed/command/master.go index c3bbf35954..359721eb22 100644 --- a/weed/command/master.go +++ b/weed/command/master.go @@ -82,6 +82,15 @@ type MasterOptions struct { // is cancelled. Used by integration tests and by weed mini; nil for // standalone weed master. shutdownCtx context.Context + // releaseAllocOnFirstReport lifts the post-mint allocation hold as soon as + // one full volume-server inventory has been applied. weed mini sets it in + // single-master mode, with singleReporterAddr pinning the embedded volume + // server as the only admitted reporter: any other volume server is refused, + // so the warmup window after a fresh topology id is dead time. + // Standalone weed master / weed server can face late-arriving veterans and + // must keep the window. + releaseAllocOnFirstReport bool + singleReporterAddr string } func init() { @@ -488,14 +497,16 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp VacuumIntervalSeconds: *m.vacuumIntervalSeconds, VacuumDeleteEmptyAfterSeconds: *m.vacuumDeleteEmptyAfterSeconds, // PulseSeconds: *m.pulseSeconds, - DefaultReplicaPlacement: *m.defaultReplication, - GarbageThreshold: *m.garbageThreshold, - WhiteList: whiteList, - DisableHttp: *m.disableHttp, - MetricsAddress: *m.metricsAddress, - MetricsIntervalSec: *m.metricsIntervalSec, - TelemetryUrl: *m.telemetryUrl, - TelemetryEnabled: *m.telemetryEnabled, - SingleMaster: isSingleMasterMode(*m.peers), + DefaultReplicaPlacement: *m.defaultReplication, + GarbageThreshold: *m.garbageThreshold, + WhiteList: whiteList, + DisableHttp: *m.disableHttp, + MetricsAddress: *m.metricsAddress, + MetricsIntervalSec: *m.metricsIntervalSec, + TelemetryUrl: *m.telemetryUrl, + TelemetryEnabled: *m.telemetryEnabled, + SingleMaster: isSingleMasterMode(*m.peers), + ReleaseAllocOnFirstReport: m.releaseAllocOnFirstReport, + SingleVolumeReporter: m.singleReporterAddr, } } diff --git a/weed/command/mini.go b/weed/command/mini.go index 9fb8cbf68d..8d213bdf70 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -1564,6 +1564,22 @@ func runMini(cmd *Command, args []string) bool { // tear those services down too. On Ctrl+C they rely on their own // OnInterrupt hooks (see grace's LIFO ordering). miniMasterOptions.shutdownCtx = MiniClusterCtx + // In single-master mode mini's embedded volume server is the only reporter + // of prior cluster state; its first heartbeat already carries the full + // inventory, so the post-mint allocation hold ends there instead of + // running the whole warmup window on every fresh start. The master pins + // the embedded address and refuses any other volume server, keeping the + // single-reporter assumption true. Multi-master minis can field + // late-reporting volume servers and keep the regular hold. + // Both fields are re-derived on every run: the package-level options + // survive in-process reruns, so a multi-master mini after a + // single-master one must not inherit the release or the stale pin. + singleMaster := isSingleMasterMode(*miniMasterOptions.peers) + miniMasterOptions.releaseAllocOnFirstReport = singleMaster + miniMasterOptions.singleReporterAddr = "" + if singleMaster { + miniMasterOptions.singleReporterAddr = util.JoinHostPort(*miniIp, *miniOptions.v.port) + } miniOptions.v.shutdownCtx = MiniClusterCtx miniFilerOptions.shutdownCtx = MiniClusterCtx // Mini is a small/dev setup with short-lived RPCs; cap the filer's diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 92a2e35161..2df9779a93 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -231,12 +231,30 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ continue } - if (dn == nil && heartbeat.Ip != "") || reportsPriorClusterState(heartbeat, ms.Topo.GetTopologyId(), ms.Topo.GetMaxVolumeId()) { + if dn == nil { + // An unregistered heartbeat carries no address to admit; under + // the pinned single reporter it can never be the embedded one, + // so it must not move allocation or topology state either. + addr := "" + if heartbeat.Ip != "" { + addr = util.JoinHostPort(heartbeat.Ip, int(heartbeat.Port)) + } + if !ms.Topo.AdmitVolumeReporter(addr) { + err := fmt.Errorf("volume server %s rejected: this master admits a single volume-server reporter", addr) + glog.Warningf("SendHeartbeat: %v", err) + return err + } + } + priorState := reportsPriorClusterState(heartbeat, ms.Topo.GetTopologyId(), ms.Topo.GetMaxVolumeId()) + if (dn == nil && heartbeat.Ip != "") || priorState { ms.Topo.NoteAllocationEvidence() } ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey) ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen)) - if len(heartbeat.Volumes) > 0 { + // A HasNoVolumes report still carries the reporter's LastSeenTopologyId; + // skipping it would let an EC-only veteran's stale id miss the mismatch + // freeze before the hold release below. + if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes { ms.Topo.CheckReportedTopologyId(heartbeat.LastSeenTopologyId) } if dn == nil { @@ -356,6 +374,15 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ } } + // A full volume list is the heartbeat section that settles what a + // reporter holds; once applied, a post-mint allocation hold kept for + // prior cluster state has nothing left to wait for. A report carrying + // such state is itself the evidence the hold exists for, so it keeps + // the window for other veterans still due. + if (len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes) && !priorState { + ms.Topo.NoteInventoryApplied() + } + if len(heartbeat.NewEcShards) > 0 || len(heartbeat.DeletedEcShards) > 0 { stats.MasterReceivedHeartbeatCounter.WithLabelValues("newEcShards").Inc() // update master internal volume layouts diff --git a/weed/server/master_grpc_server_heartbeat_admit_test.go b/weed/server/master_grpc_server_heartbeat_admit_test.go new file mode 100644 index 0000000000..8d91eeecda --- /dev/null +++ b/weed/server/master_grpc_server_heartbeat_admit_test.go @@ -0,0 +1,110 @@ +package weed_server + +import ( + "io" + "testing" + + "github.com/seaweedfs/raft" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" +) + +// admitTestRaftServer answers as leader and swallows Do so a growth freeze +// lands its local flag without needing a real raft log. +type admitTestRaftServer struct { + raft.Server +} + +func (admitTestRaftServer) State() string { return raft.Leader } +func (admitTestRaftServer) Do(raft.Command) (interface{}, error) { return nil, nil } + +type fakeHeartbeatStream struct { + grpc.ServerStream + heartbeats []*master_pb.Heartbeat + sent []*master_pb.HeartbeatResponse +} + +func (s *fakeHeartbeatStream) Recv() (*master_pb.Heartbeat, error) { + if len(s.heartbeats) == 0 { + return nil, io.EOF + } + hb := s.heartbeats[0] + s.heartbeats = s.heartbeats[1:] + return hb, nil +} + +func (s *fakeHeartbeatStream) Send(resp *master_pb.HeartbeatResponse) error { + s.sent = append(s.sent, resp) + return nil +} + +// A server the single-reporter admission refuses must not leave any applied +// state behind: its stale topology id cannot freeze growth and its seen bound +// cannot raise the master's. +func TestSendHeartbeatRejectsForeignReporterBeforeApplyingState(t *testing.T) { + ms := newLeaderMaster() + ms.Topo.RaftServer = admitTestRaftServer{} + ms.Topo.SetGrowthFreezeConfig(true, 0) + ms.Topo.SetTopologyId("cluster-b") + ms.Topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080") + + stream := &fakeHeartbeatStream{heartbeats: []*master_pb.Heartbeat{{ + Ip: "10.0.0.2", + Port: 8080, + MaxVolumeIdSeen: 7, + LastSeenTopologyId: "cluster-a", + Volumes: []*master_pb.VolumeInformationMessage{{Id: 3}}, + }}} + + err := ms.SendHeartbeat(stream) + require.ErrorContains(t, err, "rejected") + assert.False(t, ms.Topo.IsGrowthFrozen()) + assert.Zero(t, uint32(ms.Topo.GetMaxVolumeId())) + assert.Empty(t, stream.sent) +} + +// The pinned reporter is still subject to the mismatch freeze: admission gates +// who may report, not what a legitimate report can do. +func TestSendHeartbeatAdmittedReporterStillFreezesOnMismatch(t *testing.T) { + ms := newLeaderMaster() + ms.Topo.RaftServer = admitTestRaftServer{} + ms.Topo.SetGrowthFreezeConfig(true, 0) + ms.Topo.SetTopologyId("cluster-b") + ms.Topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080") + + stream := &fakeHeartbeatStream{heartbeats: []*master_pb.Heartbeat{{ + Ip: "10.0.0.1", + Port: 8080, + MaxVolumeIdSeen: 7, + LastSeenTopologyId: "cluster-a", + Volumes: []*master_pb.VolumeInformationMessage{{Id: 3}}, + }}} + + _ = ms.SendHeartbeat(stream) + assert.True(t, ms.Topo.IsGrowthFrozen()) +} + +// An unregistered heartbeat with no address can never be the pinned reporter: +// its stale topology id and seen bound must not reach the topology on the way +// to the delta-heartbeat skip. +func TestSendHeartbeatRejectsAddresslessFirstHeartbeat(t *testing.T) { + ms := newLeaderMaster() + ms.Topo.RaftServer = admitTestRaftServer{} + ms.Topo.SetGrowthFreezeConfig(true, 0) + ms.Topo.SetTopologyId("cluster-b") + ms.Topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080") + + stream := &fakeHeartbeatStream{heartbeats: []*master_pb.Heartbeat{{ + MaxVolumeIdSeen: 7, + LastSeenTopologyId: "cluster-a", + Volumes: []*master_pb.VolumeInformationMessage{{Id: 3}}, + }}} + + err := ms.SendHeartbeat(stream) + require.ErrorContains(t, err, "rejected") + assert.False(t, ms.Topo.IsGrowthFrozen()) + assert.Zero(t, uint32(ms.Topo.GetMaxVolumeId())) + assert.Empty(t, stream.sent) +} diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 0fe35ccf2e..576df80fdd 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -70,6 +70,13 @@ type MasterOption struct { TelemetryUrl string TelemetryEnabled bool VolumeGrowthDisabled bool + // ReleaseAllocOnFirstReport lifts the post-mint allocation hold as soon as + // one full volume-server inventory has been applied, for setups whose + // embedded volume server is the only possible reporter (weed mini). + // SingleVolumeReporter pins that reporter's ip:port; any other volume + // server trying to register is rejected. + ReleaseAllocOnFirstReport bool + SingleVolumeReporter string } type MasterServer struct { @@ -182,6 +189,9 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se ms.Topo.SetAllocationDeferral(d) } } + if option.ReleaseAllocOnFirstReport { + ms.Topo.ReleaseAllocationOnFirstReport(option.SingleVolumeReporter) + } ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) error { err := operation.WithVolumeServerClient(false, dn.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) diff --git a/weed/topology/growth_freeze_test.go b/weed/topology/growth_freeze_test.go index f838010bd9..13d849e4e8 100644 --- a/weed/topology/growth_freeze_test.go +++ b/weed/topology/growth_freeze_test.go @@ -191,3 +191,61 @@ func TestAllocationDeferralDuration(t *testing.T) { t.Fatalf("negative deferral should restore the default, got %v", d) } } + +func TestAllocationDeferralReleasedByInventory(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.ReleaseAllocationOnFirstReport() + + topo.DeferAllocation(time.Minute) + topo.NoteInventoryApplied() + if wait := topo.allocationDelay(); wait > 0 { + t.Fatalf("applied inventory should lift the deferral, still waiting %v", wait) + } + + fresh := NewTopology("topo", nil, 128, 3, false) + fresh.ReleaseAllocationOnFirstReport() + fresh.NoteInventoryApplied() + fresh.DeferAllocation(time.Minute) + if wait := fresh.allocationDelay(); wait > 0 { + t.Fatalf("a report applied before the mint should keep the hold unarmed, got %v", wait) + } + + walled := NewTopology("topo", nil, 128, 3, false) + walled.DeferAllocation(time.Minute) + walled.NoteInventoryApplied() + if wait := walled.allocationDelay(); wait <= 0 { + t.Fatal("without opt-in an applied inventory must not lift the deferral") + } +} + +func TestQueuedTopologyReportFreezesBeforeGateOpens(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.SetGrowthFreezeConfig(true, 0) + topo.SetTopologyInitPending(true) + + // A report arriving while the id is still being minted is stashed. + topo.CheckReportedTopologyId("cluster-a") + topo.SetTopologyId("cluster-b") + + topo.SetTopologyInitPending(false) + if topo.topologyInitPending.Load() { + t.Fatal("init gate should be open after pending=false") + } + if !topo.IsGrowthFrozen() { + t.Fatal("a stashed mismatch must freeze growth before the gate opens") + } +} + +func TestSingleReporterAdmission(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + if !topo.AdmitVolumeReporter("10.0.0.1:8080") { + t.Fatal("without opt-in any reporter is admitted") + } + topo.ReleaseAllocationOnFirstReport("10.0.0.1:8080") + if !topo.AdmitVolumeReporter("10.0.0.1:8080") { + t.Fatal("the pinned reporter should be admitted") + } + if topo.AdmitVolumeReporter("10.0.0.2:8080") { + t.Fatal("a foreign reporter should be refused") + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 017f3c8eb1..00c5f118bb 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -101,6 +101,15 @@ type Topology struct { allocationDeferral atomic.Int64 topologyInitPending atomic.Bool + // releaseAllocOnReport lifts a post-mint allocation hold as soon as a full + // volume-server inventory has been applied, for single-reporter setups + // (weed mini) whose embedded volume server is the only possible source of + // prior cluster state. inventoryApplied records that at least one full + // volume list has been synced into this topology. + releaseAllocOnReport atomic.Bool + inventoryApplied atomic.Bool + singleReporter atomic.Value // string + reportedTopologyIdsLock sync.Mutex pendingReportedTopologyIds map[string]struct{} @@ -1477,17 +1486,25 @@ func (t *Topology) CheckReportedTopologyId(reported string) { if reported == "" || !t.freezeOnMismatch.Load() { return } + t.reportedTopologyIdsLock.Lock() + defer t.reportedTopologyIdsLock.Unlock() + t.checkReportedTopologyIdLocked(reported) +} + +// checkReportedTopologyIdLocked runs the mismatch check inside +// reportedTopologyIdsLock, the same critical section the init gate drains +// under: a report received before the gate opens is either stashed or frozen +// before the gate can release. +func (t *Topology) checkReportedTopologyIdLocked(reported string) { own := t.GetTopologyId() if own == "" { // The local topology id is still being minted. Keep the report so a // later heartbeat echo cannot erase the mismatch on the server side // before the freeze check ever ran. - t.reportedTopologyIdsLock.Lock() if t.pendingReportedTopologyIds == nil { t.pendingReportedTopologyIds = make(map[string]struct{}) } t.pendingReportedTopologyIds[reported] = struct{}{} - t.reportedTopologyIdsLock.Unlock() return } if reported == own { @@ -1568,17 +1585,58 @@ func (t *Topology) applyGrowthFreeze(frozen bool) { // SetTopologyInitPending gates allocation while a topology id is being // minted, so growth cannot slip in before the warmup window is installed. +// Stashed reports are applied while the gate is still closed: clearing it +// first would let an allocation through before a queued mismatch freezes +// growth. func (t *Topology) SetTopologyInitPending(pending bool) { - t.topologyInitPending.Store(pending) if pending { + t.topologyInitPending.Store(true) return } + // The stash is drained and the gate released inside one critical section: + // a mismatching report checked here, or still in flight under the same + // lock, lands its freeze before any allocation can observe the open gate. t.reportedTopologyIdsLock.Lock() + defer t.reportedTopologyIdsLock.Unlock() reported := t.pendingReportedTopologyIds t.pendingReportedTopologyIds = nil - t.reportedTopologyIdsLock.Unlock() for id := range reported { - t.CheckReportedTopologyId(id) + t.checkReportedTopologyIdLocked(id) + } + t.topologyInitPending.Store(false) +} + +// ReleaseAllocationOnFirstReport opts the topology into lifting the post-mint +// allocation hold once one full volume-server inventory has been applied, and +// pins the admitted volume-server reporter: the early release is only sound +// when a single embedded server (weed mini) can carry prior cluster state, so +// every other reporter must be refused rather than silently trusted. +func (t *Topology) ReleaseAllocationOnFirstReport(onlyReporter ...string) { + if len(onlyReporter) > 0 { + t.singleReporter.Store(onlyReporter[0]) + } + t.releaseAllocOnReport.Store(true) +} + +// AdmitVolumeReporter enforces the single-reporter restriction the early +// release relies on: a late-arriving foreign server would carry cluster state +// the dropped hold was meant to wait for. +func (t *Topology) AdmitVolumeReporter(addr string) bool { + if !t.releaseAllocOnReport.Load() { + return true + } + expected, _ := t.singleReporter.Load().(string) + return expected == "" || expected == addr +} + +// NoteInventoryApplied marks a heartbeat's full volume list as applied to this +// topology. Under ReleaseAllocationOnFirstReport it also lifts the post-mint +// allocation hold: the bound raise and topology-id freeze checks run while the +// heartbeat is processed, so nothing protected remains pending afterwards. +func (t *Topology) NoteInventoryApplied() { + t.inventoryApplied.Store(true) + if t.releaseAllocOnReport.Load() { + t.allocationNotBefore.Store(0) } } @@ -1586,9 +1644,19 @@ func (t *Topology) SetTopologyInitPending(pending bool) { // mints a fresh topology id: on a rebuilt master that window gives veteran // servers a pulse to raise the bound or trip the freeze. func (t *Topology) DeferAllocation(d time.Duration) { + // Under release-on-report a hold is pointless once inventory has landed: + // the veteran heartbeat that mattered has already been applied. + if t.releaseAllocOnReport.Load() && t.inventoryApplied.Load() { + return + } now := time.Now() t.allocationNotBefore.Store(now.Add(d).UnixNano()) t.allocationEvidenceSince.Store(now.UnixNano()) + // Close the race with a report applied between the check above and the + // store, which would otherwise leave the full window armed. + if t.releaseAllocOnReport.Load() && t.inventoryApplied.Load() { + t.allocationNotBefore.Store(0) + } } // NoteAllocationEvidence records a heartbeat signal that could still carry