Add automatic workflow node retries via a max_retries node setting

awx/api/serializers.py

@@ -4238,6 +4238,7 @@ class Meta:
             'condition_nodes',
             'condition_edges',
             'all_parents_must_converge',
+            'max_retries',
             'identifier',
         )
 
@@ -4282,6 +4283,7 @@ class WorkflowJobNodeSerializer(LaunchConfigurationBaseSerializer):
     condition_edges = serializers.SerializerMethodField(
         help_text=_('List of condition data for each conditional edge originating from this node; id is the target node.')
     )
+    retried_jobs = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
 
     class Meta:
         model = WorkflowJobNode
@@ -4304,6 +4306,9 @@ class Meta:
             'do_not_run',
             'prior_run_succeeded',
             'prior_run_elapsed',
+            'max_retries',
+            'retry_attempts',
+            'retried_jobs',
             'identifier',
         )
 

awx/api/views/mixin.py

@@ -41,6 +41,10 @@ def destroy(self, request, *args, **kwargs):
                 raise PermissionDenied(detail=_('Cannot delete job resource when associated workflow job is running.'))
         except self.model.unified_job_node.RelatedObjectDoesNotExist:
             pass
+        # a superseded attempt of a retried workflow node no longer has a
+        # unified_job_node, but its workflow may still be running
+        if obj.retried_workflow_nodes.filter(workflow_job__status__in=ACTIVE_STATES).exists():
+            raise PermissionDenied(detail=_('Cannot delete job resource when associated workflow job is running.'))
         # Still allow deletion of new status, because these can be manually created
         if obj.status in ACTIVE_STATES and obj.status != 'new':
             raise PermissionDenied(detail=_("Cannot delete running job resource."))

awx/main/migrations/0203_workflow_node_retries.py

@@ -0,0 +1,55 @@
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('main', '0202_project_webhooks'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='workflowjobtemplatenode',
+            name='max_retries',
+            field=models.PositiveIntegerField(
+                default=0,
+                validators=[django.core.validators.MaxValueValidator(100)],
+                help_text=(
+                    "Maximum number of times this node's job is automatically retried after "
+                    "failing before its failure paths are followed. Canceled jobs are never retried."
+                ),
+            ),
+        ),
+        migrations.AddField(
+            model_name='workflowjobnode',
+            name='max_retries',
+            field=models.PositiveIntegerField(
+                default=0,
+                validators=[django.core.validators.MaxValueValidator(100)],
+                help_text=(
+                    "Maximum number of times this node's job is automatically retried after "
+                    "failing before its failure paths are followed. Canceled jobs are never retried."
+                ),
+            ),
+        ),
+        migrations.AddField(
+            model_name='workflowjobnode',
+            name='retry_attempts',
+            field=models.PositiveIntegerField(
+                default=0,
+                editable=False,
+                help_text="Number of times this node's job has been automatically retried after failing.",
+            ),
+        ),
+        migrations.AddField(
+            model_name='workflowjobnode',
+            name='retried_jobs',
+            field=models.ManyToManyField(
+                blank=True,
+                editable=False,
+                help_text='Jobs from earlier attempts of this node that were superseded by an automatic retry.',
+                related_name='retried_workflow_nodes',
+                to='main.unifiedjob',
+            ),
+        ),
+    ]

awx/main/models/unified_jobs.py

@@ -1238,7 +1238,11 @@ def get_workflow_job(self):
             try:
                 return self.unified_job_node.workflow_job
             except UnifiedJob.unified_job_node.RelatedObjectDoesNotExist:
-                pass
+                # superseded attempt of a retried node; the node points at the
+                # newest attempt but this job still belongs to its workflow
+                node = self.retried_workflow_nodes.first()
+                if node is not None:
+                    return node.workflow_job
         return None
 
     @property
@@ -1247,7 +1251,9 @@ def workflow_node_id(self):
             try:
                 return self.unified_job_node.pk
             except UnifiedJob.unified_job_node.RelatedObjectDoesNotExist:
-                pass
+                node = self.retried_workflow_nodes.first()
+                if node is not None:
+                    return node.pk
         return None
 
     def get_effective_artifacts(self, **kwargs):

awx/main/models/workflow.py

@@ -13,6 +13,7 @@
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
 from django.core.exceptions import ObjectDoesNotExist
