ambari-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aonis...@apache.org
Subject ambari git commit: AMBARI-21394. Create a topic to send hostLevelParams (aonishuk)
Date Tue, 04 Jul 2017 08:29:36 GMT
Repository: ambari
Updated Branches:
  refs/heads/branch-3.0-perf 51b1a14d7 -> 917898cdb


AMBARI-21394. Create a topic to send hostLevelParams (aonishuk)


Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/917898cd
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/917898cd
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/917898cd

Branch: refs/heads/branch-3.0-perf
Commit: 917898cdbed3bd37e1bbfa813b115f93dfb6dcaf
Parents: 51b1a14
Author: Andrew Onishuk <aonishuk@hortonworks.com>
Authored: Tue Jul 4 11:29:31 2017 +0300
Committer: Andrew Onishuk <aonishuk@hortonworks.com>
Committed: Tue Jul 4 11:29:31 2017 +0300

----------------------------------------------------------------------
 .../ambari_agent/ClusterHostLevelParamsCache.py | 45 +++++++++++++++
 .../src/main/python/ambari_agent/Constants.py   |  4 +-
 .../ambari_agent/CustomServiceOrchestrator.py   |  4 +-
 .../main/python/ambari_agent/HeartbeatThread.py | 14 +++--
 .../python/ambari_agent/InitializerModule.py    |  2 +
 .../main/python/ambari_agent/RecoveryManager.py | 16 ++---
 .../listeners/HostLevelParamsEventListener.py   | 61 ++++++++++++++++++++
 .../listeners/MetadataEventListener.py          | 11 +---
 .../ambari_agent/TestAgentStompResponses.py     | 20 +++++--
 .../dummy_files/stomp/host_level_params.json    | 37 ++++++++++++
 .../stomp/metadata_after_registration.json      |  9 +--
 .../dummy_files/stomp/topology_add_host.json    |  5 +-
 .../stomp/topology_cache_expected.json          |  6 --
 .../dummy_files/stomp/topology_create.json      | 15 +----
 14 files changed, 187 insertions(+), 62 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/ClusterHostLevelParamsCache.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/ClusterHostLevelParamsCache.py b/ambari-agent/src/main/python/ambari_agent/ClusterHostLevelParamsCache.py
