Repository: kubeflow/kfp-tekton (language: python)

Issue:
fix(sdk): Update wait_for_run_completion function to handle tekton status;
**Which issue is resolved by this Pull Request:** 
Resolves #1223 

**Description of your changes:**

**Environment tested:**

* Python Version (use `python --version`):
* Tekton Version (use `tkn version`):
* Kubernetes Version (use `kubectl version`):
* OS (e.g. from `/etc/os-release`):

**Checklist:**
- [x] The title for your pull request (PR) should follow our title convention. [Learn more about the pull request title convention used in this repository](https://github.com/kubeflow/pipelines/blob/master/CONTRIBUTING.md#pull-request-title-convention). 
<!--
   PR titles examples:
    * `fix(frontend): fixes empty page. Fixes #1234`
       Use `fix` to indicate that this PR fixes a bug.
    * `feat(backend): configurable service account. Fixes #1234, fixes #1235`
       Use `feat` to indicate that this PR adds a new feature. 
    * `chore: set up changelog generation tools`
       Use `chore` to indicate that this PR makes some changes that users don't need to know.
    * `test: fix CI failure. Part of #1234`
        Use `part of` to indicate that a PR is working on an issue, but shouldn't close the issue when merged.
-->

Update/remove all the Argo related test and mock clients
/kind feature

**Description:**
related #382 
Since Argo and Tekton are having major conflicts with the client-go in the next release, we need to completely remove all the Argo related stuff before upgrading the Tekton to the 0.18+ version.

**Additional information:**
[Miscellaneous information that will assist in solving the issue.]

fix(sdk): Update wait_for_run_completion function to handle tekton status;
**Which issue is resolved by this Pull Request:** 
Resolves #1223 

**Description of your changes:**

**Environment tested:**

* Python Version (use `python --version`):
* Tekton Version (use `tkn version`):
* Kubernetes Version (use `kubectl version`):
* OS (e.g. from `/etc/os-release`):

**Checklist:**
- [x] The title for your pull request (PR) should follow our title convention. [Learn more about the pull request title convention used in this repository](https://github.com/kubeflow/pipelines/blob/master/CONTRIBUTING.md#pull-request-title-convention). 
<!--
   PR titles examples:
    * `fix(frontend): fixes empty page. Fixes #1234`
       Use `fix` to indicate that this PR fixes a bug.
    * `feat(backend): configurable service account. Fixes #1234, fixes #1235`
       Use `feat` to indicate that this PR adds a new feature. 
    * `chore: set up changelog generation tools`
       Use `chore` to indicate that this PR makes some changes that users don't need to know.
    * `test: fix CI failure. Part of #1234`
        Use `part of` to indicate that a PR is working on an issue, but shouldn't close the issue when merged.
-->

fix(sdk): Update kfp version to fix appengine bug
**Which issue is resolved by this Pull Request:** 
Resolves #

**Description of your changes:**

**Environment tested:**

* Python Version (use `python --version`):
* Tekton Version (use `tkn version`):
* Kubernetes Version (use `kubectl version`):
* OS (e.g. from `/etc/os-release`):

**Checklist:**
- [x] The title for your pull request (PR) should follow our title convention. [Learn more about the pull request title convention used in this repository](https://github.com/kubeflow/pipelines/blob/master/CONTRIBUTING.md#pull-request-title-convention). 
<!--
   PR titles examples:
    * `fix(frontend): fixes empty page. Fixes #1234`
       Use `fix` to indicate that this PR fixes a bug.
    * `feat(backend): configurable service account. Fixes #1234, fixes #1235`
       Use `feat` to indicate that this PR adds a new feature. 
    * `chore: set up changelog generation tools`
       Use `chore` to indicate that this PR makes some changes that users don't need to know.
    * `test: fix CI failure. Part of #1234`
        Use `part of` to indicate that a PR is working on an issue, but shouldn't close the issue when merged.
-->



Discussion / hints:

Fix patch:
diff --git a/backend/src/agent/persistence/worker/metrics_reporter.go b/backend/src/agent/persistence/worker/metrics_reporter.go
index eff74016f9..9ff2a8b153 100644
--- a/backend/src/agent/persistence/worker/metrics_reporter.go
+++ b/backend/src/agent/persistence/worker/metrics_reporter.go
@@ -93,7 +93,7 @@ func (r MetricsReporter) collectNodeMetricsOrNil(
 	[]*api.RunMetric, error) {
 	defer func() {
 		if panicMessage := recover(); panicMessage != nil {
-			log.Info("nodeStatus is not yet created. Panic message: '%v'.", panicMessage)
+			log.Infof("nodeStatus is not yet created. Panic message: '%v'.", panicMessage)
 		}
 	}()
 	if nodeStatus.Status == nil ||


Test patch:
diff --git a/backend/src/agent/persistence/worker/metrics_reporter_test.go b/backend/src/agent/persistence/worker/metrics_reporter_test.go
index ff7b812ba4..ec2dc80c15 100644
--- a/backend/src/agent/persistence/worker/metrics_reporter_test.go
+++ b/backend/src/agent/persistence/worker/metrics_reporter_test.go
@@ -17,11 +17,10 @@ package worker
 import (
 	"testing"
 
-	workflowapi "github.com/argoproj/argo/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"
 	"github.com/stretchr/testify/assert"
+	workflowapi "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 )
@@ -31,20 +30,13 @@ func TestReportMetrics_NoCompletedNode_NoOP(t *testing.T) {
 
 	reporter := NewMetricsReporter(pipelineFake)
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
 			UID:       types.UID("run-1"),
 		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeRunning,
-				},
-			},
-		},
+		Status: workflowapi.PipelineRunStatus{},
 	})
 	err := reporter.ReportMetrics(workflow)
 	assert.Nil(t, err)
@@ -56,20 +48,13 @@ func TestReportMetrics_NoRunID_NoOP(t *testing.T) {
 
 	reporter := NewMetricsReporter(pipelineFake)
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
 			UID:       types.UID("run-1"),
 		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-				},
-			},
-		},
+		Status: workflowapi.PipelineRunStatus{},
 	})
 	err := reporter.ReportMetrics(workflow)
 	assert.Nil(t, err)
