feat: Add argo submit --verify hidden flag. Closes argoproj#5136 (a…

…rgoproj#5141)

Signed-off-by: Alex Collins <[email protected]>

.github/workflows/ci-build.yaml

@@ -101,8 +101,14 @@ jobs:
  echo '127.0.0.1 postgres' | sudo tee -a /etc/hosts
  echo '127.0.0.1 mysql' | sudo tee -a /etc/hosts
  git fetch --tags
- KUBECONFIG=~/.kube/config make install PROFILE=mysql E2E_EXECUTOR=${{matrix.containerRuntimeExecutor}} ALWAYS_OFFLOAD_NODE_STATUS=true DEV_IMAGE=true STATIC_FILES=false
- KUBECONFIG=~/.kube/config make start PROFILE=mysql E2E_EXECUTOR=${{matrix.containerRuntimeExecutor}} ALWAYS_OFFLOAD_NODE_STATUS=true DEV_IMAGE=true STATIC_FILES=false 2>&1 > /tmp/log/argo-e2e/argo.log &
+ PROFILE=mysql
+ ALWAYS_OFFLOAD_NODE_STATUS=true
+ if [ "${{matrix.test}}" = test-examples ]; then
+ PROFILE=minimal
+ ALWAYS_OFFLOAD_NODE_STATUS=false
+ fi
+ KUBECONFIG=~/.kube/config make install PROFILE=$PROFILE E2E_EXECUTOR=${{matrix.containerRuntimeExecutor}} ALWAYS_OFFLOAD_NODE_STATUS=${ALWAYS_OFFLOAD_NODE_STATUS} DEV_IMAGE=true STATIC_FILES=false
+ KUBECONFIG=~/.kube/config make start PROFILE=$PROFILE E2E_EXECUTOR=${{matrix.containerRuntimeExecutor}} ALWAYS_OFFLOAD_NODE_STATUS=${ALWAYS_OFFLOAD_NODE_STATUS} DEV_IMAGE=true STATIC_FILES=false 2>&1 > /tmp/log/argo-e2e/argo.log &
  - name: Wait for Argo Server to be ready
  env:
  GOPATH: /home/runner/go
@@ -112,17 +118,11 @@ jobs:
  GOPATH: /home/runner/go
  run: make ${{ matrix.test }}
  - name: Upload logs
- if: ${{ always() }}
- uses: actions/upload-artifact@v1
- with:
- name: ${{ matrix.test }}-${{matrix.containerRuntimeExecutor}}-${{ github.run_id }}-argo.log
- path: /tmp/log/argo-e2e/argo.log
- - name: Upload pod logs
  if: ${{ failure() }}
  uses: actions/upload-artifact@v1
  with:
  name: ${{ matrix.test }}-${{matrix.containerRuntimeExecutor}}-${{ github.run_id }}-pod.log
- path: /tmp/log/argo-e2e/pod.log
+ path: /tmp/log/argo-e2e
 
  codegen:
  name: Codegen

Makefile

@@ -43,6 +43,7 @@ endif
 ifneq ($(CI),)
 AUTH_MODE := client
 endif
+
 # Which mode to run in:
 # * `local` run the workflow–controller and argo-server as single replicas on the local machine (default)
 # * `kubernetes` run the workflow-controller and argo-server on the Kubernetes cluster
@@ -485,8 +486,8 @@ test-executor:
  $(GOTEST) -timeout 5m -count 1 --tags executor -p 1 ./test/e2e
 
 .PHONY: test-examples
-test-examples:
- $(GOTEST) -timeout 15m -count 1 --tags examples -p 1 ./test/e2e
+test-examples: ./dist/argo
+ ./hack/test-examples.sh
 
 .PHONY: test-functional
 test-functional:

cmd/argo/commands/submit.go