new file mode 100644
index 0000000..3e490c5
--- /dev/null
+++ b/ambari-agent/src/main/python/ambari_agent/ClusterHostLevelParamsCache.py
@@ -0,0 +1,45 @@
+#!/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.
+"""
+
+from ambari_agent.ClusterCache import ClusterCache
+import logging
+
+logger = logging.getLogger(__name__)
+
+class ClusterHostLevelParamsCache(ClusterCache):
+  """
+  Maintains an in-memory cache and disk cache of the host level params send from server for
+  every cluster. This is useful for having quick access to any of the
+  topology properties.
+
+  Host level params. Is parameters used by execution and status commands which can be generated
+  differently for every host.
+  """
+
+  def __init__(self, cluster_cache_dir):
+    """
+    Initializes the host level params cache.
+    :param cluster_cache_dir:
+    :return:
+    """
+    super(ClusterHostLevelParamsCache, self).__init__(cluster_cache_dir)
+
+  def get_cache_name(self):
+    return 'host_level_params'

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/Constants.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/Constants.py b/ambari-agent/src/main/python/ambari_agent/Constants.py
index 02945ee..17ed2be 100644
--- a/ambari-agent/src/main/python/ambari_agent/Constants.py
+++ b/ambari-agent/src/main/python/ambari_agent/Constants.py
@@ -21,16 +21,18 @@ limitations under the License.
 
 COMMANDS_TOPIC = '/user/commands'
 CONFIGURATIONS_TOPIC = '/user/configs'
+HOST_LEVEL_PARAMS_TOPIC = '/user/host_level_params'
 METADATA_TOPIC = '/events/metadata'
 TOPOLOGIES_TOPIC = '/events/topologies'
 SERVER_RESPONSES_TOPIC = '/user/'
 
 PRE_REGISTRATION_TOPICS_TO_SUBSCRIBE = [SERVER_RESPONSES_TOPIC]
-POST_REGISTRATION_TOPICS_TO_SUBSCRIBE = [COMMANDS_TOPIC, CONFIGURATIONS_TOPIC, METADATA_TOPIC,
TOPOLOGIES_TOPIC]
+POST_REGISTRATION_TOPICS_TO_SUBSCRIBE = [COMMANDS_TOPIC, CONFIGURATIONS_TOPIC, METADATA_TOPIC,
TOPOLOGIES_TOPIC, HOST_LEVEL_PARAMS_TOPIC]
 
 TOPOLOGY_REQUEST_ENDPOINT = '/agents/topologies'
 METADATA_REQUEST_ENDPOINT = '/agents/metadata'
 CONFIGURATIONS_REQUEST_ENDPOINT = '/agents/configs'
+HOST_LEVEL_PARAMS_TOPIC_ENPOINT = '/agents/host_level_params'
 COMPONENT_STATUS_REPORTS_ENDPOINT = '/reports/component_status'
 COMMANDS_STATUS_REPORTS_ENDPOINT = '/reports/commands_status'
 HOST_STATUS_REPORTS_ENDPOINT = '/reports/host_status'

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/CustomServiceOrchestrator.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/CustomServiceOrchestrator.py b/ambari-agent/src/main/python/ambari_agent/CustomServiceOrchestrator.py
index 6d1a491..c0b20ed 100644
--- a/ambari-agent/src/main/python/ambari_agent/CustomServiceOrchestrator.py
+++ b/ambari-agent/src/main/python/ambari_agent/CustomServiceOrchestrator.py
@@ -81,6 +81,7 @@ class CustomServiceOrchestrator():
     self.metadata_cache = initializer_module.metadata_cache
     self.topology_cache = initializer_module.topology_cache
     self.configurations_cache = initializer_module.configurations_cache
+    self.host_level_params_cache = initializer_module.host_level_params_cache
     self.config = initializer_module.config
     self.tmp_dir = self.config.get('agent', 'prefix')
     self.force_https_protocol = self.config.get_force_https_protocol()
@@ -459,13 +460,14 @@ class CustomServiceOrchestrator():
 
     metadata_cache = self.metadata_cache[cluster_id]
     configurations_cache = self.configurations_cache[cluster_id]
+    host_level_params_cache = self.host_level_params_cache[cluster_id]
 
     component_dict = self.topology_cache.get_component_info_by_key(cluster_id, service_name,
component_name)
 
     command_dict = {
       'clusterLevelParams': metadata_cache.clusterLevelParams,
       'serviceLevelParams': metadata_cache.serviceLevelParams[service_name],
-      'hostLevelParams': self.topology_cache.get_current_host_info(cluster_id).hostLevelParams,
+      'hostLevelParams': host_level_params_cache,
       'componentLevelParams': component_dict.componentLevelParams,
       'script_type': self.SCRIPT_TYPE_PYTHON
     }

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py b/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
index dbf4006..6ba84e6 100644
--- a/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
+++ b/ambari-agent/src/main/python/ambari_agent/HeartbeatThread.py
@@ -31,6 +31,7 @@ from ambari_agent.listeners.TopologyEventListener import TopologyEventListener
 from ambari_agent.listeners.ConfigurationEventListener import ConfigurationEventListener
 from ambari_agent.listeners.MetadataEventListener import MetadataEventListener
 from ambari_agent.listeners.CommandsEventListener import CommandsEventListener
+from ambari_agent.listeners.HostLevelParamsEventListener import HostLevelParamsEventListener
 
 HEARTBEAT_INTERVAL = 10
 REQUEST_RESPONSE_TIMEOUT = 10
@@ -49,20 +50,21 @@ class HeartbeatThread(threading.Thread):
     self.registration_builder = Register(initializer_module.config)
 
     self.initializer_module = initializer_module
-    self.caches = [initializer_module.metadata_cache, initializer_module.topology_cache,
initializer_module.configurations_cache]
 
     # listeners
     self.server_responses_listener = ServerResponsesListener()
     self.commands_events_listener = CommandsEventListener(initializer_module.action_queue)
-    self.metadata_events_listener = MetadataEventListener(initializer_module.metadata_cache,
initializer_module.recovery_manager)
+    self.metadata_events_listener = MetadataEventListener(initializer_module.metadata_cache)
     self.topology_events_listener = TopologyEventListener(initializer_module.topology_cache)
     self.configuration_events_listener = ConfigurationEventListener(initializer_module.configurations_cache)
-    self.listeners = [self.server_responses_listener, self.commands_events_listener, self.metadata_events_listener,
self.topology_events_listener, self.configuration_events_listener]
+    self.host_level_params_events_listener = HostLevelParamsEventListener(initializer_module.host_level_params_cache,
initializer_module.recovery_manager)
+    self.listeners = [self.server_responses_listener, self.commands_events_listener, self.metadata_events_listener,
self.topology_events_listener, self.configuration_events_listener, self.host_level_params_events_listener]
 
     self.post_registration_requests = [
     (Constants.TOPOLOGY_REQUEST_ENDPOINT, initializer_module.topology_cache, self.topology_events_listener),
     (Constants.METADATA_REQUEST_ENDPOINT, initializer_module.metadata_cache, self.metadata_events_listener),
-    (Constants.CONFIGURATIONS_REQUEST_ENDPOINT, initializer_module.configurations_cache,
self.configuration_events_listener)
+    (Constants.CONFIGURATIONS_REQUEST_ENDPOINT, initializer_module.configurations_cache,
self.configuration_events_listener),
+    (Constants.HOST_LEVEL_PARAMS_TOPIC_ENPOINT, initializer_module.host_level_params_cache,
self.host_level_params_events_listener)
     ]
     self.responseId = 0
 
@@ -152,6 +154,10 @@ class HeartbeatThread(threading.Thread):
     else:
       self.responseId = serverId
 
+    if 'restartAgent' in response and response['restartAgent'].lower() == "true":
+      logger.warn("Restarting the agent by the request from server")
+      Utils.restartAgent()
+
   def get_heartbeat_body(self):
     """
     Heartbeat body to be send to server

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/InitializerModule.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/InitializerModule.py b/ambari-agent/src/main/python/ambari_agent/InitializerModule.py
index 8de1fa5..dbc26e8 100644
--- a/ambari-agent/src/main/python/ambari_agent/InitializerModule.py
+++ b/ambari-agent/src/main/python/ambari_agent/InitializerModule.py
@@ -26,6 +26,7 @@ from ambari_agent.AmbariConfig import AmbariConfig
 from ambari_agent.ClusterConfigurationCache import ClusterConfigurationCache
 from ambari_agent.ClusterTopologyCache import ClusterTopologyCache
 from ambari_agent.ClusterMetadataCache import ClusterMetadataCache
