Repository: ambari Updated Branches: refs/heads/branch-3.0-perf e0e882506 -> db83ccd82
AMBARI-21646. Race condition causing execution commands to fail (aonishuk) Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/db83ccd8 Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/db83ccd8 Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/db83ccd8 Branch: refs/heads/branch-3.0-perf Commit: db83ccd8224289e80ca2da5dc95c613ccbaf2492 Parents: e0e8825 Author: Andrew Onishuk <[email protected]> Authored: Thu Aug 3 14:48:20 2017 +0300 Committer: Andrew Onishuk <[email protected]> Committed: Thu Aug 3 14:48:20 2017 +0300 ---------------------------------------------------------------------- .../main/python/ambari_agent/CommandStatusDict.py | 16 +++++++++++++--- .../python/ambari_agent/CommandStatusReporter.py | 9 +-------- 2 files changed, 14 insertions(+), 11 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/db83ccd8/ambari-agent/src/main/python/ambari_agent/CommandStatusDict.py ---------------------------------------------------------------------- diff --git a/ambari-agent/src/main/python/ambari_agent/CommandStatusDict.py b/ambari-agent/src/main/python/ambari_agent/CommandStatusDict.py index f6d304c..d6cbdcc 100644 --- a/ambari-agent/src/main/python/ambari_agent/CommandStatusDict.py +++ b/ambari-agent/src/main/python/ambari_agent/CommandStatusDict.py @@ -56,10 +56,20 @@ class CommandStatusDict(): self.current_state[key] = (command, new_report) self.reported_reports.discard(key) - self.force_update_to_server(command['clusterId'], new_report) + self.force_update_to_server({command['clusterId']: [new_report]}) - def force_update_to_server(self, cluster_id, report): - self.initializer_module.connection.send(message={'clusters':{cluster_id: [report]}}, destination=Constants.COMMANDS_STATUS_REPORTS_ENDPOINT) + def force_update_to_server(self, reports_dict): + if self.initializer_module.is_registered: + self.initializer_module.connection.send(message={'clusters':reports_dict}, destination=Constants.COMMANDS_STATUS_REPORTS_ENDPOINT) + return True + + return False + + def report(self): + report = self.generate_report() + + if report and self.force_update_to_server(report): + self.clear_reported_reports() def get_command_status(self, taskId): with self.lock: http://git-wip-us.apache.org/repos/asf/ambari/blob/db83ccd8/ambari-agent/src/main/python/ambari_agent/CommandStatusReporter.py ---------------------------------------------------------------------- diff --git a/ambari-agent/src/main/python/ambari_agent/CommandStatusReporter.py b/ambari-agent/src/main/python/ambari_agent/CommandStatusReporter.py index 5e5eb0d..f46c47a 100644 --- a/ambari-agent/src/main/python/ambari_agent/CommandStatusReporter.py +++ b/ambari-agent/src/main/python/ambari_agent/CommandStatusReporter.py @@ -21,8 +21,6 @@ limitations under the License. import logging import threading -from ambari_agent import Constants - logger = logging.getLogger(__name__) class CommandStatusReporter(threading.Thread): @@ -43,12 +41,7 @@ class CommandStatusReporter(threading.Thread): while not self.stop_event.is_set(): try: if self.initializer_module.is_registered: - report = self.commandStatuses.generate_report() - - if report: - self.initializer_module.connection.send(message={'clusters': report}, destination=Constants.COMMANDS_STATUS_REPORTS_ENDPOINT) - - self.commandStatuses.clear_reported_reports() + self.commandStatuses.report() except: logger.exception("Exception in CommandStatusReporter. Re-running it")
