fix(controller): Support int64 for param value. Fixes #4169 (#4202)

pkg/apis/workflow/v1alpha1/int64orstr.go

@@ -0,0 +1,51 @@
+package v1alpha1
+
+import (
+	"encoding/json"
+	"fmt"
+	"strconv"
+)
+
+// This is similar to `intstr.IntOrString` (which should be called `intstr.Int32OrString`!!).
+// It intended just to tolerate unmarshalling int64. Therefore:
+//
+// * It's JSON type is just string, not `int-or-string`.
+// * It will unmarshall int64 (rather than only int32) and represents it as string.
+// * It will marshall back to string - marshalling is not symmetric.
+type Int64OrString string
+
+func ParseInt64OrString(val interface{}) Int64OrString {
+	return Int64OrString(fmt.Sprintf("%v", val))
+}
+
+func Int64OrStringPtr(val interface{}) *Int64OrString {
+	i := ParseInt64OrString(val)
+	return &i
+}
+
+func (i *Int64OrString) UnmarshalJSON(value []byte) error {
+	if value[0] == '"' {
+		v := ""
+		err := json.Unmarshal(value, &v)
+		if err != nil {
+			return err
+		}
+		*i = Int64OrString(v)
+		return nil
+	}
+	v := 0
+	err := json.Unmarshal(value, &v)
+	if err != nil {
+		return err
+	}
+	*i = Int64OrString(strconv.Itoa(v))
+	return nil
+}
+
+func (i Int64OrString) MarshalJSON() ([]byte, error) {
+	return json.Marshal(string(i))
+}
+
+func (i Int64OrString) String() string {
+	return string(i)
+}

pkg/apis/workflow/v1alpha1/int64orstr_test.go