+from ambari_agent.ClusterHostLevelParamsCache import ClusterHostLevelParamsCache
 from ambari_agent.Utils import lazy_property
 from ambari_agent.security import AmbariStompConnection
 from ambari_agent.ActionQueue import ActionQueue
@@ -73,6 +74,7 @@ class InitializerModule:
     self.metadata_cache = ClusterMetadataCache(self.cluster_cache_dir)
     self.topology_cache = ClusterTopologyCache(self.cluster_cache_dir, self.config)
     self.configurations_cache = ClusterConfigurationCache(self.cluster_cache_dir)
+    self.host_level_params_cache = ClusterHostLevelParamsCache(self.cluster_cache_dir)
     self.customServiceOrchestrator = CustomServiceOrchestrator(self)
 
     self.recovery_manager = RecoveryManager(self.recovery_cache_dir)

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py b/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
index 68dd0be..19f163e 100644
--- a/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
+++ b/ambari-agent/src/main/python/ambari_agent/RecoveryManager.py
@@ -97,7 +97,6 @@ class RecoveryManager:
     self.__cache_lock = threading.RLock()
     self.active_command_count = 0
     self.paused = False
-    self.recovery_timestamp = -1
     self.cluster_id = None
 
     if not os.path.exists(cache_dir):
@@ -110,7 +109,7 @@ class RecoveryManager:
 
     self.actions = {}
 