+from django.core.validators import MaxValueValidator
 from django.utils.timezone import now, timedelta
 
 # from django import settings as tower_settings
@@ -61,6 +62,12 @@
 
 logger = logging.getLogger('awx.main.models.workflow')
 
+# Sanity ceiling for the retry budget; retries have no delay between attempts,
+# so an unbounded user value could flood the jobs table before failure paths run
+MAX_RETRIES_CEILING = 100
+# Statuses a spawned unified job cannot leave without outside action
+TERMINAL_JOB_STATUSES = ('successful', 'failed', 'error', 'canceled')
+
 
 def evaluate_artifact_condition(artifacts, artifact_key, operator, expected_value):
     """
@@ -208,6 +215,14 @@ class Meta:
     all_parents_must_converge = models.BooleanField(
         default=False, help_text=_("If enabled then the node will only run if all of the parent nodes have met the criteria to reach this node")
     )
+    max_retries = models.PositiveIntegerField(
+        default=0,
+        validators=[MaxValueValidator(MAX_RETRIES_CEILING)],
+        help_text=_(
+            "Maximum number of times this node's job is automatically retried after "
+            "failing before its failure paths are followed. Canceled jobs are never retried."
+        ),
+    )
     unified_job_template = models.ForeignKey(
         'UnifiedJobTemplate',
         related_name='%(class)ss',
@@ -239,6 +254,7 @@ def _get_workflow_job_field_names(cls):
             'credentials',
             'char_prompts',
             'all_parents_must_converge',
+            'max_retries',
             'labels',
             'instance_groups',
             'execution_environment',
@@ -291,6 +307,7 @@ class WorkflowJobTemplateNode(WorkflowNodeBase):
         'survey_passwords',
         'char_prompts',
         'all_parents_must_converge',
+        'max_retries',
         'identifier',
         'labels',
         'execution_environment',
@@ -435,6 +452,18 @@ class WorkflowJobNode(WorkflowNodeBase):
         editable=False,
         help_text=_("Elapsed time (seconds) of this node's job in the run it was carried forward from."),
     )
+    retry_attempts = models.PositiveIntegerField(
+        default=0,
+        editable=False,
+        help_text=_("Number of times this node's job has been automatically retried after failing."),
+    )
+    retried_jobs = models.ManyToManyField(
+        'UnifiedJob',
+        related_name='retried_workflow_nodes',
+        blank=True,
+        editable=False,
+        help_text=_('Jobs from earlier attempts of this node that were superseded by an automatic retry.'),
+    )
     identifier = models.CharField(
         max_length=512,
         blank=True,  # blank denotes pre-migration job nodes
@@ -463,6 +492,35 @@ class Meta:
     def event_processing_finished(self):
         return True
 
+    def retry_pending(self):
+        """True when this node's job failed but automatic retries remain, in which
+        case the scheduler treats the node as if its job were still running and
+        respawns it instead of following its failure paths. Canceled jobs are
+        explicit user actions and are never retried."""
+        if not self.job or self.job.status not in ('failed', 'error'):
+            return False
+        # a retry cannot spawn once the template is gone (deleted mid run);
+        # without this the workflow would wait on the retry forever
+        if self.unified_job_template_id is None:
+            return False
+        if self.retry_attempts >= self.max_retries:
+            return False
+        # approvals are human decisions, never retried automatically; resolve
+        # the class through the polymorphic ctype because a base UnifiedJob
+        # instance (e.g. loaded via select_related('job')) defeats isinstance
+        job_cls = self.job.get_real_instance_class() if self.job.polymorphic_ctype_id else type(self.job)
+        if issubclass(job_cls, WorkflowApproval):
+            return False
+        return True
+
+    def job_finished(self):
+        """The node's job reached a terminal status with no automatic retry pending."""
+        return bool(self.job and self.job.status in TERMINAL_JOB_STATUSES and not self.retry_pending())
+
+    def finally_failed(self):
+        """The node's job failed for good: a failure status and no retry pending."""
+        return bool(self.job and self.job.status in ('failed', 'error', 'canceled') and not self.retry_pending())
+
     def get_absolute_url(self, request=None):
         return reverse('api:workflow_job_node_detail', kwargs={'pk': self.pk}, request=request)
 

awx/main/scheduler/dag_workflow.py

