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

This reverts commit 6489987328e3f2f87e31b40dc80e39c12a7ce1bd.

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

@@ -32,7 +32,6 @@ 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,7 +18,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 '''
 import Queue
-import multiprocessing
 
 import logging
 import traceback
@@ -75,8 +74,7 @@ class ActionQueue(threading.Thread):
   def __init__(self, config, controller):
     super(ActionQueue, self).__init__()
     self.commandQueue = Queue.Queue()
-    self.statusCommandQueue = multiprocessing.Queue()
-    self.statusCommandResultQueue = multiprocessing.Queue() # this queue is filled by StatuCommandsExecutor.
+    self.statusCommandQueue = Queue.Queue()
     self.backgroundCommandQueue = Queue.Queue()
     self.commandStatuses = CommandStatusDict(callback_action =
       self.status_update_callback)
@@ -97,9 +95,8 @@ class ActionQueue(threading.Thread):
     return self._stop.isSet()
 
   def put_status(self, commands):
-    #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()
+    #Was supposed that we got all set of statuses, we don't need to keep old ones
+    self.statusCommandQueue.queue.clear()
 
     for command in commands:
       logger.info("Adding " + command['commandType'] + " for component " + \
@@ -155,7 +152,7 @@ class ActionQueue(threading.Thread):
     try:
       while not self.stopped():
         self.processBackgroundQueueSafeEmpty();
-        self.processStatusCommandResultQueueSafeEmpty();
+        self.processStatusCommandQueueSafeEmpty();
         try:
           if self.parallel_execution == 0:
             command = self.commandQueue.get(True, self.EXECUTION_COMMAND_WAIT_TIME)
@@ -199,17 +196,14 @@ class ActionQueue(threading.Thread):
       except Queue.Empty:
         pass
 
-  def processStatusCommandResultQueueSafeEmpty(self):
-    while not self.statusCommandResultQueue.empty():
+  def processStatusCommandQueueSafeEmpty(self):
+    while not self.statusCommandQueue.empty():
       try:
-        result = self.statusCommandResultQueue.get(False)
-        self.process_status_command_result(result)
+        command = self.statusCommandQueue.get(False)
+        self.process_command(command)
       except Queue.Empty:
         pass
-      except IOError:
-        # on race condition in multiprocessing.Queue if get/put and thread kill are executed at the same time.
-        # During queue.close IOError will be thrown (this prevents from permanently dead-locked get).
-        pass
+
 
   def createCommandHandle(self, command):
     if command.has_key('__handle'):
@@ -230,6 +224,8 @@ 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:
@@ -491,12 +487,11 @@ class ActionQueue(threading.Thread):
 
     self.commandStatuses.put_command_status(handle.command, roleResult)
 
-  def process_status_command_result(self, result):
+  def execute_status_command(self, command):
     '''
     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']
@@ -511,6 +506,11 @@ 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,7 +39,6 @@ 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
@@ -84,7 +83,6 @@ 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
@@ -443,18 +441,10 @@ 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,9 +53,7 @@ class PythonReflectiveExecutor(PythonExecutor):
     returncode = 1
 
     try:
-      current_context = PythonContext(script_dir, pythonCommand)
-      PythonReflectiveExecutor.last_context = current_context
-      with current_context:
+      with PythonContext(script_dir, pythonCommand):
         imp.load_source('__main__', script)
     except SystemExit as e:
       returncode = e.code
@@ -78,8 +76,6 @@ 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)
@@ -92,18 +88,12 @@ class PythonContext:
     sys.argv = self.pythonCommand[1:]
 
   def __exit__(self, exc_type, exc_val, exc_tb):
-    self.revert(is_forced_revert=False)
+    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)
     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

@@ -1,90 +0,0 @@
-#!/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 ambari_agent.PythonReflectiveExecutor import PythonReflectiveExecutor
-from ambari_agent.RemoteDebugUtils import bind_debug_signal_handlers
-
-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:
-      bind_debug_signal_handlers()
-      while True:
-        command = self.actionQueue.statusCommandQueue.get(True) # blocks until status status command appears
-        logger.debug("Running status command for {0}".format(command['componentName']))
-        
-        timeout_timer = threading.Timer( self.status_command_timeout, self.respawn, [command])
-        timeout_timer.start()
-
-        self.process_status_command(command)
-
-        timeout_timer.cancel()
-        logger.debug("Completed status command for {0}".format(command['componentName']))
-    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:
-      if hasattr(PythonReflectiveExecutor, "last_context"):
-        # 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)
-
-    # prevent queue from ending up with non-freed semaphores, locks during put. Which would result in dead-lock in process executing get.
-    self.actionQueue.statusCommandResultQueue.close()
-    self.actionQueue.statusCommandResultQueue.join_thread()

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

@@ -276,22 +276,6 @@ 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):
@@ -396,7 +380,10 @@ def main(heartbeat_stop_callback=None):
         # Set the active server
         active_server = server_hostname
         # Launch Controller communication
-        run_threads(server_hostname, heartbeat_stop_callback)
+        controller = Controller(config, server_hostname, heartbeat_stop_callback)
+        controller.start()
+        while controller.is_alive():
+          time.sleep(0.1)
 
       #
       # If Ambari Agent connected to the server or

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

@@ -310,7 +310,9 @@ class TestActionQueue(TestCase):
   @patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value))
   @patch("logging.RootLogger.exception")
   @patch.object(ActionQueue, "execute_command")
