Scheduled Events: Bug Fixes

* Fixed bug in nodewatcher slurm plugin `lock_host`. Added unit test for this function.
* Addressed comments from previous review

Signed-off-by: Rex <[email protected]>

Author: Rex

src/common/utils.py

@@ -55,6 +55,16 @@ class QueueType(Enum):
 Host = collections.namedtuple("Host", ["instance_id", "hostname", "slots", "gpus"])
 UpdateEvent = collections.namedtuple("UpdateEvent", ["action", "message", "host"])
 INSTANCE_ALIVE_STATES = ["pending", "running"]
+TREAT_DISABLED_AS_DOWN_WARNING = (
+    "Considering node as down because there is no job running and node is in a disabled state. "
+    "The node could have been put into this disabled state automatically by ParallelCluster "
+    "in response to an EC2 scheduled maintenance event, or manually by the system administrator."
+)
+POSSIBLE_LOCK_CONFLICT_WARNING = (
+    "Instance %s/%s currently in disabled state %s. "
+    "Risk of lock being released by nodewatcher if locking the node because of scheduled event now. "
+    "Marking event as failed to retry later."
+)
 
 
 def load_module(module):

src/nodewatcher/plugins/sge.py

@@ -22,7 +22,7 @@
     lock_node,
     unlock_node,
 )
-from common.utils import check_command_output
+from common.utils import TREAT_DISABLED_AS_DOWN_WARNING, check_command_output
 
 log = logging.getLogger(__name__)
 
@@ -87,13 +87,7 @@ def is_node_down():
         if all(error_state not in node.state for error_state in SGE_ERROR_STATES):
             # Consider the node down if it's in disabled state and there is no job running
             if SGE_DISABLED_STATE in node.state and not has_jobs(hostname):
-                log.warning(
-                    (
-                        "Considering node as down because there is no job running and node is in a disabled state. "
-                        "The node could have been put into this disabled state automatically by ParallelCluster "
-                        "in response to an EC2 scheduled maintenance event, or manually by the system administrator."
-                    )
-                )
+                log.warning(TREAT_DISABLED_AS_DOWN_WARNING)
                 return True
             return False
     except Exception as e:

src/nodewatcher/plugins/slurm.py

@@ -20,7 +20,7 @@
     lock_node,
     unlock_node,
 )
-from common.utils import check_command_output
+from common.utils import TREAT_DISABLED_AS_DOWN_WARNING, check_command_output
 
 log = logging.getLogger(__name__)
 
@@ -61,7 +61,7 @@ def has_pending_jobs(instance_properties, max_size):
 
 
 def lock_host(hostname, unlock=False):
-    if unlock_node:
+    if unlock:
         unlock_node(hostname)
     else:
         lock_node(hostname)
@@ -76,13 +76,7 @@ def is_node_down():
         if output and all(state not in output for state in SLURM_NODE_ERROR_STATES):
             return False
         if output and "drained" in output:
-            log.warning(
-                (
-                    "Considering node as down because there is no job running and node is in a disabled state. "
-                    "The node could have been put into this disabled state automatically by ParallelCluster in "
-                    "response to an EC2 scheduled maintenance event, or manually by the system administrator."
-                )
-            )
+            log.warning(TREAT_DISABLED_AS_DOWN_WARNING)
     except Exception as e:
         log.error("Failed when checking if node is down with exception %s. Reporting node as down.", e)
 

src/nodewatcher/plugins/torque.py

@@ -21,7 +21,7 @@
     get_pending_jobs_info,
     lock_node,
 )
-from common.utils import check_command_output
+from common.utils import TREAT_DISABLED_AS_DOWN_WARNING, check_command_output
 
 log = logging.getLogger(__name__)
 
@@ -70,13 +70,7 @@ def is_node_down():
                 # Consider the node down if it is in Disabled state placed by scheduled event
                 # and does not have job
                 if TORQUE_NODE_DISABLED_STATE in node.state and not has_jobs(hostname):
-                    log.warning(
-                        (
-                            "Considering node as down because there is no job running and node is in a disabled state. "
-                            "The node could have been put into this disabled state automatically by ParallelCluster in "
-                            "response to an EC2 scheduled maintenance event, or manually by the system administrator."
-                        )
-                    )
+                    log.warning(TREAT_DISABLED_AS_DOWN_WARNING)
                     return True
                 return False
         else:

src/sqswatcher/plugins/sge.py

@@ -23,7 +23,7 @@
     remove_hosts_from_group,
     remove_hosts_from_queue,
 )
-from common.utils import EventType
+from common.utils import POSSIBLE_LOCK_CONFLICT_WARNING, EventType
 
 log = logging.getLogger(__name__)
 
@@ -124,11 +124,7 @@ def perform_health_actions(health_events):
         try:
             if _is_node_locked(event.host.hostname):
                 log.warning(
-                    "Instance %s/%s currently in disabled state 'd'. "
-                    "Risk of lock being released by nodewatcher if locking the node because of scheduled event now. "
-                    "Marking event as failed to retry later.",
-                    event.host.instance_id,
-                    event.host.hostname,
+                    POSSIBLE_LOCK_CONFLICT_WARNING, event.host.instance_id, event.host.hostname, SGE_DISABLED_STATE,
                 )
                 failed.append(event)
                 continue

src/sqswatcher/plugins/slurm.py

@@ -24,7 +24,7 @@
 
 from common.remote_command_executor import RemoteCommandExecutor
 from common.schedulers.slurm_commands import SLURM_NODE_DISABLED_STATES, get_node_state, lock_node
