Merge branch 'master' of https://github.com/argoproj/argo

README.md

@@ -77,6 +77,7 @@ Currently **officially** using Argo:
 1. [BlackRock](https://www.blackrock.com/)
 1. [Canva](https://www.canva.com/)
 1. [Capital One](https://www.capitalone.com/tech/)
+1. [CarTrack](https://www.cartrack.com/)
 1. [CCRi](https://www.ccri.com/)
 1. [Codec](https://www.codec.ai/)
 1. [Commodus Tech](https://www.commodus.tech)

cmd/argo/commands/archive/list.go

@@ -18,7 +18,8 @@ import (
 
 func NewListCommand() *cobra.Command {
 	var (
-		output string
+		selector string
+		output   string
 	)
 	var command = &cobra.Command{
 		Use: "list",
@@ -30,6 +31,7 @@ func NewListCommand() *cobra.Command {
 			resp, err := serviceClient.ListArchivedWorkflows(ctx, &workflowarchivepkg.ListArchivedWorkflowsRequest{
 				ListOptions: &metav1.ListOptions{
 					FieldSelector: "metadata.namespace=" + namespace,
+					LabelSelector: selector,
 				},
 			})
 			errors.CheckError(err)
@@ -57,5 +59,6 @@ func NewListCommand() *cobra.Command {
 		},
 	}
 	command.Flags().StringVarP(&output, "output", "o", "wide", "Output format. One of: json|yaml|wide")
+	command.Flags().StringVarP(&selector, "selector", "l", "", "Selector (label query) to filter on, not including uninitialized ones")
 	return command
 }

cmd/argo/commands/common.go

@@ -34,6 +34,7 @@ var (
 	// DEPRECATED
 	wftmplClient     v1alpha1.WorkflowTemplateInterface
 	jobStatusIconMap map[wfv1.NodePhase]string
+	nodeTypeIconMap  map[wfv1.NodeType]string
 	noColor          bool
 	// DEPRECATED
 	namespace string
@@ -71,6 +72,9 @@ func initializeSession() {
 		wfv1.NodeFailed:    ansiFormat("✖", FgRed),
 		wfv1.NodeError:     ansiFormat("⚠", FgRed),
 	}
+	nodeTypeIconMap = map[wfv1.NodeType]string{
+		wfv1.NodeTypeSuspend: ansiFormat("ǁ", FgCyan),
+	}
 }
 
 // DEPRECATED

cmd/argo/commands/get.go

@@ -432,6 +432,9 @@ func printNode(w *tabwriter.Writer, node wfv1.NodeStatus, nodePrefix string, get
 		return
 	}
 	nodeName := fmt.Sprintf("%s %s", jobStatusIconMap[node.Phase], node.DisplayName)
+	if node.IsActiveSuspendNode() {
+		nodeName = fmt.Sprintf("%s %s", nodeTypeIconMap[node.Type], node.DisplayName)
+	}
 	if node.TemplateRef != nil {
 		nodeName = fmt.Sprintf("%s (%s/%s)", nodeName, node.TemplateRef.Name, node.TemplateRef.Template)
 	} else if node.TemplateName != "" {

cmd/argo/commands/get_test.go

@@ -0,0 +1,68 @@
+package commands
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+	"text/tabwriter"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+func testPrintNodeImpl(t *testing.T, expected string, node wfv1.NodeStatus, nodePrefix string, getArgs getFlags) {
+	var result bytes.Buffer
+	w := tabwriter.NewWriter(&result, 0, 8, 1, '\t', 0)
+	printNode(w, node, nodePrefix, getArgs)
+	w.Flush()
+	assert.Equal(t, expected, result.String())
+}
+
+// TestPrintNode
+func TestPrintNode(t *testing.T) {
+	nodeName := "testNode"
+	nodePrefix := ""
+	nodeTemplateName := "testTemplate"
+	nodeTemplateRefName := "testTemplateRef"
+	nodeID := "testID"
+	nodeMessage := "test"
+	getArgs := getFlags{
+		output: "",
+		status: "",
+	}
+	timestamp := metav1.Time{
+		Time: time.Now(),
+	}
+	node := wfv1.NodeStatus{
+		Name:        nodeName,
+		Phase:       wfv1.NodeRunning,
+		DisplayName: nodeName,
+		Type:        wfv1.NodeTypePod,
+		ID:          nodeID,
+		StartedAt:   timestamp,
+		FinishedAt:  timestamp,
+		Message:     nodeMessage,
+	}
+	testPrintNodeImpl(t, fmt.Sprintf("%s %s\t%s\t%s\t%s\n", jobStatusIconMap[wfv1.NodeRunning], nodeName, nodeID, "0s", nodeMessage), node, nodePrefix, getArgs)
+
+	node.TemplateName = nodeTemplateName
+	testPrintNodeImpl(t, fmt.Sprintf("%s %s (%s)\t%s\t%s\t%s\n", jobStatusIconMap[wfv1.NodeRunning], nodeName, nodeTemplateName, nodeID, "0s", nodeMessage), node, nodePrefix, getArgs)
+
+	node.Type = wfv1.NodeTypeSuspend
+	testPrintNodeImpl(t, fmt.Sprintf("%s %s (%s)\t%s\t%s\t%s\n", nodeTypeIconMap[wfv1.NodeTypeSuspend], nodeName, nodeTemplateName, "", "", nodeMessage), node, nodePrefix, getArgs)
+
+	node.TemplateRef = &wfv1.TemplateRef{
+		Name:     nodeTemplateRefName,
+		Template: nodeTemplateRefName,
+	}
+	testPrintNodeImpl(t, fmt.Sprintf("%s %s (%s/%s)\t%s\t%s\t%s\n", nodeTypeIconMap[wfv1.NodeTypeSuspend], nodeName, nodeTemplateRefName, nodeTemplateRefName, "", "", nodeMessage), node, nodePrefix, getArgs)
+
+	getArgs.output = "wide"
+	testPrintNodeImpl(t, fmt.Sprintf("%s %s (%s/%s)\t%s\t%s\t%s\t%s\n", nodeTypeIconMap[wfv1.NodeTypeSuspend], nodeName, nodeTemplateRefName, nodeTemplateRefName, "", "", getArtifactsString(node), nodeMessage), node, nodePrefix, getArgs)
+
+	getArgs.status = "foobar"
+	testPrintNodeImpl(t, "", node, nodePrefix, getArgs)
+}

docs/workflow-templates.md

@@ -4,7 +4,7 @@
 
 > v2.4 and after
 
-Workflow templates are reusable checks of YAML that you can use within your workflows. This allows you to have a library of templates.
+Workflow templates are reusable chunks of YAML that you can use within your workflows. This allows you to have a library of templates.
 
 You can create some example templates as follows:
 
@@ -16,4 +16,4 @@ The submit a workflow using one of those templates:
 
 ```
 argo submit https://raw.githubusercontent.com/argoproj/argo/master/examples/workflow-template/hello-world.yam
-```
+```

persist/sqldb/archived_workflow_labels.go

@@ -0,0 +1,57 @@
+package sqldb
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
+	"upper.io/db.v3"
+)
+
+func labelsClause(t dbType, requirements labels.Requirements) (db.Compound, error) {
+	var conds []db.Compound
+	for _, r := range requirements {
+		cond, err := requirementToCondition(t, r)
+		if err != nil {
+			return nil, err
+		}
+		conds = append(conds, cond)
+	}
+	return db.And(conds...), nil
+}
+
+func requirementToCondition(t dbType, r labels.Requirement) (db.Compound, error) {
+	// Should we "sanitize our inputs"? No.
+	// https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+	// Valid label values must be 63 characters or less and must be empty or begin and end with an alphanumeric character ([a-z0-9A-Z]) with dashes (-), underscores (_), dots (.), and alphanumerics between.
+	// https://kb.objectrocket.com/postgresql/casting-in-postgresql-570#string+to+integer+casting
+	switch r.Operator() {
+	case selection.DoesNotExist:
+		return db.Raw(fmt.Sprintf("not exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s')", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key())), nil
+	case selection.Equals, selection.DoubleEquals:
+		return db.Raw(fmt.Sprintf("exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s' and value = '%s')", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key(), r.Values().List()[0])), nil
+	case selection.In:
+		return db.Raw(fmt.Sprintf("exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s' and value in ('%s'))", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key(), strings.Join(r.Values().List(), "', '"))), nil
+	case selection.NotEquals:
+		return db.Raw(fmt.Sprintf("not exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s' and value = '%s')", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key(), r.Values().List()[0])), nil
+	case selection.NotIn:
+		return db.Raw(fmt.Sprintf("not exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s' and value in ('%s'))", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key(), strings.Join(r.Values().List(), "', '"))), nil
+	case selection.Exists:
+		return db.Raw(fmt.Sprintf("exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s')", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key())), nil
+	case selection.GreaterThan:
+		i, err := strconv.Atoi(r.Values().List()[0])
+		if err != nil {
+			return nil, err
+		}
+		return db.Raw(fmt.Sprintf("exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s' and cast(value as %s) > %d)", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key(), t.intType(), i)), nil
+	case selection.LessThan:
+		i, err := strconv.Atoi(r.Values().List()[0])
+		if err != nil {
+			return nil, err
+		}
+		return db.Raw(fmt.Sprintf("exists (select 1 from %s where clustername = %s.clustername and uid = %s.uid and name = '%s' and cast(value as %s) < %d)", archiveLabelsTableName, archiveTableName, archiveTableName, r.Key(), t.intType(), i)), nil
+	}
+	return nil, fmt.Errorf("operation %v is not supported", r.Operator())
+}

persist/sqldb/archived_workflow_labels_test.go

@@ -0,0 +1,47 @@
+package sqldb
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/labels"
+	"upper.io/db.v3"
+)
+
+func Test_labelsClause(t *testing.T) {
+	tests := []struct {
+		name         string
+		dbType       dbType
+		requirements labels.Requirements
+		want         db.Compound
+	}{
+		{"Empty", Postgres, requirements(""), db.And()},
+		{"DoesNotExist", Postgres, requirements("!foo"), db.And(db.Raw("not exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo')"))},
+		{"Equals", Postgres, requirements("foo=bar"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and value = 'bar')"))},
+		{"DoubleEquals", Postgres, requirements("foo==bar"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and value = 'bar')"))},
+		{"In", Postgres, requirements("foo in (bar,baz)"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and value in ('bar', 'baz'))"))},
+		{"NotEquals", Postgres, requirements("foo != bar"), db.And(db.Raw("not exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and value = 'bar')"))},
+		{"NotIn", Postgres, requirements("foo notin (bar,baz)"), db.And(db.Raw("not exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and value in ('bar', 'baz'))"))},
+		{"Exists", Postgres, requirements("foo"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo')"))},
+		{"GreaterThanPostgres", Postgres, requirements("foo>2"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and cast(value as int) > 2)"))},
+		{"GreaterThanMySQL", MySQL, requirements("foo>2"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and cast(value as signed) > 2)"))},
+		{"LessThanPostgres", Postgres, requirements("foo<2"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and cast(value as int) < 2)"))},
+		{"LessThanMySQL", MySQL, requirements("foo<2"), db.And(db.Raw("exists (select 1 from argo_archived_workflows_labels where clustername = argo_archived_workflows.clustername and uid = argo_archived_workflows.uid and name = 'foo' and cast(value as signed) < 2)"))},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := labelsClause(tt.dbType, tt.requirements)
+			if assert.NoError(t, err) {
+				assert.Equal(t, tt.want.Sentences(), got.Sentences())
+			}
+		})
+	}
+}
+
+func requirements(selector string) []labels.Requirement {
+	requirements, err := labels.ParseToRequirements(selector)
+	if err != nil {
+		panic(err)
+	}
+	return requirements
+}

persist/sqldb/db_type.go

@@ -0,0 +1,30 @@
+package sqldb
+
+import (
+	"database/sql"
+
+	"github.com/go-sql-driver/mysql"
+	"upper.io/db.v3"
+)
+
+type dbType string
+
+const (
+	MySQL    dbType = "mysql"
+	Postgres dbType = "postgres"
+)
+
+func dbTypeFor(session db.Database) dbType {
+	switch session.Driver().(*sql.DB).Driver().(type) {
+	case *mysql.MySQLDriver:
+		return MySQL
+	}
+	return Postgres
+}
+
+func (t dbType) intType() string {
+	if t == MySQL {
+		return "signed"
+	}
+	return "int"
+}