Add TTL strategy to Argo WorkflowTemplates and update CronWorkflow to use schedules

Add a configurable ttlStrategy to Argo WorkflowTemplates so completed
workflows are automatically cleaned up after 7 days (matching the default
Kubernetes job TTL). Configurable via METAFLOW_ARGO_WORKFLOWS_TTL_SECONDS_AFTER_COMPLETION
env var; set to 0 to disable.

Also update CronWorkflow creation to use the `schedules` array field instead
of the deprecated singular `schedule` field, per Argo Workflows deprecation
notice.

Closes #1231, closes #2351

Author: Nissan Pow

metaflow/metaflow_config.py

@@ -473,6 +473,13 @@
 
 ARGO_WORKFLOWS_UI_URL = from_conf("ARGO_WORKFLOWS_UI_URL")
 
+# How long (in seconds) to keep completed Argo Workflows before auto-deletion.
+# Default: 7 days (same as Kubernetes jobs). Maps to Argo's ttlStrategy.
+# Set to 0 to disable TTL (workflows persist indefinitely).
+ARGO_WORKFLOWS_TTL_SECONDS_AFTER_COMPLETION = from_conf(
+    "ARGO_WORKFLOWS_TTL_SECONDS_AFTER_COMPLETION", 7 * 24 * 60 * 60
+)
+
 ##
 # Airflow Configuration
 ##

metaflow/plugins/argo/argo_client.py

@@ -325,7 +325,7 @@ def schedule_workflow_template(self, name, schedule=None, timezone=None):
             "metadata": {"name": name},
             "spec": {
                 "suspend": schedule is None,
-                "schedule": schedule,
+                "schedules": [schedule] if schedule is not None else [],
                 "timezone": timezone,
                 "failedJobsHistoryLimit": 10000,  # default is unfortunately 1
                 "successfulJobsHistoryLimit": 10000,  # default is unfortunately 3

metaflow/plugins/argo/argo_workflows.py

@@ -25,6 +25,7 @@
     ARGO_WORKFLOWS_CAPTURE_ERROR_SCRIPT,
     ARGO_WORKFLOWS_ENV_VARS_TO_SKIP,
     ARGO_WORKFLOWS_KUBERNETES_SECRETS,
+    ARGO_WORKFLOWS_TTL_SECONDS_AFTER_COMPLETION,
     ARGO_WORKFLOWS_UI_URL,
     AWS_SECRETS_MANAGER_DEFAULT_REGION,
     AZURE_KEY_VAULT_PREFIX,
@@ -897,6 +898,8 @@ def _compile_workflow_template(self):
                 WorkflowSpec()
                 # Set overall workflow timeout.
                 .active_deadline_seconds(self.workflow_timeout)
+                # Set TTL for completed workflows (default: 7 days).
+                .ttl_strategy(ARGO_WORKFLOWS_TTL_SECONDS_AFTER_COMPLETION)
                 # TODO: Allow Argo to optionally archive all workflow execution logs
                 #       It's disabled for now since it requires all Argo installations
                 #       to enable an artifactory repository. If log archival is
@@ -4211,6 +4214,14 @@ def hooks(self, hooks):
             self.payload["hooks"].update({k: v.to_json()})
         return self
 
+    def ttl_strategy(self, seconds_after_completion):
+        # https://argoproj.github.io/argo-workflows/fields/#ttlstrategy
+        if seconds_after_completion is not None and int(seconds_after_completion) > 0:
+            self.payload["ttlStrategy"] = {
+                "secondsAfterCompletion": int(seconds_after_completion)
+            }
+        return self
+
     def to_json(self):
         return self.payload