Use a Zaqar queue to get stack events

This adds the ability to use Zaqar to retrieve Heat events instead of
polling the API.

Change-Id: I3b836637d4e72aaf7183dbabf0b0e32c2caa8270
Closes-Bug: #1639283

tripleoclient/tests/test_utils.py

@@ -22,6 +22,7 @@
 import yaml
 
 from tripleoclient import exceptions
+from tripleoclient.tests import fakes
 from tripleoclient import utils
 
 
@@ -109,6 +110,20 @@ def test_wait_for_stack_in_progress(self, mock_poll_for_events):
         result = utils.wait_for_stack_ready(self.mock_orchestration, 'stack')
         self.assertEqual(False, result)
 
+    @mock.patch("heatclient.common.event_utils.wait_for_events")
+    def test_wait_for_stack_websocket(self, mock_wait_for_events):
+
+        mock_wait_for_events.return_value = ("CREATE_IN_PROGRESS", "MESSAGE")
+
+        stack = mock.Mock()
+        stack.stack_name = 'stack'
+        stack.stack_status = 'CREATE_IN_PROGRESS'
+        self.mock_orchestration.stacks.get.return_value = stack
+
+        result = utils.wait_for_stack_ready(self.mock_orchestration, 'stack',
+                                            ws_client=fakes.FakeWebSocket())
+        self.assertEqual(False, result)
+
     @mock.patch('tripleoclient.utils.wait_for_provision_state')
     def test_set_nodes_state(self, wait_for_state_mock):
 

tripleoclient/tests/v1/overcloud_deploy/test_overcloud_deploy.py

@@ -463,10 +463,13 @@ def assertEqual(*args):
             self.assertEqual(*args)
 
         def _fake_heat_deploy(self, stack, stack_name, template_path,
-                              parameters, environments, timeout, tht_root,
-                              env, update_plan_only, run_validations):
+                              parameters, environments, timeout, tht_root, env,
+                              update_plan_only, run_validations, ws_client):
+            queue = env['event_sinks'][0]['target']
             assertEqual(
                 {'parameter_defaults': {},
+                 'event_sinks': [
+                     {'target': queue, 'ttl': 14400, 'type': 'zaqar-queue'}],
                  'resource_registry': {'Test': u'OS::Heat::None'}}, env)
 
         mock_deploy_heat.side_effect = _fake_heat_deploy
@@ -519,10 +522,13 @@ def assertEqual(*args):
             self.assertEqual(*args)
 
         def _fake_heat_deploy(self, stack, stack_name, template_path,
-                              parameters, environments, timeout, tht_root,
-                              env, update_plan_only, run_validations):
+                              parameters, environments, timeout, tht_root, env,
+                              update_plan_only, run_validations, ws_client):
+            queue = env['event_sinks'][0]['target']
             assertEqual(
                 {'parameter_defaults': {},
+                 'event_sinks': [
+                     {'target': queue, 'ttl': 14400, 'type': 'zaqar-queue'}],
                  'resource_registry': {'Test': u'OS::Heat::None'}}, env)
 
         mock_deploy_heat.side_effect = _fake_heat_deploy
@@ -760,13 +766,14 @@ def test_try_overcloud_deploy_with_first_template_existing(
             self, mock_heat_deploy_func):
         result = self.cmd._try_overcloud_deploy_with_compat_yaml(
             '/fake/path', {}, 'overcloud', {}, ['~/overcloud-env.json'], 1,
-            {}, False, True)
+            {}, False, True, mock.ANY)
         # If it returns None it succeeded
         self.assertIsNone(result)
         mock_heat_deploy_func.assert_called_once_with(
             self.cmd, {}, 'overcloud',
             '/fake/path/' + constants.OVERCLOUD_YAML_NAME, {},
-            ['~/overcloud-env.json'], 1, '/fake/path', {}, False, True)
+            ['~/overcloud-env.json'], 1, '/fake/path', {}, False, True,
+            mock.ANY)
 
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
                 '_heat_deploy', autospec=True)
