Implementation for resource templates and resource wait/fail conditions

Makefile

@@ -58,11 +58,11 @@ cli:
  go build -v -i -ldflags '${LDFLAGS}' -o ${DIST_DIR}/argo ./cmd/argo
 
 cli-linux: builder
- ${BUILDER_CMD} make cli IMAGE_TAG=$(IMAGE_TAG)
+ ${BUILDER_CMD} make cli IMAGE_TAG=$(IMAGE_TAG) IMAGE_NAMESPACE=$(IMAGE_NAMESPACE)
  mv ${DIST_DIR}/argo ${DIST_DIR}/argo-linux-amd64
 
 cli-darwin: builder
- ${BUILDER_CMD} make cli GOOS=darwin IMAGE_TAG=$(IMAGE_TAG)
+ ${BUILDER_CMD} make cli GOOS=darwin IMAGE_TAG=$(IMAGE_TAG) IMAGE_NAMESPACE=$(IMAGE_NAMESPACE)
  mv ${DIST_DIR}/argo ${DIST_DIR}/argo-darwin-amd64
 
 controller:
@@ -73,7 +73,7 @@ controller-linux: builder
 
 controller-image: controller-linux
  docker build -t $(IMAGE_PREFIX)workflow-controller:$(IMAGE_TAG) -f Dockerfile-workflow-controller .
- if [ "$(DOCKER_PUSH)" = "true" ] ; then docker push $(IMAGE_PREFIX)workflow-controller:$(IMAGE_TAG) ; fi
+ @if [ "$(DOCKER_PUSH)" = "true" ] ; then docker push $(IMAGE_PREFIX)workflow-controller:$(IMAGE_TAG) ; fi
 
 executor:
  go build -v -i -ldflags '${LDFLAGS}' -o ${DIST_DIR}/argoexec ./cmd/argoexec
@@ -83,7 +83,7 @@ executor-linux: builder
 
 executor-image: executor-linux
  docker build -t $(IMAGE_PREFIX)argoexec:$(IMAGE_TAG) -f Dockerfile-argoexec .
- if [ "$(DOCKER_PUSH)" = "true" ] ; then docker push $(IMAGE_PREFIX)argoexec:$(IMAGE_TAG) ; fi
+ @if [ "$(DOCKER_PUSH)" = "true" ] ; then docker push $(IMAGE_PREFIX)argoexec:$(IMAGE_TAG) ; fi
 
 lint:
  gometalinter --deadline 2m --config gometalinter.json --vendor ./...
@@ -103,11 +103,11 @@ ui-image:
  docker cp argo-ui-builder:/src/node_modules ./ui/tmp
  docker rm argo-ui-builder
  docker build -t $(IMAGE_PREFIX)argoui:$(IMAGE_TAG) -f ui/Dockerfile ui
- if [ "$(DOCKER_PUSH)" = "true" ] ; then docker push $(IMAGE_PREFIX)argoui:$(IMAGE_TAG) ; fi
+ @if [ "$(DOCKER_PUSH)" = "true" ] ; then docker push $(IMAGE_PREFIX)argoui:$(IMAGE_TAG) ; fi
 
 release-precheck:
- @if [ "$(TREE_STATE)" != "clean" ]; then echo 'git tree state is $(TREE_STATE)' ; exit 1; fi
- @if [ -z "$(TAG)" ]; then echo 'commit must be tagged to perform release' ; exit 1; fi
+ @if [ "$(GIT_TREE_STATE)" != "clean" ]; then echo 'git tree state is $(GIT_TREE_STATE)' ; exit 1; fi
+ @if [ -z "$(GIT_TAG)" ]; then echo 'commit must be tagged to perform release' ; exit 1; fi
 
 release: release-precheck controller-image cli-darwin cli-linux executor-image ui-image
 

api/workflow/v1alpha1/types.go

@@ -19,6 +19,17 @@ const (
  CRDFullName string = CRDPlural + "." + CRDGroup
 )
 
