Set execution status to FAILED if global/dependencies checks fail (#3083)

* Set execution status to FAILED if global/dependencies checks fail before launching the WF

* exec status after failed checks: also handle dequeueing (#3084)

* add a test

* dequeue: ignore errors, continue to the next execution

if dequeueing an execution fails, oh well. Continue to the next one.
This is because now execute_workflow can throw errors.

Co-authored-by: Lukasz Maksymczuk <tehasdf@users.noreply.github.com>

Author: glukhman

rest-service/manager_rest/resource_manager.py

@@ -235,12 +235,17 @@ def start_queued_executions(self, deployment_storage_id):
         amqp_client.add_handler(handler)
         with amqp_client:
             for execution in to_run:
-                if execution.is_system_workflow:
-                    self._execute_system_workflow(
-                        execution, queue=True, send_handler=handler)
-                else:
-                    self.execute_workflow(
-                        execution, queue=True, send_handler=handler)
+                try:
+                    if execution.is_system_workflow:
+                        self._execute_system_workflow(
+                            execution, queue=True, send_handler=handler)
+                    else:
+                        self.execute_workflow(
+                            execution, queue=True, send_handler=handler)
+                except Exception as e:
+                    current_app.logger.warning(
+                        'Could not dequeue execution %s: %s',
+                        execution, e)
 
     def _refresh_execution(self, execution: models.Execution) -> bool:
         """Prepare the execution to be started.
@@ -965,9 +970,16 @@ def execute_workflow(self, execution, *, force=False, queue=False,
                          send_handler: 'SendHandler' = None):
         with self.sm.transaction():
             if execution.deployment:
-                self._check_allow_global_execution(execution.deployment)
-                self._verify_dependencies_not_affected(
-                    execution.workflow_id, execution.deployment, force)
+                try:
+                    self._check_allow_global_execution(execution.deployment)
+                    self._verify_dependencies_not_affected(
+                        execution.workflow_id, execution.deployment, force)
+                except Exception as e:
+                    execution.status = ExecutionState.FAILED
+                    execution.error = str(e)
+                    self.sm.update(execution)
+                    db.session.commit()
+                    raise
 
             should_queue = queue
             if not allow_overlapping_running_wf:

rest-service/manager_rest/test/endpoints/test_executions.py

@@ -257,6 +257,22 @@ def test_update_execution_status(self):
             execution, ExecutionState.STARTED)
         self._modify_execution_status(execution.id, 'pending')
 
+    def test_failed_when_global_disallowed(self):
+        """When check_allow_global throws, the execution is failed"""
+        self.put_deployment(self.DEPLOYMENT_ID)
+
+        with mock.patch('manager_rest.resource_manager.ResourceManager'
+                        '._check_allow_global_execution',
+                        side_effect=manager_exceptions.ForbiddenError()):
+            with self.assertRaises(exceptions.CloudifyClientError):
+                self.client.executions.start(self.DEPLOYMENT_ID, 'install')
+        executions = self.client.executions.list(
+            deployment_id=self.DEPLOYMENT_ID,
+            workflow_id='install')
+        self.assertEqual(len(executions), 1)
+        execution = executions[0]
+        self.assertEqual(ExecutionState.FAILED, execution.status)
+
     def test_update_execution_status_with_error(self):
         (blueprint_id, deployment_id, blueprint_response,
          deployment_response) = self.put_deployment(self.DEPLOYMENT_ID)