diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index ac98b3359..f2d7defc2 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -211,20 +211,11 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ glog.V(4).Infof("master received heartbeat %s", heartbeat.String()) stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc() - if heartbeat.State != nil { + // Every heartbeat may carry the state, so a master that just took over + // learns about a server already in maintenance from the first one. + if heartbeat.State != nil && ms.Topo.SetDataNodeMaintenanceMode(dn, heartbeat.State.GetMaintenance()) { stats.MasterReceivedHeartbeatCounter.WithLabelValues("stateUpdates").Inc() - - updated := false - dn.Lock() - if dn.MaintenanceMode != heartbeat.State.GetMaintenance() { - updated = true - dn.MaintenanceMode = heartbeat.State.GetMaintenance() - } - dn.Unlock() - - if updated { - glog.V(1).Infof("master sees state update from %s: %v", dn.Url(), heartbeat.State) - } + glog.V(1).Infof("master sees state update from %s: %v", dn.Url(), heartbeat.State) } message := &master_pb.VolumeLocation{ diff --git a/weed/storage/store.go b/weed/storage/store.go index cb46a7d2a..961fd392f 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -643,6 +643,14 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { hasNoVolumes = false } + // The state rides along on every heartbeat, not only when it changes: the + // first heartbeat to a master that just took the lead is how that master + // learns this server is in maintenance mode. + var state *volume_server_pb.VolumeServerState + if s.State != nil { + state = s.State.Proto() + } + return &master_pb.Heartbeat{ Ip: s.Ip, Port: uint32(s.Port), @@ -664,6 +672,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { HasNoEcShards: len(ecVolumeMessages) == 0, LocationUuids: uuidList, DiskTags: diskTags, + State: state, } } diff --git a/weed/storage/store_heartbeat_digest_test.go b/weed/storage/store_heartbeat_digest_test.go index f98a5f041..f34c2f99a 100644 --- a/weed/storage/store_heartbeat_digest_test.go +++ b/weed/storage/store_heartbeat_digest_test.go @@ -3,6 +3,7 @@ package storage import ( "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" ) @@ -66,6 +67,28 @@ func TestCollectHeartbeatDigestsAnEmptyStore(t *testing.T) { } } +// The maintenance flag has to reach a master that never saw the change: a +// leader elected while a server sits in maintenance only hears from it through +// the regular heartbeats, so each one carries the state. +func TestCollectHeartbeatCarriesState(t *testing.T) { + store := newTestStore(t, 1) + + heartbeat := store.CollectHeartbeat() + if heartbeat.State == nil { + t.Fatal("heartbeat carried no state") + } + if heartbeat.State.GetMaintenance() { + t.Fatal("a fresh store must not report maintenance mode") + } + + if err := store.State.Update(&volume_server_pb.VolumeServerState{Maintenance: true}); err != nil { + t.Fatal(err) + } + if !store.CollectHeartbeat().GetState().GetMaintenance() { + t.Error("heartbeat did not report maintenance mode after it was switched on") + } +} + func TestCollectHeartbeatDigestFollowsVolumeChanges(t *testing.T) { store := newTestStore(t, 1) mountTestVolume(t, store.Locations[0], 1, "") diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index e9f698f56..dbe22a583 100644 --- a/weed/topology/data_node.go +++ b/weed/topology/data_node.go @@ -23,7 +23,11 @@ type DataNode struct { Counter int // in race condition, the previous dataNode was not dead IsTerminating bool - MaintenanceMode bool + // maintenanceMode mirrors the volume server's own flag, reported over the + // heartbeat. A server in maintenance is being drained: the master places + // no new volumes on it and assigns no writes to the volumes it holds. Read + // on the assign and volume-growth paths without the node lock. + maintenanceMode atomic.Bool // lookupDigest covers the volumes reachable through this node in the volume // layouts, for comparison against what its disks actually hold. lookupDigest atomic.Uint64 @@ -58,12 +62,36 @@ func (dn *DataNode) String() string { return fmt.Sprintf("Node:%s, Ip:%s, Port:%d, PublicUrl:%s", dn.NodeImpl.String(), dn.Ip, dn.Port, dn.PublicUrl) } +// InMaintenanceMode reports whether the volume server asked to be left alone +// for writes; see Topology.SetDataNodeMaintenanceMode for what that changes. +func (dn *DataNode) InMaintenanceMode() bool { + return dn.maintenanceMode.Load() +} + +// SetMaintenanceMode records the flag and reports whether it changed. Prefer +// Topology.SetDataNodeMaintenanceMode, which also updates the writable lists. +func (dn *DataNode) SetMaintenanceMode(on bool) (changed bool) { + return dn.maintenanceMode.Swap(on) != on +} + func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChangedRO, tierTransition bool) { dn.Lock() defer dn.Unlock() return dn.doAddOrUpdateVolume(v) } +// SetVolumeReadOnly records the read-only flag a volume server reported for +// one of its volumes, ahead of the heartbeat that will repeat it. +func (dn *DataNode) SetVolumeReadOnly(vid needle.VolumeId, readOnly bool) { + dn.Lock() + defer dn.Unlock() + for _, c := range dn.children { + if c.(*Disk).SetVolumeReadOnly(vid, readOnly) { + return + } + } +} + func (dn *DataNode) getOrCreateDisk(diskType string) *Disk { c, found := dn.children[NodeId(diskType)] if !found { diff --git a/weed/topology/disk.go b/weed/topology/disk.go index a34697191..f72beb72e 100644 --- a/weed/topology/disk.go +++ b/weed/topology/disk.go @@ -224,6 +224,28 @@ func (d *Disk) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged, tie return d.doAddOrUpdateVolume(v, false) } +// SetVolumeReadOnly records the read-only flag the server reported for vid, +// keeping the report digest and the active volume count in step, and reports +// whether the disk holds vid. It is not a volume report: a provisional volume +// stays protected from a stale report until one names it. +func (d *Disk) SetVolumeReadOnly(vid needle.VolumeId, readOnly bool) (found bool) { + d.Lock() + defer d.Unlock() + v, found := d.volumes[vid] + if !found || v.ReadOnly == readOnly { + return found + } + d.volumeDigest ^= v.ReportHash() + v.ReadOnly = readOnly + d.volumeDigest ^= v.ReportHash() + delta := &DiskUsageCounts{activeVolumeCount: 1} + if readOnly { + delta.activeVolumeCount = -1 + } + d.UpAdjustDiskUsageDelta(types.ToDiskType(v.DiskType), delta) + return true +} + // doAddOrUpdateVolume returns three signals about how v was installed against // any existing record: // diff --git a/weed/topology/disk_provisional_disk_id_test.go b/weed/topology/disk_provisional_disk_id_test.go index b06fd09fd..e81b0b040 100644 --- a/weed/topology/disk_provisional_disk_id_test.go +++ b/weed/topology/disk_provisional_disk_id_test.go @@ -1,6 +1,7 @@ package topology import ( + "sync/atomic" "testing" "github.com/seaweedfs/seaweedfs/weed/storage" @@ -31,3 +32,37 @@ func TestProvisionalUpdateKeepsReportedDiskId(t *testing.T) { t.Fatalf("DiskId = %d, want 1 (a real report must override)", v.DiskId) } } + +// A read-only mark from the server is not a volume report: it must not end the +// grace period that keeps a just-grown volume safe from a full report collected +// before the grow, while still keeping the digest and active count in step. +func TestSetVolumeReadOnlyKeepsProvisionalProtection(t *testing.T) { + dn := NewDataNode("dn1") + vi := storage.VolumeInfo{Id: 1, DiskType: types.HardDriveType.String()} + dn.AddProvisionalVolume(vi) + + dn.SetVolumeReadOnly(1, true) + + readOnly := vi + readOnly.ReadOnly = true + if got, want := dn.VolumeDigest(), readOnly.ReportHash(); got != want { + t.Errorf("digest = %x, want %x, the hash of the volume as the server will report it", got, want) + } + if got := atomic.LoadInt64(&dn.GetDiskUsages().getOrCreateDisk(types.HardDriveType).activeVolumeCount); got != 0 { + t.Errorf("activeVolumeCount = %d, want 0 after the read-only mark", got) + } + + // the stale report, collected before the grow, does not name the volume + if _, deleted, _ := dn.UpdateVolumes(nil); len(deleted) != 0 { + t.Fatalf("a report that raced the grow removed the volume: %v", deleted) + } + if v, err := dn.GetVolumesById(1); err != nil || !v.ReadOnly { + t.Fatalf("GetVolumesById = %+v, %v; want the volume kept, read-only", v, err) + } + + // the server's own report confirms it, and from then on absence counts + dn.UpdateVolumes([]storage.VolumeInfo{readOnly}) + if _, deleted, _ := dn.UpdateVolumes(nil); len(deleted) != 1 { + t.Fatalf("a report after confirmation should remove the volume, got %v", deleted) + } +} diff --git a/weed/topology/maintenance_mode_test.go b/weed/topology/maintenance_mode_test.go new file mode 100644 index 000000000..27c3f580a --- /dev/null +++ b/weed/topology/maintenance_mode_test.go @@ -0,0 +1,331 @@ +package topology + +import ( + "slices" + "testing" + + "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" +) + +// addMaintenanceTestNode links a data node with maxVol free slots under rack. +func addMaintenanceTestNode(rack *Rack, id string, maxVol int64) *DataNode { + dn := NewDataNode(id) + dn.Ip = id + rack.LinkChildNode(dn) + dn.getOrCreateDisk("").UpAdjustDiskUsageDelta("", &DiskUsageCounts{maxVolumeCount: maxVol}) + return dn +} + +func newMaintenanceTestTopology() (*Topology, *DataCenter) { + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dc := NewDataCenter("dc1") + topo.LinkChildNode(dc) + return topo, dc +} + +func TestMaintenanceModeReportsNoFreeSlots(t *testing.T) { + rack := NewRack("rack1") + dn := addMaintenanceTestNode(rack, "server1", 10) + option := &VolumeGrowOption{DiskType: types.HardDriveType} + + if got := dn.AvailableSpaceFor(option); got != 10 { + t.Fatalf("AvailableSpaceFor = %d, want 10", got) + } + if !dn.SetMaintenanceMode(true) { + t.Fatal("first switch should report a change") + } + if dn.SetMaintenanceMode(true) { + t.Fatal("repeating the same value should not report a change") + } + if !dn.InMaintenanceMode() { + t.Fatal("InMaintenanceMode = false after switching on") + } + if got := dn.AvailableSpaceFor(option); got != 0 { + t.Errorf("AvailableSpaceFor in maintenance = %d, want 0: no new volumes go to a server in maintenance", got) + } + if got := dn.AvailableSpaceForReservation(option); got != 0 { + t.Errorf("AvailableSpaceForReservation in maintenance = %d, want 0", got) + } + if _, ok := dn.TryReserveCapacity(types.HardDriveType, 1); ok { + t.Error("reserved capacity on a server in maintenance") + } + if got := rack.AvailableSpaceFor(option); got != 10 { + t.Errorf("rack AvailableSpaceFor = %d, want 10: the node's capacity still rolls up to its parents", got) + } + + if !dn.SetMaintenanceMode(false) { + t.Fatal("switching off should report a change") + } + if got := dn.AvailableSpaceFor(option); got != 10 { + t.Errorf("AvailableSpaceFor after maintenance = %d, want 10", got) + } +} + +// A server in maintenance mode is the one being evacuated, so volume growth in +// its rack must land every replica on its siblings even when it has the most +// free slots by far. +func TestVolumeGrowthSkipsMaintenanceNodeInSameRack(t *testing.T) { + topo, dc := newMaintenanceTestTopology() + rack := NewRack("rack1") + dc.LinkChildNode(rack) + busy := addMaintenanceTestNode(rack, "busy", 1000) + busy.SetMaintenanceMode(true) + addMaintenanceTestNode(rack, "small1", 2) + addMaintenanceTestNode(rack, "small2", 2) + + vg := NewDefaultVolumeGrowth() + rp, _ := super_block.NewReplicaPlacementFromString("001") + option := &VolumeGrowOption{ReplicaPlacement: rp, DiskType: types.HardDriveType} + for i := 0; i < 50; i++ { + servers, _, err := vg.findEmptySlotsForOneVolume(topo, option, false) + if err != nil { + t.Fatalf("iteration %d: %v", i, err) + } + if len(servers) != 2 { + t.Fatalf("iteration %d: got %d servers, want 2", i, len(servers)) + } + for _, s := range servers { + if s.Id() == busy.Id() { + t.Fatalf("iteration %d: picked the server in maintenance mode", i) + } + } + } + + // Pinning the request to the maintenance node must fail rather than + // create a volume there. + rp0, _ := super_block.NewReplicaPlacementFromString("000") + pinned := &VolumeGrowOption{ReplicaPlacement: rp0, DiskType: types.HardDriveType, DataNode: "busy"} + if _, _, err := vg.findEmptySlotsForOneVolume(topo, pinned, false); err == nil { + t.Fatal("a volume pinned to a server in maintenance mode must not be created") + } +} + +// The replica on another rack is chosen by walking a random offset through the +// rack's rolled-up free slots, which still count a node in maintenance. The walk +// has to land on an eligible sibling every time, not fall off the end. +func TestVolumeGrowthSkipsMaintenanceNodeInOtherRack(t *testing.T) { + topo, dc := newMaintenanceTestTopology() + rack1 := NewRack("rack1") + dc.LinkChildNode(rack1) + addMaintenanceTestNode(rack1, "r1n1", 100) + rack2 := NewRack("rack2") + dc.LinkChildNode(rack2) + busy := addMaintenanceTestNode(rack2, "r2busy", 1000) + busy.SetMaintenanceMode(true) + ok := addMaintenanceTestNode(rack2, "r2ok", 1) + + vg := NewDefaultVolumeGrowth() + rp, _ := super_block.NewReplicaPlacementFromString("010") + option := &VolumeGrowOption{ReplicaPlacement: rp, DiskType: types.HardDriveType} + for i := 0; i < 50; i++ { + servers, _, err := vg.findEmptySlotsForOneVolume(topo, option, false) + if err != nil { + t.Fatalf("iteration %d: %v", i, err) + } + if len(servers) != 2 { + t.Fatalf("iteration %d: got %d servers, want 2", i, len(servers)) + } + onSibling := 0 + for _, s := range servers { + if s.Id() == busy.Id() { + t.Fatalf("iteration %d: picked the server in maintenance mode", i) + } + if s.Id() == ok.Id() { + onSibling++ + } + } + if onSibling != 1 { + t.Fatalf("iteration %d: the other-rack replica must land on the eligible sibling, got %v", i, servers) + } + } +} + +// A replica on a server in maintenance mode takes no writes, so its volume +// leaves the writable list while the mode is on and returns once it is off. +func TestMaintenanceModeTogglesVolumeWritability(t *testing.T) { + topo, dc := newMaintenanceTestTopology() + rack := NewRack("rack1") + dc.LinkChildNode(rack) + dn1 := addMaintenanceTestNode(rack, "dn1", 10) + dn2 := addMaintenanceTestNode(rack, "dn2", 10) + + rp, _ := super_block.NewReplicaPlacementFromString("001") + shared := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()} + dn1.AddOrUpdateVolume(shared) + dn2.AddOrUpdateVolume(shared) + topo.RegisterVolumeLayout(shared, dn1) + topo.RegisterVolumeLayout(shared, dn2) + + rp0, _ := super_block.NewReplicaPlacementFromString("000") + alone := storage.VolumeInfo{Id: 2, Size: 100, ReplicaPlacement: rp0, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()} + dn2.AddOrUpdateVolume(alone) + topo.RegisterVolumeLayout(alone, dn2) + + vlShared := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType) + vlAlone := topo.GetVolumeLayout("", rp0, needle.EMPTY_TTL, types.HardDriveType) + expectWritables := func(step string, vl *VolumeLayout, want ...needle.VolumeId) { + t.Helper() + got := vl.CloneWritableVolumes() + slices.Sort(got) + if !slices.Equal(got, want) { + t.Errorf("%s: writables = %v, want %v", step, got, want) + } + } + expectWritables("initial shared", vlShared, 1) + expectWritables("initial alone", vlAlone, 2) + + if !topo.SetDataNodeMaintenanceMode(dn1, true) { + t.Fatal("switching dn1 on should report a change") + } + expectWritables("dn1 in maintenance: a volume with a replica there takes no writes", vlShared) + expectWritables("dn1 in maintenance: volumes elsewhere are unaffected", vlAlone, 2) + if locations := vlShared.Lookup(1); len(locations) != 2 { + t.Errorf("reads must still see every replica, got %v", locations) + } + + if topo.SetDataNodeMaintenanceMode(dn1, true) { + t.Error("repeating the same state should not report a change") + } + + if !topo.SetDataNodeMaintenanceMode(dn1, false) { + t.Fatal("switching dn1 off should report a change") + } + expectWritables("dn1 back: the volume is writable again", vlShared, 1) + + topo.SetDataNodeMaintenanceMode(dn2, true) + expectWritables("dn2 in maintenance: shared", vlShared) + expectWritables("dn2 in maintenance: alone", vlAlone) +} + +// A vacuum commit or a mark-writable that lands after the server entered +// maintenance mode must not put the volume back on the writable list. Both +// used to check only the replica count, and since heartbeats carry only changed +// volumes, nothing would have taken it off again. +func TestMaintenanceModeHoldsThroughVacuumAndMarkWritable(t *testing.T) { + topo, dc := newMaintenanceTestTopology() + rack := NewRack("rack1") + dc.LinkChildNode(rack) + dn1 := addMaintenanceTestNode(rack, "dn1", 10) + dn2 := addMaintenanceTestNode(rack, "dn2", 10) + + rp, _ := super_block.NewReplicaPlacementFromString("001") + vi := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()} + dn1.AddOrUpdateVolume(vi) + dn2.AddOrUpdateVolume(vi) + topo.RegisterVolumeLayout(vi, dn1) + topo.RegisterVolumeLayout(vi, dn2) + vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType) + expectWritableCount := func(step string, want int) { + t.Helper() + if got, _ := vl.GetWritableVolumeCount(); got != want { + t.Errorf("%s: writable count = %d, want %d", step, got, want) + } + } + + // vacuum: taken off the writable list before compaction, dn1 enters + // maintenance meanwhile, then every replica commits + vl.DrainAndRemoveFromWritable(1) + topo.SetDataNodeMaintenanceMode(dn1, true) + for _, dn := range []*DataNode{dn1, dn2} { + if vl.SetVolumeAvailable(dn, 1, false, false) { + t.Errorf("vacuum commit on %s made the volume writable with a replica in maintenance", dn.Id()) + } + } + expectWritableCount("after vacuum commit", 0) + topo.SetDataNodeMaintenanceMode(dn1, false) + expectWritableCount("after maintenance", 1) + + // mark-writable from a vacuum worker, after dn1 entered maintenance + vl.SetVolumeReadOnly(dn2, 1) + topo.SetDataNodeMaintenanceMode(dn1, true) + if vl.SetVolumeWritable(dn2, 1) { + t.Error("mark-writable made the volume writable with a replica in maintenance") + } + expectWritableCount("after mark-writable", 0) + topo.SetDataNodeMaintenanceMode(dn1, false) + expectWritableCount("after maintenance", 1) +} + +// A volume server notifies the master the moment it flips a volume between +// read-only and writable, ahead of the heartbeat that repeats the flag. The +// master must act on the notification, not on its heartbeat copy: mark-writable +// takes effect at once, and a maintenance toggle landing between a mark-readonly +// and its heartbeat must not put the volume back. +func TestMarkReadOnlyAndWritableAheadOfHeartbeat(t *testing.T) { + topo, dc := newMaintenanceTestTopology() + rack := NewRack("rack1") + dc.LinkChildNode(rack) + dn1 := addMaintenanceTestNode(rack, "dn1", 10) + dn2 := addMaintenanceTestNode(rack, "dn2", 10) + + rp, _ := super_block.NewReplicaPlacementFromString("001") + vi := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()} + dn1.AddOrUpdateVolume(vi) + dn2.AddOrUpdateVolume(vi) + topo.RegisterVolumeLayout(vi, dn1) + topo.RegisterVolumeLayout(vi, dn2) + vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType) + expectWritableCount := func(step string, want int) { + t.Helper() + if got, _ := vl.GetWritableVolumeCount(); got != want { + t.Errorf("%s: writable count = %d, want %d", step, got, want) + } + } + + // the heartbeat has reported dn2's replica read-only + readOnly := vi + readOnly.ReadOnly = true + dn2.AddOrUpdateVolume(readOnly) + vl.EnsureCorrectWritables(&readOnly) + expectWritableCount("heartbeat says read-only", 0) + + // dn2 marks it writable and notifies, before the next heartbeat + if !vl.SetVolumeWritable(dn2, 1) { + t.Error("mark-writable from the replica's own server should take effect at once") + } + expectWritableCount("after mark-writable", 1) + if v, err := dn2.GetVolumesById(1); err != nil || v.ReadOnly { + t.Errorf("the node's record should say writable, got %+v, %v", v, err) + } + + // dn2 marks it read-only and notifies; dn1 leaves maintenance before the + // heartbeat repeats the flag + vl.SetVolumeReadOnly(dn2, 1) + expectWritableCount("after mark-readonly", 0) + topo.SetDataNodeMaintenanceMode(dn1, true) + topo.SetDataNodeMaintenanceMode(dn1, false) + expectWritableCount("re-evaluated before the heartbeat", 0) +} + +// SetDataNodeMaintenanceMode walks a snapshot of the node's volumes, so a +// concurrent disconnect can have removed one from its layout by the time it is +// re-evaluated. That must be a no-op, not a crash or a resurrected entry. +func TestMaintenanceModeAfterVolumeLeftLayout(t *testing.T) { + topo, dc := newMaintenanceTestTopology() + rack := NewRack("rack1") + dc.LinkChildNode(rack) + dn := addMaintenanceTestNode(rack, "dn1", 10) + + rp, _ := super_block.NewReplicaPlacementFromString("000") + vi := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()} + dn.AddOrUpdateVolume(vi) + topo.RegisterVolumeLayout(vi, dn) + vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType) + vl.UnRegisterVolume(&vi, dn) + + for _, on := range []bool{true, false} { + if !topo.SetDataNodeMaintenanceMode(dn, on) { + t.Fatalf("SetDataNodeMaintenanceMode(%v) should report a change", on) + } + if got, _ := vl.GetWritableVolumeCount(); got != 0 { + t.Errorf("maintenance %v: writable count = %d, want 0", on, got) + } + if locations := vl.Lookup(1); locations != nil { + t.Errorf("maintenance %v: re-evaluating a departed volume re-created its location list: %v", on, locations) + } + } +} diff --git a/weed/topology/node.go b/weed/topology/node.go index f0bb84297..a603d7d87 100644 --- a/weed/topology/node.go +++ b/weed/topology/node.go @@ -306,7 +306,22 @@ func (n *NodeImpl) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts { return n.diskUsages.getOrCreateDisk(diskType) } +// inMaintenanceMode is true for a data node whose volume server is in +// maintenance mode. Racks and data centers never are: their rolled-up counters +// still include such a node, so their free-slot totals may exceed what their +// children will actually hand out; see reserveOneVolumeInternal. +func (n *NodeImpl) inMaintenanceMode() bool { + dn, ok := n.value.(*DataNode) + return ok && dn.InMaintenanceMode() +} + +// AvailableSpaceFor is the free volume slots on this node for the option's disk +// type. A data node in maintenance mode reports none: it is being drained, so +// it is neither a volume-growth candidate nor reservable. func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 { + if n.inMaintenanceMode() { + return 0 + } t := n.getOrCreateDisk(option.DiskType) freeVolumeSlotCount := atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount) - atomic.LoadInt64(&t.volumeCount) freeVolumeSlotCount -= erasure_coding.VolumeSlots(atomic.LoadInt64(&t.ecShardCount)) @@ -400,13 +415,29 @@ func (n *NodeImpl) ReserveOneVolumeForReservation(r int64, option *VolumeGrowOpt func (n *NodeImpl) reserveOneVolumeInternal(r int64, option *VolumeGrowOption, useReservations bool) (assignedNode *DataNode, err error) { n.RLock() defer n.RUnlock() - for _, node := range n.children { - var freeSpace int64 + freeSpaceOf := func(node Node) int64 { if useReservations { - freeSpace = node.AvailableSpaceForReservation(option) - } else { - freeSpace = node.AvailableSpaceFor(option) + return node.AvailableSpaceForReservation(option) } + return node.AvailableSpaceFor(option) + } + // The caller draws r from this node's rolled-up free slots, which still + // count children the walk below skips: a data node in maintenance mode + // reports no space, and an over-committed one reports less than zero. Fold + // r into the space that is actually on offer so it cannot walk off the end + // and fail with slots still free. + var eligible int64 + for _, node := range n.children { + if freeSpace := freeSpaceOf(node); freeSpace > 0 { + eligible += freeSpace + } + } + if eligible <= 0 { + return nil, errors.New("No free volume slot found!") + } + r %= eligible + for _, node := range n.children { + freeSpace := freeSpaceOf(node) // fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace) if freeSpace <= 0 { continue diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index b0894313e..665dfe16c 100644 --- a/weed/topology/topology_event_handling.go +++ b/weed/topology/topology_event_handling.go @@ -89,6 +89,25 @@ func (t *Topology) SetVolumeCrowded(volumeInfo storage.VolumeInfo) { vl.SetVolumeCrowded(volumeInfo.Id) } +// SetDataNodeMaintenanceMode records the maintenance flag a volume server +// reported and returns whether it changed. On a change every volume the node +// holds is re-evaluated: while the flag is on, a volume with a replica there +// leaves the writable list, so no writes are assigned to it and the server can +// be drained; once the flag is off the volumes are writable again. Reads are +// untouched, and volume growth skips the node through AvailableSpaceFor. +func (t *Topology) SetDataNodeMaintenanceMode(dn *DataNode, on bool) (changed bool) { + if !dn.SetMaintenanceMode(on) { + return false + } + glog.V(0).Infof("volume server %s maintenance mode: %v", dn.Id(), on) + for _, v := range dn.GetVolumes() { + diskType := types.ToDiskType(v.DiskType) + vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType) + vl.EnsureCorrectWritables(&v) + } + return true +} + func (t *Topology) UnRegisterDataNode(dn *DataNode) { dn.IsTerminating = true for _, v := range dn.GetVolumes() { diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 165005ca3..17897cbbf 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -404,9 +404,15 @@ func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) { } } +// isAllWritable reports whether every replica of vid can take writes. A write +// lands on one replica and is forwarded to the rest, so one read-only replica, +// or one on a server in maintenance mode, holds the whole volume out. func (vl *VolumeLayout) isAllWritable(vid needle.VolumeId) bool { if location, ok := vl.vid2location[vid]; ok { for _, dn := range location.list { + if dn.InMaintenanceMode() { + return false + } if v, getError := dn.GetVolumesById(vid); getError == nil { if v.ReadOnly { return false @@ -923,7 +929,11 @@ func (vl *VolumeLayout) setVolumeWritable(vid needle.VolumeId) bool { return true } +// SetVolumeReadOnly and SetVolumeWritable apply what dn itself said about its +// replica of vid. That is recorded on the node first, so isAllWritable judges +// the volume by it now rather than by the heartbeat that has yet to repeat it. func (vl *VolumeLayout) SetVolumeReadOnly(dn *DataNode, vid needle.VolumeId) bool { + dn.SetVolumeReadOnly(vid, true) vl.accessLock.Lock() defer vl.accessLock.Unlock() @@ -935,6 +945,7 @@ func (vl *VolumeLayout) SetVolumeReadOnly(dn *DataNode, vid needle.VolumeId) boo } func (vl *VolumeLayout) SetVolumeWritable(dn *DataNode, vid needle.VolumeId) bool { + dn.SetVolumeReadOnly(vid, false) vl.accessLock.Lock() defer vl.accessLock.Unlock() @@ -942,7 +953,7 @@ func (vl *VolumeLayout) SetVolumeWritable(dn *DataNode, vid needle.VolumeId) boo location.SetReadOnly(dn, false) } - if vl.enoughCopies(vid) { + if vl.enoughCopies(vid) && vl.isAllWritable(vid) { return vl.setVolumeWritable(vid) } return false @@ -1002,7 +1013,7 @@ func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, is } vl.initSizeTracking(vid, vInfo.Size, vInfo.CompactRevision) - if vl.enoughCopies(vid) { + if vl.enoughCopies(vid) && vl.isAllWritable(vid) { becameWritable = vl.setVolumeWritable(vid) if becameWritable { if st := vl.sizeTracking[vid]; st != nil && !st.fullSince.IsZero() {