@@ -2,13 +2,12 @@ package commands
 
 import (
  "context"
- "log"
  "os"
  "strings"
 
  "github.com/argoproj/pkg/errors"
  argoJson "github.com/argoproj/pkg/json"
- "github.com/sirupsen/logrus"
+ log "github.com/sirupsen/logrus"
  "github.com/spf13/cobra"
  corev1 "k8s.io/api/core/v1"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -25,6 +24,7 @@ type cliSubmitOpts struct {
  output string // --output
  wait bool // --wait
  watch bool // --watch
+ verify bool // --verify
  log bool // --log
  strict bool // --strict
  priority *int32 // --priority
@@ -67,7 +67,7 @@ func NewSubmitCommand() *cobra.Command {
  }
 
  if !cliSubmitOpts.watch && len(cliSubmitOpts.getArgs.status) > 0 {
- logrus.Warn("--status should only be used with --watch")
+ log.Warn("--status should only be used with --watch")
  }
 
  ctx, apiClient := client.NewAPIClient()
@@ -88,6 +88,8 @@ func NewSubmitCommand() *cobra.Command {
  command.Flags().StringVarP(&cliSubmitOpts.output, "output", "o", "", "Output format. One of: name|json|yaml|wide")
  command.Flags().BoolVarP(&cliSubmitOpts.wait, "wait", "w", false, "wait for the workflow to complete")
  command.Flags().BoolVar(&cliSubmitOpts.watch, "watch", false, "watch the workflow until it completes")
+ command.Flags().BoolVar(&cliSubmitOpts.verify, "verify", false, "verify completed workflows by running the Python code in the workflows.argoproj.io/verify.py annotation")
+ errors.CheckError(command.Flags().MarkHidden("verify"))
  command.Flags().BoolVar(&cliSubmitOpts.log, "log", false, "log the workflow until it completes")
  command.Flags().BoolVar(&cliSubmitOpts.strict, "strict", true, "perform strict workflow validation")
  command.Flags().Int32Var(&priority, "priority", 0, "workflow priority")
@@ -258,4 +260,7 @@ func waitWatchOrLog(ctx context.Context, serviceClient workflowpkg.WorkflowServi
  watchWorkflow(ctx, serviceClient, namespace, workflow, cliSubmitOpts.getArgs)
  }
  }
+ if cliSubmitOpts.verify {
+ verifyWorkflows(ctx, serviceClient, namespace, workflowNames)
+ }
 }

cmd/argo/commands/verify.go

@@ -0,0 +1,34 @@
+package commands
+
+import (
+ "context"
+ "fmt"
+ "os"
+
+ "github.com/TwinProduction/go-color"
+ "github.com/argoproj/pkg/errors"
+
+ workflowpkg "github.com/argoproj/argo-workflows/v3/pkg/apiclient/workflow"
+ "github.com/argoproj/argo-workflows/v3/workflow/verify"
+)
+
+func verifyWorkflows(ctx context.Context, serviceClient workflowpkg.WorkflowServiceClient, namespace string, workflowNames []string) {
+ _, _ = fmt.Fprintln(os.Stderr, "verifying workflows...")
+ failAtEnd := false
+ for _, name := range workflowNames {
+ wf, err := serviceClient.GetWorkflow(ctx, &workflowpkg.WorkflowGetRequest{
+ Namespace: namespace,
+ Name: name,
+ })
+ errors.CheckError(err)
+ if err := verify.Workflow(wf); err != nil {
+ _, _ = fmt.Fprintf(os.Stdout, "%s %s: %v\n", color.Ize(color.Red, "✖"), name, err)
+ failAtEnd = true
+ } else {
+ _, _ = fmt.Fprintf(os.Stdout, "%s %s\n", color.Ize(color.Green, "✔"), name)
+ }
+ }
+ if failAtEnd {
+ os.Exit(1)
+ }
+}

cmd/argo/commands/wait.go

@@ -67,8 +67,6 @@ func waitWorkflows(ctx context.Context, serviceClient workflowpkg.WorkflowServic
 }
 
 func waitOnOne(serviceClient workflowpkg.WorkflowServiceClient, ctx context.Context, wfName, namespace string, ignoreNotFound, quiet bool) bool {
- ctx, cancel := context.WithCancel(ctx)
- defer cancel()
  req := &workflowpkg.WatchWorkflowsRequest{
  Namespace: namespace,
  ListOptions: &metav1.ListOptions{

cmd/argo/commands/watch.go

@@ -50,8 +50,6 @@ func NewWatchCommand() *cobra.Command {
 }
 
 func watchWorkflow(ctx context.Context, serviceClient workflowpkg.WorkflowServiceClient, namespace string, workflow string, getArgs getFlags) {
- ctx, cancel := context.WithCancel(ctx)
- defer cancel()
  req := &workflowpkg.WatchWorkflowsRequest{
  Namespace: namespace,
  ListOptions: &metav1.ListOptions{

hack/test-examples.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+set -eu -o pipefail
+
+kubectl delete wf -l workflows.argoproj.io/test
+
+grep -lR 'workflows.argoproj.io/test' examples/* | while read f ; do
+ ./dist/argo submit --watch --verify $f
+done

server/auth/gatekeeper.go

@@ -2,7 +2,6 @@ package auth
 
 import (
  "context"
- "encoding/json"
  "fmt"
  "net/http"
  "sort"
@@ -27,6 +26,7 @@ import (
  "github.com/argoproj/argo-workflows/v3/server/auth/sso"
  "github.com/argoproj/argo-workflows/v3/server/auth/types"
  servertypes "github.com/argoproj/argo-workflows/v3/server/types"
+ jsonutil "github.com/argoproj/argo-workflows/v3/util/json"
  "github.com/argoproj/argo-workflows/v3/util/kubeconfig"
  "github.com/argoproj/argo-workflows/v3/workflow/common"
 )
@@ -203,15 +203,10 @@ func (s *gatekeeper) rbacAuthorization(ctx context.Context, claims *types.Claims
  sort.Slice(serviceAccounts, func(i, j int) bool { return precedence(serviceAccounts[i]) > precedence(serviceAccounts[j]) })
  for _, serviceAccount := range serviceAccounts {
  rule := serviceAccount.Annotations[common.AnnotationKeyRBACRule]
- data, err := json.Marshal(claims)
+ v, err := jsonutil.Jsonify(claims)
  if err != nil {
  return nil, fmt.Errorf("failed to marshall claims: %w", err)
  }
- v := make(map[string]interface{})
- err = json.Unmarshal(data, &v)
- if err != nil {
- return nil, fmt.Errorf("failed to unmarshall claims: %w", err)
- }
  result, err := expr.Eval(rule, v)
  if err != nil {
  return nil, fmt.Errorf("failed to evaluate rule: %w", err)

server/event/dispatch/operation.go

@@ -20,6 +20,7 @@ import (
  "github.com/argoproj/argo-workflows/v3/server/auth"
  errorsutil "github.com/argoproj/argo-workflows/v3/util/errors"
  "github.com/argoproj/argo-workflows/v3/util/instanceid"
+ jsonutil "github.com/argoproj/argo-workflows/v3/util/json"
  "github.com/argoproj/argo-workflows/v3/util/labels"
  waitutil "github.com/argoproj/argo-workflows/v3/util/wait"
  "github.com/argoproj/argo-workflows/v3/workflow/common"
@@ -192,12 +193,7 @@ func expressionEnvironment(ctx context.Context, namespace, discriminator string,
  "metadata": metaData(ctx),
  "payload": payload,
  }
- data, err := json.Marshal(src)
- if err != nil {
- return nil, err
- }
- env := make(map[string]interface{})
- return env, json.Unmarshal(data, &env)
+ return jsonutil.Jsonify(src)
 }
 
 func metaData(ctx context.Context) map[string]interface{} {

util/flatten/flatten.go

@@ -1,16 +1,15 @@
 package flatten
 
 import (
- "encoding/json"
  "fmt"
  "reflect"
+
+ jsonutil "github.com/argoproj/argo-workflows/v3/util/json"
 )
 
 func toMap(in interface{}) map[string]interface{} {
- data, _ := json.Marshal(in)
- out := make(map[string]interface{})
- _ = json.Unmarshal(data, &out)
- return out
+ v, _ := jsonutil.Jsonify(in)
+ return v
 }
 
 func flattenWithPrefix(in map[string]interface{}, out map[string]string, prefix string) {

util/json/jsonify.go

@@ -0,0 +1,12 @@
+package json
+
+import "encoding/json"
+
+func Jsonify(v interface{}) (map[string]interface{}, error) {
+ data, err := json.Marshal(v)
+ if err != nil {
+ return nil, err
+ }
+ x := make(map[string]interface{})
+ return x, json.Unmarshal(data, &x)
+}

test/e2e/py.go → util/python/conversion.go

@@ -1,9 +1,9 @@
-package e2e
+package python
 
 import (
- "reflect"
-
  "github.com/go-python/gpython/py"
+
+ "github.com/argoproj/argo-workflows/v3/util/json"
 )
 
 func obj(v interface{}) py.Object {
@@ -19,7 +19,8 @@ func obj(v interface{}) py.Object {
  case map[string]interface{}:
  return dict(x)
  default:
- panic(reflect.TypeOf(x).String())
+ v, _ := json.Jsonify(v)
+ return obj(v)
  }
 }
 

util/python/python.go

@@ -0,0 +1,24 @@
+package python
+
+import (
+ "fmt"
+
+ _ "github.com/go-python/gpython/builtin"
+ "github.com/go-python/gpython/compile"
+ "github.com/go-python/gpython/py"
+ "github.com/go-python/gpython/vm"
+)
+
+func Run(s string, globals map[string]interface{}) error {
+ x, err := compile.Compile(s, "<stdin>", "exec", 0, true)
+ if err != nil {
+ return err
+ }
+ m := py.NewModule("__main__", "", nil, dict(globals))
+ code, ok := x.(*py.Code)
+ if !ok {
+ return fmt.Errorf("obj cannot be cast to code")
+ }
+ _, err = vm.EvalCode(code, m.Globals, nil)
+ return err
+}