Separate run resources in namespaces

backend/src/apiserver/client/BUILD.bazel

@@ -7,18 +7,22 @@ go_library(
         "pod.go",
         "scheduled_workflow.go",
         "sql.go",
-        "workflow.go",
+        "argo.go",
+        "argo_fake.go",
+        "workflow_fake.go",
         "kfam.go",
         "kfam_fake.go",
     ],
     importpath = "github.com/kubeflow/pipelines/backend/src/apiserver/client",
     visibility = ["//visibility:public"],
     deps = [
         "//backend/src/common/util:go_default_library",
+        "//backend/src/apiserver/common:go_default_library",
         "//backend/src/crd/pkg/client/clientset/versioned:go_default_library",
         "//backend/src/crd/pkg/client/clientset/versioned/typed/scheduledworkflow/v1beta1:go_default_library",
         "@com_github_argoproj_argo//pkg/client/clientset/versioned:go_default_library",
         "@com_github_argoproj_argo//pkg/client/clientset/versioned/typed/workflow/v1alpha1:go_default_library",
+        "@com_github_argoproj_argo//pkg/apis/workflow/v1alpha1:go_default_library",
         "@com_github_cenkalti_backoff//:go_default_library",
         "@com_github_go_sql_driver_mysql//:go_default_library",
         "@com_github_golang_glog//:go_default_library",
@@ -27,6 +31,9 @@ go_library(
         "@io_k8s_client_go//kubernetes:go_default_library",
         "@io_k8s_client_go//kubernetes/typed/core/v1:go_default_library",
         "@io_k8s_client_go//rest:go_default_library",
+        "@io_k8s_apimachinery//pkg/apis/meta/v1:go_default_library",
+        "@io_k8s_apimachinery//pkg/types:go_default_library",
+        "@io_k8s_apimachinery//pkg/watch:go_default_library",
     ],
 )
 

backend/src/apiserver/client/argo.go

@@ -0,0 +1,67 @@
+// Copyright 2018 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+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"
+	"github.com/cenkalti/backoff"
+	"github.com/golang/glog"
+	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
+	"github.com/pkg/errors"
+	"k8s.io/client-go/rest"
+)
+
+const (
+	PodNamespace = "POD_NAMESPACE"
+)
+
+type ArgoClientInterface interface {
+	Workflow(namespace string) argoprojv1alpha1.WorkflowInterface
+}
+
+type ArgoClient struct {
+	argoProjClient argoprojv1alpha1.ArgoprojV1alpha1Interface
+}
+
+func (argoClient *ArgoClient) Workflow(namespace string) argoprojv1alpha1.WorkflowInterface {
+	if len(namespace) == 0 {
+		namespace = common.GetStringConfig(PodNamespace)
+	}
+	return argoClient.argoProjClient.Workflows(namespace)
+}
+
+func NewArgoClientOrFatal(initConnectionTimeout time.Duration) *ArgoClient {
+	var argoProjClient argoprojv1alpha1.ArgoprojV1alpha1Interface
+	var operation = func() error {
+		restConfig, err := rest.InClusterConfig()
+		if err != nil {
+			return errors.Wrap(err, "Failed to initialize the RestConfig")
+		}
+		argoProjClient = argoclient.NewForConfigOrDie(restConfig).ArgoprojV1alpha1()
+		return nil
+	}
+
+	b := backoff.NewExponentialBackOff()
+	b.MaxElapsedTime = initConnectionTimeout
+	err := backoff.Retry(operation, b)
+
+	if err != nil {
+		glog.Fatalf("Failed to create ArgoClient. Error: %v", err)
+	}
+	return &ArgoClient{argoProjClient}
+}

backend/src/apiserver/client/argo_fake.go

@@ -0,0 +1,39 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package client
+
+import (
+	argoprojv1alpha1 "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
+)
+
+type FakeArgoClient struct {
+	workflowClientFake FakeWorkflowClientInterface
+}
+
+func NewFakeArgoClient() *FakeArgoClient {
+	return &FakeArgoClient{NewWorkflowClientFake()}
+}
+
+func (c *FakeArgoClient) Workflow(namespace string) argoprojv1alpha1.WorkflowInterface {
+	return c.workflowClientFake
+}
+
+func (c *FakeArgoClient) FakeWorkflow() FakeWorkflowClientInterface {
+	return c.workflowClientFake
+}
+
+func (c *FakeArgoClient) SetFakeWorkflow(fakeWorkflow FakeWorkflowClientInterface) {
+	c.workflowClientFake = fakeWorkflow
+}

backend/src/apiserver/resource/workflow_fake.go → backend/src/apiserver/client/workflow_fake.go

