feat: Support for data sourcing and transformation with data template

Makefile

@@ -593,6 +593,7 @@ prepare-release: check-version-warning clean codegen manifests
 publish-release: check-version-warning clis checksums
 	git push
 	git push $(GIT_REMOTE) $(VERSION)
+
 endif
 
 .PHONY: check-version-warning

api/jsonschema/schema.json

@@ -2811,6 +2811,87 @@
       },
       "type": "object"
     },
+    "io.argoproj.workflow.v1alpha1.ArtifactPaths": {
+      "description": "ArtifactPaths expands a step from a collection of artifacts",
+      "properties": {
+        "archive": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArchiveStrategy",
+          "description": "Archive controls how the artifact will be saved to the artifact repository."
+        },
+        "archiveLogs": {
+          "description": "ArchiveLogs indicates if the container logs should be archived",
+          "type": "boolean"
+        },
+        "artifactory": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArtifactoryArtifact",
+          "description": "Artifactory contains artifactory artifact location details"
+        },
+        "from": {
+          "description": "From allows an artifact to reference an artifact from a previous step",
+          "type": "string"
+        },
+        "gcs": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.GCSArtifact",
+          "description": "GCS contains GCS artifact location details"
+        },
+        "git": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.GitArtifact",
+          "description": "Git contains git artifact location details"
+        },
+        "globalName": {
+          "description": "GlobalName exports an output artifact to the global scope, making it available as '{{io.argoproj.workflow.v1alpha1.outputs.artifacts.XXXX}} and in workflow.status.outputs.artifacts",
+          "type": "string"
+        },
+        "hdfs": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.HDFSArtifact",
+          "description": "HDFS contains HDFS artifact location details"
+        },
+        "http": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.HTTPArtifact",
+          "description": "HTTP contains HTTP artifact location details"
+        },
+        "mode": {
+          "description": "mode bits to use on this file, must be a value between 0 and 0777 set when loading input artifacts.",
+          "type": "integer"
+        },
+        "name": {
+          "description": "name of the artifact. must be unique within a template's inputs/outputs.",
+          "type": "string"
+        },
+        "optional": {
+          "description": "Make Artifacts optional, if Artifacts doesn't generate or exist",
+          "type": "boolean"
+        },
+        "oss": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.OSSArtifact",
+          "description": "OSS contains OSS artifact location details"
+        },
+        "path": {
+          "description": "Path is the container path to the artifact",
+          "type": "string"
+        },
+        "raw": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.RawArtifact",
+          "description": "Raw contains raw artifact location details"
+        },
+        "recurseMode": {
+          "description": "If mode is set, apply the permission recursively into the artifact if it is a folder",
+          "type": "boolean"
+        },
+        "s3": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.S3Artifact",
+          "description": "S3 contains S3 artifact location details"
+        },
+        "subPath": {
+          "description": "SubPath allows an artifact to be sourced from a subpath within the specified source",
+          "type": "string"
+        }
+      },
+      "required": [
+        "name"
+      ],
+      "type": "object"
+    },
     "io.argoproj.workflow.v1alpha1.ArtifactRepositoryRef": {
       "properties": {
         "configMap": {
@@ -3298,6 +3379,37 @@
       ],
       "type": "object"
     },
+    "io.argoproj.workflow.v1alpha1.Data": {
+      "description": "Data is a data template",
+      "properties": {
+        "source": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.DataSource",
+          "description": "Source sources external data into a data template"
+        },
+        "transformation": {
+          "description": "Transformation applies a set of transformations",
+          "items": {
+            "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.TransformationStep"
+          },
+          "type": "array"
+        }
+      },
+      "required": [
+        "source",
+        "transformation"
+      ],
+      "type": "object"
+    },
+    "io.argoproj.workflow.v1alpha1.DataSource": {
+      "description": "DataSource sources external data into a data template",
+      "properties": {
+        "artifactPaths": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArtifactPaths",
+          "description": "ArtifactPaths is a data transformation that collects a list of artifact paths"
+        }
+      },
+      "type": "object"
+    },
     "io.argoproj.workflow.v1alpha1.Event": {
       "properties": {
         "selector": {
@@ -4198,9 +4310,6 @@
           "type": "boolean"
         }
       },
-      "required": [
-        "key"
-      ],
       "type": "object"
     },
     "io.argoproj.workflow.v1alpha1.ScriptTemplate": {
@@ -4577,6 +4686,10 @@
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.DAGTemplate",
           "description": "DAG template subtype which runs a DAG"
         },
+        "data": {
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.Data",
+          "description": "Data is a data template"
+        },
         "executor": {
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ExecutorConfig",
           "description": "Executor holds configurations of the executor container."
@@ -4740,6 +4853,18 @@
       },
       "type": "object"
     },
+    "io.argoproj.workflow.v1alpha1.TransformationStep": {
+      "properties": {
+        "expression": {
+          "description": "Expression defines an expr expression to apply",
+          "type": "string"
+        }
+      },
+      "required": [
+        "expression"
+      ],
+      "type": "object"
+    },
     "io.argoproj.workflow.v1alpha1.UpdateCronWorkflowRequest": {
       "properties": {
         "cronWorkflow": {

api/openapi-spec/swagger.json

@@ -6229,6 +6229,87 @@
         }
       }
     },
