feat: Add argo submit --verify hidden flag. Closes #5136

.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
+}