@@ -82,21 +67,14 @@ func TestReportMetrics_NoArtifact_NoOP(t *testing.T) {
 
 	reporter := NewMetricsReporter(pipelineFake)
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
 			UID:       types.UID("run-1"),
 			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
 		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-				},
-			},
-		},
+		Status: workflowapi.PipelineRunStatus{},
 	})
 	err := reporter.ReportMetrics(workflow)
 	assert.Nil(t, err)
@@ -109,24 +87,14 @@ func TestReportMetrics_NoMetricsArtifact_NoOP(t *testing.T) {
 
 	reporter := NewMetricsReporter(pipelineFake)
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
 			UID:       types.UID("run-1"),
 			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
 		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-ui-metadata"}},
-					},
-				},
-			},
-		},
+		Status: workflowapi.PipelineRunStatus{},
 	})
 	err := reporter.ReportMetrics(workflow)
 	assert.Nil(t, err)
@@ -134,364 +102,16 @@ func TestReportMetrics_NoMetricsArtifact_NoOP(t *testing.T) {
 	assert.Nil(t, pipelineFake.GetReportedMetricsRequest())
 }
 
-func TestReportMetrics_Succeed(t *testing.T) {
-	pipelineFake := client.NewPipelineClientFake()
-	reporter := NewMetricsReporter(pipelineFake)
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "MY_NAMESPACE",
-			Name:      "MY_NAME",
-			UID:       types.UID("run-1"),
-			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
-		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-			},
-		},
-	})
-	metricsJSON := `{"metrics": [{"name": "accuracy", "numberValue": 0.77}, {"name": "logloss", "numberValue": 1.2}]}`
-	artifactData, _ := util.ArchiveTgz(map[string]string{"file": metricsJSON})
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-1",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte(artifactData),
-		})
-	pipelineFake.StubReportRunMetrics(&api.ReportRunMetricsResponse{
-		Results: []*api.ReportRunMetricsResponse_ReportRunMetricResult{},
-	}, nil)
-
-	err := reporter.ReportMetrics(workflow)
-
-	assert.Nil(t, err)
-	expectedMetricsRequest := &api.ReportRunMetricsRequest{
-		RunId: "run-1",
-		Metrics: []*api.RunMetric{
-			{
-				Name:   "accuracy",
-				NodeId: "node-1",
-				Value:  &api.RunMetric_NumberValue{NumberValue: 0.77},
-			},
-			{
-				Name:   "logloss",
-				NodeId: "node-1",
-				Value:  &api.RunMetric_NumberValue{NumberValue: 1.2},
-			},
-		},
-	}
-	assert.Equal(t, expectedMetricsRequest, pipelineFake.GetReportedMetricsRequest())
-}
-
-func TestReportMetrics_EmptyArchive_Fail(t *testing.T) {
-	pipelineFake := client.NewPipelineClientFake()
-	reporter := NewMetricsReporter(pipelineFake)
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "MY_NAMESPACE",
-			Name:      "MY_NAME",
-			UID:       types.UID("run-1"),
-			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
-		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-			},
-		},
-	})
-	artifactData, _ := util.ArchiveTgz(map[string]string{})
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-1",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte(artifactData),
-		})
-
-	err := reporter.ReportMetrics(workflow)
-
-	assert.NotNil(t, err)
-	assert.True(t, util.HasCustomCode(err, util.CUSTOM_CODE_PERMANENT))
-	// Verify that ReportRunMetrics is not called.
-	assert.Nil(t, pipelineFake.GetReportedMetricsRequest())
-}
-
-func TestReportMetrics_MultipleFilesInArchive_Fail(t *testing.T) {
-	pipelineFake := client.NewPipelineClientFake()
-	reporter := NewMetricsReporter(pipelineFake)
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "MY_NAMESPACE",
-			Name:      "MY_NAME",
-			UID:       types.UID("run-1"),
-			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
-		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-			},
-		},
-	})
-	validMetricsJSON := `{"metrics": [{"name": "accuracy", "numberValue": 0.77}, {"name": "logloss", "numberValue": 1.2}]}`
-	invalidMetricsJSON := `invalid JSON`
-	artifactData, _ := util.ArchiveTgz(map[string]string{"file1": validMetricsJSON, "file2": invalidMetricsJSON})
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-1",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte(artifactData),
-		})
-
-	err := reporter.ReportMetrics(workflow)
-
-	assert.NotNil(t, err)
-	assert.True(t, util.HasCustomCode(err, util.CUSTOM_CODE_PERMANENT))
-	// Verify that ReportRunMetrics is not called.
-	assert.Nil(t, pipelineFake.GetReportedMetricsRequest())
-}
-
-func TestReportMetrics_InvalidMetricsJSON_Fail(t *testing.T) {
-	pipelineFake := client.NewPipelineClientFake()
-	reporter := NewMetricsReporter(pipelineFake)
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "MY_NAMESPACE",
-			Name:      "MY_NAME",
-			UID:       types.UID("run-1"),
-			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
-		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-			},
-		},
-	})
-	metricsJSON := `invalid JSON`
-	artifactData, _ := util.ArchiveTgz(map[string]string{"file": metricsJSON})
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-1",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte(artifactData),
-		})
-
-	err := reporter.ReportMetrics(workflow)
-
-	assert.NotNil(t, err)
-	assert.True(t, util.HasCustomCode(err, util.CUSTOM_CODE_PERMANENT))
-	// Verify that ReportRunMetrics is not called.
-	assert.Nil(t, pipelineFake.GetReportedMetricsRequest())
-}
-
-func TestReportMetrics_InvalidMetricsJSON_PartialFail(t *testing.T) {
-	pipelineFake := client.NewPipelineClientFake()
-	reporter := NewMetricsReporter(pipelineFake)
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "MY_NAMESPACE",
-			Name:      "MY_NAME",
-			UID:       types.UID("run-1"),
-			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
-		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-				"node-2": workflowapi.NodeStatus{
-					ID:    "node-2",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-			},
-		},
-	})
-	validMetricsJSON := `{"metrics": [{"name": "accuracy", "numberValue": 0.77}, {"name": "logloss", "numberValue": 1.2}]}`
-	invalidMetricsJSON := `invalid JSON`
-	validArtifactData, _ := util.ArchiveTgz(map[string]string{"file": validMetricsJSON})
-	invalidArtifactData, _ := util.ArchiveTgz(map[string]string{"file": invalidMetricsJSON})
-	// Stub two artifacts, node-1 is invalid, node-2 is valid.
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-1",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte(invalidArtifactData),
-		})
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-2",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte(validArtifactData),
-		})
+// Removed TestReportMetrics_Succeed - specific to argo artifact spec
 
