diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index ac9e0577ef..38092211b0 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -184,6 +184,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ ms.Topo.Sequence.SetMax(heartbeat.MaxFileKey) ms.Topo.UpAdjustMaxVolumeId(needle.VolumeId(heartbeat.MaxVolumeIdSeen)) + ms.Topo.CheckReportedTopologyId(heartbeat.LastSeenTopologyId) if dn == nil { // Skip delta heartbeat for volume server versions better than 3.28 https://github.com/seaweedfs/seaweedfs/pull/3630 if heartbeat.Ip == "" { diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 716d06779f..53bd2c27f0 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -471,6 +471,13 @@ func (ms *MasterServer) VolumeMarkReadonly(ctx context.Context, req *master_pb.V return resp, nil } +func (ms *MasterServer) ResumeVolumeGrowth(ctx context.Context, req *master_pb.ResumeVolumeGrowthRequest) (*master_pb.ResumeVolumeGrowthResponse, error) { + if !ms.Topo.IsLeader() { + return nil, raft.NotLeaderError + } + return &master_pb.ResumeVolumeGrowthResponse{WasFrozen: ms.Topo.ResumeVolumeGrowth()}, nil +} + func (ms *MasterServer) VolumeGrow(ctx context.Context, req *master_pb.VolumeGrowRequest) (*master_pb.VolumeGrowResponse, error) { if !ms.Topo.IsLeader() { return nil, raft.NotLeaderError diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 22f33176e4..b2e0de3874 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -122,6 +122,8 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se v.SetDefault("master.volume_growth.reservation_timeout", "5m") v.SetDefault("master.volume_growth.disable", false) option.VolumeGrowthDisabled = v.GetBool("master.volume_growth.disable") + v.SetDefault("master.volume_growth.freeze_on_topology_mismatch", true) + v.SetDefault("master.volume_growth.state_loss_recovery_wait", "0s") topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1") topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2") @@ -157,6 +159,9 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se glog.Fatalf("create sequencer failed.") } ms.Topo = topology.NewTopology("topo", seq, uint64(ms.option.VolumeSizeLimitMB)*1024*1024, 5, replicationAsMin) + ms.Topo.SetGrowthFreezeConfig( + v.GetBool("master.volume_growth.freeze_on_topology_mismatch"), + v.GetDuration("master.volume_growth.state_loss_recovery_wait")) ms.vg = topology.NewDefaultVolumeGrowth() glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB") diff --git a/weed/server/raft_common.go b/weed/server/raft_common.go index 9a87a9952b..4e4d7afc7d 100644 --- a/weed/server/raft_common.go +++ b/weed/server/raft_common.go @@ -59,6 +59,7 @@ func EnsureTopologyId(topo *topology.Topology, checkLeaderFn func() bool, persis } else { glog.V(0).Infof("TopologyId generated: %s", topologyId) } + topo.DeferAllocation(topo.WarmupDuration()) break } } diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index 7b39b943be..2b37256f7a 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -101,6 +101,14 @@ var ( Help: "admin lock", }, []string{"client"}) + MasterVolumeGrowthFrozen = prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemMaster, + Name: "volume_growth_frozen", + Help: "1 while volume growth is frozen after detected master state loss", + }) + MasterReceivedHeartbeatCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ Namespace: Namespace, @@ -991,6 +999,7 @@ func init() { Gather.MustRegister(MasterClientConnectCounter) Gather.MustRegister(MasterRaftIsleader) Gather.MustRegister(MasterAdminLock) + Gather.MustRegister(MasterVolumeGrowthFrozen) Gather.MustRegister(MasterReceivedHeartbeatCounter) Gather.MustRegister(MasterLeaderChangeCounter) Gather.MustRegister(MasterReplicaPlacementMismatch) diff --git a/weed/topology/growth_freeze_test.go b/weed/topology/growth_freeze_test.go new file mode 100644 index 0000000000..2a6d250148 --- /dev/null +++ b/weed/topology/growth_freeze_test.go @@ -0,0 +1,85 @@ +package topology + +import ( + "testing" + "time" +) + +func TestGrowthFreezeOnTopologyMismatch(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.SetGrowthFreezeConfig(true, 0) + topo.SetTopologyId("cluster-a") + + topo.CheckReportedTopologyId("cluster-a") + if topo.IsGrowthFrozen() { + t.Fatal("matching topology id should not freeze growth") + } + + topo.CheckReportedTopologyId("") + topo.CheckReportedTopologyId("cluster-b") + if !topo.IsGrowthFrozen() { + t.Fatal("mismatching topology id should freeze growth") + } + + if !topo.ResumeVolumeGrowth() { + t.Fatal("resume should report the frozen state") + } + if topo.IsGrowthFrozen() { + t.Fatal("growth should be unfrozen after resume") + } + if topo.ResumeVolumeGrowth() { + t.Fatal("second resume should report no frozen state") + } +} + +func TestGrowthFreezeDisabled(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.SetGrowthFreezeConfig(false, 0) + topo.SetTopologyId("cluster-a") + + topo.CheckReportedTopologyId("cluster-b") + if topo.IsGrowthFrozen() { + t.Fatal("freeze_on_topology_mismatch=false should not freeze") + } +} + +func TestGrowthFreezeAutoRecovery(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.SetGrowthFreezeConfig(true, 50*time.Millisecond) + topo.SetTopologyId("cluster-a") + + topo.CheckReportedTopologyId("cluster-b") + if !topo.IsGrowthFrozen() { + t.Fatal("mismatching topology id should freeze growth") + } + time.Sleep(60 * time.Millisecond) + if topo.IsGrowthFrozen() { + t.Fatal("growth should auto-resume past the recovery wait") + } +} + +func TestCheckReportedTopologyIdBeforeTopologyIdKnown(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + topo.SetGrowthFreezeConfig(true, 0) + + topo.CheckReportedTopologyId("cluster-b") + if topo.IsGrowthFrozen() { + t.Fatal("empty local topology id should not trigger the freeze") + } +} + +func TestAllocationDelay(t *testing.T) { + topo := NewTopology("topo", nil, 128, 3, false) + + if wait := topo.allocationDelay(); wait > 0 { + t.Fatalf("unexpected deferral: %v", wait) + } + topo.DeferAllocation(50 * time.Millisecond) + if wait := topo.allocationDelay(); wait <= 0 { + t.Fatal("expected a positive deferral after DeferAllocation") + } + topo.DeferAllocation(-time.Second) + if wait := topo.allocationDelay(); wait > 0 { + t.Fatalf("expired deferral should not delay: %v", wait) + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index cd4d92ab37..e53b4dea9f 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -88,6 +88,12 @@ type Topology struct { maxVolumeIdFileLock sync.Mutex lastPersistedMaxVolumeId atomic.Uint32 lastPersistedTopologyId atomic.Value + + growthFrozen atomic.Bool + growthFrozenAt atomic.Int64 + freezeOnMismatch atomic.Bool + stateLossRecoveryWait atomic.Int64 + allocationNotBefore atomic.Int64 } func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology { @@ -394,6 +400,13 @@ func (t *Topology) NextVolumeId() (needle.VolumeId, error) { return 0, fmt.Errorf("as leader can not read yet") } + if t.IsGrowthFrozen() { + return 0, fmt.Errorf("volume growth is frozen after master state loss; verify the cluster inventory, then run: volume.growth.resume") + } + if wait := t.allocationDelay(); wait > 0 { + glog.V(0).Infof("deferring volume id allocation for %v", wait) + time.Sleep(wait) + } vid := t.GetMaxVolumeId() next := vid.Next() @@ -950,6 +963,66 @@ func (t *Topology) SetTopologyId(topologyId string) { } } +func (t *Topology) SetGrowthFreezeConfig(freezeOnMismatch bool, recoveryWait time.Duration) { + t.freezeOnMismatch.Store(freezeOnMismatch) + t.stateLossRecoveryWait.Store(int64(recoveryWait)) +} + +// CheckReportedTopologyId freezes growth when a volume server reports a +// topology id different from this cluster's: the topology id survives raft +// state cleanup, so a mismatch means the master's state was rebuilt (or a +// foreign disk was cloned in). +func (t *Topology) CheckReportedTopologyId(reported string) { + if reported == "" || !t.freezeOnMismatch.Load() { + return + } + own := t.GetTopologyId() + if own == "" || reported == own { + return + } + if t.growthFrozen.CompareAndSwap(false, true) { + t.growthFrozenAt.Store(time.Now().UnixNano()) + stats.MasterVolumeGrowthFrozen.Set(1) + glog.Errorf("volume growth frozen: a volume server reports topology id %s but this cluster is %s; the master state was rebuilt. Verify the cluster inventory, then run: volume.growth.resume", reported, own) + } +} + +func (t *Topology) IsGrowthFrozen() bool { + if !t.growthFrozen.Load() { + return false + } + if wait := t.stateLossRecoveryWait.Load(); wait > 0 { + if time.Duration(time.Now().UnixNano()-t.growthFrozenAt.Load()) > time.Duration(wait) { + if t.growthFrozen.CompareAndSwap(true, false) { + stats.MasterVolumeGrowthFrozen.Set(0) + glog.V(0).Infof("volume growth unfrozen after %v", time.Duration(wait)) + } + return false + } + } + return true +} + +func (t *Topology) ResumeVolumeGrowth() bool { + if t.growthFrozen.CompareAndSwap(true, false) { + stats.MasterVolumeGrowthFrozen.Set(0) + glog.V(0).Infof("volume growth resumed by operator") + return true + } + return false +} + +// DeferAllocation holds off new id allocation for d. Called when the leader +// 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) { + t.allocationNotBefore.Store(time.Now().Add(d).UnixNano()) +} + +func (t *Topology) allocationDelay() time.Duration { + return time.Until(time.Unix(0, t.allocationNotBefore.Load())) +} + // SetMaxVolumeIdFile mirrors the allocation bound to a file outside raft, so // a master whose raft state is gone still restores a safe boundary. An // existing file seeds both the bound and the topology id.