如何在嵌套的 argo 步骤中传递参数？答案

【问题标题】：How to pass parameters in nested argo steps?如何在嵌套的 argo 步骤中传递参数？
【发布时间】：2021-11-29 08:01:46
【问题描述】：

我正在尝试将参数从外部步骤模板传递到 argo 中的内部步骤模板。以下是我的工作流程定义。

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: fanout-
spec:
  templates:
    - name: process-workflows
      steps:
        - - name: fanout
            template: fan-out
        - - name: fanout-step
            template: parallel-process
            arguments:
              parameters:
                - name: batch
                  value: '{{item}}'
            withParam: '{{steps.fanout.outputs.result}}'
    - name: fan-out
      script:
        name: main
        image: 'node:lts-alpine3.14'
        command:
          - node
        resources: {}
        source: |
          inputlist = JSON.parse({{=toJson(workflow.parameters.inputlist)}})
          fanout = {{workflow.parameters.fanout}}
          var i,j, result=[];
          for (i = 0,j = inputlist.length; i < j; i += fanout) {
              result.push(inputlist.slice(i, i + fanout));
          }
          console.log(JSON.stringify(result))
    - name: parallel-process
      inputs:
        parameters:
          - name: batch
      steps:
        - - name: actualprocessor
            template: process
            arguments:
              parameters:
                - name: input
                  value: {{inputs.parameters.batch}}
        - - name: aggregate-result
            template: aggregate
            arguments:
              parameters:
                - name: aggregate
                  value: {{steps.actualprocessor.outputs.parameters.res}}
    - name: process
      inputs:
        parameters:
          - name: input
      outputs:
        parameters:
          - name: res
            valueFrom:
              path: /tmp/res.txt
      script:
        name: main
        image: 'alpine:latest'
        command:
          - sh
        source: |
          sleep 5
          echo 'awakened...'
          echo processing-{{=toJson(inputs.parameters.input)}}
          echo {{=toJson(inputs.parameters.input)}} > /tmp/res.txt
    - name: aggregate
      inputs:
        parameters:
          - name: aggregate
      container:
        name: main
        image: 'alpine:latest'
        command:
          - sh
          - '-c'
        args:
          - 'echo received {{inputs.parameters.aggregate}}'
  entrypoint: process-workflows
  arguments:
    parameters:
      - name: inputlist
        value: |
          [
            {"k" : "v1", "a" : [{ "k": true}]},
            {"k" : "v2", "a" : [{ "k": true}]}
          ]
      - name: fanout
        value: '1'

用例：

扇出步骤（外部步骤）使用并行处理模板（内部步骤）。它为并行进程模板提供了一个batch 参数。并行处理模板需要将batch的值提供给目标步骤中的input参数。

问题：实际处理器步骤中的输入参数为空。我可以看到批处理输入参数已正确填充。

我在这里错过了什么？

【问题讨论】：

我能找到的唯一问题是一些模板没有被引用。这是您使用的确切 YAML 吗？
是的，这个模板可以通过argo UI提交。
您的意思是{{inputs.parameters.batch}} 在actualprocessor 启动时为空？你能试试"{{inputs.parameters.batch}}"吗？我记得当参数解析器没有包含在字符串中时遇到问题
谢谢，@TomSlabbaert 引用似乎是缺失的部分。

标签： argo-workflows

【解决方案1】：

通过将参数括在引号中来解决问题。感谢 Tom Slabbaer 指出。

下面是工作模板。

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: fanout-
spec:
  templates:
    - name: process-workflows
      steps:
        - - name: fanout
            template: fan-out
        - - name: fanout-step
            template: parallel-process
            arguments:
              parameters:
                - name: batch
                  value: '{{item}}'
            withParam: '{{steps.fanout.outputs.result}}'
    - name: fan-out
      script:
        name: main
        image: 'node:lts-alpine3.14'
        command:
          - node
        resources: {}
        source: |
          inputlist = JSON.parse({{=toJson(workflow.parameters.inputlist)}})
          fanout = {{workflow.parameters.fanout}}
          var i,j, result=[];
          for (i = 0,j = inputlist.length; i < j; i += fanout) {
              result.push(inputlist.slice(i, i + fanout));
          }
          console.log(JSON.stringify(result))
    - name: parallel-process
      inputs:
        parameters:
          - name: batch
      steps:
        - - name: actualprocessor
            template: process
            arguments:
              parameters:
                - name: input
                  value: "{{inputs.parameters.batch}}"
        - - name: aggregate-result
            template: aggregate
            arguments:
              parameters:
                - name: aggregate
                  value: "{{steps.actualprocessor.outputs.parameters.res}}"
    - name: process
      inputs:
        parameters:
          - name: input
      outputs:
        parameters:
          - name: res
            valueFrom:
              path: /tmp/res.txt
      script:
        name: main
        image: 'alpine:latest'
        command:
          - sh
        source: |
          sleep 5
          echo 'awakened...'
          echo processing-{{=toJson(inputs.parameters.input)}}
          echo {{=toJson(inputs.parameters.input)}} > /tmp/res.txt
    - name: aggregate
      inputs:
        parameters:
          - name: aggregate
      container:
        name: main
        image: 'alpine:latest'
        command:
          - sh
          - '-c'
        args:
          - 'echo received {{inputs.parameters.aggregate}}'
  entrypoint: process-workflows
  arguments:
    parameters:
      - name: inputlist
        value: |
          [
            {"k" : "v1", "a" : [{ "k": true}]},
            {"k" : "v2", "a" : [{ "k": true}]}
          ]
      - name: fanout
        value: '1'

【讨论】：