Healthz check for deadlocks (#4558)

2 files changed, 41 insertions, 0 deletions

diff --git a/weed/topology/node.go b/weed/topology/node.go
index 1c7e3e468..851e71385 100644
--- a/weed/topology/node.go
+++ b/weed/topology/node.go
@@ -34,11 +34,13 @@ type Node interface {
 	IsDataNode() bool
 	IsRack() bool
 	IsDataCenter() bool
+	IsLocked() bool
 	Children() []Node
 	Parent() Node
 
 	GetValue() interface{} //get reference to the topology,dc,rack,datanode
 }
+
 type NodeImpl struct {
 	diskUsages   *DiskUsages
 	id           NodeId
@@ -122,24 +124,37 @@ func (n *NodeImpl) PickNodesByWeight(numberOfNodes int, option *VolumeGrowOption
 func (n *NodeImpl) IsDataNode() bool {
 	return n.nodeType == "DataNode"
 }
+
 func (n *NodeImpl) IsRack() bool {
 	return n.nodeType == "Rack"
 }
+
 func (n *NodeImpl) IsDataCenter() bool {
 	return n.nodeType == "DataCenter"
 }
+
+func (n *NodeImpl) IsLocked() (isTryLock bool) {
+	if isTryLock = n.TryRLock(); isTryLock {
+		n.RUnlock()
+	}
+	return !isTryLock
+}
+
 func (n *NodeImpl) String() string {
 	if n.parent != nil {
 		return n.parent.String() + ":" + string(n.id)
 	}
 	return string(n.id)
 }
+
 func (n *NodeImpl) Id() NodeId {
 	return n.id
 }
+
 func (n *NodeImpl) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts {
 	return n.diskUsages.getOrCreateDisk(diskType)
 }
+
 func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 {
 	t := n.getOrCreateDisk(option.DiskType)
 	freeVolumeSlotCount := atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount) - atomic.LoadInt64(&t.volumeCount)
@@ -152,6 +167,7 @@ func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 {
 func (n *NodeImpl) SetParent(node Node) {
 	n.parent = node
 }
+
 func (n *NodeImpl) Children() (ret []Node) {
 	n.RLock()
 	defer n.RUnlock()
@@ -160,12 +176,15 @@ func (n *NodeImpl) Children() (ret []Node) {
 	}
 	return ret
 }
+
 func (n *NodeImpl) Parent() Node {
 	return n.parent
 }
+
 func (n *NodeImpl) GetValue() interface{} {
 	return n.value
 }
+
 func (n *NodeImpl) ReserveOneVolume(r int64, option *VolumeGrowOption) (assignedNode *DataNode, err error) {
 	n.RLock()
 	defer n.RUnlock()
diff --git a/weed/topology/topology.go b/weed/topology/topology.go
index fee4d6a03..017208b81 100644
--- a/weed/topology/topology.go
+++ b/weed/topology/topology.go
@@ -76,6 +76,28 @@ func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, puls
 	return t
 }
 
+func (t *Topology) IsChildLocked() (bool, error) {
+	if t.IsLocked() {
+		return true, errors.New("topology is locked")
+	}
+	for _, dcNode := range t.Children() {
+		if dcNode.IsLocked() {
+			return true, fmt.Errorf("topology child %s is locked", dcNode.String())
+		}
+		for _, rackNode := range dcNode.Children() {
+			if rackNode.IsLocked() {
+				return true, fmt.Errorf("dc %s child %s is locked", dcNode.String(), rackNode.String())
+			}
+			for _, dataNode := range rackNode.Children() {
+				if dataNode.IsLocked() {
+					return true, fmt.Errorf("rack %s child %s is locked", rackNode.String(), dataNode.Id())
+				}
+			}
+		}
+	}
+	return false, nil
+}
+
 func (t *Topology) IsLeader() bool {
 	t.RaftServerAccessLock.RLock()
 	defer t.RaftServerAccessLock.RUnlock()