如何使用 Argo Workflows 将上一步输出作为输入？

Question

我正在尝试按照这些说明 (https://argoproj.github.io/argo-workflows/workflow-inputs/#using-previous-step-outputs-as-inputs) 格式化我的工作流程，但似乎无法正确完成。具体来说，我试图模仿“使用上一步输出作为输入”

我在下面包含了我的工作流程。在这个版本中，我添加了一个指向 inputs.artifacts 的路径，因为错误要求一个。我现在收到的错误是：

ATA[2022-02-28T14:14:45.933Z] Failed to submit workflow: templates.entrypoint.tasks.print1 templates.print1.inputs.artifacts.result.from not valid in inputs

有人可以告诉我如何更正此工作流程以使其正常工作吗？

---
{
   "apiVersion": "argoproj.io/v1alpha1",
   "kind": "Workflow",
   "metadata": {
      "annotations": {
         "workflows.argoproj.io/description": "Building from the ground up",
         "workflows.argoproj.io/version": ">= 3.1.0"
      },
      "labels": {
         "workflows.argoproj.io/archive-strategy": "false"
      },
      "name": "data-passing",
      "namespace": "sandbox"
   },
   "spec": {
      "artifactRepositoryRef": {
         "configMap": "my-config",
         "key": "data"
      },
      "entrypoint": "entrypoint",
      "nodeSelector": {
         "kubernetes.io/os": "linux"
      },
      "parallelism": 3,
      "securityContext": {
         "fsGroup": 2000,
         "fsGroupChangePolicy": "OnRootMismatch",
         "runAsGroup": 3000,
         "runAsNonRoot": true,
         "runAsUser": 1000
      },
      "templates": [
         {
            "container": {
               "args": [
                  "Hello World"
               ],
               "command": [
                  "cowsay"
               ],
               "image": "docker/whalesay:latest",
               "imagePullPolicy": "IfNotPresent"
            },
            "name": "whalesay",
            "outputs": {
               "artifacts": [
                  {
                     "name": "msg",
                     "path": "/tmp/raw"
                  }
               ]
            },
            "securityContext": {
               "fsGroup": 2000,
               "fsGroupChangePolicy": "OnRootMismatch",
               "runAsGroup": 3000,
               "runAsNonRoot": true,
               "runAsUser": 1000
            }
         },
         {
            "inputs": {
               "artifacts": [
                  {
                     "from": "{{tasks.whalesay.outputs.artifacts.msg}}",
                     "name": "result",
                     "path": "/tmp/raw"
                  }
               ]
            },
            "name": "print1",
            "script": {
               "command": [
                  "python"
               ],
               "image": "python:alpine3.6",
               "imagePullPolicy": "IfNotPresent",
               "source": "cat {{inputs.artifacts.result}}\n"
            },
            "securityContext": {
               "fsGroup": 2000,
               "fsGroupChangePolicy": "OnRootMismatch",
               "runAsGroup": 3000,
               "runAsNonRoot": true,
               "runAsUser": 1000
            }
         },
         {
            "dag": {
               "tasks": [
                  {
                     "name": "whalesay",
                     "template": "whalesay"
                  },
                  {
                     "arguments": {
                        "artifacts": [
                           {
                              "from": "{{tasks.whalesay.outputs.artifacts.msg}}",
                              "name": "result",
                              "path": "/tmp/raw"
                           }
                        ]
                     },
                     "dependencies": [
                        "whalesay"
                     ],
                     "name": "print1",
                     "template": "print1"
                  }
               ]
            },
            "name": "entrypoint"
         }
      ]
   }
}
...

Answer 1

可以在此处找到与 Argo developers/maintainers 非常相似的工作流程：

https://github.com/argoproj/argo-workflows/blob/master/examples/README.md#artifacts

Answer 2

在 print1 的 artifact 参数中，您应该只放置 name 和 from 参数

例如：

- name: print1
  arguments:
    artifacts: [{name: results, from: "{{tasks.whalesay.outputs.artifacts.msg}}"}]

然后在您的模板声明中，您应该将 name 和 path 放入您的工件输入中，如下所示：

- name: input1
  inputs:
    artifacts:
    - name: result
      path: /tmp/raw
...

这是有效的，因为在你的任务参数中（在 dag 声明中）你告诉程序你希望如何调用该输入以及从哪里提取它，并且在模板声明中你从 name 接收输入并告诉程序暂时把它放在哪里。（这是我自己的理解）

我看到的另一个问题是 print1 而不是打印到 stdout 或使用 sys 到运行 cat 命令，你直接运行 cat，这（我认为）不是可能的。

你应该做类似的事情

import sys

sys.stdout.write("{{inputs.artifacts.result}}\n")

或

import os

os.system("cat {{inputs.artifacts.result}}\n")

如何使用 Argo Workflows 将上一步输出作为输入？

How do I use Argo Workflows Using Previous Step Outputs As Inputs?

argo-workflows