chore: partially revert 9afe23e (kubeflow#11713)

In 9afe23e we introduced blackend dropping of unknown fields for
unmarshalling, but going forward we want to handle this more on a case
by case basis. In the case for driver we should drop them because by
this point the api server has declare the pipeline spec is acceptable,
so the driver should not fail here. As such we keep driver changes, but
revert those utilized by the api server.

Signed-off-by: Humair Khan <[email protected]>

Author: HumairAK

backend/src/apiserver/server/list_request_util.go

@@ -21,6 +21,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/golang/protobuf/jsonpb"
 	apiv1beta1 "github.com/kubeflow/pipelines/backend/api/v1beta1/go_client"
 	apiv2beta1 "github.com/kubeflow/pipelines/backend/api/v2beta1/go_client"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
@@ -147,13 +148,13 @@ func parseAPIFilter(encoded string, apiVersion string) (interface{}, error) {
 	switch apiVersion {
 	case "v2beta1":
 		f := &apiv2beta1.Filter{}
-		if err := util.UnmarshalString(decoded, f); err != nil {
+		if err := jsonpb.UnmarshalString(decoded, f); err != nil {
 			return nil, util.NewInvalidInputError("failed to parse valid filter from %q: %v", encoded, err)
 		}
 		return f, nil
 	case "v1beta1":
 		f := &apiv1beta1.Filter{}
-		if err := util.UnmarshalString(decoded, f); err != nil {
+		if err := jsonpb.UnmarshalString(decoded, f); err != nil {
 			return nil, util.NewInvalidInputError("failed to parse valid filter from %q: %v", encoded, err)
 		}
 		return f, nil

backend/src/common/util/pipelinerun.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/golang/protobuf/jsonpb"
 	"sort"
 	"strings"
 	"time"
@@ -663,7 +664,7 @@ func collectTaskRunMetricsOrNil(
 	// ReportRunMetricsRequest as a workaround to hold user's metrics, which is a superset of what
 	// user can provide.
 	reportMetricsRequest := new(api.ReportRunMetricsRequest)
-	err = UnmarshalString(metricsJSON, reportMetricsRequest)
+	err = jsonpb.UnmarshalString(metricsJSON, reportMetricsRequest)
 	if err != nil {
 		// User writes invalid metrics JSON.
 		// TODO(#1426): report the error back to api server to notify user

backend/src/common/util/workflow.go

@@ -29,6 +29,7 @@ import (
 	"github.com/argoproj/argo-workflows/v3/workflow/packer"
 	"github.com/argoproj/argo-workflows/v3/workflow/validate"
 	"github.com/golang/glog"
+	"github.com/golang/protobuf/jsonpb"
 	api "github.com/kubeflow/pipelines/backend/api/v1beta1/go_client"
 	exec "github.com/kubeflow/pipelines/backend/src/common"
 	swfregister "github.com/kubeflow/pipelines/backend/src/crd/pkg/apis/scheduledworkflow"
@@ -516,7 +517,7 @@ func collectNodeMetricsOrNil(runID string, nodeStatus *workflowapi.NodeStatus, r
 	// ReportRunMetricsRequest as a workaround to hold user's metrics, which is a superset of what
 	// user can provide.
 	reportMetricsRequest := new(api.ReportRunMetricsRequest)
-	err = UnmarshalString(metricsJSON, reportMetricsRequest)
+	err = jsonpb.UnmarshalString(metricsJSON, reportMetricsRequest)
 	if err != nil {
 		// User writes invalid metrics JSON.
 		// TODO(#1426): report the error back to api server to notify user

backend/src/v2/compiler/argocompiler/container.go

@@ -16,13 +16,13 @@ package argocompiler
 
 import (
 	"fmt"
-	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"os"
 	"strconv"
 	"strings"
 
 	wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/golang/glog"
+	"github.com/golang/protobuf/jsonpb"
 	"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
 	"github.com/kubeflow/pipelines/backend/src/v2/component"
 	"github.com/kubeflow/pipelines/kubernetes_platform/go/kubernetesplatform"
@@ -419,7 +419,7 @@ func (c *workflowCompiler) addContainerExecutorTemplate(refName string) string {
 
 	if kubernetesConfigParam != nil {
 		k8sExecCfg := &kubernetesplatform.KubernetesExecutorConfig{}
-		if err := util.UnmarshalString(string(*kubernetesConfigParam.Value), k8sExecCfg); err == nil {
+		if err := jsonpb.UnmarshalString(string(*kubernetesConfigParam.Value), k8sExecCfg); err == nil {
 			extendPodMetadata(&executor.Metadata, k8sExecCfg)
 		}
 	}

backend/src/v2/compiler/visitor.go

@@ -24,7 +24,6 @@ package compiler
 import (
 	"bytes"
 	"fmt"
-	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"sort"
 
 	"github.com/golang/protobuf/jsonpb"
@@ -188,7 +187,7 @@ func GetPipelineSpec(job *pipelinespec.PipelineJob) (*pipelinespec.PipelineSpec,
 		return nil, fmt.Errorf("failed marshal pipeline spec to json: %w", err)
 	}
 	spec := &pipelinespec.PipelineSpec{}
-	if err := util.UnmarshalString(json, spec); err != nil {
+	if err := jsonpb.UnmarshalString(json, spec); err != nil {
 		return nil, fmt.Errorf("failed to parse pipeline spec: %v", err)
 	}
 	return spec, nil

backend/src/v2/compiler/visitor_test.go

@@ -15,10 +15,10 @@ package compiler_test
 
 import (
 	"fmt"
-	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"os"
 	"testing"
 
+	"github.com/golang/protobuf/jsonpb"
 	"github.com/google/go-cmp/cmp"
 	"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
 	"github.com/kubeflow/pipelines/backend/src/v2/compiler"
@@ -93,7 +93,7 @@ func load(t *testing.T, path string) *pipelinespec.PipelineJob {
 	}
 	json := string(content)
 	job := &pipelinespec.PipelineJob{}
-	if err := util.UnmarshalString(json, job); err != nil {
+	if err := jsonpb.UnmarshalString(json, job); err != nil {
 		t.Errorf("Failed to parse pipeline job, error: %s, job: %v", err, json)
 	}
 	return job