feat(backend): upgrade argo go module to V3. Part of #5718 (#5792)

* updating argo

* updated fix deps and code

* clean up

backend/src/agent/persistence/client/workflow_client.go

@@ -15,7 +15,7 @@
 package client
 
 import (
-	"github.com/argoproj/argo/pkg/client/informers/externalversions/workflow/v1alpha1"
+	"github.com/argoproj/argo-workflows/v3/pkg/client/informers/externalversions/workflow/v1alpha1"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	"k8s.io/client-go/tools/cache"

backend/src/agent/persistence/main.go

@@ -18,8 +18,8 @@ import (
 	"flag"
 	"time"
 
-	workflowclientSet "github.com/argoproj/argo/pkg/client/clientset/versioned"
-	workflowinformers "github.com/argoproj/argo/pkg/client/informers/externalversions"
+	workflowclientSet "github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned"
+	workflowinformers "github.com/argoproj/argo-workflows/v3/pkg/client/informers/externalversions"
 	"github.com/kubeflow/pipelines/backend/src/agent/persistence/client"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	swfclientset "github.com/kubeflow/pipelines/backend/src/crd/pkg/client/clientset/versioned"

backend/src/agent/persistence/persistence_agent.go

@@ -18,8 +18,8 @@ import (
 	"fmt"
 	"time"
 
-	workflowregister "github.com/argoproj/argo/pkg/apis/workflow"
-	workflowinformers "github.com/argoproj/argo/pkg/client/informers/externalversions"
+	workflowregister "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow"
+	workflowinformers "github.com/argoproj/argo-workflows/v3/pkg/client/informers/externalversions"
 	"github.com/kubeflow/pipelines/backend/src/agent/persistence/client"
 	"github.com/kubeflow/pipelines/backend/src/agent/persistence/worker"
 	"github.com/kubeflow/pipelines/backend/src/common/util"

backend/src/agent/persistence/worker/metrics_reporter.go

@@ -18,7 +18,7 @@ import (
 	"errors"
 	"strings"
 
-	workflowapi "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	workflowapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/golang/protobuf/jsonpb"
 	api "github.com/kubeflow/pipelines/backend/api/go_client"
 	"github.com/kubeflow/pipelines/backend/src/agent/persistence/client"

backend/src/agent/persistence/worker/metrics_reporter_test.go

@@ -17,7 +17,7 @@ package worker
 import (
 	"testing"
 
-	workflowapi "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	workflowapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	api "github.com/kubeflow/pipelines/backend/api/go_client"
 	"github.com/kubeflow/pipelines/backend/src/agent/persistence/client"
 	"github.com/kubeflow/pipelines/backend/src/common/util"

backend/src/agent/persistence/worker/persistence_worker_test.go

@@ -18,7 +18,7 @@ import (
 	"fmt"
 	"testing"
 
-	workflowapi "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	workflowapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	client "github.com/kubeflow/pipelines/backend/src/agent/persistence/client"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/stretchr/testify/assert"

backend/src/agent/persistence/worker/workflow_saver_test.go

@@ -19,7 +19,7 @@ import (
 	"testing"
 	"time"
 
-	workflowapi "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	workflowapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/kubeflow/pipelines/backend/src/agent/persistence/client"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/stretchr/testify/assert"
@@ -170,8 +170,8 @@ func TestWorkflow_Save_FinalStatueNotSkippedDueToExceedTTL(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
-			Labels:    map[string]string{
-				util.LabelKeyWorkflowRunId: "MY_UUID",
+			Labels: map[string]string{
+				util.LabelKeyWorkflowRunId:               "MY_UUID",
 				util.LabelKeyWorkflowPersistedFinalState: "true",
 			},
 		},

backend/src/apiserver/archive/log_test.go

@@ -19,12 +19,13 @@ import (
 	"bytes"
 	"compress/gzip"
 	"encoding/json"
-	workflowapi "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	"testing"
+	"time"
+
+	workflowapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"testing"
-	"time"
 )
 
 func compressInput(t *testing.T, content string) []byte {

backend/src/apiserver/auth/authenticator_token_review.go

@@ -21,6 +21,7 @@ import (
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/pkg/errors"
 	authv1 "k8s.io/api/authentication/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 type TokenReviewAuthenticator struct {
@@ -49,7 +50,7 @@ func (tra *TokenReviewAuthenticator) GetUserIdentity(ctx context.Context) (strin
 		return "", err
 	}
 
-	userInfo, err := tra.doTokenReview(token)
+	userInfo, err := tra.doTokenReview(ctx, token)
 	if err != nil {
 		return "", util.Wrap(err, "Authentication failure")
 	}
@@ -73,14 +74,17 @@ func (tra *TokenReviewAuthenticator) ensureAudience(audience []string) bool {
 	return true
 }
 
-func (tra *TokenReviewAuthenticator) doTokenReview(userIdentity string) (*authv1.UserInfo, error) {
+func (tra *TokenReviewAuthenticator) doTokenReview(ctx context.Context, userIdentity string) (*authv1.UserInfo, error) {
 	review, err := tra.client.Create(
+		ctx,
 		&authv1.TokenReview{
 			Spec: authv1.TokenReviewSpec{
 				Token:     userIdentity,
 				Audiences: tra.audiences,
 			},
-		})
+		},
+		v1.CreateOptions{},
+	)
 	if err != nil {
 		return nil, util.NewUnauthenticatedError(err, "Request header error: Failed to review the token provided")
 	}

backend/src/apiserver/client/argo.go

@@ -17,8 +17,8 @@ package client
 import (
 	"time"
 
-	argoclient "github.com/argoproj/argo/pkg/client/clientset/versioned"
-	argoprojv1alpha1 "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
+	argoclient "github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned"
+	argoprojv1alpha1 "github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
 	"github.com/cenkalti/backoff"
 	"github.com/golang/glog"
 	"github.com/kubeflow/pipelines/backend/src/common/util"

backend/src/apiserver/client/argo_fake.go

@@ -15,7 +15,7 @@
 package client
 
 import (
-	argoprojv1alpha1 "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
+	argoprojv1alpha1 "github.com/argoproj/argo-workflows/v3/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/pkg/errors"
 )

backend/src/apiserver/client/pod_fake.go

@@ -1,6 +1,7 @@
 package client
 
 import (
+	"context"
 	"errors"
 
 	"github.com/golang/glog"
@@ -15,66 +16,66 @@ import (
 type FakePodClient struct {
 }
 
-func (FakePodClient) GetEphemeralContainers(string, v1.GetOptions) (*corev1.EphemeralContainers, error) {
+func (FakePodClient) GetEphemeralContainers(context.Context, string, v1.GetOptions) (*corev1.EphemeralContainers, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) UpdateEphemeralContainers(string, *corev1.EphemeralContainers) (*corev1.EphemeralContainers, error) {
+func (FakePodClient) UpdateEphemeralContainers(context.Context, string, *corev1.EphemeralContainers, v1.UpdateOptions) (*corev1.EphemeralContainers, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) Create(*corev1.Pod) (*corev1.Pod, error) {
+func (FakePodClient) Create(context.Context, *corev1.Pod, v1.CreateOptions) (*corev1.Pod, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) Update(*corev1.Pod) (*corev1.Pod, error) {
+func (FakePodClient) Update(context.Context, *corev1.Pod, v1.UpdateOptions) (*corev1.Pod, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) UpdateStatus(*corev1.Pod) (*corev1.Pod, error) {
+func (FakePodClient) UpdateStatus(context.Context, *corev1.Pod, v1.UpdateOptions) (*corev1.Pod, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) Delete(name string, options *v1.DeleteOptions) error {
+func (FakePodClient) Delete(ctx context.Context, name string, options v1.DeleteOptions) error {
 	return nil
 }
 
-func (FakePodClient) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
+func (FakePodClient) DeleteCollection(ctx context.Context, options v1.DeleteOptions, listOptions v1.ListOptions) error {
 	glog.Error("This fake method is not yet implemented.")
 	return nil
 }
 
-func (FakePodClient) Get(name string, options v1.GetOptions) (*corev1.Pod, error) {
+func (FakePodClient) Get(ctx context.Context, name string, options v1.GetOptions) (*corev1.Pod, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) List(opts v1.ListOptions) (*corev1.PodList, error) {
+func (FakePodClient) List(ctx context.Context, opts v1.ListOptions) (*corev1.PodList, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) Watch(opts v1.ListOptions) (watch.Interface, error) {
+func (FakePodClient) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *corev1.Pod, err error) {
+func (FakePodClient) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *corev1.Pod, err error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (FakePodClient) Bind(binding *corev1.Binding) error {
+func (FakePodClient) Bind(ctx context.Context, binding *corev1.Binding, opts v1.CreateOptions) error {
 	glog.Error("This fake method is not yet implemented.")
 	return nil
 }
 
-func (FakePodClient) Evict(eviction *v1beta1.Eviction) error {
+func (FakePodClient) Evict(ctx context.Context, eviction *v1beta1.Eviction) error {
 	glog.Error("This fake method is not yet implemented.")
 	return nil
 }
@@ -84,10 +85,15 @@ func (FakePodClient) GetLogs(name string, opts *corev1.PodLogOptions) *rest.Requ
 	return nil
 }
 
+func (FakePodClient) ProxyGet(scheme, name, port, path string, params map[string]string) rest.ResponseWrapper {
+	glog.Error("This fake method is not yet implemented.")
+	return nil
+}
+
 type FakeBadPodClient struct {
 	FakePodClient
 }
 
-func (FakeBadPodClient) Delete(name string, options *v1.DeleteOptions) error {
+func (FakeBadPodClient) Delete(ctx context.Context, name string, options v1.DeleteOptions) error {
 	return errors.New("failed to delete pod")
 }

backend/src/apiserver/client/scheduled_workflow_fake.go

@@ -15,6 +15,7 @@
 package client
 
 import (
+	"context"
 	"errors"
 
 	"github.com/golang/glog"
@@ -36,15 +37,15 @@ func NewScheduledWorkflowClientFake() *FakeScheduledWorkflowClient {
 	}
 }
 
-func (c *FakeScheduledWorkflowClient) Create(scheduledWorkflow *v1beta1.ScheduledWorkflow) (*v1beta1.ScheduledWorkflow, error) {
+func (c *FakeScheduledWorkflowClient) Create(ctx context.Context, scheduledWorkflow *v1beta1.ScheduledWorkflow) (*v1beta1.ScheduledWorkflow, error) {
 	scheduledWorkflow.UID = "123e4567-e89b-12d3-a456-426655440000"
 	scheduledWorkflow.Namespace = "ns1"
 	scheduledWorkflow.Name = scheduledWorkflow.GenerateName
 	c.scheduledWorkflows[scheduledWorkflow.Name] = scheduledWorkflow
 	return scheduledWorkflow, nil
 }
 
-func (c *FakeScheduledWorkflowClient) Delete(name string, options *v1.DeleteOptions) error {
+func (c *FakeScheduledWorkflowClient) Delete(ctx context.Context, name string, options *v1.DeleteOptions) error {
 	_, ok := c.scheduledWorkflows[name]
 	if ok {
 		delete(c.scheduledWorkflows, name)
@@ -53,34 +54,34 @@ func (c *FakeScheduledWorkflowClient) Delete(name string, options *v1.DeleteOpti
 	return k8errors.NewNotFound(k8schema.ParseGroupResource("scheduledworkflows.kubeflow.org"), name)
 }
 
-func (c *FakeScheduledWorkflowClient) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1beta1.ScheduledWorkflow, err error) {
+func (c *FakeScheduledWorkflowClient) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, subresources ...string) (result *v1beta1.ScheduledWorkflow, err error) {
 	return nil, nil
 }
 
-func (c *FakeScheduledWorkflowClient) Get(name string, options v1.GetOptions) (*v1beta1.ScheduledWorkflow, error) {
+func (c *FakeScheduledWorkflowClient) Get(ctx context.Context, name string, options v1.GetOptions) (*v1beta1.ScheduledWorkflow, error) {
 	scheduledWorkflow, ok := c.scheduledWorkflows[name]
 	if ok {
 		return scheduledWorkflow, nil
 	}
 	return nil, k8errors.NewNotFound(k8schema.ParseGroupResource("scheduledworkflows.kubeflow.org"), name)
 }
 
-func (c *FakeScheduledWorkflowClient) Update(*v1beta1.ScheduledWorkflow) (*v1beta1.ScheduledWorkflow, error) {
+func (c *FakeScheduledWorkflowClient) Update(context.Context, *v1beta1.ScheduledWorkflow) (*v1beta1.ScheduledWorkflow, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (c *FakeScheduledWorkflowClient) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
+func (c *FakeScheduledWorkflowClient) DeleteCollection(ctx context.Context, options *v1.DeleteOptions, listOptions v1.ListOptions) error {
 	glog.Error("This fake method is not yet implemented.")
 	return nil
 }
 
-func (c *FakeScheduledWorkflowClient) List(opts v1.ListOptions) (*v1beta1.ScheduledWorkflowList, error) {
+func (c *FakeScheduledWorkflowClient) List(ctx context.Context, opts v1.ListOptions) (*v1beta1.ScheduledWorkflowList, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
 
-func (c *FakeScheduledWorkflowClient) Watch(opts v1.ListOptions) (watch.Interface, error) {
+func (c *FakeScheduledWorkflowClient) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
 	glog.Error("This fake method is not yet implemented.")
 	return nil, nil
 }
@@ -89,18 +90,18 @@ type FakeBadScheduledWorkflowClient struct {
 	FakeScheduledWorkflowClient
 }
 
-func (FakeBadScheduledWorkflowClient) Create(workflow *v1beta1.ScheduledWorkflow) (*v1beta1.ScheduledWorkflow, error) {
+func (FakeBadScheduledWorkflowClient) Create(ctx context.Context, workflow *v1beta1.ScheduledWorkflow) (*v1beta1.ScheduledWorkflow, error) {
 	return nil, errors.New("some error")
 }
 
-func (FakeBadScheduledWorkflowClient) Get(name string, options v1.GetOptions) (*v1beta1.ScheduledWorkflow, error) {
+func (FakeBadScheduledWorkflowClient) Get(ctx context.Context, name string, options v1.GetOptions) (*v1beta1.ScheduledWorkflow, error) {
 	return nil, errors.New("some error")
 }
 
-func (FakeBadScheduledWorkflowClient) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1beta1.ScheduledWorkflow, err error) {
+func (FakeBadScheduledWorkflowClient) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, subresources ...string) (result *v1beta1.ScheduledWorkflow, err error) {
 	return nil, errors.New("some error")
 }
 
-func (c *FakeBadScheduledWorkflowClient) Delete(name string, options *v1.DeleteOptions) error {
+func (c *FakeBadScheduledWorkflowClient) Delete(ctx context.Context, name string, options *v1.DeleteOptions) error {
 	return errors.New("some error")
 }

backend/src/apiserver/client/subject_access_review.go

@@ -15,16 +15,18 @@
 package client
 
 import (
+	"context"
 	"time"
 
 	"github.com/cenkalti/backoff"
 	"github.com/golang/glog"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	authzv1 "k8s.io/api/authorization/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 type SubjectAccessReviewInterface interface {
-	Create(sar *authzv1.SubjectAccessReview) (result *authzv1.SubjectAccessReview, err error)
+	Create(ctx context.Context, sar *authzv1.SubjectAccessReview, opts v1.CreateOptions) (result *authzv1.SubjectAccessReview, err error)
 }
 
 func createSubjectAccessReviewClient(clientParams util.ClientParameters) (SubjectAccessReviewInterface, error) {

backend/src/apiserver/client/subject_access_review_fake.go

@@ -15,14 +15,17 @@
 package client
 
 import (
+	"context"
+
 	"github.com/pkg/errors"
 	authzv1 "k8s.io/api/authorization/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 type FakeSubjectAccessReviewClient struct {
 }
 
-func (FakeSubjectAccessReviewClient) Create(*authzv1.SubjectAccessReview) (*authzv1.SubjectAccessReview, error) {
+func (FakeSubjectAccessReviewClient) Create(context.Context, *authzv1.SubjectAccessReview, v1.CreateOptions) (*authzv1.SubjectAccessReview, error) {
 	return &authzv1.SubjectAccessReview{Status: authzv1.SubjectAccessReviewStatus{
 		Allowed:         true,
 		Denied:          false,
@@ -38,7 +41,7 @@ func NewFakeSubjectAccessReviewClient() FakeSubjectAccessReviewClient {
 type FakeSubjectAccessReviewClientUnauthorized struct {
 }
 
-func (FakeSubjectAccessReviewClientUnauthorized) Create(*authzv1.SubjectAccessReview) (*authzv1.SubjectAccessReview, error) {
+func (FakeSubjectAccessReviewClientUnauthorized) Create(context.Context, *authzv1.SubjectAccessReview, v1.CreateOptions) (*authzv1.SubjectAccessReview, error) {
 	return &authzv1.SubjectAccessReview{Status: authzv1.SubjectAccessReviewStatus{
 		Allowed:         false,
 		Denied:          false,