@@ -0,0 +1,63 @@
+package v1alpha1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestInt64OrString(t *testing.T) {
+	t.Run("Empty", func(t *testing.T) {
+		x := Int64OrStringPtr("")
+		data, err := json.Marshal(x)
+		if assert.NoError(t, err) {
+			assert.Equal(t, `""`, string(data), "string value has quotes")
+		}
+		i := Int64OrStringPtr("")
+		err = i.UnmarshalJSON([]byte(`""`))
+		if assert.NoError(t, err) {
+			assert.Equal(t, Int64OrStringPtr(""), i)
+		}
+		assert.Equal(t, "", i.String(), "string value does not have quotes")
+	})
+	t.Run("String", func(t *testing.T) {
+		x := Int64OrStringPtr("my-string")
+		data, err := json.Marshal(x)
+		if assert.NoError(t, err) {
+			assert.Equal(t, `"my-string"`, string(data), "string value has quotes")
+		}
+		i := Int64OrStringPtr("")
+		err = i.UnmarshalJSON([]byte(`"my-string"`))
+		if assert.NoError(t, err) {
+			assert.Equal(t, Int64OrStringPtr("my-string"), i)
+		}
+		assert.Equal(t, "my-string", i.String(), "string value does not have quotes")
+	})
+	t.Run("StringNumber", func(t *testing.T) {
+		x := Int64OrStringPtr(1)
+		data, err := json.Marshal(x)
+		if assert.NoError(t, err) {
+			assert.Equal(t, `"1"`, string(data), "number value has quotes")
+		}
+		i := Int64OrStringPtr("")
+		err = i.UnmarshalJSON([]byte(`"1"`))
+		if assert.NoError(t, err) {
+			assert.Equal(t, Int64OrStringPtr("1"), i)
+		}
+		assert.Equal(t, "1", i.String(), "number value does not have quotes")
+	})
+	t.Run("LargeNumber", func(t *testing.T) {
+		x := ParseInt64OrString(881217801864)
+		data, err := json.Marshal(x)
+		if assert.NoError(t, err) {
+			assert.Equal(t, `"881217801864"`, string(data))
+		}
+		i := Int64OrStringPtr("")
+		err = i.UnmarshalJSON([]byte("881217801864"))
+		if assert.NoError(t, err) {
+			assert.Equal(t, Int64OrStringPtr("881217801864"), i)
+		}
+		assert.Equal(t, "881217801864", i.String())
+	})
+}

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -612,11 +612,11 @@ type Parameter struct {
 	Name string `json:"name" protobuf:"bytes,1,opt,name=name"`
 
 	// Default is the default value to use for an input parameter if a value was not supplied
-	Default *string `json:"default,omitempty" protobuf:"bytes,2,opt,name=default"`
+	Default *Int64OrString `json:"default,omitempty" protobuf:"bytes,2,opt,name=default"`
 
 	// Value is the literal value to use for the parameter.
 	// If specified in the context of an input parameter, the value takes precedence over any passed values
-	Value *string `json:"value,omitempty" protobuf:"bytes,3,opt,name=value"`
+	Value *Int64OrString `json:"value,omitempty" protobuf:"bytes,3,opt,name=value"`
 
 	// ValueFrom is the source for the output parameter's value
 	ValueFrom *ValueFrom `json:"valueFrom,omitempty" protobuf:"bytes,4,opt,name=valueFrom"`
@@ -648,7 +648,7 @@ type ValueFrom struct {
 	Supplied *SuppliedValueFrom `json:"supplied,omitempty" protobuf:"bytes,6,opt,name=supplied"`
 
 	// Default specifies a value to be used if retrieving the value from the specified source fails
-	Default *string `json:"default,omitempty" protobuf:"bytes,5,opt,name=default"`
+	Default *Int64OrString `json:"default,omitempty" protobuf:"bytes,5,opt,name=default"`
 }
 
 // SuppliedValueFrom is a placeholder for a value to be filled in directly, either through the CLI, API, etc.

server/event/dispatch/operation.go

@@ -16,7 +16,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/retry"
-	"k8s.io/utils/pointer"
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/server/auth"
@@ -114,7 +113,7 @@ func (o *Operation) dispatch(wfeb wfv1.WorkflowEventBinding, nameSuffix string)
 				if err != nil {
 					return nil, fmt.Errorf("failed to evaluate workflow template parameter \"%s\" expression: %w", p.Name, err)
 				}
-				wf.Spec.Arguments.Parameters = append(wf.Spec.Arguments.Parameters, wfv1.Parameter{Name: p.Name, Value: pointer.StringPtr(fmt.Sprintf("%v", result))})
+				wf.Spec.Arguments.Parameters = append(wf.Spec.Arguments.Parameters, wfv1.Parameter{Name: p.Name, Value: wfv1.Int64OrStringPtr(result)})
 			}
 		}
 		wf, err = client.ArgoprojV1alpha1().Workflows(wfeb.Namespace).Create(wf)

server/event/dispatch/operation_test.go

@@ -8,7 +8,6 @@ import (
 	"google.golang.org/grpc/metadata"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/record"
-	"k8s.io/utils/pointer"
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/pkg/client/clientset/versioned/fake"
@@ -84,7 +83,7 @@ func TestNewOperation(t *testing.T) {
 			assert.Equal(t, "my-instanceid", wf.Labels[common.LabelKeyControllerInstanceID])
 			assert.Equal(t, "my-sub", wf.Labels[common.LabelKeyCreator])
 			assert.Contains(t, wf.Labels, common.LabelKeyWorkflowEventBinding)
-			assert.Equal(t, []wfv1.Parameter{{Name: "my-param", Value: pointer.StringPtr(`foo`)}}, wf.Spec.Arguments.Parameters)
+			assert.Equal(t, []wfv1.Parameter{{Name: "my-param", Value: wfv1.Int64OrStringPtr(`foo`)}}, wf.Spec.Arguments.Parameters)
 		}
 	}
 	assert.Equal(t, "Warning WorkflowEventBindingError failed to dispatch event: failed to evaluate workflow template expression: unexpected token EOF (1:1)", <-recorder.Events)

test/e2e/cli_with_server_test.go

