Skip to content

Commit d35334f

Browse files
authored
chore: re-enable conditionals in foreach on argo workflows (#2560)
draft for now, as this approach had a bunch of issues caused by the "leftover" template literals in input-paths causing havoc to argo adjacent systems.
1 parent 1dbaada commit d35334f

2 files changed

Lines changed: 23 additions & 37 deletions

File tree

‎metaflow/plugins/argo/argo_workflows.py‎

Lines changed: 17 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -1156,32 +1156,15 @@ def _visit(
11561156
else:
11571157
# Every other node needs only input-paths
11581158
parameters = [
1159-
(
1160-
Parameter("input-paths").value(
1161-
compress_list(
1162-
[
1163-
"argo-{{workflow.name}}/%s/{{tasks.%s.outputs.parameters.task-id}}"
1164-
% (n, self._sanitize(n))
1165-
for n in node.in_funcs
1166-
],
1167-
# NOTE: We set zlibmin to infinite because zlib compression for the Argo input-paths breaks template value substitution.
1168-
zlibmin=inf,
1169-
)
1170-
)
1171-
if not self._is_conditional_join_node(node)
1172-
# The value fetching for input-paths from conditional steps has to be quite involved
1173-
# in order to avoid issues with replacements due to missing step outputs.
1174-
# NOTE: we differentiate the input-path expression only for conditional joins so we can still utilize the list compression,
1175-
# but do not have to rework all decompress usage due to the need for a custom separator
1176-
else Parameter("input-paths").value(
1177-
compress_list(
1178-
[
1179-
"argo-{{workflow.name}}/%s/{{=(get(tasks['%s']?.outputs?.parameters, 'task-id') ?? 'no-task')}}"
1180-
% (n, self._sanitize(n))
1181-
for n in node.in_funcs
1182-
],
1183-
separator="%", # non-default separator is required due to commas in the value expression
1184-
)
1159+
Parameter("input-paths").value(
1160+
compress_list(
1161+
[
1162+
"argo-{{workflow.name}}/%s/{{tasks.%s.outputs.parameters.task-id}}"
1163+
% (n, self._sanitize(n))
1164+
for n in node.in_funcs
1165+
],
1166+
# NOTE: We set zlibmin to infinite because zlib compression for the Argo input-paths breaks template value substitution.
1167+
zlibmin=inf,
11851168
)
11861169
)
11871170
]
@@ -1279,13 +1262,6 @@ def _visit(
12791262
parent_foreach,
12801263
)
12811264
elif node.type == "split-switch":
1282-
if node.is_inside_foreach:
1283-
# TODO: Fix this. The issue is with conditional branches nested inside a foreach branch. The value expression for the input-paths parameter
1284-
# on Argo fails completely for the nested structure (though the identical shape outside of nesting works fine).
1285-
raise MetaflowException(
1286-
"*%s* is a switch step inside a foreach. Conditional steps are not supported inside a foreach on Argo Workflows yet."
1287-
% node.name
1288-
)
12891265
for n in node.out_funcs:
12901266
_visit(
12911267
self.graph[n],
@@ -1602,6 +1578,14 @@ def _container_templates(self):
16021578
input_paths_expr = (
16031579
"export INPUT_PATHS={{inputs.parameters.input-paths}}"
16041580
)
1581+
if self._is_conditional_join_node(node):
1582+
# NOTE: Argo template expressions that fail to resolve, output the expression itself as a value.
1583+
# With conditional steps, some of the input-paths are therefore 'broken' due to containing a nil expression
1584+
# e.g. "{{ tasks['A'].outputs.parameters.task-id }}" when task A never executed.
1585+
# We base64 encode the input-paths in order to not pollute the execution environment with templating expressions.
1586+
# NOTE: Adding conditionals that check if a key exists or not does not work either, due to an issue with how Argo
1587+
# handles tasks in a nested foreach (withParam template) leading to all such expressions getting evaluated as false.
1588+
input_paths_expr = "export INPUT_PATHS={{=toBase64(inputs.parameters['input-paths'])}}"
16051589
input_paths = "$(echo $INPUT_PATHS)"
16061590
if any(self.graph[n].type == "foreach" for n in node.in_funcs):
16071591
task_idx = "{{inputs.parameters.split-index}}"
@@ -1617,7 +1601,6 @@ def _container_templates(self):
16171601
# foreaches
16181602
task_idx = "{{inputs.parameters.split-index}}"
16191603
root_input = "{{inputs.parameters.root-input-path}}"
1620-
16211604
# Task string to be hashed into an ID
16221605
task_str = "-".join(
16231606
[

‎metaflow/plugins/argo/conditional_input_paths.py‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,19 @@
11
from math import inf
22
import sys
33
from metaflow.util import decompress_list, compress_list
4+
import base64
45

56

67
def generate_input_paths(input_paths):
7-
# Note the non-default separator due to difficulties setting parameter values from conditional step outputs.
8-
paths = decompress_list(input_paths, separator="%")
8+
# => run_id/step/:foo,bar
9+
# input_paths are base64 encoded due to Argo shenanigans
10+
decoded = base64.b64decode(input_paths).decode("utf-8")
11+
paths = decompress_list(decoded)
912

1013
# some of the paths are going to be malformed due to never having executed per conditional.
1114
# strip these out of the list.
1215

13-
trimmed = [path for path in paths if not path.endswith("/no-task")]
16+
trimmed = [path for path in paths if not "{{" in path]
1417
return compress_list(trimmed, zlibmin=inf)
1518

1619

0 commit comments

Comments
 (0)