-  def test_process_command(self, execute_command_mock, log_exc_mock):
+  @patch.object(ActionQueue, "execute_status_command")
+  def test_process_command(self, execute_status_command_mock,
+                           execute_command_mock, log_exc_mock):
     dummy_controller = MagicMock()
     config = AmbariConfig()
     config.set('agent', 'tolerate_download_failures', "true")
@@ -327,19 +329,29 @@ 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
@@ -351,6 +363,7 @@ 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)
 
@@ -822,11 +835,14 @@ 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,
+                                  build_mock, execute_command_mock, requestComponentSecurityState_mock,
+                                  requestComponentStatus_mock,
                                   status_update_callback):
     CustomServiceOrchestrator_mock.return_value = None
     dummy_controller = MagicMock()
@@ -836,25 +852,33 @@ class TestActionQueue(TestCase):
 
     dummy_controller.recovery_manager = RecoveryManager(tempfile.mktemp())
 
-    result = (self.status_command, {'exitcode': 0 }, 'UNKNOWN')
+    requestComponentStatus_mock.reset_mock()
+    requestComponentStatus_mock.return_value = {'exitcode': 0 }
 
-    actionQueue.process_status_command_result(result)
+    requestComponentSecurityState_mock.reset_mock()
+    requestComponentSecurityState_mock.return_value = 'UNKNOWN'
+
+    actionQueue.execute_status_command(self.status_command)
     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_process_status_command_result_recovery(self, CustomServiceOrchestrator_mock,
-                                  build_mock, execute_command_mock,
+  def test_execute_status_command_recovery(self, CustomServiceOrchestrator_mock,
+                                  build_mock, execute_command_mock, requestComponentSecurityState_mock,
+                                  requestComponentStatus_mock,
                                   status_update_callback, requires_recovery_mock,
                                   command_exists_mock):
     CustomServiceOrchestrator_mock.return_value = None
@@ -867,9 +891,13 @@ class TestActionQueue(TestCase):
 
     dummy_controller.recovery_manager = RecoveryManager(tempfile.mktemp(), True, False)
 
-    result = (self.status_command, {'exitcode': 0 }, 'UNKNOWN')
+    requestComponentStatus_mock.reset_mock()
+    requestComponentStatus_mock.return_value = {'exitcode': 0 }
+
+    requestComponentSecurityState_mock.reset_mock()
+    requestComponentSecurityState_mock.return_value = 'UNKNOWN'
 
-    actionQueue.process_status_command_result(result)
+    actionQueue.execute_status_command(self.status_command)
     report = actionQueue.result()
     expected = {'dummy report': '',
                 'securityState' : 'UNKNOWN',
@@ -877,13 +905,17 @@ 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
-    
-    result = (self.status_command, {'exitcode': 0 }, 'UNKNOWN')
+    requestComponentStatus_mock.reset_mock()
+    requestComponentStatus_mock.return_value = {'exitcode': 0 }
+
+    requestComponentSecurityState_mock.reset_mock()
+    requestComponentSecurityState_mock.return_value = 'UNKNOWN'
 
-    actionQueue.process_status_command_result(result)
+    actionQueue.execute_status_command(self.status_command)
     report = actionQueue.result()
     expected = {'dummy report': '',
                 'securityState' : 'UNKNOWN',
@@ -891,33 +923,39 @@ 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_process_status_command_result_with_alerts(self, CustomServiceOrchestrator_mock,
+  def test_execute_status_command_with_alerts(self, CustomServiceOrchestrator_mock,
+                                              requestComponentSecurityState_mock,
                                   build_mock, execute_command_mock,
+                                  requestComponentStatus_mock,
                                   status_update_callback):
     CustomServiceOrchestrator_mock.return_value = None
     dummy_controller = MagicMock()
     actionQueue = ActionQueue(AmbariConfig(), dummy_controller)
-    command_return_value = {
+
+
+    requestComponentStatus_mock.reset_mock()
+    requestComponentStatus_mock.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.process_status_command_result(result)
+    actionQueue.execute_status_command(self.status_command_for_alerts)
 
     report = actionQueue.result()
 
+    self.assertTrue(requestComponentStatus_mock.called)
     self.assertEqual(len(report['componentStatus']), 1)
     self.assertTrue(report['componentStatus'][0].has_key('alerts'))
 
@@ -1177,7 +1215,7 @@ class TestActionQueue(TestCase):
     execute_command = copy.deepcopy(self.background_command)
     actionQueue.put([execute_command])
     actionQueue.processBackgroundQueueSafeEmpty();
-    actionQueue.processStatusCommandResultQueueSafeEmpty();
+    actionQueue.processStatusCommandQueueSafeEmpty();
     
     #assert that python execturor start
     self.assertTrue(runCommand_mock.called)
@@ -1221,7 +1259,7 @@ class TestActionQueue(TestCase):
                                                                  None, command_complete_w)
     actionQueue.put([self.background_command])
     actionQueue.processBackgroundQueueSafeEmpty();
-    actionQueue.processStatusCommandResultQueueSafeEmpty();
+    actionQueue.processStatusCommandQueueSafeEmpty();
     
     with lock:
       complete_done.wait(0.1)

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

@@ -44,7 +44,6 @@ 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

@@ -324,7 +324,6 @@ 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__")
@@ -333,7 +332,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_getStatusCommandsExecutor, Controller_is_alive_mock, Controller_init_mock, try_to_connect_mock,
+                parse_args_mock, start_mock, 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,