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=5
 alert_kinit_timeout=14400000
 system_resource_overrides=/etc/resource_overrides
 ; memory_threshold_soft_mb=400

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

@@ -18,6 +18,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 '''
 import Queue
+import multiprocessing
 
 import logging
 import traceback
@@ -74,7 +75,8 @@ class ActionQueue(threading.Thread):
   def __init__(self, config, controller):
     super(ActionQueue, self).__init__()
     self.commandQueue = Queue.Queue()
-    self.statusCommandQueue = Queue.Queue()
+    self.statusCommandQueue = multiprocessing.Queue()
+    self.statusCommandResultQueue = multiprocessing.Queue() # this queue is filled by StatuCommandsExecutor.
     self.backgroundCommandQueue = Queue.Queue()
     self.commandStatuses = CommandStatusDict(callback_action =
       self.status_update_callback)
@@ -95,8 +97,9 @@ class ActionQueue(threading.Thread):
     return self._stop.isSet()
 
   def put_status(self, commands):
-    #Was supposed that we got all set of statuses, we don't need to keep old ones
-    self.statusCommandQueue.queue.clear()
+    #Clear all status commands. Was supposed that we got all set of statuses, we don't need to keep old ones
+    while not self.statusCommandQueue.empty():
+      self.statusCommandQueue.get()
 
     for command in commands:
       logger.info("Adding " + command['commandType'] + " for component " + \
@@ -152,7 +155,7 @@ class ActionQueue(threading.Thread):
     try:
       while not self.stopped():
         self.processBackgroundQueueSafeEmpty();
-        self.processStatusCommandQueueSafeEmpty();
+        self.processStatusCommandResultQueueSafeEmpty();
         try:
           if self.parallel_execution == 0:
             command = self.commandQueue.get(True, self.EXECUTION_COMMAND_WAIT_TIME)
@@ -196,15 +199,14 @@ class ActionQueue(threading.Thread):
       except Queue.Empty:
         pass
 
-  def processStatusCommandQueueSafeEmpty(self):
-    while not self.statusCommandQueue.empty():
+  def processStatusCommandResultQueueSafeEmpty(self):
+    while not self.statusCommandResultQueue.empty():
       try:
-        command = self.statusCommandQueue.get(False)
-        self.process_command(command)
+        result = self.statusCommandResultQueue.get(False)
+        self.process_status_command_result(result)
       except Queue.Empty:
         pass
 
-
   def createCommandHandle(self, command):
     if command.has_key('__handle'):
       raise AgentException("Command already has __handle")
@@ -224,8 +226,6 @@ class ActionQueue(threading.Thread):
         finally:
           if self.controller.recovery_manager.enabled():
             self.controller.recovery_manager.stop_execution_command()
-      elif commandType == self.STATUS_COMMAND:
-        self.execute_status_command(command)
       else:
         logger.error("Unrecognized command " + pprint.pformat(command))
     except Exception:
@@ -487,11 +487,12 @@ class ActionQueue(threading.Thread):
 
     self.commandStatuses.put_command_status(handle.command, roleResult)
 
-  def execute_status_command(self, command):
+  def process_status_command_result(self, result):
     '''
     Executes commands of type STATUS_COMMAND
     '''
     try:
+      command, component_status_result, component_security_status_result = result
       cluster = command['clusterName']
       service = command['serviceName']
       component = command['componentName']
@@ -506,11 +507,6 @@ class ActionQueue(threading.Thread):
 
       component_extra = None
 
-      # For custom services, responsibility to determine service status is
-      # delegated to python scripts
-      component_status_result = self.customServiceOrchestrator.requestComponentStatus(command)
-      component_security_status_result = self.customServiceOrchestrator.requestComponentSecurityState(command)
-
       if component_status_result['exitcode'] == 0:
         component_status = LiveStatus.LIVE_STATUS
         if self.controller.recovery_manager.enabled() \

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

@@ -39,6 +39,7 @@ import AmbariConfig
 from ambari_agent.Heartbeat import Heartbeat
 from ambari_agent.Register import Register
 from ambari_agent.ActionQueue import ActionQueue
+from ambari_agent.StatusCommandsExecutor import StatusCommandsExecutor
 from ambari_agent.FileCache import FileCache
 from ambari_agent.NetUtil import NetUtil
 from ambari_agent.LiveStatus import LiveStatus
@@ -83,6 +84,7 @@ class Controller(threading.Thread):
     self.cachedconnect = None
     self.range = range
     self.hasMappedComponents = True
+    self.statusCommandsExecutor = None
     # Event is used for synchronizing heartbeat iterations (to make possible
     # manual wait() interruption between heartbeats )
     self.heartbeat_stop_callback = heartbeat_stop_callback
@@ -441,10 +443,18 @@ class Controller(threading.Thread):
         logger.info("Stop event received")
         self.DEBUG_STOP_HEARTBEATING=True
 
+  def spawnStatusCommandsExecutorProcess(self):
+    self.statusCommandsExecutor = StatusCommandsExecutor(self.config, self.actionQueue)
+    self.statusCommandsExecutor.start()
+
+  def getStatusCommandsExecutor(self):
+    return self.statusCommandsExecutor
+
   def run(self):
     try:
       self.actionQueue = ActionQueue(self.config, controller=self)
       self.actionQueue.start()
+      self.spawnStatusCommandsExecutorProcess()
       self.register = Register(self.config)
       self.heartbeat = Heartbeat(self.actionQueue, self.config, self.alert_scheduler_handler.collector())
   

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
@@ -76,6 +78,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 +92,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/main/python/ambari_agent/StatusCommandsExecutor.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python
+'''
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+'''
+
+import os
+import signal
+import threading
+import logging
+import multiprocessing
+from PythonReflectiveExecutor import PythonReflectiveExecutor
+
+logger = logging.getLogger(__name__)
+
+class StatusCommandsExecutor(multiprocessing.Process):
+  """
+  A process which executes status/security status commands.
+
+  It dies and respawns itself on timeout of the command. Which is the most graceful way to end the currently running status command.
+  """
+  def __init__(self, config, actionQueue):
+    multiprocessing.Process.__init__(self)
+
+    self.config = config
+    self.actionQueue = actionQueue
+
+    self.status_command_timeout = int(self.config.get('agent', 'status_command_timeout', 5)) # in seconds
+    self.hasTimeoutedEvent = multiprocessing.Event()
+
+  def run(self):
+    try:
+      while True:
+        command = self.actionQueue.statusCommandQueue.get(True) # blocks until status status command appears
+        
+        timeout_timer = threading.Timer( self.status_command_timeout, self.respawn, [command])
+        timeout_timer.start()
+
+        self.process_status_command(command)
+
+        timeout_timer.cancel()
+    except:
+      logger.exception("StatusCommandsExecutor process failed with exception:")
+      raise
+
+    logger.warn("StatusCommandsExecutor process has finished")
+
+  def process_status_command(self, command):
+    component_status_result = self.actionQueue.customServiceOrchestrator.requestComponentStatus(command)
+    component_security_status_result = self.actionQueue.customServiceOrchestrator.requestComponentSecurityState(command)
+    result = (command, component_status_result, component_security_status_result)
+
+    self.actionQueue.statusCommandResultQueue.put(result)
+
+  def respawn(self, command):
+    try:
+      # Force context to reset to normal. By context we mean sys.path, imports, 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. Terminating it due to timeout.".format(command['commandType'], command['componentName'], self.status_command_timeout))
+
+      self.hasTimeoutedEvent.set()
+    except:
+      logger.exception("StatusCommandsExecutor.finish thread failed with exception:")
+      raise
+
+  def kill(self):
+    os.kill(self.pid, signal.SIGKILL)

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

@@ -264,6 +264,22 @@ def reset_agent(options):
 
 MAX_RETRIES = 10
 
+def run_threads(server_hostname, heartbeat_stop_callback):
+  # Launch Controller communication
+  controller = Controller(config, server_hostname, heartbeat_stop_callback)
+  controller.start()
+  while controller.is_alive():
+    time.sleep(0.1)
+
+    if controller.getStatusCommandsExecutor() is not None and (not controller.getStatusCommandsExecutor().is_alive() or controller.getStatusCommandsExecutor().hasTimeoutedEvent.is_set()):
+      if controller.getStatusCommandsExecutor().is_alive():
+        logger.info("Terminating statusCommandsExecutor")
+        controller.getStatusCommandsExecutor().kill()
+      logger.info("Respawning statusCommandsExecutor")
+      controller.spawnStatusCommandsExecutorProcess()
+
+  controller.getStatusCommandsExecutor().kill()
+
 # event - event, that will be passed to Controller and NetUtil to make able to interrupt loops form outside process
 # we need this for windows os, where no sigterm available
 def main(heartbeat_stop_callback=None):
@@ -360,10 +376,7 @@ def main(heartbeat_stop_callback=None):
         # Set the active server
         active_server = server_hostname
         # Launch Controller communication
-        controller = Controller(config, server_hostname, heartbeat_stop_callback)
-        controller.start()
-        while controller.is_alive():
-          time.sleep(0.1)
+        run_threads(server_hostname, heartbeat_stop_callback)
 
       #
       # If Ambari Agent connected to the server or

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

@@ -310,9 +310,7 @@ class TestActionQueue(TestCase):
   @patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value))
   @patch("logging.RootLogger.exception")
   @patch.object(ActionQueue, "execute_command")
-  @patch.object(ActionQueue, "execute_status_command")
-  def test_process_command(self, execute_status_command_mock,
-                           execute_command_mock, log_exc_mock):
+  def test_process_command(self, execute_command_mock, log_exc_mock):
     dummy_controller = MagicMock()
     config = AmbariConfig()
     config.set('agent', 'tolerate_download_failures', "true")
@@ -329,29 +327,19 @@ class TestActionQueue(TestCase):
     # Try wrong command
     actionQueue.process_command(wrong_command)
     self.assertFalse(execute_command_mock.called)
-    self.assertFalse(execute_status_command_mock.called)
     self.assertFalse(log_exc_mock.called)
 
     execute_command_mock.reset_mock()
-    execute_status_command_mock.reset_mock()
     log_exc_mock.reset_mock()
     # Try normal execution
     actionQueue.process_command(execution_command)
     self.assertTrue(execute_command_mock.called)
-    self.assertFalse(execute_status_command_mock.called)
     self.assertFalse(log_exc_mock.called)
 
     execute_command_mock.reset_mock()
-    execute_status_command_mock.reset_mock()
     log_exc_mock.reset_mock()
 
-    actionQueue.process_command(status_command)
-    self.assertFalse(execute_command_mock.called)
-    self.assertTrue(execute_status_command_mock.called)
-    self.assertFalse(log_exc_mock.called)
-
     execute_command_mock.reset_mock()
-    execute_status_command_mock.reset_mock()
     log_exc_mock.reset_mock()
 
     # Try exception to check proper logging
@@ -363,7 +351,6 @@ class TestActionQueue(TestCase):
 
     log_exc_mock.reset_mock()
 
-    execute_status_command_mock.side_effect = side_effect
     actionQueue.process_command(execution_command)
     self.assertTrue(log_exc_mock.called)
 
@@ -835,14 +822,11 @@ class TestActionQueue(TestCase):
 
   @patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value))
   @patch.object(ActionQueue, "status_update_callback")
-  @patch.object(CustomServiceOrchestrator, "requestComponentStatus")
-  @patch.object(CustomServiceOrchestrator, "requestComponentSecurityState")
   @patch.object(ActionQueue, "execute_command")
   @patch.object(LiveStatus, "build")
   @patch.object(CustomServiceOrchestrator, "__init__")
   def test_execute_status_command(self, CustomServiceOrchestrator_mock,
-                                  build_mock, execute_command_mock, requestComponentSecurityState_mock,
-                                  requestComponentStatus_mock,
+                                  build_mock, execute_command_mock,
                                   status_update_callback):
     CustomServiceOrchestrator_mock.return_value = None
     dummy_controller = MagicMock()
@@ -852,33 +836,25 @@ class TestActionQueue(TestCase):
 
     dummy_controller.recovery_manager = RecoveryManager(tempfile.mktemp())
 
-    requestComponentStatus_mock.reset_mock()
-    requestComponentStatus_mock.return_value = {'exitcode': 0 }
+    result = (self.status_command, {'exitcode': 0 }, 'UNKNOWN')
 
-    requestComponentSecurityState_mock.reset_mock()
-    requestComponentSecurityState_mock.return_value = 'UNKNOWN'
-
-    actionQueue.execute_status_command(self.status_command)
+    actionQueue.process_status_command_result(result)
     report = actionQueue.result()
     expected = {'dummy report': '',
                 'securityState' : 'UNKNOWN'}
 
     self.assertEqual(len(report['componentStatus']), 1)
     self.assertEqual(report['componentStatus'][0], expected)
-    self.assertTrue(requestComponentStatus_mock.called)
 
   @patch.object(RecoveryManager, "command_exists")
   @patch.object(RecoveryManager, "requires_recovery")
   @patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value))
   @patch.object(ActionQueue, "status_update_callback")
-  @patch.object(CustomServiceOrchestrator, "requestComponentStatus")
-  @patch.object(CustomServiceOrchestrator, "requestComponentSecurityState")
   @patch.object(ActionQueue, "execute_command")
   @patch.object(LiveStatus, "build")
   @patch.object(CustomServiceOrchestrator, "__init__")
-  def test_execute_status_command_recovery(self, CustomServiceOrchestrator_mock,
-                                  build_mock, execute_command_mock, requestComponentSecurityState_mock,
-                                  requestComponentStatus_mock,
+  def test_process_status_command_result_recovery(self, CustomServiceOrchestrator_mock,
+                                  build_mock, execute_command_mock,
                                   status_update_callback, requires_recovery_mock,
                                   command_exists_mock):
     CustomServiceOrchestrator_mock.return_value = None
@@ -891,13 +867,9 @@ class TestActionQueue(TestCase):
 
     dummy_controller.recovery_manager = RecoveryManager(tempfile.mktemp(), True, False)
 
-    requestComponentStatus_mock.reset_mock()
-    requestComponentStatus_mock.return_value = {'exitcode': 0 }
-
-    requestComponentSecurityState_mock.reset_mock()
-    requestComponentSecurityState_mock.return_value = 'UNKNOWN'
+    result = (self.status_command, {'exitcode': 0 }, 'UNKNOWN')
 
-    actionQueue.execute_status_command(self.status_command)
+    actionQueue.process_status_command_result(result)
     report = actionQueue.result()
     expected = {'dummy report': '',
                 'securityState' : 'UNKNOWN',
@@ -905,17 +877,13 @@ class TestActionQueue(TestCase):
 
     self.assertEqual(len(report['componentStatus']), 1)
     self.assertEqual(report['componentStatus'][0], expected)
-    self.assertTrue(requestComponentStatus_mock.called)
 
     requires_recovery_mock.return_value = True
     command_exists_mock.return_value = True
-    requestComponentStatus_mock.reset_mock()
-    requestComponentStatus_mock.return_value = {'exitcode': 0 }
-
-    requestComponentSecurityState_mock.reset_mock()
-    requestComponentSecurityState_mock.return_value = 'UNKNOWN'
+    
+    result = (self.status_command, {'exitcode': 0 }, 'UNKNOWN')
 
-    actionQueue.execute_status_command(self.status_command)
+    actionQueue.process_status_command_result(result)
     report = actionQueue.result()
     expected = {'dummy report': '',
                 'securityState' : 'UNKNOWN',
@@ -923,39 +891,33 @@ class TestActionQueue(TestCase):
 
     self.assertEqual(len(report['componentStatus']), 1)
     self.assertEqual(report['componentStatus'][0], expected)
-    self.assertTrue(requestComponentStatus_mock.called)
 
   @patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value))
   @patch.object(ActionQueue, "status_update_callback")
-  @patch.object(CustomServiceOrchestrator, "requestComponentStatus")
-  @patch.object(CustomServiceOrchestrator, "requestComponentSecurityState")
   @patch.object(ActionQueue, "execute_command")
   @patch.object(LiveStatus, "build")
   @patch.object(CustomServiceOrchestrator, "__init__")
-  def test_execute_status_command_with_alerts(self, CustomServiceOrchestrator_mock,
-                                              requestComponentSecurityState_mock,
+  def test_process_status_command_result_with_alerts(self, CustomServiceOrchestrator_mock,
                                   build_mock, execute_command_mock,
-                                  requestComponentStatus_mock,
                                   status_update_callback):
     CustomServiceOrchestrator_mock.return_value = None
     dummy_controller = MagicMock()
     actionQueue = ActionQueue(AmbariConfig(), dummy_controller)
-
-
-    requestComponentStatus_mock.reset_mock()
-    requestComponentStatus_mock.return_value = {
+    command_return_value = {
       'exitcode': 0,
       'stdout': 'out',
       'stderr': 'err',
       'structuredOut': {'alerts': [ {'name': 'flume_alert'} ] }
     }
+    
+    result = (self.status_command_for_alerts, command_return_value, command_return_value)
+    
     build_mock.return_value = {'somestatusresult': 'aresult'}
 
-    actionQueue.execute_status_command(self.status_command_for_alerts)
+    actionQueue.process_status_command_result(result)
 
     report = actionQueue.result()
 
-    self.assertTrue(requestComponentStatus_mock.called)
     self.assertEqual(len(report['componentStatus']), 1)
     self.assertTrue(report['componentStatus'][0].has_key('alerts'))
 
@@ -1215,7 +1177,7 @@ class TestActionQueue(TestCase):
     execute_command = copy.deepcopy(self.background_command)
     actionQueue.put([execute_command])
     actionQueue.processBackgroundQueueSafeEmpty();
-    actionQueue.processStatusCommandQueueSafeEmpty();
+    actionQueue.processStatusCommandResultQueueSafeEmpty();
     
     #assert that python execturor start
     self.assertTrue(runCommand_mock.called)
@@ -1259,7 +1221,7 @@ class TestActionQueue(TestCase):
                                                                  None, command_complete_w)
     actionQueue.put([self.background_command])
     actionQueue.processBackgroundQueueSafeEmpty();
-    actionQueue.processStatusCommandQueueSafeEmpty();
+    actionQueue.processStatusCommandResultQueueSafeEmpty();
     
     with lock:
       complete_done.wait(0.1)

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

@@ -44,6 +44,7 @@ import ambari_commons
 
 @not_for_platform(PLATFORM_WINDOWS)
 @patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value))
+@patch.object(Controller.Controller, "spawnStatusCommandsExecutorProcess", new = MagicMock())
 class TestController(unittest.TestCase):
 
   logger = logging.getLogger()

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

@@ -322,6 +322,7 @@ class TestMain(unittest.TestCase):
   @patch.object(Controller, "__init__")
   @patch.object(Controller, "is_alive")
   @patch.object(Controller, "start")
+  @patch.object(Controller, "getStatusCommandsExecutor")
   @patch("optparse.OptionParser.parse_args")
   @patch.object(DataCleaner,"start")
   @patch.object(DataCleaner,"__init__")
@@ -330,7 +331,7 @@ class TestMain(unittest.TestCase):
   @patch.object(ExitHelper,"execute_cleanup")
   @patch.object(ExitHelper, "exit")
   def test_main(self, exithelper_exit_mock, cleanup_mock, ping_port_init_mock, ping_port_start_mock, data_clean_init_mock,data_clean_start_mock,
-                parse_args_mock, start_mock, Controller_is_alive_mock, Controller_init_mock, try_to_connect_mock,
+                parse_args_mock, start_mock, Controller_getStatusCommandsExecutor, Controller_is_alive_mock, Controller_init_mock, try_to_connect_mock,
                 update_log_level_mock, daemonize_mock, perform_prestart_checks_mock,
                 ambari_config_mock,
                 stop_mock, bind_signal_handlers_mock,