AMBARI-18505. Ambari Status commands should enforce a timeout < heartbeat interval (aonishuk)

ambari-agent/conf/unix/ambari-agent.ini

@@ -32,6 +32,7 @@ tolerate_download_failures=true
 run_as_user=root
 parallel_execution=0
 alert_grace_period=5
+status_command_timeout=2
 alert_kinit_timeout=14400000
 system_resource_overrides=/etc/resource_overrides
 ; memory_threshold_soft_mb=400

ambari-agent/src/main/python/ambari_agent/ActionQueue.py

@@ -29,6 +29,7 @@ import time
 import signal
 
 from AgentException import AgentException
+from PythonReflectiveExecutor import PythonReflectiveExecutor
 from LiveStatus import LiveStatus
 from ActualConfigHandler import ActualConfigHandler
 from CommandStatusDict import CommandStatusDict
@@ -82,9 +83,11 @@ class ActionQueue(threading.Thread):
     self.controller = controller
     self.configTags = {}
     self._stop = threading.Event()
+    self.hangingStatusCommands = {}
     self.tmpdir = config.get('agent', 'prefix')
     self.customServiceOrchestrator = CustomServiceOrchestrator(config, controller)
     self.parallel_execution = config.get_parallel_exec_option()
+    self.status_command_timeout = int(self.config.get('agent', 'status_command_timeout', 5))
     if self.parallel_execution == 1:
       logger.info("Parallel execution is enabled, will execute agent commands in parallel")
 
@@ -225,7 +228,24 @@ class ActionQueue(threading.Thread):
           if self.controller.recovery_manager.enabled():
             self.controller.recovery_manager.stop_execution_command()
       elif commandType == self.STATUS_COMMAND:
-        self.execute_status_command(command)
+        component_name = command['componentName']
+
+        if component_name in self.hangingStatusCommands and not self.hangingStatusCommands[component_name].isAlive():
+          del self.hangingStatusCommands[component_name]
+
+        if not component_name in self.hangingStatusCommands:
+          thread = threading.Thread(target = self.execute_status_command, args = (command,))
+          thread.daemon = True # hanging status commands should not be prevent ambari-agent from stopping
+          thread.start()
+          thread.join(timeout=self.status_command_timeout)
+
+          if thread.isAlive():
+            # Force context to reset to normal. By context we mean sys.path, imports, logger setting, etc. They are set by specific status command, and are not relevant to ambari-agent.
+            PythonReflectiveExecutor.last_context.revert()
+            logger.warn("Command {0} for {1} is running for more than {2} seconds. Skipping it for current pack of status commands.".format(commandType, component_name, self.status_command_timeout))
+            self.hangingStatusCommands[component_name] = thread
+        else:
+          logger.info("Not running {0} for {1}, because previous one is still running.".format(commandType, component_name))
       else:
         logger.error("Unrecognized command " + pprint.pformat(command))
     except Exception:

ambari-agent/src/main/python/ambari_agent/PythonReflectiveExecutor.py

@@ -53,7 +53,9 @@ class PythonReflectiveExecutor(PythonExecutor):
     returncode = 1
 
     try:
-      with PythonContext(script_dir, pythonCommand):
+      current_context = PythonContext(script_dir, pythonCommand)
+      PythonReflectiveExecutor.last_context = current_context
+      with current_context:
         imp.load_source('__main__', script)
     except SystemExit as e:
       returncode = e.code
@@ -62,7 +64,10 @@ class PythonReflectiveExecutor(PythonExecutor):
     except (ClientComponentHasNoStatus, ComponentIsNotRunning):
       logger.debug("Reflective command failed with exception:", exc_info=1)
     except Exception:
-      logger.info("Reflective command failed with exception:", exc_info=1)
+      if current_context.is_forced_revert:
+        logger.info("Hanging status command finished its execution")
+      else:
+        logger.info("Reflective command failed with exception:", exc_info=1)
     else: 
       returncode = 0
       
@@ -76,6 +81,8 @@ class PythonContext:
   def __init__(self, script_dir, pythonCommand):
     self.script_dir = script_dir
     self.pythonCommand = pythonCommand
+    self.is_reverted = False
+    self.is_forced_revert = False
     
   def __enter__(self):
     self.old_sys_path = copy.copy(sys.path)
@@ -88,12 +95,18 @@ class PythonContext:
     sys.argv = self.pythonCommand[1:]
 
   def __exit__(self, exc_type, exc_val, exc_tb):
-    sys.path = self.old_sys_path
-    sys.argv = self.old_agv
-    logging.disable(self.old_logging_disable)
-    self.revert_sys_modules(self.old_sys_modules)
+    self.revert(is_forced_revert=False)
     return False
   
+  def revert(self, is_forced_revert=True):
+    if not self.is_reverted:
+      self.is_forced_revert = is_forced_revert
+      self.is_reverted = True
+      sys.path = self.old_sys_path
+      sys.argv = self.old_agv
+      logging.disable(self.old_logging_disable)
+      self.revert_sys_modules(self.old_sys_modules)
+
   def revert_sys_modules(self, value):
     sys.modules.update(value)
     

ambari-agent/src/test/python/ambari_agent/TestActionQueue.py

@@ -225,6 +225,7 @@ class TestActionQueue(TestCase):
   retryable_command = {
     'commandType': 'EXECUTION_COMMAND',
     'role': 'NAMENODE',
+    'componentName': 'NAMENODE',
     'roleCommand': 'INSTALL',
     'commandId': '1-1',
     'taskId': 19,
@@ -322,6 +323,7 @@ class TestActionQueue(TestCase):
     }
     status_command = {
       'commandType' : ActionQueue.STATUS_COMMAND,
+      'componentName': 'NAMENODE'
     }
     wrong_command = {
       'commandType' : "SOME_WRONG_COMMAND",
@@ -1126,7 +1128,6 @@ class TestActionQueue(TestCase):
     self.assertTrue(runCommand_mock.called)
     self.assertEqual(2, runCommand_mock.call_count)
     self.assertEqual(1, sleep_mock.call_count)
-    sleep_mock.assert_has_calls([call(1)], False)
     runCommand_mock.assert_has_calls([
       call(command, os.sep + 'tmp' + os.sep + 'ambari-agent' + os.sep + 'output-19.txt',
            os.sep + 'tmp' + os.sep + 'ambari-agent' + os.sep + 'errors-19.txt', override_output_files=True, retry=False),