+    "io.argoproj.workflow.v1alpha1.ArtifactPaths": {
+      "description": "ArtifactPaths expands a step from a collection of artifacts",
+      "type": "object",
+      "required": [
+        "name"
+      ],
+      "properties": {
+        "archive": {
+          "description": "Archive controls how the artifact will be saved to the artifact repository.",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArchiveStrategy"
+        },
+        "archiveLogs": {
+          "description": "ArchiveLogs indicates if the container logs should be archived",
+          "type": "boolean"
+        },
+        "artifactory": {
+          "description": "Artifactory contains artifactory artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArtifactoryArtifact"
+        },
+        "from": {
+          "description": "From allows an artifact to reference an artifact from a previous step",
+          "type": "string"
+        },
+        "gcs": {
+          "description": "GCS contains GCS artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.GCSArtifact"
+        },
+        "git": {
+          "description": "Git contains git artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.GitArtifact"
+        },
+        "globalName": {
+          "description": "GlobalName exports an output artifact to the global scope, making it available as '{{io.argoproj.workflow.v1alpha1.outputs.artifacts.XXXX}} and in workflow.status.outputs.artifacts",
+          "type": "string"
+        },
+        "hdfs": {
+          "description": "HDFS contains HDFS artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.HDFSArtifact"
+        },
+        "http": {
+          "description": "HTTP contains HTTP artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.HTTPArtifact"
+        },
+        "mode": {
+          "description": "mode bits to use on this file, must be a value between 0 and 0777 set when loading input artifacts.",
+          "type": "integer"
+        },
+        "name": {
+          "description": "name of the artifact. must be unique within a template's inputs/outputs.",
+          "type": "string"
+        },
+        "optional": {
+          "description": "Make Artifacts optional, if Artifacts doesn't generate or exist",
+          "type": "boolean"
+        },
+        "oss": {
+          "description": "OSS contains OSS artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.OSSArtifact"
+        },
+        "path": {
+          "description": "Path is the container path to the artifact",
+          "type": "string"
+        },
+        "raw": {
+          "description": "Raw contains raw artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.RawArtifact"
+        },
+        "recurseMode": {
+          "description": "If mode is set, apply the permission recursively into the artifact if it is a folder",
+          "type": "boolean"
+        },
+        "s3": {
+          "description": "S3 contains S3 artifact location details",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.S3Artifact"
+        },
+        "subPath": {
+          "description": "SubPath allows an artifact to be sourced from a subpath within the specified source",
+          "type": "string"
+        }
+      }
+    },
     "io.argoproj.workflow.v1alpha1.ArtifactRepositoryRef": {
       "type": "object",
       "properties": {
@@ -6712,6 +6793,37 @@
         }
       }
     },
+    "io.argoproj.workflow.v1alpha1.Data": {
+      "description": "Data is a data template",
+      "type": "object",
+      "required": [
+        "source",
+        "transformation"
+      ],
+      "properties": {
+        "source": {
+          "description": "Source sources external data into a data template",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.DataSource"
+        },
+        "transformation": {
+          "description": "Transformation applies a set of transformations",
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.TransformationStep"
+          }
+        }
+      }
+    },
+    "io.argoproj.workflow.v1alpha1.DataSource": {
+      "description": "DataSource sources external data into a data template",
+      "type": "object",
+      "properties": {
+        "artifactPaths": {
+          "description": "ArtifactPaths is a data transformation that collects a list of artifact paths",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArtifactPaths"
+        }
+      }
+    },
     "io.argoproj.workflow.v1alpha1.Event": {
       "type": "object",
       "required": [
@@ -7571,9 +7683,6 @@
     "io.argoproj.workflow.v1alpha1.S3Artifact": {
       "description": "S3Artifact is the location of an S3 artifact",
       "type": "object",
-      "required": [
-        "key"
-      ],
       "properties": {
         "accessKeySecret": {
           "description": "AccessKeySecret is the secret selector to the bucket's access key",
@@ -7995,6 +8104,10 @@
           "description": "DAG template subtype which runs a DAG",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.DAGTemplate"
         },
+        "data": {
+          "description": "Data is a data template",
+          "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.Data"
+        },
         "executor": {
           "description": "Executor holds configurations of the executor container.",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ExecutorConfig"
@@ -8154,6 +8267,18 @@
         }
       }
     },
+    "io.argoproj.workflow.v1alpha1.TransformationStep": {
+      "type": "object",
+      "required": [
+        "expression"
+      ],
+      "properties": {
+        "expression": {
+          "description": "Expression defines an expr expression to apply",
+          "type": "string"
+        }
+      }
+    },
     "io.argoproj.workflow.v1alpha1.UpdateCronWorkflowRequest": {
       "type": "object",
       "properties": {

cmd/argoexec/commands/data.go

@@ -0,0 +1,20 @@
+package commands
+
+import (
+	"context"
+
+	"github.com/spf13/cobra"
+)
+
+func NewDataCommand() *cobra.Command {
+	command := cobra.Command{
+		Use:   "data",
+		Short: "Process data",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := context.Background()
+			wfExecutor := initExecutor()
+			return wfExecutor.Data(ctx)
+		},
+	}
+	return &command
+}

cmd/argoexec/commands/root.go

@@ -64,6 +64,7 @@ func NewRootCommand() *cobra.Command {
 	command.AddCommand(NewInitCommand())
 	command.AddCommand(NewResourceCommand())
 	command.AddCommand(NewWaitCommand())
+	command.AddCommand(NewDataCommand())
 	command.AddCommand(cmd.NewVersionCmd(CLIName))
 
 	clientConfig = kubecli.AddKubectlFlagsToCmd(&command)

config/config.go

@@ -365,7 +365,7 @@ type MetricsConfig struct {
 }
 
 type WorkflowRestrictions struct {
-	TemplateReferencing TemplateReferencing `json:"templateReferencing"`
+	TemplateReferencing TemplateReferencing `json:"templateReferencing,omitempty"`
 }
 
 type TemplateReferencing string