-	err := reporter.ReportMetrics(workflow)
+// Removed TestReportMetrics_EmptyArchive_Fail - specific to argo artifact spec
 
-	// Partial failure is reported while valid metrics are reported.
-	assert.NotNil(t, err)
-	assert.True(t, util.HasCustomCode(err, util.CUSTOM_CODE_PERMANENT))
-	expectedMetricsRequest := &api.ReportRunMetricsRequest{
-		RunId: "run-1",
-		Metrics: []*api.RunMetric{
-			&api.RunMetric{
-				Name:   "accuracy",
-				NodeId: "node-2",
-				Value:  &api.RunMetric_NumberValue{NumberValue: 0.77},
-			},
-			&api.RunMetric{
-				Name:   "logloss",
-				NodeId: "node-2",
-				Value:  &api.RunMetric_NumberValue{NumberValue: 1.2},
-			},
-		},
-	}
-	assert.Equal(t, expectedMetricsRequest, pipelineFake.GetReportedMetricsRequest())
-}
+// Removed TestReportMetrics_MultipleFilesInArchive_Fail - specific to argo artifact spec
 
-func TestReportMetrics_CorruptedArchiveFile_Fail(t *testing.T) {
-	pipelineFake := client.NewPipelineClientFake()
-	reporter := NewMetricsReporter(pipelineFake)
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "MY_NAMESPACE",
-			Name:      "MY_NAME",
-			UID:       types.UID("run-1"),
-			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
-		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-			},
-		},
-	})
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-1",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte("invalid tgz content"),
-		})
+// Removed TestReportMetrics_InvalidMetricsJSON_Fail - specific to argo artifact spec
 