@@ -226,7 +226,7 @@ spec:
 			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
 			nodeStatus := status.Nodes.FindByDisplayName("release")
 			if assert.NotNil(t, nodeStatus) {
-				assert.Equal(t, "Hello, World!", *nodeStatus.Inputs.Parameters[0].Value)
+				assert.Equal(t, "Hello, World!", nodeStatus.Inputs.Parameters[0].Value.String())
 			}
 			nodeStatus = status.Nodes.FindByDisplayName("approve")
 			if assert.NotNil(t, nodeStatus) {

test/e2e/functional_test.go

@@ -591,7 +591,7 @@ spec:
 			assert.True(t, status.Nodes.Any(func(node wfv1.NodeStatus) bool {
 				if node.Outputs != nil {
 					for _, param := range node.Outputs.Parameters {
-						if param.Value != nil && *param.Value == "Default value" {
+						if param.Value != nil && param.Value.String() == "Default value" {
 							return true
 						}
 					}

util/printer/workflow-printer.go

@@ -116,7 +116,7 @@ func parameterString(params []wfv1.Parameter) string {
 	pStrs := make([]string, 0)
 	for _, p := range params {
 		if p.Value != nil {
-			str := fmt.Sprintf("%s=%s", p.Name, truncateString(*p.Value, 50))
+			str := fmt.Sprintf("%s=%s", p.Name, truncateString(p.Value.String(), 50))
 			pStrs = append(pStrs, str)
 		}
 	}

util/printer/workflow-printer_test.go

@@ -20,7 +20,7 @@ func TestPrintWorkflows(t *testing.T) {
 			ObjectMeta: metav1.ObjectMeta{Name: "my-wf", Namespace: "my-ns", CreationTimestamp: metav1.Time{Time: now}},
 			Spec: wfv1.WorkflowSpec{
 				Arguments: wfv1.Arguments{Parameters: []wfv1.Parameter{
-					{Name: "my-param", Value: pointer.StringPtr("my-value")},
+					{Name: "my-param", Value: wfv1.Int64OrStringPtr("my-value")},
 				}},
 				Priority: pointer.Int32Ptr(2),
 				Templates: []wfv1.Template{

workflow/common/util.go

@@ -325,7 +325,7 @@ func SubstituteParams(tmpl *wfv1.Template, globalParams, localParams Parameters)
 		if inParam.Value == nil {
 			return nil, errors.InternalErrorf("inputs.parameters.%s had no value", inParam.Name)
 		}
-		replaceMap["inputs.parameters."+inParam.Name] = *inParam.Value
+		replaceMap["inputs.parameters."+inParam.Name] = inParam.Value.String()
 	}
 	//allow {{inputs.parameters}} to fetch the entire input parameters list as JSON
 	jsonInputParametersBytes, err := json.Marshal(globalReplacedTmpl.Inputs.Parameters)

workflow/controller/cache_test.go

@@ -49,7 +49,7 @@ func TestConfigMapCacheLoadHit(t *testing.T) {
 	assert.NoError(t, err)
 	if assert.Len(t, outputs.Parameters, 1) {
 		assert.Equal(t, "hello", outputs.Parameters[0].Name)
-		assert.Equal(t, sampleOutput, *outputs.Parameters[0].Value)
+		assert.Equal(t, "foobar", outputs.Parameters[0].Value.String())
 	}
 }
 
@@ -68,7 +68,7 @@ func TestConfigMapCacheSave(t *testing.T) {
 	var MockParamValue string = "Hello world"
 	var MockParam = wfv1.Parameter{
 		Name:  "hello",
-		Value: &MockParamValue,
+		Value: wfv1.Int64OrStringPtr(MockParamValue),
 	}
 	cancel, controller := newController()
 	defer cancel()

workflow/controller/controller.go

@@ -464,7 +464,7 @@ func (wfc *WorkflowController) processNextItem() bool {
 	if err != nil {
 		log.WithFields(log.Fields{"key": key, "error": err}).Warn("Failed to unmarshal key to workflow object")
 		woc := newWorkflowOperationCtx(wf, wfc)
-		woc.markWorkflowFailed(fmt.Sprintf("invalid spec: %s", err.Error()))
+		woc.markWorkflowFailed(fmt.Sprintf("cannot unmarshall spec: %s", err.Error()))
 		woc.persistUpdates()
 		wfc.throttler.Remove(key)
 		return true

workflow/controller/operator.go

@@ -463,7 +463,7 @@ func (woc *wfOperationCtx) setGlobalParameters(executionParameters wfv1.Argument
 		woc.globalParams[common.GlobalVarWorkflowParameters] = string(workflowParameters)
 	}
 	for _, param := range executionParameters.Parameters {
-		woc.globalParams["workflow.parameters."+param.Name] = *param.Value
+		woc.globalParams["workflow.parameters."+param.Name] = param.Value.String()
 	}
 	for k, v := range woc.wf.ObjectMeta.Annotations {
 		woc.globalParams["workflow.annotations."+k] = v
@@ -473,7 +473,7 @@ func (woc *wfOperationCtx) setGlobalParameters(executionParameters wfv1.Argument
 	}
 	if woc.wf.Status.Outputs != nil {
 		for _, param := range woc.wf.Status.Outputs.Parameters {
-			woc.globalParams["workflow.outputs.parameters."+param.Name] = *param.Value
+			woc.globalParams["workflow.outputs.parameters."+param.Name] = param.Value.String()
 		}
 	}
 }
@@ -2161,12 +2161,12 @@ func getTemplateOutputsFromScope(tmpl *wfv1.Template, scope *wfScope) (*wfv1.Out
 			if err != nil {
 				// We have a default value to use instead of returning an error
 				if param.ValueFrom.Default != nil {
-					val = *param.ValueFrom.Default
+					val = param.ValueFrom.Default.String()
 				} else {
 					return nil, err
 				}
 			}
-			param.Value = &val
+			param.Value = wfv1.Int64OrStringPtr(val)
 			param.ValueFrom = nil
 			outputs.Parameters = append(outputs.Parameters, param)
 		}
@@ -2305,7 +2305,7 @@ func (woc *wfOperationCtx) addOutputsToLocalScope(prefix string, outputs *wfv1.O
 	}
 	for _, param := range outputs.Parameters {
 		if param.Value != nil {
-			scope.addParamToScope(fmt.Sprintf("%s.outputs.parameters.%s", prefix, param.Name), *param.Value)
+			scope.addParamToScope(fmt.Sprintf("%s.outputs.parameters.%s", prefix, param.Name), param.Value.String())
 		}
 	}
 	for _, art := range outputs.Artifacts {
@@ -2367,7 +2367,7 @@ func (woc *wfOperationCtx) processAggregateNodeOutputs(tmpl *wfv1.Template, scop
 		if len(node.Outputs.Parameters) > 0 {
 			param := make(map[string]string)
 			for _, p := range node.Outputs.Parameters {
-				param[p.Name] = *p.Value
+				param[p.Name] = p.Value.String()
 			}
 			paramList = append(paramList, param)
 		}
@@ -2415,16 +2415,16 @@ func (woc *wfOperationCtx) addParamToGlobalScope(param wfv1.Parameter) {
 		woc.wf.Status.Outputs = &wfv1.Outputs{}
 	}
 	paramName := fmt.Sprintf("workflow.outputs.parameters.%s", param.GlobalName)
-	woc.globalParams[paramName] = *param.Value
+	woc.globalParams[paramName] = param.Value.String()
 	if index == -1 {
-		woc.log.Infof("setting %s: '%s'", paramName, *param.Value)
+		woc.log.Infof("setting %s: '%s'", paramName, param.Value)
 		gParam := wfv1.Parameter{Name: param.GlobalName, Value: param.Value}
 		woc.wf.Status.Outputs.Parameters = append(woc.wf.Status.Outputs.Parameters, gParam)
 		woc.updated = true
 	} else {
 		prevVal := *woc.wf.Status.Outputs.Parameters[index].Value
 		if prevVal != *param.Value {
-			woc.log.Infof("overwriting %s: '%s' -> '%s'", paramName, *woc.wf.Status.Outputs.Parameters[index].Value, *param.Value)
+			woc.log.Infof("overwriting %s: '%s' -> '%s'", paramName, woc.wf.Status.Outputs.Parameters[index].Value, param.Value)
 			woc.wf.Status.Outputs.Parameters[index].Value = param.Value
 			woc.updated = true
 		}