@@ -103,8 +103,9 @@ def _are_relevant_parents_finished(self, node):
             # do_not_run is False, node might still run a job and thus blocks children
             elif not p.job:
                 return False
-            # Node decidedly got a job; check if job is done
-            elif p.job and p.job.status not in ['successful', 'failed', 'error', 'canceled']:
+            # Node decidedly got a job; check if job is done. A failed job with
+            # automatic retries left is not done, a new attempt will be spawned.
+            elif p.job and not p.job_finished():
                 return False
         return True
 
@@ -116,8 +117,9 @@ def _all_parents_met_convergence_criteria(self, node):
         parent_nodes = [p['node_object'] for p in self.get_parents(obj)]
         for p in parent_nodes:
             status = None
-            # node has a status
-            if p.job and p.job.status in ["successful", "failed"]:
+            # node has a status; a failed job with automatic retries left has
+            # no final status yet
+            if p.job_finished() and p.job.status in ["successful", "failed"]:
                 if p.job and p.job.status == "successful":
                     status = "success_nodes"
                 elif p.job and p.job.status == "failed":
@@ -157,7 +159,11 @@ def bfs_nodes_to_run(self):
                     self.get_children(obj, 'success_nodes') + self.get_children(obj, 'always_nodes') + self._passing_condition_children(obj, 'success')
                 )
             elif obj.job:
-                if obj.job.status in ['failed', 'error', 'canceled']:
+                # a failed job with automatic retries left is respawned instead
+                # of following its failure paths
+                if obj.retry_pending():
+                    nodes_found.append(n)
+                elif obj.job.status in ['failed', 'error', 'canceled']:
                     nodes.extend(
                         self.get_children(obj, 'failure_nodes') + self.get_children(obj, 'always_nodes') + self._passing_condition_children(obj, 'failure')
                     )
@@ -198,7 +204,7 @@ def is_workflow_done(self):
             obj = node['node_object']
             if obj.do_not_run is False and not obj.prior_run_succeeded and not obj.job and obj.unified_job_template:
                 return False
-            elif obj.job and obj.job.status not in ['successful', 'failed', 'canceled', 'error']:
+            elif obj.job and not obj.job_finished():
                 return False
         return True
 
@@ -212,7 +218,7 @@ def has_workflow_failed(self):
             obj = node['node_object']
             if obj.do_not_run is False and not obj.prior_run_succeeded and obj.unified_job_template is None:
                 failed_nodes.append(node)
-            elif obj.job and obj.job.status in ['failed', 'canceled', 'error']:
+            elif obj.finally_failed():
                 failed_nodes.append(node)
 
         for node in failed_nodes:
@@ -262,7 +268,7 @@ def has_workflow_failed(self):
 
     def _are_all_nodes_dnr_decided(self, workflow_nodes):
         for n in workflow_nodes:
-            if n.do_not_run is False and not n.prior_run_succeeded and not n.job and n.unified_job_template:
+            if n.do_not_run is False and not n.prior_run_succeeded and (not n.job or n.retry_pending()) and n.unified_job_template:
                 return False
         return True
 
@@ -287,7 +293,10 @@ def _should_mark_node_dnr(self, node, parent_nodes):
                 ):
                     return False
             elif p.job:
-                if p.job.status == 'successful':
+                # a failed job with automatic retries left is not decided yet
+                if p.retry_pending():
+                    return False
+                elif p.job.status == 'successful':
                     if node in (self.get_children(p, 'success_nodes') + self.get_children(p, 'always_nodes')) or self._edge_condition_passes(
                         p, node['node_object'], 'success'
                     ):

awx/main/scheduler/task_manager.py

@@ -225,11 +225,29 @@ def spawn_workflow_graph_jobs(self):
                 for spawn_node in spawn_nodes:
                     if spawn_node.unified_job_template is None:
                         continue
+                    # a node that already has a (failed) job is being automatically retried
+                    superseded_job_id = spawn_node.job_id
+                    is_retry = superseded_job_id is not None
                     kv = spawn_node.get_job_kwargs()
                     job = spawn_node.unified_job_template.create_unified_job(**kv)
                     spawn_node.job = job
+                    if is_retry:
+                        spawn_node.retry_attempts += 1
                     spawn_node.save()
