Rebase on upstream KFP 7838009

Author: mprahl

.github/workflows/validate-generated-files.yml

@@ -1,13 +1,20 @@
 name: Validate Generated Files
 on:
+  push:
+    branches:
+      - master
   pull_request:
     paths:
-      - 'backend/api/**/*.proto'
       - '.github/workflows/validate-generated-files.yml'
+      - 'backend/api/**/*.proto'
       - 'backend/api/**/go_http_client/**'
       - 'backend/api/**/go_client/**'
       - 'backend/api/**/python_http_client/**'
       - 'backend/api/**/swagger/**'
+      - 'api/**/*.proto'
+      - 'api/**/*.go'
+      - 'kubernetes_platform/**/*.proto'
+      - 'kubernetes_platform/**/*.go'
 
 jobs:
   validate-generated-files:

api/v2alpha1/go/pipelinespec/pipeline_spec.pb.go

@@ -292,7 +292,7 @@ message TaskInputsSpec {
 
   // Represents an input parameter. The value can be taken from an upstream
   // task's output parameter (if specifying `producer_task` and
-  // `output_parameter_key`, or it can be a runtime value, which can either be
+  // `output_parameter_key`), or it can be a runtime value, which can either be
   // determined at compile-time, or from a pipeline parameter.
   message InputParameterSpec {
     // Represents an upstream task's output parameter.
@@ -307,9 +307,9 @@ message TaskInputsSpec {
 
     // Represents an upstream task's final status. The field can only be set if
     // the schema version is `2.0.0`. The resolved input parameter will be a
-    // json payload in string type.
+    // JSON payload in string type.
     message TaskFinalStatus {
-      // The name of the upsteram task where the final status is coming from.
+      // The name of the upstream task where the final status is coming from.
       string producer_task = 1;
     }
 

api/v2alpha1/pipeline_spec.proto

@@ -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/apiserver/server/list_request_util.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/pipelinerun.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/common/util/workflow.go

@@ -20,10 +20,9 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/kubeflow/pipelines/backend/src/common/util"
-
 	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/apiserver/common"
 	"github.com/kubeflow/pipelines/backend/src/v2/component"
@@ -431,7 +430,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/argocompiler/container.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.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

backend/src/v2/compiler/visitor_test.go

@@ -60,7 +60,7 @@ clean-python:
 .PHONY: generate-go-in-container
 generate-go-in-container:
 	mkdir -p go/kubernetesplatform
-	cd proto && protoc -I=. \
+	cd proto && protoc -I=. -I=../../api/v2alpha1 \
 		--go_out=../go/kubernetesplatform \
 		--go_opt=paths=source_relative \
 		kubernetes_executor_config.proto