You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by jo...@apache.org on 2016/11/10 16:58:40 UTC
[12/50] [abbrv] ambari git commit: AMBARI-18505. Ambari Status
commands should enforce a timeout < heartbeat interval (aonishuk)
AMBARI-18505. Ambari Status commands should enforce a timeout < heartbeat interval (aonishuk)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/e158756e
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/e158756e
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/e158756e
Branch: refs/heads/branch-feature-AMBARI-18456
Commit: e158756e1c778a99e6f016fbd8bdcd7ce1033dec
Parents: ffad4a3
Author: Andrew Onishuk <ao...@hortonworks.com>
Authored: Tue Nov 8 11:32:46 2016 +0200
Committer: Andrew Onishuk <ao...@hortonworks.com>
Committed: Tue Nov 8 11:32:46 2016 +0200
----------------------------------------------------------------------
ambari-agent/conf/unix/ambari-agent.ini | 1 +
.../src/main/python/ambari_agent/ActionQueue.py | 36 ++++----
.../src/main/python/ambari_agent/Controller.py | 10 +++
.../ambari_agent/PythonReflectiveExecutor.py | 20 +++--
.../ambari_agent/StatusCommandsExecutor.py | 91 ++++++++++++++++++++
.../src/main/python/ambari_agent/main.py | 21 ++++-
.../test/python/ambari_agent/TestActionQueue.py | 76 ++++------------
.../test/python/ambari_agent/TestController.py | 1 +
.../src/test/python/ambari_agent/TestMain.py | 3 +-
9 files changed, 175 insertions(+), 84 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/conf/unix/ambari-agent.ini
----------------------------------------------------------------------
diff --git a/ambari-agent/conf/unix/ambari-agent.ini b/ambari-agent/conf/unix/ambari-agent.ini
index 914e09a..9a0b537 100644
--- a/ambari-agent/conf/unix/ambari-agent.ini
+++ b/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
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/main/python/ambari_agent/ActionQueue.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/ActionQueue.py b/ambari-agent/src/main/python/ambari_agent/ActionQueue.py
index f104939..a9f1084 100644
--- a/ambari-agent/src/main/python/ambari_agent/ActionQueue.py
+++ b/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,14 +199,19 @@ 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
-
+ 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
+ except UnicodeDecodeError:
+ pass
def createCommandHandle(self, command):
if command.has_key('__handle'):
@@ -224,8 +232,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 +493,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 +513,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() \
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/main/python/ambari_agent/Controller.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/Controller.py b/ambari-agent/src/main/python/ambari_agent/Controller.py
index 2a4d384..cece30c 100644
--- a/ambari-agent/src/main/python/ambari_agent/Controller.py
+++ b/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())
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/main/python/ambari_agent/PythonReflectiveExecutor.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/PythonReflectiveExecutor.py b/ambari-agent/src/main/python/ambari_agent/PythonReflectiveExecutor.py
index 655b2fc..b27d7d1 100644
--- a/ambari-agent/src/main/python/ambari_agent/PythonReflectiveExecutor.py
+++ b/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)
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/main/python/ambari_agent/StatusCommandsExecutor.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/StatusCommandsExecutor.py b/ambari-agent/src/main/python/ambari_agent/StatusCommandsExecutor.py
new file mode 100644
index 0000000..5d3607a
--- /dev/null
+++ b/ambari-agent/src/main/python/ambari_agent/StatusCommandsExecutor.py
@@ -0,0 +1,91 @@
+#!/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()
+ self.actionQueue.statusCommandResultQueue = multiprocessing.Queue()
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/main/python/ambari_agent/main.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/main.py b/ambari-agent/src/main/python/ambari_agent/main.py
index a5edfe3..04e0c2c 100644
--- a/ambari-agent/src/main/python/ambari_agent/main.py
+++ b/ambari-agent/src/main/python/ambari_agent/main.py
@@ -294,6 +294,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):
@@ -400,10 +416,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
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py b/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py
index 7d04d42..4a63f7c 100644
--- a/ambari-agent/src/test/python/ambari_agent/TestActionQueue.py
+++ b/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)
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/test/python/ambari_agent/TestController.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/TestController.py b/ambari-agent/src/test/python/ambari_agent/TestController.py
index 59b41cd..b47af03 100644
--- a/ambari-agent/src/test/python/ambari_agent/TestController.py
+++ b/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()
http://git-wip-us.apache.org/repos/asf/ambari/blob/e158756e/ambari-agent/src/test/python/ambari_agent/TestMain.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/TestMain.py b/ambari-agent/src/test/python/ambari_agent/TestMain.py
index 01028ec..6f38410 100644
--- a/ambari-agent/src/test/python/ambari_agent/TestMain.py
+++ b/ambari-agent/src/test/python/ambari_agent/TestMain.py
@@ -340,6 +340,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__")
@@ -348,7 +349,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,