-    self.update_config(6, 60, 5, 12, recovery_enabled, auto_start_only, auto_install_start,
"", -1)
+    self.update_config(6, 60, 5, 12, recovery_enabled, auto_start_only, auto_install_start,
"")
 
     pass
 
@@ -560,8 +559,7 @@ class RecoveryManager:
       "maxCount" : 10,
       "windowInMinutes" : 60,
       "retryGap" : 0,
-      "components" : "a,b",
-      "recoveryTimestamp" : 1458150424380
+      "components" : "a,b"
       }
     """
 
@@ -573,7 +571,6 @@ class RecoveryManager:
     retry_gap = 5
     max_lifetime_count = 12
     enabled_components = ""
-    recovery_timestamp = -1 # Default value if recoveryTimestamp is not available.
 
 
     if dictionary and "recoveryConfig" in dictionary:
@@ -599,11 +596,8 @@ class RecoveryManager:
       if 'components' in config:
         enabled_components = config['components']
 
-      if 'recoveryTimestamp' in config:
-        recovery_timestamp = config['recoveryTimestamp']
-
     self.update_config(max_count, window_in_min, retry_gap, max_lifetime_count, recovery_enabled,
auto_start_only,
-                       auto_install_start, enabled_components, recovery_timestamp)
+                       auto_install_start, enabled_components)
     pass
 
   """
@@ -617,10 +611,9 @@ class RecoveryManager:
   auto_start_only - True if AUTO_START recovery type was specified. False otherwise.
   auto_install_start - True if AUTO_INSTALL_START recovery type was specified. False otherwise.
   enabled_components - CSV of componenents enabled for auto start.
-  recovery_timestamp - Timestamp when the recovery values were last updated. -1 on start
up.
   """
   def update_config(self, max_count, window_in_min, retry_gap, max_lifetime_count, recovery_enabled,
-                    auto_start_only, auto_install_start, enabled_components, recovery_timestamp):
+                    auto_start_only, auto_install_start, enabled_components):
     """
     Update recovery configuration, recovery is disabled if configuration values
     are not correct
@@ -653,7 +646,6 @@ class RecoveryManager:
     self.auto_install_start = auto_install_start
     self.max_lifetime_count = max_lifetime_count
     self.enabled_components = []