@@ -775,7 +782,7 @@ def test_try_overcloud_deploy_with_no_templates_existing(
         mock_heat_deploy_func.side_effect = ObjectClientException('error')
         self.assertRaises(ValueError,
                           self.cmd._try_overcloud_deploy_with_compat_yaml,
-                          '/fake/path', mock.ANY, mock.ANY, mock.ANY,
+                          '/fake/path', mock.ANY, mock.ANY, mock.ANY, mock.ANY,
                           mock.ANY, mock.ANY, mock.ANY, mock.ANY, mock.ANY)
 
     @mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
@@ -786,8 +793,8 @@ def test_try_overcloud_deploy_show_missing_file(
             ObjectClientException('/fake/path not found')
         try:
             self.cmd._try_overcloud_deploy_with_compat_yaml(
-                '/fake/path', mock.ANY, mock.ANY, mock.ANY,
-                mock.ANY, mock.ANY, mock.ANY, mock.ANY, mock.ANY)
+                '/fake/path', mock.ANY, mock.ANY, mock.ANY, mock.ANY, mock.ANY,
+                mock.ANY, mock.ANY, mock.ANY, mock.ANY)
         except ValueError as value_error:
             self.assertIn('/fake/path', str(value_error))
 
@@ -1256,7 +1263,7 @@ def test_heat_deploy_update_plan_only(self, mock_breakpoints_cleanup,
         mock_get_template_contents.return_value = [{}, {}]
 
         self.cmd._heat_deploy(mock_stack, 'mock_stack', '/tmp', {},
-                              {}, 1, '/tmp', {}, True, False)
+                              {}, 1, '/tmp', {}, True, False, mock.ANY)
 
         self.assertFalse(mock_deploy_and_wait.called)
 

tripleoclient/utils.py

@@ -153,7 +153,7 @@ def check_hypervisor_stats(compute_client, nodes=1, memory=0, vcpu=0):
 
 
 def wait_for_stack_ready(orchestration_client, stack_name, marker=None,
-                         action='CREATE', verbose=False):
+                         action='CREATE', verbose=False, ws_client=None):
     """Check the status of an orchestration stack
 
     Get the status of an orchestration stack and check whether it is complete
@@ -183,9 +183,14 @@ def wait_for_stack_ready(orchestration_client, stack_name, marker=None,
         out = sys.stdout
     else:
         out = open(os.devnull, "w")
-    stack_status, msg = event_utils.poll_for_events(
-        orchestration_client, stack_name, action=action,
-        poll_period=5, marker=marker, out=out, nested_depth=2)
+    if ws_client is not None:
+        with ws_client:
+            stack_status, msg = event_utils.wait_for_events(
+                ws_client, stack_name, out=out)
+    else:
+        stack_status, msg = event_utils.poll_for_events(
+            orchestration_client, stack_name, action=action,
+            poll_period=5, marker=marker, out=out, nested_depth=2)
     print(msg)
     return stack_status == '%s_COMPLETE' % action
 

tripleoclient/v1/overcloud_deploy.py

@@ -192,7 +192,7 @@ def _process_multiple_environments(self, created_env_files, added_files,
 
     def _heat_deploy(self, stack, stack_name, template_path, parameters,
                      env_files, timeout, tht_root, env, update_plan_only,
-                     run_validations):
+                     run_validations, ws_client):
         """Verify the Baremetal nodes are available and do a stack update"""
 
         clients = self.app.client_manager
@@ -242,7 +242,8 @@ def do_object_request(method='GET', object_path=None):
             deployment.deploy_and_wait(self.log, clients, stack, stack_name,
                                        self.app_args.verbose_level,
                                        timeout=timeout,
-                                       run_validations=run_validations)
+                                       run_validations=run_validations,
+                                       ws_client=ws_client)
 
     def _load_environment_directories(self, directories):
         if os.environ.get('TRIPLEO_ENVIRONMENT_DIRECTORY'):
@@ -433,6 +434,13 @@ def _deploy_tripleo_heat_templates(self, stack, parsed_args,
                 parsed_args.environment_directories))
         env.update(self._create_parameters_env(parameters))
 
+        event_queue = str(uuid.uuid4())
+        env['event_sinks'] = [
+            {'type': 'zaqar-queue', 'target': event_queue,
+             'ttl': parsed_args.timeout * 60}]
+        clients = self.app.client_manager
+        ws_client = clients.tripleoclient.messaging_websocket(event_queue)
+
         if parsed_args.rhel_reg:
             reg_env_files, reg_env = self._create_registration_env(parsed_args)
             created_env_files.extend(reg_env_files)
@@ -449,19 +457,19 @@ def _deploy_tripleo_heat_templates(self, stack, parsed_args,
         self._try_overcloud_deploy_with_compat_yaml(
             tht_root, stack, parsed_args.stack, parameters, env_files,
             parsed_args.timeout, env, parsed_args.update_plan_only,
-            parsed_args.run_validations)
+            parsed_args.run_validations, ws_client)
 
     def _try_overcloud_deploy_with_compat_yaml(self, tht_root, stack,
                                                stack_name, parameters,
                                                env_files, timeout,
                                                env, update_plan_only,
-                                               run_validations):
+                                               run_validations, ws_client):
         overcloud_yaml = os.path.join(tht_root, constants.OVERCLOUD_YAML_NAME)
         try:
             self._heat_deploy(stack, stack_name, overcloud_yaml,
                               parameters, env_files, timeout,
                               tht_root, env, update_plan_only,
-                              run_validations)
+                              run_validations, ws_client)
         except ClientException as e:
             messages = 'Failed to deploy: %s' % str(e)
             raise ValueError(messages)

tripleoclient/workflows/deployment.py

@@ -45,7 +45,7 @@ def deploy(clients, **workflow_input):
 
 
 def deploy_and_wait(log, clients, stack, plan_name, verbose_level,
-                    timeout=None, run_validations=False):
+                    timeout=None, run_validations=False, ws_client=None):
     """Start the deploy and wait for it to finish"""
 
     workflow_input = {
@@ -79,7 +79,8 @@ def deploy_and_wait(log, clients, stack, plan_name, verbose_level,
     time.sleep(10)
     verbose_events = verbose_level > 0
     create_result = utils.wait_for_stack_ready(
-        orchestration_client, plan_name, marker, action, verbose_events)
+        orchestration_client, plan_name, marker, action, verbose_events,
+        ws_client)
     if not create_result:
         shell.OpenStackShell().run(["stack", "failures", "list", plan_name])
         if stack is None: