feat: let a relaunch from failed nodes overwrite the workflow variables

ascender/api/serializers/workflow.py

@@ -65,6 +65,7 @@ class Meta:
             'organization',
             'survey_enabled',
             'allow_simultaneous',
+            'allow_overwrite_flow_vars_on_relaunch',
             'ask_variables_on_launch',
             'inventory',
             'limit',
@@ -160,6 +161,7 @@ class Meta:
             'workflow_job_template',
             'extra_vars',
             'allow_simultaneous',
+            'allow_overwrite_flow_vars_on_relaunch',
             'job_template',
             'is_sliced_job',
             '-execution_environment',

ascender/api/templates/api/workflow_job_relaunch.md

@@ -1,5 +1,10 @@
 Relaunch a workflow job:
 
-Make a POST request to this endpoint to launch a workflow job identical to the parent workflow job. This will spawn jobs, project updates, or inventory updates based on the unified job templates referenced in the workflow nodes in the workflow job. No POST data is accepted for this action.
+Make a POST request to this endpoint to launch a workflow job identical to the parent workflow job. This will spawn jobs, project updates, or inventory updates based on the unified job templates referenced in the workflow nodes in the workflow job.
 
-If successful, the response status code will be 201 and serialized data of the new workflow job will be returned.
+Two optional parameters are accepted:
+
+* `nodes`: set it to `failed` to run the nodes that did not succeed and everything downstream of them, carrying the nodes that already succeeded forward instead of running them again.
+* `extra_vars`: variables that overwrite the ones the original run used. They are only accepted together with `nodes=failed`, and only when this workflow job has `allow_overwrite_flow_vars_on_relaunch` set. That field is copied onto the job from its workflow job template when the job is launched, so it reflects what the template said at launch time rather than what it says now: enabling it on the template makes the runs started after that eligible, and disabling it leaves the runs already started eligible. Answers to password questions of the survey may be left as `$encrypted$` to keep the value the original run used; anywhere else `$encrypted$` is rejected as a reserved word.
+
+If successful, the response status code will be 201 and serialized data of the new workflow job will be returned.

ascender/api/views/workflow.py

@@ -29,7 +29,7 @@
 from ascender.api.views.mixin import RelatedJobsPreventDeleteMixin, UnifiedJobDeletionMixin
 from ascender.main import models
 from ascender.main.scheduler.dag_workflow import WorkflowDAG
-from ascender.main.utils import ScheduleWorkflowManager, getattrd
+from ascender.main.utils import ScheduleWorkflowManager, getattrd, parse_yaml_or_json
 from collections import OrderedDict
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import Q
@@ -418,9 +418,39 @@ def check_object_permissions(self, request, obj):
     def get(self, request, *args, **kwargs):
         return Response({})
 
+    def _relaunch_extra_vars(self, request, obj, from_failed):
+        """Read the variables a relaunch wants to overwrite, if it sends any.
+
+        Overwriting is off unless the workflow job template turned it on, and it
+        only applies to a relaunch from failed nodes: that is the run you cannot
+        redo any other way, since a plain launch starts the whole workflow over
+        and can be prompted normally."""
+        if 'extra_vars' not in request.data:
+            return {}
+        # Whatever was sent goes to the parser as it came, so a payload that is
+        # not a mapping (a list, a bare null) is rejected rather than read as
+        # "no variables".
+        extra_vars = parse_yaml_or_json(request.data['extra_vars'], silent_failure=False)
+        if not extra_vars:
+            return {}
+        if not from_failed:
+            raise ParseError(_('Variables can only be overwritten when relaunching from failed nodes.'))
+        if not obj.allow_overwrite_flow_vars_on_relaunch:
+            raise ParseError(
+                _(
+                    'This workflow job does not allow overwriting variables on relaunch. '
+                    'Enable allow_overwrite_flow_vars_on_relaunch on its workflow job template; runs started after that carry the setting.'
+                )
+            )
+        errors = obj.validate_relaunch_extra_vars(extra_vars)
+        if errors:
+            raise ParseError({'variables_needed_to_start': errors})
+        return extra_vars
+
     def post(self, request, *args, **kwargs):
         obj = self.get_object()
         from_failed = request.data.get('nodes') == 'failed'
+        extra_vars = self._relaunch_extra_vars(request, obj, from_failed)
         if obj.is_sliced_job:
             if from_failed:
                 raise ParseError(_('Cannot relaunch a sliced workflow job from failed nodes.'))
@@ -441,7 +471,7 @@ def post(self, request, *args, **kwargs):
                         _('Cannot relaunch from failed nodes: this workflow failed because a node has no job template, which relaunching cannot recover.')
                     )
                 raise ParseError(_('This workflow job has no failed nodes to relaunch from.'))
-        new_workflow_job = obj.create_relaunch_workflow_job(from_failed=from_failed)
+        new_workflow_job = obj.create_relaunch_workflow_job(from_failed=from_failed, extra_vars=extra_vars)
         new_workflow_job.signal_start()
 
         data = serializers.WorkflowJobSerializer(new_workflow_job, context=self.get_serializer_context()).data

ascender/main/access.py

@@ -2179,7 +2179,13 @@ def can_start(self, obj):
                 raise PermissionDenied(_("Job was launched with secret prompts provided by another user."))
             if not JobLaunchConfigAccess(self.user).can_add({'reference_obj': config}):
                 raise PermissionDenied(_('Job was launched with prompts you lack access to.'))
-            if config.has_unprompted(template):
+            # A run corrected by a relaunch from failed nodes keeps the variables
+            # it was given in its launch config, so relaunching it again runs
+            # with them. While the template still allows that overwrite, anyone
+            # who can relaunch could hand those same variables in themselves,
+            # so they are not prompts the template stopped accepting.
+            ignore_variables = getattr(template, 'allow_overwrite_flow_vars_on_relaunch', False)
+            if config.has_unprompted(template, ignore_variables=ignore_variables):
                 raise PermissionDenied(_('Job was launched with prompts no longer accepted.'))
 
         return True  # passed config checks

ascender/main/migrations/0220_workflow_allow_overwrite_flow_vars_on_relaunch.py

@@ -0,0 +1,26 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('main', '0219_source_project_update_set_null'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='workflowjob',
+            name='allow_overwrite_flow_vars_on_relaunch',
+            field=models.BooleanField(
+                default=False,
+                help_text='Allow a relaunch from failed nodes to be given variables that overwrite the ones carried over from the original run.',
+            ),
+        ),
+        migrations.AddField(
+            model_name='workflowjobtemplate',
+            name='allow_overwrite_flow_vars_on_relaunch',
+            field=models.BooleanField(
+                default=False,
+                help_text='Allow a relaunch from failed nodes to be given variables that overwrite the ones carried over from the original run.',
+            ),
+        ),
+    ]

ascender/main/models/jobs.py

