github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-09-14 13:45:06 +00:00
Code Issues Projects Releases Wiki Activity

use kube clientset to access k8s clusters

Browse Source
This commit is contained in:
jianhuiz
2016-08-19 15:00:15 -07:00
parent 982640f73e
commit 65cb176572
4 changed files with 33 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
federation/pkg/federation-controller/util/federated_informer.go
View File

@@ -27,6 +27,7 @@ import (
api "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache"
kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
"k8s.io/kubernetes/pkg/client/restclient"
"k8s.io/kubernetes/pkg/controller/framework"
pkg_runtime "k8s.io/kubernetes/pkg/runtime"
@@ -72,7 +73,7 @@ type FederatedReadOnlyStore interface {
// An interface to access federation members and clients.
type FederationView interface {
// GetClientsetForCluster returns a clientset for the cluster, if present.
GetClientsetForCluster(clusterName string) (federation_release_1_4.Interface, error)
GetClientsetForCluster(clusterName string) (kube_release_1_4.Interface, error)
// GetReadyClusers returns all clusters for which the sub-informers are run.
GetReadyClusters() ([]*federation_api.Cluster, error)
@@ -106,12 +107,12 @@ type FederatedInformer interface {
type FederatedInformerForTestOnly interface {
FederatedInformer
SetClientFactory(func(*federation_api.Cluster) (federation_release_1_4.Interface, error))
SetClientFactory(func(*federation_api.Cluster) (kube_release_1_4.Interface, error))
}
// A function that should be used to create an informer on the target object. Store should use
// framework.DeletionHandlingMetaNamespaceKeyFunc as a keying function.
type TargetInformerFactory func(*federation_api.Cluster, federation_release_1_4.Interface) (cache.Store, framework.ControllerInterface)
type TargetInformerFactory func(*federation_api.Cluster, kube_release_1_4.Interface) (cache.Store, framework.ControllerInterface)
// A structure with cluster lifecycle handler functions. Cluster is available (and ClusterAvailable is fired)
// when it is created in federated etcd and ready. Cluster becomes unavailable (and ClusterUnavailable is fired)
@@ -133,10 +134,10 @@ func NewFederatedInformer(
federatedInformer := &federatedInformerImpl{
targetInformerFactory: targetInformerFactory,
clientFactory: func(cluster *federation_api.Cluster) (federation_release_1_4.Interface, error) {
clientFactory: func(cluster *federation_api.Cluster) (kube_release_1_4.Interface, error) {
clusterConfig, err := BuildClusterConfig(cluster)
if err == nil && clusterConfig != nil {
clientset := federation_release_1_4.NewForConfigOrDie(restclient.AddUserAgent(clusterConfig, userAgentName))
clientset := kube_release_1_4.NewForConfigOrDie(restclient.AddUserAgent(clusterConfig, userAgentName))
return clientset, nil
}
return nil, err
@@ -249,7 +250,7 @@ type federatedInformerImpl struct {
targetInformers map[string]informer
// A function to build clients.
clientFactory func(*federation_api.Cluster) (federation_release_1_4.Interface, error)
clientFactory func(*federation_api.Cluster) (kube_release_1_4.Interface, error)
}
type federatedStoreImpl struct {
@@ -274,7 +275,7 @@ func (f *federatedInformerImpl) Start() {
go f.clusterInformer.controller.Run(f.clusterInformer.stopChan)
}
func (f *federatedInformerImpl) SetClientFactory(clientFactory func(*federation_api.Cluster) (federation_release_1_4.Interface, error)) {
func (f *federatedInformerImpl) SetClientFactory(clientFactory func(*federation_api.Cluster) (kube_release_1_4.Interface, error)) {
f.Lock()
defer f.Unlock()
@@ -282,13 +283,13 @@ func (f *federatedInformerImpl) SetClientFactory(clientFactory func(*federation_
}
// GetClientsetForCluster returns a clientset for the cluster, if present.
func (f *federatedInformerImpl) GetClientsetForCluster(clusterName string) (federation_release_1_4.Interface, error) {
func (f *federatedInformerImpl) GetClientsetForCluster(clusterName string) (kube_release_1_4.Interface, error) {
f.Lock()
defer f.Unlock()
return f.getClientsetForClusterUnlocked(clusterName)
}
func (f *federatedInformerImpl) getClientsetForClusterUnlocked(clusterName string) (federation_release_1_4.Interface, error) {
func (f *federatedInformerImpl) getClientsetForClusterUnlocked(clusterName string) (kube_release_1_4.Interface, error) {
// No locking needed. Will happen in f.GetCluster.
if cluster, found, err := f.getReadyClusterUnlocked(clusterName); found && err == nil {
return f.clientFactory(cluster)

22
federation/pkg/federation-controller/util/federated_informer_test.go
View File

@@ -21,11 +21,12 @@ import (
"time"
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1"
federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4"
fake_federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4/fake"
api "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache"
kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
fake_kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4/fake"
"k8s.io/kubernetes/pkg/client/testing/core"
"k8s.io/kubernetes/pkg/controller/framework"
"k8s.io/kubernetes/pkg/runtime"
@@ -37,7 +38,7 @@ import (
// Basic test for Federated Informer. Checks whether the subinformer are added and deleted
// when the corresponding cluster entries appear and dissapear from etcd.
func TestFederatedInformer(t *testing.T) {
fakeClient := &fake_federation_release_1_4.Clientset{}
fakeFederationClient := &fake_federation_release_1_4.Clientset{}
// Add a single cluster to federation and remove it when needed.
cluster := federation_api.Cluster{
@@ -50,11 +51,11 @@ func TestFederatedInformer(t *testing.T) {
},
},
}
fakeClient.AddReactor("list", "clusters", func(action core.Action) (bool, runtime.Object, error) {
fakeFederationClient.AddReactor("list", "clusters", func(action core.Action) (bool, runtime.Object, error) {
return true, &federation_api.ClusterList{Items: []federation_api.Cluster{cluster}}, nil
})
deleteChan := make(chan struct{})
fakeClient.AddWatchReactor("clusters", func(action core.Action) (bool, watch.Interface, error) {
fakeFederationClient.AddWatchReactor("clusters", func(action core.Action) (bool, watch.Interface, error) {
fakeWatch := watch.NewFake()
go func() {
<-deleteChan
@@ -63,6 +64,7 @@ func TestFederatedInformer(t *testing.T) {
return true, fakeWatch, nil
})
fakeKubeClient := &fake_kube_release_1_4.Clientset{}
// There is a single service ns1/s1 in cluster mycluster.
service := api_v1.Service{
ObjectMeta: api_v1.ObjectMeta{
@@ -70,14 +72,14 @@ func TestFederatedInformer(t *testing.T) {
Name: "s1",
},
}
fakeClient.AddReactor("list", "services", func(action core.Action) (bool, runtime.Object, error) {
fakeKubeClient.AddReactor("list", "services", func(action core.Action) (bool, runtime.Object, error) {
return true, &api_v1.ServiceList{Items: []api_v1.Service{service}}, nil
})
fakeClient.AddWatchReactor("services", func(action core.Action) (bool, watch.Interface, error) {
fakeKubeClient.AddWatchReactor("services", func(action core.Action) (bool, watch.Interface, error) {
return true, watch.NewFake(), nil
})
targetInformerFactory := func(cluster *federation_api.Cluster, clientset federation_release_1_4.Interface) (cache.Store, framework.ControllerInterface) {
targetInformerFactory := func(cluster *federation_api.Cluster, clientset kube_release_1_4.Interface) (cache.Store, framework.ControllerInterface) {
return framework.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
@@ -105,9 +107,9 @@ func TestFederatedInformer(t *testing.T) {
},
}
informer := NewFederatedInformer(fakeClient, targetInformerFactory, &lifecycle).(*federatedInformerImpl)
informer.clientFactory = func(cluster *federation_api.Cluster) (federation_release_1_4.Interface, error) {
return fakeClient, nil
informer := NewFederatedInformer(fakeFederationClient, targetInformerFactory, &lifecycle).(*federatedInformerImpl)
informer.clientFactory = func(cluster *federation_api.Cluster) (kube_release_1_4.Interface, error) {
return fakeKubeClient, nil
}
assert.NotNil(t, informer)
informer.Start()

4
federation/pkg/federation-controller/util/federated_updater.go
View File

@@ -20,7 +20,7 @@ import (
"fmt"
"time"
federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4"
kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
pkg_runtime "k8s.io/kubernetes/pkg/runtime"
)
@@ -50,7 +50,7 @@ type FederatedUpdater interface {
}
// A function that executes some operation using the passed client and object.
type FederatedOperationHandler func(federation_release_1_4.Interface, pkg_runtime.Object) error
type FederatedOperationHandler func(kube_release_1_4.Interface, pkg_runtime.Object) error
type federatedUpdaterImpl struct {
federation FederationView

18
federation/pkg/federation-controller/util/federated_updater_test.go
View File

@@ -22,9 +22,9 @@ import (
"time"
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1"
federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4"
fake_federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4/fake"
api_v1 "k8s.io/kubernetes/pkg/api/v1"
kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
fake_kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4/fake"
pkg_runtime "k8s.io/kubernetes/pkg/runtime"
"github.com/stretchr/testify/assert"
@@ -34,8 +34,8 @@ import (
type fakeFederationView struct {
}
func (f fakeFederationView) GetClientsetForCluster(clusterName string) (federation_release_1_4.Interface, error) {
return &fake_federation_release_1_4.Clientset{}, nil
func (f fakeFederationView) GetClientsetForCluster(clusterName string) (kube_release_1_4.Interface, error) {
return &fake_kube_release_1_4.Clientset{}, nil
}
func (f *fakeFederationView) GetReadyClusters() ([]*federation_api.Cluster, error) {
@@ -55,12 +55,12 @@ func TestFederatedUpdaterOK(t *testing.T) {
updateChan := make(chan string, 5)
updater := NewFederatedUpdater(&fakeFederationView{},
func(_ federation_release_1_4.Interface, obj pkg_runtime.Object) error {
func(_ kube_release_1_4.Interface, obj pkg_runtime.Object) error {
service := obj.(*api_v1.Service)
addChan <- service.Name
return nil
},
func(_ federation_release_1_4.Interface, obj pkg_runtime.Object) error {
func(_ kube_release_1_4.Interface, obj pkg_runtime.Object) error {
service := obj.(*api_v1.Service)
updateChan <- service.Name
return nil
@@ -86,7 +86,7 @@ func TestFederatedUpdaterOK(t *testing.T) {
func TestFederatedUpdaterError(t *testing.T) {
updater := NewFederatedUpdater(&fakeFederationView{},
func(_ federation_release_1_4.Interface, obj pkg_runtime.Object) error {
func(_ kube_release_1_4.Interface, obj pkg_runtime.Object) error {
return fmt.Errorf("boom")
}, noop, noop)
@@ -106,7 +106,7 @@ func TestFederatedUpdaterError(t *testing.T) {
func TestFederatedUpdaterTimeout(t *testing.T) {
start := time.Now()
updater := NewFederatedUpdater(&fakeFederationView{},
func(_ federation_release_1_4.Interface, obj pkg_runtime.Object) error {
func(_ kube_release_1_4.Interface, obj pkg_runtime.Object) error {
time.Sleep(time.Minute)
return nil
},
@@ -136,6 +136,6 @@ func makeService(cluster, name string) *api_v1.Service {
}
}
func noop(_ federation_release_1_4.Interface, _ pkg_runtime.Object) error {
func noop(_ kube_release_1_4.Interface, _ pkg_runtime.Object) error {
return nil
}