-	err := reporter.ReportMetrics(workflow)
+// Removed TestReportMetrics_InvalidMetricsJSON_PartialFail - specific to argo artifact spec
 
-	assert.NotNil(t, err)
-	assert.True(t, util.HasCustomCode(err, util.CUSTOM_CODE_PERMANENT))
-	// Verify that ReportRunMetrics is not called.
-	assert.Nil(t, pipelineFake.GetReportedMetricsRequest())
-}
+// Removed TestReportMetrics_CorruptedArchiveFile_Fail - specific to argo artifact spec
 
-func TestReportMetrics_MultiplMetricErrors_TransientErrowWin(t *testing.T) {
-	pipelineFake := client.NewPipelineClientFake()
-	reporter := NewMetricsReporter(pipelineFake)
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: "MY_NAMESPACE",
-			Name:      "MY_NAME",
-			UID:       types.UID("run-1"),
-			Labels:    map[string]string{util.LabelKeyWorkflowRunId: "run-1"},
-		},
-		Status: workflowapi.WorkflowStatus{
-			Nodes: map[string]workflowapi.NodeStatus{
-				"node-1": workflowapi.NodeStatus{
-					ID:    "node-1",
-					Phase: workflowapi.NodeSucceeded,
-					Outputs: &workflowapi.Outputs{
-						Artifacts: []workflowapi.Artifact{{Name: "mlpipeline-metrics"}},
-					},
-				},
-			},
-		},
-	})
-	metricsJSON :=
-		`{"metrics": [{"name": "accuracy", "numberValue": 0.77}, {"name": "log loss", "numberValue": 1.2}, {"name": "accuracy", "numberValue": 1.2}]}`
-	artifactData, _ := util.ArchiveTgz(map[string]string{"file": metricsJSON})
-	pipelineFake.StubArtifact(
-		&api.ReadArtifactRequest{
-			RunId:        "run-1",
-			NodeId:       "node-1",
-			ArtifactName: "mlpipeline-metrics",
-		},
-		&api.ReadArtifactResponse{
-			Data: []byte(artifactData),
-		})
-	pipelineFake.StubReportRunMetrics(&api.ReportRunMetricsResponse{
-		Results: []*api.ReportRunMetricsResponse_ReportRunMetricResult{
-			&api.ReportRunMetricsResponse_ReportRunMetricResult{
-				MetricNodeId: "node-1",
-				MetricName:   "accuracy",
-				Status:       api.ReportRunMetricsResponse_ReportRunMetricResult_OK,
-			},
-			// Invalid argument error triggers permanent error
-			&api.ReportRunMetricsResponse_ReportRunMetricResult{
-				MetricNodeId: "node-1",
-				MetricName:   "log loss",
-				Status:       api.ReportRunMetricsResponse_ReportRunMetricResult_INVALID_ARGUMENT,
-			},
-			// Internal error triggers transient error
-			&api.ReportRunMetricsResponse_ReportRunMetricResult{
-				MetricNodeId: "node-1",
-				MetricName:   "accuracy",
-				Status:       api.ReportRunMetricsResponse_ReportRunMetricResult_INTERNAL_ERROR,
-			},
-		},
-	}, nil)
-
-	err := reporter.ReportMetrics(workflow)
-
-	assert.NotNil(t, err)
-	assert.True(t, util.HasCustomCode(err, util.CUSTOM_CODE_TRANSIENT))
-}
+// Removed TestReportMetrics_MultiplMetricErrors_TransientErrowWin - specific to argo artifact spec
diff --git a/backend/src/agent/persistence/worker/persistence_worker_test.go b/backend/src/agent/persistence/worker/persistence_worker_test.go
index 75141f00d8..3113a591cc 100644
--- a/backend/src/agent/persistence/worker/persistence_worker_test.go
+++ b/backend/src/agent/persistence/worker/persistence_worker_test.go
@@ -18,10 +18,10 @@ import (
 	"fmt"
 	"testing"
 
-	workflowapi "github.com/argoproj/argo/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"
+	workflowapi "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	"k8s.io/client-go/tools/cache"
@@ -41,7 +41,7 @@ func (h *FakeEventHandler) AddEventHandler(handler cache.ResourceEventHandler) {
 
 func TestPersistenceWorker_Success(t *testing.T) {
 	// Set up workflow client
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
@@ -73,7 +73,7 @@ func TestPersistenceWorker_Success(t *testing.T) {
 
 func TestPersistenceWorker_NotFoundError(t *testing.T) {
 	// Set up workflow client
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
@@ -103,7 +103,7 @@ func TestPersistenceWorker_NotFoundError(t *testing.T) {
 
 func TestPersistenceWorker_GetWorklowError(t *testing.T) {
 	// Set up workflow client
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
@@ -134,7 +134,7 @@ func TestPersistenceWorker_GetWorklowError(t *testing.T) {
 
 func TestPersistenceWorker_ReportWorkflowRetryableError(t *testing.T) {
 	// Set up workflow client
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
@@ -168,7 +168,7 @@ func TestPersistenceWorker_ReportWorkflowRetryableError(t *testing.T) {
 
 func TestPersistenceWorker_ReportWorkflowNonRetryableError(t *testing.T) {
 	// Set up workflow client
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
diff --git a/backend/src/agent/persistence/worker/workflow_saver_test.go b/backend/src/agent/persistence/worker/workflow_saver_test.go
index 22926fbc3d..8cd76eb349 100644
--- a/backend/src/agent/persistence/worker/workflow_saver_test.go
+++ b/backend/src/agent/persistence/worker/workflow_saver_test.go
@@ -19,10 +19,10 @@ import (
 	"testing"
 	"time"
 
-	workflowapi "github.com/argoproj/argo/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"
+	workflowapi "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 )
@@ -31,7 +31,7 @@ func TestWorkflow_Save_Success(t *testing.T) {
 	workflowFake := client.NewWorkflowClientFake()
 	pipelineFake := client.NewPipelineClientFake()
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
@@ -84,7 +84,7 @@ func TestWorkflow_Save_PermanentFailureWhileReporting(t *testing.T) {
 	pipelineFake.SetError(util.NewCustomError(fmt.Errorf("Error"), util.CUSTOM_CODE_PERMANENT,
 		"My Permanent Error"))
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
@@ -110,7 +110,7 @@ func TestWorkflow_Save_TransientFailureWhileReporting(t *testing.T) {
 	pipelineFake.SetError(util.NewCustomError(fmt.Errorf("Error"), util.CUSTOM_CODE_TRANSIENT,
 		"My Transient Error"))
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
@@ -137,16 +137,19 @@ func TestWorkflow_Save_SkippedDueToFinalStatue(t *testing.T) {
 	pipelineFake.SetError(util.NewCustomError(fmt.Errorf("Error"), util.CUSTOM_CODE_PERMANENT,
 		"My Permanent Error"))
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	currentTime := metav1.Now()
+
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",
 			Labels:    map[string]string{util.LabelKeyWorkflowPersistedFinalState: "true"},
 		},
-		Status: workflowapi.WorkflowStatus{
-			FinishedAt: metav1.Now(),
-		},
-	})
+		Status: workflowapi.PipelineRunStatus{
+			PipelineRunStatusFields: workflowapi.PipelineRunStatusFields{
+				CompletionTime: &currentTime,
+			},
+		}})
 
 	workflowFake.Put("MY_NAMESPACE", "MY_NAME", workflow)
 
@@ -166,17 +169,21 @@ func TestWorkflow_Save_FinalStatueNotSkippedDueToExceedTTL(t *testing.T) {
 	pipelineFake.SetError(util.NewCustomError(fmt.Errorf("Error"), util.CUSTOM_CODE_PERMANENT,
 		"My Permanent Error"))
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	currentTime := metav1.Now()
+
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		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",
 			},
 		},
-		Status: workflowapi.WorkflowStatus{
-			FinishedAt: metav1.Now(),
+		Status: workflowapi.PipelineRunStatus{
+			PipelineRunStatusFields: workflowapi.PipelineRunStatusFields{
+				CompletionTime: &currentTime,
+			},
 		},
 	})
 
@@ -202,7 +209,7 @@ func TestWorkflow_Save_SkippedDDueToMissingRunID(t *testing.T) {
 	pipelineFake.SetError(util.NewCustomError(fmt.Errorf("Error"), util.CUSTOM_CODE_PERMANENT,
 		"My Permanent Error"))
 
-	workflow := util.NewWorkflow(&workflowapi.Workflow{
+	workflow := util.NewWorkflow(&workflowapi.PipelineRun{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: "MY_NAMESPACE",
 			Name:      "MY_NAME",