@@ -12,21 +12,29 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package resource
+package client
 
 import (
 	"encoding/json"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"strconv"
 
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	workflowclient "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
+type FakeWorkflowClientInterface interface {
+	workflowclient.WorkflowInterface
+	GetWorkflowCount() int
+	GetWorkflowKeys() map[string]bool
+	IsTerminated(name string) (bool, error)
+}
+
 type FakeWorkflowClient struct {
 	workflows       map[string]*v1alpha1.Workflow
 	lastGeneratedId int
@@ -129,7 +137,7 @@ func (c *FakeWorkflowClient) Patch(name string, pt types.PatchType, data []byte,
 	return nil, errors.New("Failed to patch workflow")
 }
 
-func (c *FakeWorkflowClient) isTerminated(name string) (bool, error) {
+func (c *FakeWorkflowClient) IsTerminated(name string) (bool, error) {
 	workflow, ok := c.workflows[name]
 	if !ok {
 		return false, errors.New("No workflow found with name: " + name)

backend/src/apiserver/client_manager.go

@@ -17,12 +17,12 @@ package main
 import (
 	"database/sql"
 	"fmt"
-	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
-	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"os"
 	"time"
 
-	workflowclient "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
+	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+
 	"github.com/cenkalti/backoff"
 	"github.com/golang/glog"
 
@@ -45,13 +45,12 @@ const (
 	mysqlPassword          = "DBConfig.Password"
 	mysqlDBName            = "DBConfig.DBName"
 	mysqlGroupConcatMaxLen = "DBConfig.GroupConcatMaxLen"
-	kfamServiceHost        = "KFAM_SERVICE_HOST"
-	kfamServicePort        = "KFAM_SERVICE_PORT"
+	kfamServiceHost        = "PROFILES_KFAM_SERVICE_HOST"
+	kfamServicePort        = "PROFILES_KFAM_SERVICE_PORT"
 
 	visualizationServiceHost = "ML_PIPELINE_VISUALIZATIONSERVER_SERVICE_HOST"
 	visualizationServicePort = "ML_PIPELINE_VISUALIZATIONSERVER_SERVICE_PORT"
 
-	podNamespace          = "POD_NAMESPACE"
 	initConnectionTimeout = "InitConnectionTimeout"
 )
 
@@ -66,7 +65,7 @@ type ClientManager struct {
 	dBStatusStore          storage.DBStatusStoreInterface
 	defaultExperimentStore storage.DefaultExperimentStoreInterface
 	objectStore            storage.ObjectStoreInterface
-	wfClient               workflowclient.WorkflowInterface
+	argoClient             client.ArgoClientInterface
 	swfClient              scheduledworkflowclient.ScheduledWorkflowInterface
 	podClient              v1.PodInterface
 	kfamClient             client.KFAMClientInterface
@@ -106,8 +105,8 @@ func (c *ClientManager) ObjectStore() storage.ObjectStoreInterface {
 	return c.objectStore
 }
 
-func (c *ClientManager) Workflow() workflowclient.WorkflowInterface {
-	return c.wfClient
+func (c *ClientManager) ArgoClient() client.ArgoClientInterface {
+	return c.argoClient
 }
 
 func (c *ClientManager) ScheduledWorkflow() scheduledworkflowclient.ScheduledWorkflowInterface {
@@ -149,14 +148,13 @@ func (c *ClientManager) init() {
 	c.defaultExperimentStore = storage.NewDefaultExperimentStore(db)
 	c.objectStore = initMinioClient(common.GetDurationConfig(initConnectionTimeout))
 
-	c.wfClient = client.CreateWorkflowClientOrFatal(
-		common.GetStringConfig(podNamespace), common.GetDurationConfig(initConnectionTimeout))
+	c.argoClient = client.NewArgoClientOrFatal(common.GetDurationConfig(initConnectionTimeout))
 
 	c.swfClient = client.CreateScheduledWorkflowClientOrFatal(
-		common.GetStringConfig(podNamespace), common.GetDurationConfig(initConnectionTimeout))
+		common.GetStringConfig(client.PodNamespace), common.GetDurationConfig(initConnectionTimeout))
 
 	c.podClient = client.CreatePodClientOrFatal(
-		common.GetStringConfig(podNamespace), common.GetDurationConfig(initConnectionTimeout))
+		common.GetStringConfig(client.PodNamespace), common.GetDurationConfig(initConnectionTimeout))
 
 	runStore := storage.NewRunStore(db, c.time)
 	c.runStore = runStore

backend/src/apiserver/common/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
@@ -8,6 +8,7 @@ go_library(
         "filter_context.go",
         "pagination_context.go",
         "paths.go",
+        "util.go",
     ],
     importpath = "github.com/kubeflow/pipelines/backend/src/apiserver/common",
     visibility = ["//visibility:public"],
@@ -18,3 +19,15 @@ go_library(
         "@com_github_spf13_viper//:go_default_library",
     ],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "util_test.go",
+    ],
+    embed = [":go_default_library"],
+    deps = [
+        "//backend/api:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+    ],
+)

backend/src/apiserver/common/util.go

@@ -0,0 +1,30 @@
+// Copyright 2019 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package common
+
+import (
+	api "github.com/kubeflow/pipelines/backend/api/go_client"
+)
+
+func GetNamespaceFromResourceReferences(resourceRefs []*api.ResourceReference) string {
+	namespace := ""
+	for _, resourceRef := range resourceRefs {
+		if resourceRef.Key.Type == api.ResourceType_NAMESPACE {
+			namespace = resourceRef.Key.Id
+			break
+		}
+	}
+	return namespace
+}