-                    logger.debug('Spawned %s in %s for node %s', job.log_format, workflow_job.log_format, spawn_node.pk)
+                    if is_retry:
+                        # keep the superseded attempt linked to the node so it stays
+                        # protected from deletion and traceable to this workflow
+                        spawn_node.retried_jobs.add(superseded_job_id)
+                        logger.info(
+                            'Spawned %s in %s for node %s, retry %s of %s after failed attempt',
+                            job.log_format,
+                            workflow_job.log_format,
+                            spawn_node.pk,
+                            spawn_node.retry_attempts,
+                            spawn_node.max_retries,
+                        )
+                    else:
+                        logger.debug('Spawned %s in %s for node %s', job.log_format, workflow_job.log_format, spawn_node.pk)
                     can_start = True
                     if isinstance(spawn_node.unified_job_template, WorkflowJobTemplate):
                         workflow_ancestors = job.get_ancestor_workflows()
@@ -267,6 +285,12 @@ def spawn_workflow_graph_jobs(self):
                                 "Job spawned from workflow could not start because it was not in the right state or required manual credentials"
                             )
                     if not can_start:
+                        # the job never started for a structural reason (missing
+                        # resources, recursion, credentials); retrying cannot help,
+                        # so exhaust the node's automatic retry budget
+                        if spawn_node.retry_attempts < spawn_node.max_retries:
+                            spawn_node.retry_attempts = spawn_node.max_retries
+                            spawn_node.save(update_fields=['retry_attempts'])
                         job.status = 'failed'
                         job.save(update_fields=['status', 'job_explanation'])
                         job.websocket_emit_status('failed')

awx/main/tests/functional/api/test_workflow_node.py

@@ -52,6 +52,44 @@ def test_node_accepts_prompted_fields(inventory, project, workflow_job_template,
     post(url, {'unified_job_template': job_template.pk, 'limit': 'webservers'}, user=admin_user, expect=201)
 
 
+@pytest.mark.django_db
+def test_node_max_retries_settable(inventory, project, workflow_job_template, post, admin_user):
+    job_template = JobTemplate.objects.create(inventory=inventory, project=project, playbook='helloworld.yml')
+    url = reverse('api:workflow_job_template_workflow_nodes_list', kwargs={'pk': workflow_job_template.pk})
+    r = post(url, {'unified_job_template': job_template.pk, 'max_retries': 2}, user=admin_user, expect=201)
+    node = WorkflowJobTemplateNode.objects.get(pk=r.data['id'])
+    assert node.max_retries == 2
+
+
+@pytest.mark.django_db
+def test_node_max_retries_rejects_out_of_range_values(inventory, project, workflow_job_template, post, admin_user):
+    # retries have no delay between attempts, so the budget is capped to keep a
+    # misconfigured node from flooding the jobs table before failure paths run
+    job_template = JobTemplate.objects.create(inventory=inventory, project=project, playbook='helloworld.yml')
+    url = reverse('api:workflow_job_template_workflow_nodes_list', kwargs={'pk': workflow_job_template.pk})
+    r = post(url, {'unified_job_template': job_template.pk, 'max_retries': 101}, user=admin_user, expect=400)
+    assert 'max_retries' in r.data
+    r = post(url, {'unified_job_template': job_template.pk, 'max_retries': -1}, user=admin_user, expect=400)
+    assert 'max_retries' in r.data
+
+
+@pytest.mark.django_db
+def test_superseded_retry_attempt_protected_while_workflow_runs(delete, admin_user, inventory, project, workflow_job_template):
+    # a job superseded by an automatic retry loses its unified_job_node link,
+    # but must stay undeletable while its workflow is still running
+    job_template = JobTemplate.objects.create(name='retried-jt', inventory=inventory, project=project, playbook='helloworld.yml')
+    wfj = WorkflowJob.objects.create(workflow_job_template=workflow_job_template, status='running')
+    node = wfj.workflow_nodes.create(unified_job_template=job_template)
+    old_job = job_template.create_job()
+    old_job.status = 'failed'
+    old_job.launch_type = 'workflow'
+    old_job.save()
+    node.retried_jobs.add(old_job)
+
+    url = reverse('api:job_detail', kwargs={'pk': old_job.pk})
+    delete(url, user=admin_user, expect=403)
+
+
 @pytest.mark.django_db
 @pytest.mark.parametrize(
     "field_name, field_value",