fix(node_controller): check node responsibility before deletion

Signed-off-by: David Cheng <davidcheng0922@gmail.com>

Author: davidcheng0922

controller/monitor/disk_monitor.go

@@ -141,6 +141,11 @@ func (m *DiskMonitor) run(value interface{}) error {
 		return errors.Wrapf(err, "failed to get longhorn node %v", m.nodeName)
 	}
 
+	if node.DeletionTimestamp != nil {
+		logrus.Infof("Skipping disk monitoring for node %v: node is being deleted", m.nodeName)
+		return nil
+	}
+
 	collectedData := m.collectDiskData(node)
 	if !reflect.DeepEqual(m.collectedData, collectedData) {
 		func() {

controller/node_controller.go

@@ -1585,20 +1585,22 @@ func (nc *NodeController) alignDiskSpecAndStatus(node *longhorn.Node) {
 				continue
 			}
 
-			if err := nc.deleteDisk(diskStatus.Type, diskName, diskStatus.DiskUUID, diskStatus.DiskPath, string(diskStatus.DiskDriver)); err != nil {
+			if err := nc.deleteDisk(node.Name, diskStatus.Type, diskName, diskStatus.DiskUUID, diskStatus.DiskPath, string(diskStatus.DiskDriver)); err != nil {
 				nc.logger.WithError(err).Warnf("Failed to delete disk %v", diskName)
 			}
 			delete(node.Status.DiskStatus, diskName)
 		}
 	}
 }
 
-func (nc *NodeController) deleteDisk(diskType longhorn.DiskType, diskName, diskUUID, diskPath, diskDriver string) error {
+func (nc *NodeController) deleteDisk(nodeName string, diskType longhorn.DiskType, diskName, diskUUID, diskPath, diskDriver string) error {
 	nc.logger.Infof("Deleting disk %v with diskUUID %v", diskName, diskUUID)
 
 	dataEngine := util.GetDataEngineForDiskType(diskType)
 
-	im, err := nc.ds.GetRunningInstanceManagerByNodeRO(nc.controllerID, dataEngine)
+	// use nodeName instead of nc.controllerID to get the instance manager
+	// allow to delete the non-own node disks
+	im, err := nc.ds.GetRunningInstanceManagerByNodeRO(nodeName, dataEngine)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get running instance manager for data engine %v", dataEngine)
 	}
@@ -1613,6 +1615,7 @@ func (nc *NodeController) deleteDisk(diskType longhorn.DiskType, diskName, diskU
 		return errors.Wrapf(err, "failed to delete disk %v", diskName)
 	}
 
+	nc.logger.Infof("Deleted disk %v with diskUUID %v", diskName, diskUUID)
 	return nil
 }
 
@@ -1649,7 +1652,7 @@ func (nc *NodeController) cleanupDisksBeforeNodeDeletion(node *longhorn.Node) er
 		// Note: Use diskName instead of diskStatus.Name since they are different:
 		//   diskName:        default-disk-block-1
 		//   diskStatus.Name: default-disk-block-1n1
-		if err := nc.deleteDisk(diskStatus.Type, diskName, diskStatus.DiskUUID, diskStatus.DiskPath, string(diskStatus.DiskDriver)); err != nil {
+		if err := nc.deleteDisk(node.Name, diskStatus.Type, diskName, diskStatus.DiskUUID, diskStatus.DiskPath, string(diskStatus.DiskDriver)); err != nil {
 			nc.logger.WithError(err).Errorf("Failed to delete SPDK disk %v during node cleanup", diskStatus.DiskName)
 			errs.Append("errors", errors.Wrapf(err, "failed to delete SPDK disk %v during node cleanup", diskStatus.DiskName))
 			continue