-    self.recovery_timestamp = recovery_timestamp
 
     self.allowed_desired_states = [self.STARTED, self.INSTALLED]
     self.allowed_current_states = [self.INIT, self.INSTALL_FAILED, self.INSTALLED, self.STARTED]

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/listeners/HostLevelParamsEventListener.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/listeners/HostLevelParamsEventListener.py
b/ambari-agent/src/main/python/ambari_agent/listeners/HostLevelParamsEventListener.py
new file mode 100644
index 0000000..5aee634
--- /dev/null
+++ b/ambari-agent/src/main/python/ambari_agent/listeners/HostLevelParamsEventListener.py
@@ -0,0 +1,61 @@
+#!/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 logging
+import ambari_stomp
+
+from ambari_agent.listeners import EventListener
+from ambari_agent import Constants
+
+logger = logging.getLogger(__name__)
+
+class HostLevelParamsEventListener(EventListener):
+  """
+  Listener of Constants.HOST_LEVEL_PARAMS_TOPIC events from server.
+  """
+  def __init__(self, host_level_params_cache, recovery_manager):
+    self.host_level_params_cache = host_level_params_cache
+    self.recovery_manager = recovery_manager
+
+  def on_event(self, headers, message):
+    """
+    Is triggered when an event to Constants.CONFIGURATIONS_TOPIC topic is received from server.
+
+    @param headers: headers dictionary
+    @param message: message payload dictionary
+    """
+    # this kind of response is received if hash was identical. And server does not need to
change anything
+    if message == {}:
+      return
+
+    self.host_level_params_cache.rewrite_cache(message['clusters'])
+    self.host_level_params_cache.hash = message['hash']
+
+    if message['clusters']:
+      # FIXME: Recovery manager does not support multiple cluster as of now.
+      cluster_id = message['clusters'].keys()[0]
+
+      if 'recoveryConfig' in message['clusters'][cluster_id]:
+        logging.info("Updating recoveryConfig from metadata")
+        self.recovery_manager.update_recovery_config(self.host_level_params_cache[cluster_id])
+        self.recovery_manager.cluster_id = cluster_id
+
+  def get_handled_path(self):
+    return Constants.HOST_LEVEL_PARAMS_TOPIC
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/main/python/ambari_agent/listeners/MetadataEventListener.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/ambari_agent/listeners/MetadataEventListener.py
b/ambari-agent/src/main/python/ambari_agent/listeners/MetadataEventListener.py
index 364d8af..5802ffe 100644
--- a/ambari-agent/src/main/python/ambari_agent/listeners/MetadataEventListener.py
+++ b/ambari-agent/src/main/python/ambari_agent/listeners/MetadataEventListener.py
@@ -32,9 +32,8 @@ class MetadataEventListener(EventListener):
   """
   Listener of Constants.METADATA_TOPIC events from server.
   """
-  def __init__(self, metadata_cache, recovery_manager):
+  def __init__(self, metadata_cache):
     self.metadata_cache = metadata_cache