+// TemplateType is the type of a template
+type TemplateType string
+
+// Possible template types
+const (
+ TemplateTypeContainer TemplateType = "Container"
+ TemplateTypeSteps TemplateType = "Steps"
+ TemplateTypeScript TemplateType = "Script"
+ TemplateTypeResource TemplateType = "Resource"
+)
+
 // NodePhase is a label for the condition of a node at the current time.
 type NodePhase string
 
@@ -104,6 +115,8 @@ type Template struct {
  // Sidecar containers
  Sidecars []Sidecar `json:"sidecars,omitempty"`
 
+ Resource *ResourceTemplate `json:"resource,omitempty"`
+
  // Location in which all files related to the step will be stored (logs, artifacts, etc...).
  // Can be overridden by individual items in Outputs. If omitted, will use the default
  // artifact repository location configured in the controller, appended with the
@@ -127,7 +140,9 @@ type Parameter struct {
  Name string `json:"name"`
  Value *string `json:"value,omitempty"`
  Default *string `json:"default,omitempty"`
- Path string `json:"path,omitempty"`
+
+ // Path describes the location in which to retrieve the output parameter value from
+ Path string `json:"path,omitempty"`
 }
 
 // Artifact indicates an artifact to place at a specified path
@@ -159,11 +174,14 @@ type ArtifactLocation struct {
 }
 
 type Outputs struct {
+ // Parameters holds the list of output parameters produced by a step
  Parameters []Parameter `json:"parameters,omitempty"`
- Artifacts []Artifact `json:"artifacts,omitempty"`
- Result *string `json:"result,omitempty"`
- // TODO:
- // - Logs (log artifact(s) from the container?)
+
+ // Artifacts holds the list of output artifacts produced by a step
+ Artifacts []Artifact `json:"artifacts,omitempty"`
+
+ // Result holds the result (stdout) of a script template
+ Result *string `json:"result,omitempty"`
 }
 
 // WorkflowStep is a template ref
@@ -320,6 +338,40 @@ type Script struct {
  Source string `json:"source"`
 }
 
+// ResourceTemplate is a template subtype to manipulate kubernetes resources
+type ResourceTemplate struct {
+ // Action is the action to perform to the resource.
+ // Must be one of: create, apply, delete
+ Action string `json:"action"`
+
+ // Manifest contains the kubernetes manifest
+ Manifest string `json:"manifest"`
+
+ // SuccessCondition is a label selector expression which describes the conditions
+ // of the k8s resource in which it is acceptable to proceed to the following step
+ SuccessCondition string `json:"successCondition,omitempty"`
+
+ // FailureCondition is a label selector expression which describes the conditions
+ // of the k8s resource in which the step was considered failed
+ FailureCondition string `json:"failureCondition,omitempty"`
+}
+
+func (tmpl *Template) GetType() TemplateType {
+ if tmpl.Container != nil {
+ return TemplateTypeContainer
+ }
+ if tmpl.Steps != nil {
+ return TemplateTypeSteps
+ }
+ if tmpl.Script != nil {
+ return TemplateTypeScript
+ }
+ if tmpl.Resource != nil {
+ return TemplateTypeResource
+ }
+ return "Unknown"
+}
+
 func (in *Inputs) GetArtifactByName(name string) *Artifact {
  for _, art := range in.Artifacts {
  if art.Name == name {

cmd/argoexec/commands/init.go

@@ -19,10 +19,10 @@ var initCmd = &cobra.Command{
 func loadArtifacts(cmd *cobra.Command, args []string) {
  wfExecutor := initExecutor()
  // Download input artifacts
- err := wfExecutor.LoadScriptSource()
+ err := wfExecutor.StageFiles()
  if err != nil {
  _ = wfExecutor.AddAnnotation(common.AnnotationKeyNodeMessage, err.Error())
- log.Fatalf("Error loading script: %+v", err)
+ log.Fatalf("Error loading staging files: %+v", err)
  }
  err = wfExecutor.LoadArtifacts()
  if err != nil {

cmd/argoexec/commands/resource.go

@@ -0,0 +1,43 @@
+package commands
+
+import (
+ "os"
+
+ "github.com/argoproj/argo/workflow/common"
+ log "github.com/sirupsen/logrus"
+ "github.com/spf13/cobra"
+)
+
+func init() {
+ RootCmd.AddCommand(resourceCmd)
+}
+
+var resourceCmd = &cobra.Command{
+ Use: "resource (get|create|apply|delete) MANIFEST",
+ Short: "update a resource and wait for resource conditions",
+ Run: execResource,
+}
+
+func execResource(cmd *cobra.Command, args []string) {
+ if len(args) != 1 {
+ cmd.HelpFunc()(cmd, args)
+ os.Exit(1)
+ }
+
+ wfExecutor := initExecutor()
+ err := wfExecutor.StageFiles()
+ if err != nil {
+ _ = wfExecutor.AddAnnotation(common.AnnotationKeyNodeMessage, err.Error())
+ log.Fatalf("Error staing resource: %+v", err)
+ }
+ resourceName, err := wfExecutor.ExecResource(args[0], common.ExecutorResourceManifestPath)
+ if err != nil {
+ _ = wfExecutor.AddAnnotation(common.AnnotationKeyNodeMessage, err.Error())
+ log.Fatalf("Error running %s resource: %+v", args[0], err)
+ }
+ err = wfExecutor.WaitResource(resourceName)
+ if err != nil {
+ _ = wfExecutor.AddAnnotation(common.AnnotationKeyNodeMessage, err.Error())
+ log.Fatalf("Error waiting for resource %s: %+v", resourceName, err)
+ }
+}

cmd/argoexec/commands/root.go

@@ -20,6 +20,7 @@ import (
  "github.com/spf13/cobra"
  "k8s.io/client-go/kubernetes"
  "k8s.io/client-go/rest"
+ "k8s.io/client-go/tools/clientcmd"
 )
 
 const (
@@ -32,7 +33,14 @@ var (
  GlobalArgs globalFlags
 )
 
+type globalFlags struct {
+ podAnnotationsPath string // --pod-annotations
+ kubeConfig string // --kubeconfig
+}
+
 func init() {
+ RootCmd.PersistentFlags().StringVar(&GlobalArgs.kubeConfig, "kubeconfig", "", "Kubernetes config (used when running outside of cluster)")
+ RootCmd.PersistentFlags().StringVar(&GlobalArgs.podAnnotationsPath, "pod-annotations", common.PodMetadataAnnotationsPath, "Pod annotations file from k8s downward API")
  RootCmd.AddCommand(cmd.NewVersionCmd(CLIName))
 }
 
@@ -45,14 +53,12 @@ var RootCmd = &cobra.Command{
  },
 }
 
-type globalFlags struct {
- hostIP string // --host-ip
- podAnnotationsPath string // --pod-annotations
-}
-
-func init() {
- RootCmd.PersistentFlags().StringVar(&GlobalArgs.hostIP, "host-ip", common.EnvVarHostIP, fmt.Sprintf("IP of host. (Default: %s)", common.EnvVarHostIP))
- RootCmd.PersistentFlags().StringVar(&GlobalArgs.podAnnotationsPath, "pod-annotations", common.PodMetadataAnnotationsPath, fmt.Sprintf("Pod annotations fiel from k8s downward API. (Default: %s)", common.PodMetadataAnnotationsPath))
+// getClientConfig return rest config, if path not specified, assume in cluster config
+func getClientConfig(kubeconfig string) (*rest.Config, error) {
+ if kubeconfig != "" {
+ return clientcmd.BuildConfigFromFlags("", kubeconfig)
+ }
+ return rest.InClusterConfig()
 }
 
 func initExecutor() *executor.WorkflowExecutor {
@@ -71,8 +77,7 @@ func initExecutor() *executor.WorkflowExecutor {
  log.Fatalf("Error getting template %v", err)
  }
 
- // Initialize in-cluster Kubernetes client
- config, err := rest.InClusterConfig()
+ config, err := getClientConfig(GlobalArgs.kubeConfig)
  if err != nil {
  panic(err.Error())
  }

examples/k8s-jobs.yaml

@@ -0,0 +1,41 @@
+# This example demonstrates the 'resource' template type, which provides a
+# convenient way to create/update/delete any type of kubernetes resources
+# in a workflow. The resource template type accepts any k8s manifest
+# (including CRDs) and can perform any kubectl action against it (e.g. create,
+# apply, delete, patch).
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+ generateName: k8s-jobs-
+spec:
+ entrypoint: pi-tmpl
+ templates:
+ - name: pi-tmpl
+ resource:
+ action: create
+ # successCondition and failureCondition are expressions which are evaluated
+ # upon every update of the resource. If failureCondition is ever evaluated
+ # to true, the step is considered failed. Likewise, if successCondition is
+ # ever evaluated to true the step is considered successful. It uses kubernetes
+ # label selection syntax and can be applied against any field of the resource
+ # (not just labels). Multiple AND conditions can be represented by comma
+ # delimited expressions. For more details, see:
+ # https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+ successCondition: status.succeeded > 0
+ failureCondition: status.failed > 3
+ manifest: |
+ apiVersion: batch/v1
+ kind: Job
+ metadata:
+ generateName: pi-job-
+ spec:
+ template:
+ metadata:
+ name: pi
+ spec:
+ containers:
+ - name: pi
+ image: perl
+ command: ["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"]
+ restartPolicy: Never
+ backoffLimit: 4

workflow/common/common.go

@@ -67,17 +67,15 @@ const (
  // (when there is overlapping paths between artifacts and volume mounts)
  InitContainerMainFilesystemDir = "/mainctrfs"
 
- // ScriptTemplateEmptyDir is the path of the emptydir which will be shared between init/main container for script templates
- ScriptTemplateEmptyDir = "/argo/script"
- // ScriptTemplateSourcePath is the path which init will write the source file to and the main container will execute
- ScriptTemplateSourcePath = "/argo/script/source"
+ // ExecutorStagingEmptyDir is the path of the emptydir which is used as a staging area to transfer a file between init/main container for script/resource templates
+ ExecutorStagingEmptyDir = "/argo/staging"
+ // ExecutorScriptSourcePath is the path which init will write the script source file to for script templates
+ ExecutorScriptSourcePath = "/argo/staging/script"
+ // ExecutorResourceManifestPath is the path which init will write the a manifest file to for resource templates
+ ExecutorResourceManifestPath = "/tmp/manifest.yaml"
 
  // Various environment variables containing pod information exposed to the executor container(s)
 
- // EnvVarHostIP contains the host IP which the container is executing on.
- // Used to communicate with kubelet directly. Kubelet enables the wait sidecar
- // to query pod state without burdening the k8s apiserver.
- EnvVarHostIP = "ARGO_HOST_IP"
  // EnvVarPodIP contains the IP of the pod (currently unused)
  EnvVarPodIP = "ARGO_POD_IP"
  // EnvVarPodName contains the name of the pod (currently unused)

workflow/common/validate.go

@@ -109,12 +109,15 @@ func (ctx *wfValidationCtx) validateTemplate(tmpl *wfv1.Template, args wfv1.Argu
  if tmpl.Script != nil {
  tmplTypes++
  }
+ if tmpl.Resource != nil {
+ tmplTypes++
+ }
  switch tmplTypes {
  case 0:
- return errors.New(errors.CodeBadRequest, "template type unspecified. choose one of: container, steps, script")
+ return errors.New(errors.CodeBadRequest, "template type unspecified. choose one of: container, steps, script, resource")
  case 1:
  default:
- return errors.New(errors.CodeBadRequest, "multiple template types specified. choose one of: container, steps, script")
+ return errors.New(errors.CodeBadRequest, "multiple template types specified. choose one of: container, steps, script, resource")
  }
  if tmpl.Steps == nil {
  err = validateLeaf(scope, tmpl)