-from common.utils import EventType, run_command
+from common.utils import POSSIBLE_LOCK_CONFLICT_WARNING, EventType, run_command
 
 log = logging.getLogger(__name__)
 
@@ -231,11 +231,10 @@ def perform_health_actions(health_events):
         try:
             if _is_node_locked(event.host.hostname):
                 log.warning(
-                    "Instance %s/%s currently in disabled state 'draining/drained'. "
-                    "Risk of lock being released by nodewatcher if locking the node because of scheduled event now. "
-                    "Marking event as failed to retry later.",
+                    POSSIBLE_LOCK_CONFLICT_WARNING,
                     event.host.instance_id,
                     event.host.hostname,
+                    SLURM_NODE_DISABLED_STATES,
                 )
                 failed.append(event)
                 continue

src/sqswatcher/plugins/torque.py

@@ -23,7 +23,7 @@
     update_cluster_limits,
     wakeup_scheduler,
 )
-from common.utils import EventType
+from common.utils import POSSIBLE_LOCK_CONFLICT_WARNING, EventType
 
 log = logging.getLogger(__name__)
 
@@ -75,11 +75,10 @@ def perform_health_actions(health_events):
         try:
             if _is_node_locked(event.host.hostname):
                 log.warning(
-                    "Instance %s/%s currently in disabled state 'offline'. "
-                    "Risk of lock being released by nodewatcher if locking the node because of scheduled event now. "
-                    "Marking event as failed to retry later.",
+                    POSSIBLE_LOCK_CONFLICT_WARNING,
                     event.host.instance_id,
                     event.host.hostname,
+                    TORQUE_NODE_DISABLED_STATE,
                 )
                 failed.append(event)
                 continue

tests/nodewatcher/plugins/test_sge.py

@@ -12,7 +12,7 @@
 
 from assertpy import assert_that
 from common.schedulers.sge_commands import SGE_HOLD_STATE, SgeHost, SgeJob
-from nodewatcher.plugins.sge import has_jobs, has_pending_jobs, is_node_down
+from nodewatcher.plugins.sge import has_jobs, has_pending_jobs, is_node_down, lock_host
 
 
 @pytest.mark.parametrize(
@@ -184,3 +184,15 @@ def test_has_jobs(jobs, expected_result, mocker):
 
     assert_that(has_jobs(hostname)).is_equal_to(expected_result)
     mock.assert_called_with(hostname_filter=hostname, job_state_filter="rs")
+
+
+@pytest.mark.parametrize(
+    "hostname, unlock", [("ip-10-0-0-166", False), ("ip-10-0-0-166", True)],
+)
+def test_lock_host(hostname, unlock, mocker):
+    if unlock:
+        mock = mocker.patch("nodewatcher.plugins.sge.unlock_node", autospec=True)
+    else:
+        mock = mocker.patch("nodewatcher.plugins.sge.lock_node", autospec=True)
+    lock_host(hostname, unlock)
+    mock.assert_called_with(hostname)

tests/nodewatcher/plugins/test_slurm.py

@@ -12,7 +12,7 @@
 
 from assertpy import assert_that
 from common.schedulers.slurm_commands import PENDING_RESOURCES_REASONS, SlurmJob
-from nodewatcher.plugins.slurm import has_pending_jobs, is_node_down
+from nodewatcher.plugins.slurm import has_pending_jobs, is_node_down, lock_host
 
 
 @pytest.mark.parametrize(
@@ -88,3 +88,15 @@ def test_is_node_down(hostname, get_node_state_output, expected_result, mocker):
 
     assert_that(is_node_down()).is_equal_to(expected_result)
     mock.assert_called_with(hostname)
+
+
+@pytest.mark.parametrize(
+    "hostname, unlock", [("ip-10-0-0-166", False), ("ip-10-0-0-166", True)],
+)
+def test_lock_host(hostname, unlock, mocker):
+    if unlock:
+        mock = mocker.patch("nodewatcher.plugins.slurm.unlock_node", autospec=True)
+    else:
+        mock = mocker.patch("nodewatcher.plugins.slurm.lock_node", autospec=True)
+    lock_host(hostname, unlock)
+    mock.assert_called_with(hostname)

tests/nodewatcher/plugins/test_torque.py

@@ -12,7 +12,7 @@
 
 from assertpy import assert_that
 from common.schedulers.torque_commands import TorqueHost, TorqueJob, TorqueResourceList
-from nodewatcher.plugins.torque import has_jobs, has_pending_jobs, is_node_down
+from nodewatcher.plugins.torque import has_jobs, has_pending_jobs, is_node_down, lock_host
 
 
 @pytest.mark.parametrize(
@@ -171,3 +171,12 @@ def test_has_jobs(jobs, expected_result, mocker):
 
     assert_that(has_jobs(hostname)).is_equal_to(expected_result)
     mock.assert_called_with(filter_by_exec_hosts={hostname.split(".")[0]}, filter_by_states=["R", "S"])
+
+
+@pytest.mark.parametrize(
+    "hostname, unlock", [("ip-10-0-0-166", False), ("ip-10-0-0-166", True)],
+)
+def test_lock_host(hostname, unlock, mocker):
+    mock = mocker.patch("nodewatcher.plugins.torque.lock_node", autospec=True)
+    lock_host(hostname, unlock)
+    mock.assert_called_with(hostname=hostname, unlock=unlock)