-    self.recovery_manager = recovery_manager
 
   def on_event(self, headers, message):
     """
@@ -50,13 +49,5 @@ class MetadataEventListener(EventListener):
     self.metadata_cache.rewrite_cache(message['clusters'])
     self.metadata_cache.hash = message['hash']
 
-    # FIXME: Recovery manager does not support multiple cluster as of now.
-    cluster_id = message['clusters'].keys()[0]
-
-    if 'recoveryConfig' in message['clusters'][cluster_id]:
-      logging.info("Updating recoveryConfig from metadata")
-      self.recovery_manager.update_recovery_config(self.metadata_cache[cluster_id])
-      self.recovery_manager.cluster_id = cluster_id
-
   def get_handled_path(self):
     return Constants.METADATA_TOPIC
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/test/python/ambari_agent/TestAgentStompResponses.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/TestAgentStompResponses.py b/ambari-agent/src/test/python/ambari_agent/TestAgentStompResponses.py
index c41f87e..feaf7dd 100644
--- a/ambari-agent/src/test/python/ambari_agent/TestAgentStompResponses.py
+++ b/ambari-agent/src/test/python/ambari_agent/TestAgentStompResponses.py
@@ -40,7 +40,7 @@ from mock.mock import MagicMock, patch
 @patch("ambari_agent.hostname.hostname", new=MagicMock(return_value="c6401.ambari.apache.org"))
 class TestAgentStompResponses(BaseStompServerTestCase):
   def setUp(self):
-    self.remove_files(['/tmp/cluster_cache/configurations.json', '/tmp/cluster_cache/metadata.json',
'/tmp/cluster_cache/topology.json'])
+    self.remove_files(['/tmp/cluster_cache/configurations.json', '/tmp/cluster_cache/metadata.json',
'/tmp/cluster_cache/topology.json', '/tmp/host_level_params.json'])
 
     if not os.path.exists("/tmp/ambari-agent"):
       os.mkdir("/tmp/ambari-agent")
@@ -83,9 +83,13 @@ class TestAgentStompResponses(BaseStompServerTestCase):
     f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '3'}, body=self.get_json("configurations_update.json"))
     self.server.topic_manager.send(f)
 
+    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '4'}, body=self.get_json("host_level_params.json"))
+    self.server.topic_manager.send(f)
+
     initial_topology_request = self.server.frames_queue.get()
     initial_metadata_request = self.server.frames_queue.get()
     initial_configs_request = self.server.frames_queue.get()
+    initial_host_level_params_request = self.server.frames_queue.get()
 
     while not initializer_module.is_registered:
       time.sleep(0.1)
@@ -103,6 +107,7 @@ class TestAgentStompResponses(BaseStompServerTestCase):
     configurations_subscribe_frame = self.server.frames_queue.get()
     metadata_subscribe_frame = self.server.frames_queue.get()
     topologies_subscribe_frame = self.server.frames_queue.get()
+    host_level_params_subscribe_frame = self.server.frames_queue.get()
     heartbeat_frame = self.server.frames_queue.get()
     dn_install_in_progress_frame = json.loads(self.server.frames_queue.get().body)
     dn_install_failed_frame = json.loads(self.server.frames_queue.get().body)
@@ -116,7 +121,7 @@ class TestAgentStompResponses(BaseStompServerTestCase):
 
     initializer_module.stop_event.set()
 
-    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '4'}, body=json.dumps({'id':'1'}))
+    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '5'}, body=json.dumps({'id':'1'}))
     self.server.topic_manager.send(f)
 
     command_status_reporter.join()
@@ -179,6 +184,9 @@ class TestAgentStompResponses(BaseStompServerTestCase):
     f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '3'}, body='{}')
     self.server.topic_manager.send(f)
 
+    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '4'}, body='{}')
+    self.server.topic_manager.send(f)
+
     commands_subscribe_frame = self.server.frames_queue.get()
     configurations_subscribe_frame = self.server.frames_queue.get()
     metadata_subscribe_frame = self.server.frames_queue.get()
@@ -188,7 +196,7 @@ class TestAgentStompResponses(BaseStompServerTestCase):
 
     initializer_module.stop_event.set()
 
-    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '4'}, body=json.dumps({'id':'1'}))
+    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '5'}, body=json.dumps({'id':'1'}))
     self.server.topic_manager.send(f)
 
     heartbeat_thread.join()
@@ -222,9 +230,13 @@ class TestAgentStompResponses(BaseStompServerTestCase):
     f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '3'}, body='{}')
     self.server.topic_manager.send(f)
 
+    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '4'}, body='{}')
+    self.server.topic_manager.send(f)
+
     initial_topology_request = self.server.frames_queue.get()
     initial_metadata_request = self.server.frames_queue.get()
     initial_configs_request = self.server.frames_queue.get()
+    initial_host_level_params_request = self.server.frames_queue.get()
 
     while not initializer_module.is_registered:
       time.sleep(0.1)
@@ -262,7 +274,7 @@ class TestAgentStompResponses(BaseStompServerTestCase):
 
     initializer_module.stop_event.set()
 
-    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '4'}, body=json.dumps({'id':'1'}))
+    f = Frame(frames.MESSAGE, headers={'destination': '/user/', 'correlationId': '5'}, body=json.dumps({'id':'1'}))
     self.server.topic_manager.send(f)
 
     heartbeat_thread.join()

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/host_level_params.json
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/host_level_params.json
b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/host_level_params.json
new file mode 100644
index 0000000..7447e93
--- /dev/null
+++ b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/host_level_params.json
@@ -0,0 +1,37 @@
+{
+  "hash":"aa4cd03688d36f18fc2cbba06614bafe",
+  "clusters":{
+    "0":{
+      "repoInfo": [
+        {
+          "defaultBaseUrl": "http://public-repo-1.hortonworks.com/HDP/centos6/2.x/updates/2.6.0.3",
+          "ambariManagedRepositories": true,
+          "baseUrl": "http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129/",
+          "latestBaseUrl": "http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129",
+          "repoSaved": true,
+          "repoName": "HDP",
+          "osType": "redhat6",
+          "unique": true,
+          "repoId": "HDP-2.6"
+        },
+        {
+          "defaultBaseUrl": "http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6",
+          "ambariManagedRepositories": true,
+          "baseUrl": "http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6",
+          "latestBaseUrl": "http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6",
+          "repoSaved": true,
+          "repoName": "HDP-UTILS",
+          "osType": "redhat6",
+          "unique": false,
+          "repoId": "HDP-UTILS-1.1.0.21"
+        }
+      ],
+      "recoveryConfig": {
+        "type" : "AUTO_INSTALL_START",
+        "maxCount" : 10,
+        "windowInMinutes" : 60,
+        "components" : "NAMENODE,DATANODE"
+      }
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/metadata_after_registration.json
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/metadata_after_registration.json
b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/metadata_after_registration.json
index 6462ccf..c1aba28 100644
--- a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/metadata_after_registration.json
+++ b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/metadata_after_registration.json
@@ -38,14 +38,7 @@
       },
       "status_commands_to_run": [
         "STATUS"
-      ],
-      "recoveryConfig": {
-        "type" : "AUTO_INSTALL_START",
-        "maxCount" : 10,
-        "windowInMinutes" : 60,
-        "components" : "NAMENODE,DATANODE",
-        "recoveryTimestamp" : 1458150424380
-      }
+      ]
     }
   }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_add_host.json
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_add_host.json
b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_add_host.json
index 2458f08..a9407c3 100644
--- a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_add_host.json
+++ b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_add_host.json
@@ -8,10 +8,7 @@
           "hostId":2,
           "hostName":"c6403.ambari.apache.org",
           "rackName":"/default-rack",
-          "ipv4":"192.168.64.103",
-          "hostLevelParams": {
-            "repo_info": "[{\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129/\",\"osType\":\"redhat6\",\"repoId\":\"HDP-2.6\",\"repoName\":\"HDP\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP/centos6/2.x/updates/2.6.0.3\",\"latestBaseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129\",\"repoSaved\":true,\"unique\":true,\"ambariManagedRepositories\":true},{\"baseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"osType\":\"redhat6\",\"repoId\":\"HDP-UTILS-1.1.0.21\",\"repoName\":\"HDP-UTILS\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"latestBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"repoSaved\":true,\"unique\":false,\"ambariManagedRepositories\":true}]"
-          }
+          "ipv4":"192.168.64.103"
         }
       ]
     }

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_cache_expected.json
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_cache_expected.json
b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_cache_expected.json
index 53d0e0d..ff2b3fd 100644
--- a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_cache_expected.json
+++ b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_cache_expected.json
@@ -46,18 +46,12 @@
     "hosts": [
       {
         "hostId": 1,
-        "hostLevelParams": {
-          "repo_info": "[{\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129/\",\"osType\":\"redhat6\",\"repoId\":\"HDP-2.6\",\"repoName\":\"HDP\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP/centos6/2.x/updates/2.6.0.3\",\"latestBaseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129\",\"repoSaved\":true,\"unique\":true,\"ambariManagedRepositories\":true},{\"baseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"osType\":\"redhat6\",\"repoId\":\"HDP-UTILS-1.1.0.21\",\"repoName\":\"HDP-UTILS\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"latestBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"repoSaved\":true,\"unique\":false,\"ambariManagedRepositories\":true}]"
-        },
         "hostName": "c6402.ambari.apache.org",
         "ipv4": "192.168.64.102",
         "rackName": "/default-rack"
       },
       {
         "hostId": 2,
-        "hostLevelParams": {
-          "repo_info": "[{\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129/\",\"osType\":\"redhat6\",\"repoId\":\"HDP-2.6\",\"repoName\":\"HDP\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP/centos6/2.x/updates/2.6.0.3\",\"latestBaseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129\",\"repoSaved\":true,\"unique\":true,\"ambariManagedRepositories\":true},{\"baseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"osType\":\"redhat6\",\"repoId\":\"HDP-UTILS-1.1.0.21\",\"repoName\":\"HDP-UTILS\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"latestBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"repoSaved\":true,\"unique\":false,\"ambariManagedRepositories\":true}]"
-        },
         "hostName": "c6403.ambari.apache.org",
         "ipv4": "192.168.64.103",
         "rackName": "/default-rack"

http://git-wip-us.apache.org/repos/asf/ambari/blob/917898cd/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_create.json
----------------------------------------------------------------------
diff --git a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_create.json
b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_create.json
index dfe17b9..6df4bc3 100644
--- a/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_create.json
+++ b/ambari-agent/src/test/python/ambari_agent/dummy_files/stomp/topology_create.json
@@ -50,19 +50,13 @@
           "hostId":0,
           "hostName":"c6401.ambari.apache.org",
           "rackName":"/default-rack",
-          "ipv4":"192.168.64.101",
-          "hostLevelParams": {
-            "repo_info": "[{\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129/\",\"osType\":\"redhat6\",\"repoId\":\"HDP-2.6\",\"repoName\":\"HDP\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP/centos6/2.x/updates/2.6.0.3\",\"latestBaseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129\",\"repoSaved\":true,\"unique\":true,\"ambariManagedRepositories\":true},{\"baseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"osType\":\"redhat6\",\"repoId\":\"HDP-UTILS-1.1.0.21\",\"repoName\":\"HDP-UTILS\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"latestBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"repoSaved\":true,\"unique\":false,\"ambariManagedRepositories\":true}]"
-          }
+          "ipv4":"192.168.64.101"
         },
         {
           "hostId":1,
           "hostName":"c6402.ambari.apache.org",
           "rackName":"/default-rack",
-          "ipv4":"192.168.64.102",
-          "hostLevelParams": {
-            "repo_info": "[{\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129/\",\"osType\":\"redhat6\",\"repoId\":\"HDP-2.6\",\"repoName\":\"HDP\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP/centos6/2.x/updates/2.6.0.3\",\"latestBaseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129\",\"repoSaved\":true,\"unique\":true,\"ambariManagedRepositories\":true},{\"baseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"osType\":\"redhat6\",\"repoId\":\"HDP-UTILS-1.1.0.21\",\"repoName\":\"HDP-UTILS\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"latestBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"repoSaved\":true,\"unique\":false,\"ambariManagedRepositories\":true}]"
-          }
+          "ipv4":"192.168.64.102"
         }
       ]
     },
@@ -87,10 +81,7 @@
           "hostId":0,
           "hostName":"c6401.ambari.apache.org",
           "rackName":"/default-rack",
-          "ipv4":"192.168.64.101",
-          "hostLevelParams": {
-            "repo_info": "[{\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129/\",\"osType\":\"redhat6\",\"repoId\":\"HDP-2.6\",\"repoName\":\"HDP\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP/centos6/2.x/updates/2.6.0.3\",\"latestBaseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos6/2.x/BUILDS/2.6.1.0-129\",\"repoSaved\":true,\"unique\":true,\"ambariManagedRepositories\":true},{\"baseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"osType\":\"redhat6\",\"repoId\":\"HDP-UTILS-1.1.0.21\",\"repoName\":\"HDP-UTILS\",\"defaultBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"latestBaseUrl\":\"http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.21/repos/centos6\",\"repoSaved\":true,\"unique\":false,\"ambariManagedRepositories\":true}]"
-          }
+          "ipv4":"192.168.64.101"
         }
       ]
     }


Mime
View raw message