Introduce workflow failure condition

awx/api/serializers.py

@@ -2249,11 +2249,15 @@ class WorkflowNodeBaseSerializer(BaseSerializer):
     success_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
     failure_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
     always_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
+    fail_on_job_failure = serializers.BooleanField(
+        help_text=('If set to true, and if the job runs and fails, '
+                   'the workflow will also be marked as failed.'),
+        default=True)
 
     class Meta:
         fields = ('*', '-name', '-description', 'id', 'url', 'related',
                   'unified_job_template', 'success_nodes', 'failure_nodes', 'always_nodes',
-                  'inventory', 'credential', 'job_type', 'job_tags', 'skip_tags', 'limit', 'skip_tags')
+                  'inventory', 'credential', 'job_type', 'job_tags', 'skip_tags', 'limit', 'skip_tags', 'fail_on_job_failure')
 
     def get_related(self, obj):
         res = super(WorkflowNodeBaseSerializer, self).get_related(obj)

awx/main/migrations/0041_v310_workflow_failure_condition.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('main', '0040_v310_artifacts'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='workflowjobnode',
+            name='fail_on_job_failure',
+            field=models.BooleanField(default=True),
+        ),
+        migrations.AddField(
+            model_name='workflowjobtemplatenode',
+            name='fail_on_job_failure',
+            field=models.BooleanField(default=True),
+        ),
+    ]

awx/main/models/workflow.py

@@ -60,6 +60,10 @@ class WorkflowNodeBase(CreatedModifiedModel):
         default=None,
         on_delete=models.SET_NULL,
     )
+    fail_on_job_failure = models.BooleanField(
+        blank=True,
+        default=True,
+    )
     # Prompting-related fields
     inventory = models.ForeignKey(
         'Inventory',
@@ -137,7 +141,7 @@ class WorkflowNodeBase(CreatedModifiedModel):
         Return field names that should be copied from template node to job node.
         '''
         return ['workflow_job', 'unified_job_template',
-                'inventory', 'credential', 'char_prompts']
+                'inventory', 'credential', 'char_prompts', 'fail_on_job_failure']
 
 class WorkflowJobTemplateNode(WorkflowNodeBase):
     # TODO: Ensure the API forces workflow_job_template being set
@@ -383,6 +387,9 @@ class WorkflowJob(UnifiedJob, WorkflowJobOptions, JobNotificationMixin, Workflow
         from awx.main.tasks import RunWorkflowJob
         return RunWorkflowJob
 
+    def _has_failed(self):
+        return self.workflow_job_nodes.filter(job__status='failed', fail_on_job_failure=True).exists()
+
     def socketio_emit_data(self):
         return {}
 

awx/main/scheduler/__init__.py

@@ -73,10 +73,12 @@ def process_finished_workflow_jobs(workflow_jobs):
         dag = WorkflowDAG(workflow_job)
         if dag.is_workflow_done():
             with transaction.atomic():
-                # TODO: detect if wfj failed
+                if workflow_job._has_failed():
-                workflow_job.status = 'completed'
+                    workflow_job.status = 'failed'
+                else:
+                    workflow_job.status = 'successful'
                 workflow_job.save()
-                workflow_job.websocket_emit_status('completed')
+                workflow_job.websocket_emit_status(workflow_job.status)
 
 def rebuild_graph():
     """Regenerate the task graph by refreshing known tasks from Tower, purging

awx/main/tests/functional/models/test_workflow.py

@@ -90,3 +90,42 @@ class TestWorkflowJobTemplate:
         assert len(parent_qs) == 1
         assert parent_qs[0] == wfjt.workflow_job_template_nodes.all()[1]
 
+@pytest.mark.django_db
+class TestWorkflowJobFailure:
+    @pytest.fixture
+    def wfj(self):
+        return WorkflowJob.objects.create(name='test-wf-job')
+
+    def test_workflow_has_failed(self, wfj):
+        """
+        Test that a single failed node with fail_on_job_failure = true
+        leads to the entire WF being marked as failed
+        """
+        job = Job.objects.create(name='test-job', status='failed')
+        # Node has a failed job connected
+        WorkflowJobNode.objects.create(workflow_job=wfj, job=job)
+        assert wfj._has_failed()
+
+    def test_workflow_not_failed_unran_job(self, wfj):
+        """
+        Test that an un-ran node will not mark workflow job as failed
+        """
+        WorkflowJobNode.objects.create(workflow_job=wfj)
+        assert not wfj._has_failed()
+
+    def test_workflow_not_failed_successful_job(self, wfj):
+        """
+        Test that a sucessful node will not mark workflow job as failed
+        """
+        job = Job.objects.create(name='test-job', status='successful')
+        WorkflowJobNode.objects.create(workflow_job=wfj, job=job)
+        assert not wfj._has_failed()
+
+    def test_workflow_not_failed_failed_job_but_okay(self, wfj):
+        """
+        Test that a failed node will not mark workflow job as failed
+        if the fail_on_job_failure is set to false
+        """
+        job = Job.objects.create(name='test-job', status='failed')
+        WorkflowJobNode.objects.create(workflow_job=wfj, job=job, fail_on_job_failure=False)
+        assert not wfj._has_failed()