issue-12678 - refactored client_manager to reuse KubernetesCoreClient for blob storage initialization; added GetClientSet method to KubernetesCoreInterface

Signed-off-by: Helber Belmiro <[email protected]>

Author: hbelmiro

backend/src/apiserver/client/kubernetes_core.go

@@ -6,27 +6,37 @@ import (
 	"github.com/cenkalti/backoff"
 	"github.com/golang/glog"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
+	"k8s.io/client-go/kubernetes"
 	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
 type KubernetesCoreInterface interface {
 	PodClient(namespace string) v1.PodInterface
+	GetClientSet() kubernetes.Interface
 }
 
 type KubernetesCore struct {
 	coreV1Client v1.CoreV1Interface
+	clientSet    kubernetes.Interface
 }
 
 func (c *KubernetesCore) PodClient(namespace string) v1.PodInterface {
 	return c.coreV1Client.Pods(namespace)
 }
 
+func (c *KubernetesCore) GetClientSet() kubernetes.Interface {
+	return c.clientSet
+}
+
 func createKubernetesCore(clientParams util.ClientParameters) (KubernetesCoreInterface, error) {
 	clientSet, err := getKubernetesClientset(clientParams)
 	if err != nil {
 		return nil, err
 	}
-	return &KubernetesCore{clientSet.CoreV1()}, nil
+	return &KubernetesCore{
+		coreV1Client: clientSet.CoreV1(),
+		clientSet:    clientSet,
+	}, nil
 }
 
 // CreateKubernetesCoreOrFatal creates a new client for the Kubernetes pod.

backend/src/apiserver/client/kubernetes_core_fake.go

@@ -20,6 +20,7 @@ import (
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	policyv1 "k8s.io/api/policy/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
+	"k8s.io/client-go/kubernetes"
 	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
@@ -34,6 +35,11 @@ func (c *FakeKuberneteCoreClient) PodClient(namespace string) v1.PodInterface {
 	return c.podClientFake
 }
 
+func (c *FakeKuberneteCoreClient) GetClientSet() kubernetes.Interface {
+	// Return nil for fake implementation - tests that need this should use a mock
+	return nil
+}
+
 func NewFakeKuberneteCoresClient() *FakeKuberneteCoreClient {
 	return &FakeKuberneteCoreClient{&FakePodClient{}}
 }
@@ -50,6 +56,11 @@ func (c *FakeKubernetesCoreClientWithBadPodClient) PodClient(namespace string) v
 	return c.podClientFake
 }
 
+func (c *FakeKubernetesCoreClientWithBadPodClient) GetClientSet() kubernetes.Interface {
+	// Return nil for fake implementation
+	return nil
+}
+
 func (c *FakePodClient) EvictV1(context.Context, *policyv1.Eviction) error {
 	return nil
 }

backend/src/apiserver/client_manager/client_manager.go

@@ -301,19 +301,8 @@ func (c *ClientManager) init(options *Options) error {
 
 	c.k8sCoreClient = client.CreateKubernetesCoreOrFatal(common.GetDurationConfig(initConnectionTimeout), clientParams)
 
-	restConfig, err := util.GetKubernetesConfig()
-	if err != nil {
-		return err
-	}
-	restConfig.QPS = float32(clientParams.QPS)
-	restConfig.Burst = clientParams.Burst
-	k8sClient, err := kubernetes.NewForConfig(restConfig)
-	if err != nil {
-		return util.Wrap(err, "Failed to create Kubernetes client for blob storage")
-	}
-
 	glog.Info("Initializing Object store client...")
-	c.objectStore = initBlobObjectStore(options.Context, common.GetDurationConfig(initConnectionTimeout), k8sClient)
+	c.objectStore = initBlobObjectStore(options.Context, common.GetDurationConfig(initConnectionTimeout), c.k8sCoreClient.GetClientSet())
 	glog.Info("Object store client initialized successfully")
 
 	runStore := storage.NewRunStore(db, c.time)