WIP namespace isolation for artifacts and pipeline definitions

backend/api/run.proto

@@ -353,6 +353,10 @@ message ReadArtifactRequest {
   string node_id = 2;
   // The name of the artifact.
   string artifact_name = 3;
+  // What resource reference to filter on.
+  // For Experiment, the only valid resource type is Namespace. An sample query string could be
+  // resource_reference_key.type=NAMESPACE&resource_reference_key.id=ns1
+  ResourceKey resource_reference_key = 4;
 }
 
 message ReadArtifactResponse {

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

@@ -54,10 +54,11 @@ func (r MetricsReporter) ReportMetrics(workflow *util.Workflow) error {
 		return nil
 	}
 	runID := workflow.ObjectMeta.Labels[util.LabelKeyWorkflowRunId]
+	namespace := workflow.ObjectMeta.Namespace
 	runMetrics := []*api.RunMetric{}
 	partialFailures := []error{}
 	for _, nodeStatus := range workflow.Status.Nodes {
-		nodeMetrics, err := r.collectNodeMetricsOrNil(runID, nodeStatus)
+		nodeMetrics, err := r.collectNodeMetricsOrNil(namespace, runID, nodeStatus)
 		if err != nil {
 			partialFailures = append(partialFailures, err)
 			continue
@@ -88,13 +89,11 @@ func (r MetricsReporter) ReportMetrics(workflow *util.Workflow) error {
 	return aggregateErrors(partialFailures)
 }
 
-func (r MetricsReporter) collectNodeMetricsOrNil(
-	runID string, nodeStatus workflowapi.NodeStatus) (
-	[]*api.RunMetric, error) {
+func (r MetricsReporter) collectNodeMetricsOrNil(namespace string, runID string, nodeStatus workflowapi.NodeStatus) ([]*api.RunMetric, error) {
 	if !nodeStatus.Completed() {
 		return nil, nil
 	}
-	metricsJSON, err := r.readNodeMetricsJSONOrEmpty(runID, nodeStatus)
+	metricsJSON, err := r.readNodeMetricsJSONOrEmpty(namespace, runID, nodeStatus)
 	if err != nil || metricsJSON == "" {
 		return nil, err
 	}
@@ -126,7 +125,7 @@ func (r MetricsReporter) collectNodeMetricsOrNil(
 	return reportMetricsRequest.GetMetrics(), nil
 }
 
-func (r MetricsReporter) readNodeMetricsJSONOrEmpty(runID string, nodeStatus workflowapi.NodeStatus) (string, error) {
+func (r MetricsReporter) readNodeMetricsJSONOrEmpty(namespace string, runID string, nodeStatus workflowapi.NodeStatus) (string, error) {
 	if nodeStatus.Outputs == nil || nodeStatus.Outputs.Artifacts == nil {
 		return "", nil // No output artifacts, skip the reporting
 	}
@@ -145,6 +144,10 @@ func (r MetricsReporter) readNodeMetricsJSONOrEmpty(runID string, nodeStatus wor
 		RunId:        runID,
 		NodeId:       nodeStatus.ID,
 		ArtifactName: metricsArtifactName,
+		ResourceReferenceKey: &api.ResourceKey{
+			Id:   namespace,
+			Type: api.ResourceType_NAMESPACE,
+		},
 	}
 	artifactResponse, err := r.pipelineClient.ReadArtifact(artifactRequest)
 	if err != nil {

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

@@ -17,10 +17,11 @@ package worker
 import (
 	"encoding/json"
 	"fmt"
+	"testing"
+
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/protobuf/testing/protocmp"
-	"testing"
 
 	workflowapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	api "github.com/kubeflow/pipelines/backend/api/go_client"
@@ -168,6 +169,10 @@ func TestReportMetrics_Succeed(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-1",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte(artifactData),
@@ -197,7 +202,7 @@ func TestReportMetrics_Succeed(t *testing.T) {
 	got := pipelineFake.GetReportedMetricsRequest()
 	if diff := cmp.Diff(expectedMetricsRequest, got, cmpopts.EquateEmpty(), protocmp.Transform()); diff != "" {
 		t.Errorf("parseRuntimeInfo() = %+v, want %+v\nDiff (-want, +got)\n%s", got, expectedMetricsRequest, diff)
-		s, _ := json.MarshalIndent(expectedMetricsRequest ,"", "  ")
+		s, _ := json.MarshalIndent(expectedMetricsRequest, "", "  ")
 		fmt.Printf("Want %s", s)
 	}
 }
@@ -230,6 +235,10 @@ func TestReportMetrics_EmptyArchive_Fail(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-1",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte(artifactData),
@@ -273,6 +282,10 @@ func TestReportMetrics_MultipleFilesInArchive_Fail(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-1",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte(artifactData),
@@ -315,6 +328,10 @@ func TestReportMetrics_InvalidMetricsJSON_Fail(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-1",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte(artifactData),
@@ -367,6 +384,10 @@ func TestReportMetrics_InvalidMetricsJSON_PartialFail(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-1",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte(invalidArtifactData),
@@ -376,6 +397,10 @@ func TestReportMetrics_InvalidMetricsJSON_PartialFail(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-2",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte(validArtifactData),
@@ -404,7 +429,7 @@ func TestReportMetrics_InvalidMetricsJSON_PartialFail(t *testing.T) {
 	got := pipelineFake.GetReportedMetricsRequest()
 	if diff := cmp.Diff(expectedMetricsRequest, got, cmpopts.EquateEmpty(), protocmp.Transform()); diff != "" {
 		t.Errorf("parseRuntimeInfo() = %+v, want %+v\nDiff (-want, +got)\n%s", got, expectedMetricsRequest, diff)
-		s, _ := json.MarshalIndent(expectedMetricsRequest ,"", "  ")
+		s, _ := json.MarshalIndent(expectedMetricsRequest, "", "  ")
 		fmt.Printf("Want %s", s)
 	}
 }
@@ -436,6 +461,10 @@ func TestReportMetrics_CorruptedArchiveFile_Fail(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-1",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte("invalid tgz content"),
@@ -479,6 +508,10 @@ func TestReportMetrics_MultiplMetricErrors_TransientErrowWin(t *testing.T) {
 			RunId:        "run-1",
 			NodeId:       "node-1",
 			ArtifactName: "mlpipeline-metrics",
+			ResourceReferenceKey: &api.ResourceKey{
+				Type: api.ResourceType_NAMESPACE,
+				Id:   "MY_NAMESPACE",
+			},
 		},
 		&api.ReadArtifactResponse{
 			Data: []byte(artifactData),