diff --git a/weed/topology/node.go b/weed/topology/node.go index 66d44a8e1..6b82dd223 100644 --- a/weed/topology/node.go +++ b/weed/topology/node.go @@ -145,12 +145,16 @@ type Node interface { } type NodeImpl struct { - diskUsages *DiskUsages - id NodeId - parent Node + diskUsages *DiskUsages + id NodeId + parent Node sync.RWMutex // lock children - children map[NodeId]Node - maxVolumeId needle.VolumeId + children map[NodeId]Node + // maxVolumeId uses atomic ops so UpAdjustMaxVolumeId (called from the + // volume server heartbeat path) and GetMaxVolumeId (called from the + // master's assign / warmup checks) can run concurrently without a + // data race on NodeImpl. + maxVolumeId atomic.Uint32 //for rack, data center, topology nodeType string @@ -391,16 +395,23 @@ func (n *NodeImpl) UpAdjustDiskUsageDelta(diskType types.DiskType, diskUsage *Di n.parent.UpAdjustDiskUsageDelta(diskType, diskUsage) } } -func (n *NodeImpl) UpAdjustMaxVolumeId(vid needle.VolumeId) { //can be negative - if n.maxVolumeId < vid { - n.maxVolumeId = vid - if n.parent != nil { - n.parent.UpAdjustMaxVolumeId(vid) +func (n *NodeImpl) UpAdjustMaxVolumeId(vid needle.VolumeId) { + target := uint32(vid) + for { + current := n.maxVolumeId.Load() + if current >= target { + return } + if n.maxVolumeId.CompareAndSwap(current, target) { + break + } + } + if n.parent != nil { + n.parent.UpAdjustMaxVolumeId(vid) } } func (n *NodeImpl) GetMaxVolumeId() needle.VolumeId { - return n.maxVolumeId + return needle.VolumeId(n.maxVolumeId.Load()) } func (n *NodeImpl) LinkChildNode(node Node) { diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 9d133530c..30fbfc1f0 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -71,6 +71,7 @@ type Topology struct { topologyIdLock sync.RWMutex lastLeaderChangeTime time.Time + hadVolumesAtLeaderChange bool lastLeaderChangeTimeLock sync.RWMutex } @@ -121,10 +122,17 @@ func (t *Topology) IsChildLocked() (bool, error) { } // SetLastLeaderChangeTime records the time of the most recent leader transition. +// It also snapshots whether the topology already had known volumes at that +// moment. IsWarmingUp uses the snapshot instead of the live MaxVolumeId so a +// fresh cluster that happens to grow its first volume inside the warmup window +// does not retroactively flip into "warming up" state — there is no prior +// topology to wait for on a bootstrap. func (t *Topology) SetLastLeaderChangeTime(ts time.Time) { + hadVolumes := t.GetMaxVolumeId() > 0 t.lastLeaderChangeTimeLock.Lock() defer t.lastLeaderChangeTimeLock.Unlock() t.lastLeaderChangeTime = ts + t.hadVolumesAtLeaderChange = hadVolumes } // GetLastLeaderChangeTime returns the time of the most recent leader transition. @@ -137,15 +145,21 @@ func (t *Topology) GetLastLeaderChangeTime() time.Time { // IsWarmingUp returns true if the master recently became leader and may not yet // have a complete topology. After a leader change or restart, volume servers need // up to WarmupPulseMultiplier heartbeat intervals to reconnect and report their volumes. -// Returns false on a fresh cluster start (MaxVolumeId == 0) since there are no -// existing volumes to wait for. +// Returns false on a fresh cluster start — i.e. when no volumes existed at the +// time of the leader change — since there is no prior topology state to wait for. +// Checking the *live* MaxVolumeId here would make a bootstrapping cluster flip +// into warming-up the moment its first volume is grown, which manifested as a +// 15-second window of spurious Unavailable errors on AssignVolume for workloads +// that start writing immediately (see #8777). func (t *Topology) IsWarmingUp() bool { - if t.GetMaxVolumeId() == 0 { + t.lastLeaderChangeTimeLock.RLock() + lastChange := t.lastLeaderChangeTime + hadVolumes := t.hadVolumesAtLeaderChange + t.lastLeaderChangeTimeLock.RUnlock() + if !hadVolumes || lastChange.IsZero() { return false } - warmupDuration := time.Duration(t.pulse*WarmupPulseMultiplier) * time.Second - lastChange := t.GetLastLeaderChangeTime() - return !lastChange.IsZero() && time.Since(lastChange) < warmupDuration + return time.Since(lastChange) < t.WarmupDuration() } // WarmupDuration returns the configured warmup duration based on pulse interval.