@@ -1150,15 +1150,17 @@ def has_user_prompts(self, template):
         """
         return self._has_user_prompts(template, only_unprompted=False)
 
-    def has_unprompted(self, template):
+    def has_unprompted(self, template, ignore_variables=False):
         """
         returns True if the template has set ask_ fields to False after
         launching with those prompts
         """
-        return self._has_user_prompts(template, only_unprompted=True)
+        return self._has_user_prompts(template, only_unprompted=True, ignore_variables=ignore_variables)
 
-    def _has_user_prompts(self, template, only_unprompted=True):
+    def _has_user_prompts(self, template, only_unprompted=True, ignore_variables=False):
         prompts = self.prompts_dict()
+        if ignore_variables:
+            prompts.pop('extra_vars', None)
         ask_mapping = template.get_ask_mapping()
         if template.survey_enabled and (not template.ask_variables_on_launch):
             ask_mapping.pop('extra_vars')

ascender/main/models/workflow.py

@@ -48,7 +48,8 @@
 from ascender.main.models.jobs import LaunchTimeConfigBase, LaunchTimeConfig, JobTemplate
 from ascender.main.models.credential import Credential
 from ascender.main.redact import REPLACE_STR
-from ascender.main.utils import ScheduleWorkflowManager
+from ascender.main.utils import ScheduleWorkflowManager, parse_yaml_or_json
+from ascender.main.utils.encryption import encrypt_dict
 
 __all__ = [
     'WorkflowJobTemplate',
@@ -667,6 +668,10 @@ class Meta:
         'InstanceGroup', related_name='workflow_job_instance_groups', blank=True, editable=False, through='WorkflowJobInstanceGroupMembership'
     )
     allow_simultaneous = models.BooleanField(default=False)
+    allow_overwrite_flow_vars_on_relaunch = models.BooleanField(
+        default=False,
+        help_text=_("Allow a relaunch from failed nodes to be given variables that overwrite the ones carried over from the original run."),
+    )
 
     extra_vars_dict = VarsDictProperty('extra_vars', True)
 
@@ -774,8 +779,10 @@ def mark_prior_succeeded_nodes_from(self, original):
                 new_node.ancestor_artifacts = artifacts
                 new_node.save(update_fields=['prior_run_succeeded', 'prior_run_elapsed', 'ancestor_artifacts'])
 
-    def create_relaunch_workflow_job(self, from_failed=False):
+    def create_relaunch_workflow_job(self, from_failed=False, extra_vars=None):
         new_workflow_job = self.copy_unified_job()
+        if extra_vars:
+            new_workflow_job.apply_relaunch_extra_vars(extra_vars)
         if self.unified_job_template_id is None:
             new_workflow_job.copy_nodes_from_original(original=self, mark_succeeded_as_prior=from_failed)
         elif from_failed:
@@ -1002,6 +1009,84 @@ def event_processing_finished(self):
     def has_unpartitioned_events(self):
         return False  # workflow jobs do not have events
 
+    def relaunch_password_variables(self):
+        """The keys of this run whose value is a survey password.
+
+        Both the ones the run itself recorded and the ones the template asks for
+        now: the survey may have been edited since, and a key that was a
+        password on either side is treated as one, so a relaunch never turns a
+        stored secret back into plain text."""
+        keys = set(self.survey_passwords or {})
+        wfjt = self.workflow_job_template
+        if wfjt is not None:
+            keys |= set(wfjt.survey_password_variables())
+        return keys
+
+    def validate_relaunch_extra_vars(self, extra_vars):
+        """Check the variables handed to a relaunch.
+
+        The answers the survey defines are validated as a launch validates them,
+        and $encrypted$ is refused wherever it is not a survey password. Every
+        other key is a plain workflow variable, and
+        allow_overwrite_flow_vars_on_relaunch is what decides whether those are
+        allowed at all. Returns a list of error strings, empty when the
+        variables can be used."""
+        errors = []
+        password_keys = self.relaunch_password_variables()
+        # $encrypted$ means "keep the value the run used", which only a survey
+        # password has. Anywhere else it is a reserved word rather than a value,
+        # as it is on launch.
+        reserved = set()
+        for key, value in extra_vars.items():
+            if value == REPLACE_STR and key not in password_keys:
+                reserved.add(key)
+                errors.append(_('"$encrypted$" is a reserved keyword, may not be used for {key}.').format(key=key))
+        wfjt = self.workflow_job_template
+        if wfjt is None or not (wfjt.survey_enabled and wfjt.survey_spec):
+            return errors
+        for survey_element in wfjt.survey_spec.get('spec', []):
+            key = survey_element.get('variable')
+            if key not in extra_vars or key in reserved:
+                continue
+            # A password answer left at $encrypted$ keeps its stored value, so
+            # there is nothing new to validate for it.
+            if survey_element.get('type') == 'password' and extra_vars[key] == REPLACE_STR:
+                continue
+            errors += wfjt._survey_element_validation(survey_element, extra_vars)
+        return errors
+
+    def apply_relaunch_extra_vars(self, extra_vars):
+        """Overwrite this job's variables with the ones given at relaunch time.
+
+        The job was copied from the original run, so its extra_vars already hold
+        what that run used; only the keys handed in here change. Survey password
+        answers are stored encrypted, exactly as they would be on a launch."""
+        password_keys = self.relaunch_password_variables()
+        # A password answer sent back as $encrypted$ is the run's own value
+        # coming home; leave the stored one alone. Every other key is taken as
+        # it was sent.
+        overrides = {key: value for key, value in extra_vars.items() if not (value == REPLACE_STR and key in password_keys)}
+        if not overrides:
+            return
+        changed_passwords = password_keys & set(overrides)
+        encrypt_dict(overrides, changed_passwords)
+        masked = {key: REPLACE_STR for key in changed_passwords}
+        merged = self.extra_vars_dict
+        merged.update(overrides)
+        self.extra_vars = json.dumps(merged)
+        self.survey_passwords = {**(self.survey_passwords or {}), **masked}
+        self.save(update_fields=['extra_vars', 'survey_passwords'])
+        # A later relaunch of this job rebuilds it from its launch config, not
+        # from extra_vars, so the overrides go there too. Otherwise relaunching
+        # the corrected run would bring back the values it replaced.
+        try:
+            config = self.launch_config
+        except ObjectDoesNotExist:
+            return
+        config.extra_data = {**parse_yaml_or_json(config.extra_data), **overrides}
+        config.survey_passwords = {**(config.survey_passwords or {}), **masked}
+        config.save(update_fields=['extra_data', 'survey_passwords'])
+
     def _get_parent_field_name(self):
         if self.job_template_id:
             # This is a workflow job which is a container for slice jobs