fix(controller): Tolerate int64 parameters. Fixes argoproj#4359 (argo…

…proj#4401)

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

workflow/controller/estimation/estimator_factory.go

@@ -6,14 +6,14 @@ import (
 
  "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
  "k8s.io/apimachinery/pkg/labels"
- "k8s.io/apimachinery/pkg/runtime"
  "k8s.io/client-go/tools/cache"
 
  "github.com/argoproj/argo/persist/sqldb"
  wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
  "github.com/argoproj/argo/workflow/common"
  "github.com/argoproj/argo/workflow/controller/indexes"
  "github.com/argoproj/argo/workflow/hydrator"
+ "github.com/argoproj/argo/workflow/util"
 )
 
 type EstimatorFactory interface {
@@ -55,8 +55,7 @@ func (f *estimatorFactory) NewEstimator(wf *wfv1.Workflow) (Estimator, error) {
  if un.GetLabels()[common.LabelKeyPhase] != string(wfv1.NodeSucceeded) {
  continue
  }
- candidateWf := &wfv1.Workflow{}
- err := runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, candidateWf)
+ candidateWf, err := util.FromUnstructured(un)
  if err != nil {
  return defaultEstimator, fmt.Errorf("failed convert unstructured to workflow: %w", err)
  }

workflow/controller/informer/cluster_workflow_template_convert.go

@@ -8,6 +8,7 @@ import (
  "k8s.io/apimachinery/pkg/runtime"
 
  wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+ "github.com/argoproj/argo/workflow/util"
 )
 
 // this function always tries to return a value, even if it is badly formed
@@ -17,7 +18,7 @@ func objectToClusterWorkflowTemplate(object runtime.Object) (*wfv1.ClusterWorkfl
  if !ok {
  return v, fmt.Errorf("malformed cluster workflow template: expected \"*unstructured.Unstructured\", got \"%s\"", reflect.TypeOf(object).String())
  }
- err := runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, v)
+ err := util.FromUnstructuredObj(un, v)
  if err != nil {
  return v, fmt.Errorf("malformed cluster workflow template \"%s\": %w", un.GetName(), err)
  }

workflow/controller/informer/workflow_template_convert.go

@@ -8,6 +8,7 @@ import (
  "k8s.io/apimachinery/pkg/runtime"
 
  wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+ "github.com/argoproj/argo/workflow/util"
 )
 
 func objectToWorkflowTemplate(object runtime.Object) (*wfv1.WorkflowTemplate, error) {
@@ -16,7 +17,7 @@ func objectToWorkflowTemplate(object runtime.Object) (*wfv1.WorkflowTemplate, er
  if !ok {
  return v, fmt.Errorf("malformed workflow template: expected \"*unstructured.Unstructured\", got \"%s\"", reflect.TypeOf(object).String())
  }
- err := runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, v)
+ err := util.FromUnstructuredObj(un, v)
  if err != nil {
  return v, fmt.Errorf("malformed workflow template \"%s/%s\": %w", un.GetNamespace(), un.GetName(), err)
  }

workflow/cron/controller.go

@@ -12,7 +12,6 @@ import (
  "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
  "k8s.io/apimachinery/pkg/fields"
  "k8s.io/apimachinery/pkg/labels"
- "k8s.io/apimachinery/pkg/runtime"
  "k8s.io/apimachinery/pkg/runtime/schema"
  "k8s.io/apimachinery/pkg/selection"
  "k8s.io/apimachinery/pkg/util/wait"
@@ -140,7 +139,7 @@ func (cc *Controller) processNextCronItem() bool {
  return true
  }
  cronWf := &v1alpha1.CronWorkflow{}
- err = runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, cronWf)
+ err = util.FromUnstructuredObj(un, cronWf)
  if err != nil {
  cc.eventRecorderManager.Get(un.GetNamespace()).Event(un, apiv1.EventTypeWarning, "Malformed", err.Error())
  logCtx.WithError(err).Error("malformed cron workflow: could not convert from unstructured")

workflow/util/util.go

@@ -119,10 +119,25 @@ func NewWorkflowLister(informer cache.SharedIndexInformer) WorkflowLister {
 // FromUnstructured converts an unstructured object to a workflow
 func FromUnstructured(un *unstructured.Unstructured) (*wfv1.Workflow, error) {
  var wf wfv1.Workflow
- err := runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, &wf)
+ err := FromUnstructuredObj(un, &wf)
  return &wf, err
 }
 
+func FromUnstructuredObj(un *unstructured.Unstructured, v interface{}) error {
+ err := runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, v)
+ if err != nil {
+ if err.Error() == "cannot convert int64 to v1alpha1.Int64OrString" {
+ data, err := json.Marshal(un)
+ if err != nil {
+ return err
+ }
+ return json.Unmarshal(data, v)
+ }
+ return err
+ }
+ return nil
+}
+
 // ToUnstructured converts an workflow to an Unstructured object
 func ToUnstructured(wf *wfv1.Workflow) (*unstructured.Unstructured, error) {
  obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(wf)

workflow/util/util_test.go

@@ -10,6 +10,7 @@ import (
  "github.com/stretchr/testify/assert"
  v1 "k8s.io/api/core/v1"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
  kubefake "k8s.io/client-go/kubernetes/fake"
  "sigs.k8s.io/yaml"
 
@@ -749,6 +750,30 @@ func TestRetryWorkflow(t *testing.T) {
  }
 }
 
+func TestFromUnstructuredObj(t *testing.T) {
+ un := &unstructured.Unstructured{}
+ err := yaml.Unmarshal([]byte(`apiVersion: argoproj.io/v1alpha1
+kind: CronWorkflow
+metadata:
+ name: example-integers
+spec:
+ schedule: "* * * * *"
+ workflowSpec:
+ entrypoint: whalesay
+ templates:
+ - name: whalesay
+ inputs:
+ parameters:
+ - name: age
+ value: 20
+ container:
+ image: my-image`), un)
+ assert.NoError(t, err)
+ x := &wfv1.CronWorkflow{}
+ err = FromUnstructuredObj(un, x)
+ assert.NoError(t, err)
+}
+
 func TestToUnstructured(t *testing.T) {
  un, err := ToUnstructured(&wfv1.Workflow{})
  if assert.NoError(t, err) {