Skip to content

K
k3s
Commit eaa2b6f5 authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #31042 from jianhuiz/federation-informer-kube-client

Automatic merge from submit-queue Federation informer use kube clientset for target This is to use kubernetes clientset as the federation informer target clientset as it's used to talking the k8s cluster. #29939 #30669 #30207 @mwielgus @quinton-hoole @kshafiee @deepak-vij
parents c6bc6546 47f317ee
Hide whitespace changes
Inline Side-by-side
Showing with 87 additions and 80 deletions
federation/pkg/federation-controller/namespace/namespace_controller.go
View file @ eaa2b6f5
...@@ -26,6 +26,7 @@ import ( ...@@ -26,6 +26,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1" api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
"k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller"
"k8s.io/kubernetes/pkg/controller/framework" "k8s.io/kubernetes/pkg/controller/framework"
pkg_runtime "k8s.io/kubernetes/pkg/runtime" pkg_runtime "k8s.io/kubernetes/pkg/runtime"
...@@ -102,7 +103,7 @@ func NewNamespaceController(client federation_release_1_4.Interface) *NamespaceC ...@@ -102,7 +103,7 @@ func NewNamespaceController(client federation_release_1_4.Interface) *NamespaceC
// Federated informer on namespaces in members of federation. // Federated informer on namespaces in members of federation.
nc.namespaceFederatedInformer = util.NewFederatedInformer( nc.namespaceFederatedInformer = util.NewFederatedInformer(
client, client,
func(cluster *federation_api.Cluster, targetClient federation_release_1_4.Interface) (cache.Store, framework.ControllerInterface) { func(cluster *federation_api.Cluster, targetClient kube_release_1_4.Interface) (cache.Store, framework.ControllerInterface) {
return framework.NewInformer( return framework.NewInformer(
&cache.ListWatch{ &cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) { ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
...@@ -131,17 +132,17 @@ func NewNamespaceController(client federation_release_1_4.Interface) *NamespaceC ...@@ -131,17 +132,17 @@ func NewNamespaceController(client federation_release_1_4.Interface) *NamespaceC
// Federated updeater along with Create/Update/Delete operations. // Federated updeater along with Create/Update/Delete operations.
nc.federatedUpdater = util.NewFederatedUpdater(nc.namespaceFederatedInformer, nc.federatedUpdater = util.NewFederatedUpdater(nc.namespaceFederatedInformer,
func(client federation_release_1_4.Interface, obj pkg_runtime.Object) error { func(client kube_release_1_4.Interface, obj pkg_runtime.Object) error {
namespace := obj.(*api_v1.Namespace) namespace := obj.(*api_v1.Namespace)
_, err := client.Core().Namespaces().Create(namespace) _, err := client.Core().Namespaces().Create(namespace)
return err return err
}, },
func(client federation_release_1_4.Interface, obj pkg_runtime.Object) error { func(client kube_release_1_4.Interface, obj pkg_runtime.Object) error {
namespace := obj.(*api_v1.Namespace) namespace := obj.(*api_v1.Namespace)
_, err := client.Core().Namespaces().Update(namespace) _, err := client.Core().Namespaces().Update(namespace)
return err return err
}, },
func(client federation_release_1_4.Interface, obj pkg_runtime.Object) error { func(client kube_release_1_4.Interface, obj pkg_runtime.Object) error {
namespace := obj.(*api_v1.Namespace) namespace := obj.(*api_v1.Namespace)
err := client.Core().Namespaces().Delete(namespace.Name, &api.DeleteOptions{}) err := client.Core().Namespaces().Delete(namespace.Name, &api.DeleteOptions{})
return err return err
......
federation/pkg/federation-controller/namespace/namespace_controller_test.go
View file @ eaa2b6f5
...@@ -22,10 +22,11 @@ import ( ...@@ -22,10 +22,11 @@ import (
"time" "time"
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1" 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" fake_federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4/fake"
"k8s.io/kubernetes/federation/pkg/federation-controller/util" "k8s.io/kubernetes/federation/pkg/federation-controller/util"
api_v1 "k8s.io/kubernetes/pkg/api/v1" 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"
"k8s.io/kubernetes/pkg/client/testing/core" "k8s.io/kubernetes/pkg/client/testing/core"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/watch" "k8s.io/kubernetes/pkg/watch"
...@@ -38,25 +39,25 @@ func TestNamespaceController(t *testing.T) { ...@@ -38,25 +39,25 @@ func TestNamespaceController(t *testing.T) {
cluster2 := mkCluster("cluster2", api_v1.ConditionTrue) cluster2 := mkCluster("cluster2", api_v1.ConditionTrue)
fakeClient := &fake_federation_release_1_4.Clientset{} fakeClient := &fake_federation_release_1_4.Clientset{}
RegisterList("clusters", fakeClient, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}}) RegisterList("clusters", &fakeClient.Fake, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}})
RegisterList("namespaces", fakeClient, &api_v1.NamespaceList{Items: []api_v1.Namespace{}}) RegisterList("namespaces", &fakeClient.Fake, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
namespaceWatch := RegisterWatch("namespaces", fakeClient) namespaceWatch := RegisterWatch("namespaces", &fakeClient.Fake)
clusterWatch := RegisterWatch("clusters", fakeClient) clusterWatch := RegisterWatch("clusters", &fakeClient.Fake)
cluster1Client := &fake_federation_release_1_4.Clientset{} cluster1Client := &fake_kube_release_1_4.Clientset{}
cluster1Watch := RegisterWatch("namespaces", cluster1Client) cluster1Watch := RegisterWatch("namespaces", &cluster1Client.Fake)
RegisterList("namespaces", cluster1Client, &api_v1.NamespaceList{Items: []api_v1.Namespace{}}) RegisterList("namespaces", &cluster1Client.Fake, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
cluster1CreateChan := RegisterCopyOnCreate("namespaces", cluster1Client, cluster1Watch) cluster1CreateChan := RegisterCopyOnCreate("namespaces", &cluster1Client.Fake, cluster1Watch)
cluster1UpdateChan := RegisterCopyOnUpdate("namespaces", cluster1Client, cluster1Watch) cluster1UpdateChan := RegisterCopyOnUpdate("namespaces", &cluster1Client.Fake, cluster1Watch)
cluster2Client := &fake_federation_release_1_4.Clientset{} cluster2Client := &fake_kube_release_1_4.Clientset{}
cluster2Watch := RegisterWatch("namespaces", cluster2Client) cluster2Watch := RegisterWatch("namespaces", &cluster2Client.Fake)
RegisterList("namespaces", cluster2Client, &api_v1.NamespaceList{Items: []api_v1.Namespace{}}) RegisterList("namespaces", &cluster2Client.Fake, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
cluster2CreateChan := RegisterCopyOnCreate("namespaces", cluster2Client, cluster2Watch) cluster2CreateChan := RegisterCopyOnCreate("namespaces", &cluster2Client.Fake, cluster2Watch)
namespaceController := NewNamespaceController(fakeClient) namespaceController := NewNamespaceController(fakeClient)
informer := toFederatedInformerForTestOnly(namespaceController.namespaceFederatedInformer) informer := toFederatedInformerForTestOnly(namespaceController.namespaceFederatedInformer)
informer.SetClientFactory(func(cluster *federation_api.Cluster) (federation_release_1_4.Interface, error) { informer.SetClientFactory(func(cluster *federation_api.Cluster) (kube_release_1_4.Interface, error) {
switch cluster.Name { switch cluster.Name {
case cluster1.Name: case cluster1.Name:
return cluster1Client, nil return cluster1Client, nil
...@@ -124,19 +125,19 @@ func mkCluster(name string, readyStatus api_v1.ConditionStatus) *federation_api. ...@@ -124,19 +125,19 @@ func mkCluster(name string, readyStatus api_v1.ConditionStatus) *federation_api.
} }
} }
func RegisterWatch(resource string, client *fake_federation_release_1_4.Clientset) *watch.FakeWatcher { func RegisterWatch(resource string, client *core.Fake) *watch.FakeWatcher {
watcher := watch.NewFake() watcher := watch.NewFake()
client.AddWatchReactor(resource, func(action core.Action) (bool, watch.Interface, error) { return true, watcher, nil }) client.AddWatchReactor(resource, func(action core.Action) (bool, watch.Interface, error) { return true, watcher, nil })
return watcher return watcher
} }
func RegisterList(resource string, client *fake_federation_release_1_4.Clientset, obj runtime.Object) { func RegisterList(resource string, client *core.Fake, obj runtime.Object) {
client.AddReactor("list", resource, func(action core.Action) (bool, runtime.Object, error) { client.AddReactor("list", resource, func(action core.Action) (bool, runtime.Object, error) {
return true, obj, nil return true, obj, nil
}) })
} }
func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.Clientset, watcher *watch.FakeWatcher) chan runtime.Object { func RegisterCopyOnCreate(resource string, client *core.Fake, watcher *watch.FakeWatcher) chan runtime.Object {
objChan := make(chan runtime.Object, 100) objChan := make(chan runtime.Object, 100)
client.AddReactor("create", resource, func(action core.Action) (bool, runtime.Object, error) { client.AddReactor("create", resource, func(action core.Action) (bool, runtime.Object, error) {
createAction := action.(core.CreateAction) createAction := action.(core.CreateAction)
...@@ -150,7 +151,7 @@ func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.C ...@@ -150,7 +151,7 @@ func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.C
return objChan return objChan
} }
func RegisterCopyOnUpdate(resource string, client *fake_federation_release_1_4.Clientset, watcher *watch.FakeWatcher) chan runtime.Object { func RegisterCopyOnUpdate(resource string, client *core.Fake, watcher *watch.FakeWatcher) chan runtime.Object {
objChan := make(chan runtime.Object, 100) objChan := make(chan runtime.Object, 100)
client.AddReactor("update", resource, func(action core.Action) (bool, runtime.Object, error) { client.AddReactor("update", resource, func(action core.Action) (bool, runtime.Object, error) {
updateAction := action.(core.UpdateAction) updateAction := action.(core.UpdateAction)
......
federation/pkg/federation-controller/secret/secret_controller.go
View file @ eaa2b6f5
...@@ -27,6 +27,7 @@ import ( ...@@ -27,6 +27,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1" api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
kube_release_1_4 "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4"
"k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller"
"k8s.io/kubernetes/pkg/controller/framework" "k8s.io/kubernetes/pkg/controller/framework"
pkg_runtime "k8s.io/kubernetes/pkg/runtime" pkg_runtime "k8s.io/kubernetes/pkg/runtime"
...@@ -103,7 +104,7 @@ func NewSecretController(client federation_release_1_4.Interface) *SecretControl ...@@ -103,7 +104,7 @@ func NewSecretController(client federation_release_1_4.Interface) *SecretControl
// Federated informer on secrets in members of federation. // Federated informer on secrets in members of federation.
secretcontroller.secretFederatedInformer = util.NewFederatedInformer( secretcontroller.secretFederatedInformer = util.NewFederatedInformer(
client, client,
func(cluster *federation_api.Cluster, targetClient federation_release_1_4.Interface) (cache.Store, framework.ControllerInterface) { func(cluster *federation_api.Cluster, targetClient kube_release_1_4.Interface) (cache.Store, framework.ControllerInterface) {
return framework.NewInformer( return framework.NewInformer(
&cache.ListWatch{ &cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) { ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
...@@ -134,17 +135,17 @@ func NewSecretController(client federation_release_1_4.Interface) *SecretControl ...@@ -134,17 +135,17 @@ func NewSecretController(client federation_release_1_4.Interface) *SecretControl
// Federated updeater along with Create/Update/Delete operations. // Federated updeater along with Create/Update/Delete operations.
secretcontroller.federatedUpdater = util.NewFederatedUpdater(secretcontroller.secretFederatedInformer, secretcontroller.federatedUpdater = util.NewFederatedUpdater(secretcontroller.secretFederatedInformer,
func(client federation_release_1_4.Interface, obj pkg_runtime.Object) error { func(client kube_release_1_4.Interface, obj pkg_runtime.Object) error {
secret := obj.(*api_v1.Secret) secret := obj.(*api_v1.Secret)
_, err := client.Core().Secrets(secret.Namespace).Create(secret) _, err := client.Core().Secrets(secret.Namespace).Create(secret)
return err return err
}, },
func(client federation_release_1_4.Interface, obj pkg_runtime.Object) error { func(client kube_release_1_4.Interface, obj pkg_runtime.Object) error {
secret := obj.(*api_v1.Secret) secret := obj.(*api_v1.Secret)
_, err := client.Core().Secrets(secret.Namespace).Update(secret) _, err := client.Core().Secrets(secret.Namespace).Update(secret)
return err return err
}, },
func(client federation_release_1_4.Interface, obj pkg_runtime.Object) error { func(client kube_release_1_4.Interface, obj pkg_runtime.Object) error {
secret := obj.(*api_v1.Secret) secret := obj.(*api_v1.Secret)
err := client.Core().Secrets(secret.Namespace).Delete(secret.Name, &api.DeleteOptions{}) err := client.Core().Secrets(secret.Namespace).Delete(secret.Name, &api.DeleteOptions{})
return err return err
......
federation/pkg/federation-controller/secret/secret_controller_test.go
View file @ eaa2b6f5
...@@ -23,10 +23,11 @@ import ( ...@@ -23,10 +23,11 @@ import (
"time" "time"
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1" 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" fake_federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4/fake"
"k8s.io/kubernetes/federation/pkg/federation-controller/util" "k8s.io/kubernetes/federation/pkg/federation-controller/util"
api_v1 "k8s.io/kubernetes/pkg/api/v1" 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"
"k8s.io/kubernetes/pkg/client/testing/core" "k8s.io/kubernetes/pkg/client/testing/core"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/watch" "k8s.io/kubernetes/pkg/watch"
...@@ -39,25 +40,25 @@ func TestSecretController(t *testing.T) { ...@@ -39,25 +40,25 @@ func TestSecretController(t *testing.T) {
cluster2 := mkCluster("cluster2", api_v1.ConditionTrue) cluster2 := mkCluster("cluster2", api_v1.ConditionTrue)
fakeClient := &fake_federation_release_1_4.Clientset{} fakeClient := &fake_federation_release_1_4.Clientset{}
RegisterList("clusters", fakeClient, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}}) RegisterList("clusters", &fakeClient.Fake, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}})
RegisterList("secrets", fakeClient, &api_v1.SecretList{Items: []api_v1.Secret{}}) RegisterList("secrets", &fakeClient.Fake, &api_v1.SecretList{Items: []api_v1.Secret{}})
secretWatch := RegisterWatch("secrets", fakeClient) secretWatch := RegisterWatch("secrets", &fakeClient.Fake)
clusterWatch := RegisterWatch("clusters", fakeClient) clusterWatch := RegisterWatch("clusters", &fakeClient.Fake)
cluster1Client := &fake_federation_release_1_4.Clientset{} cluster1Client := &fake_kube_release_1_4.Clientset{}
cluster1Watch := RegisterWatch("secrets", cluster1Client) cluster1Watch := RegisterWatch("secrets", &cluster1Client.Fake)
RegisterList("secrets", cluster1Client, &api_v1.SecretList{Items: []api_v1.Secret{}}) RegisterList("secrets", &cluster1Client.Fake, &api_v1.SecretList{Items: []api_v1.Secret{}})
cluster1CreateChan := RegisterCopyOnCreate("secrets", cluster1Client, cluster1Watch) cluster1CreateChan := RegisterCopyOnCreate("secrets", &cluster1Client.Fake, cluster1Watch)
cluster1UpdateChan := RegisterCopyOnUpdate("secrets", cluster1Client, cluster1Watch) cluster1UpdateChan := RegisterCopyOnUpdate("secrets", &cluster1Client.Fake, cluster1Watch)
cluster2Client := &fake_federation_release_1_4.Clientset{} cluster2Client := &fake_kube_release_1_4.Clientset{}
cluster2Watch := RegisterWatch("secrets", cluster2Client) cluster2Watch := RegisterWatch("secrets", &cluster2Client.Fake)
RegisterList("secrets", cluster2Client, &api_v1.SecretList{Items: []api_v1.Secret{}}) RegisterList("secrets", &cluster2Client.Fake, &api_v1.SecretList{Items: []api_v1.Secret{}})
cluster2CreateChan := RegisterCopyOnCreate("secrets", cluster2Client, cluster2Watch) cluster2CreateChan := RegisterCopyOnCreate("secrets", &cluster2Client.Fake, cluster2Watch)
secretController := NewSecretController(fakeClient) secretController := NewSecretController(fakeClient)
informer := toFederatedInformerForTestOnly(secretController.secretFederatedInformer) informer := toFederatedInformerForTestOnly(secretController.secretFederatedInformer)
informer.SetClientFactory(func(cluster *federation_api.Cluster) (federation_release_1_4.Interface, error) { informer.SetClientFactory(func(cluster *federation_api.Cluster) (kube_release_1_4.Interface, error) {
switch cluster.Name { switch cluster.Name {
case cluster1.Name: case cluster1.Name:
return cluster1Client, nil return cluster1Client, nil
...@@ -136,19 +137,19 @@ func mkCluster(name string, readyStatus api_v1.ConditionStatus) *federation_api. ...@@ -136,19 +137,19 @@ func mkCluster(name string, readyStatus api_v1.ConditionStatus) *federation_api.
} }
} }
func RegisterWatch(resource string, client *fake_federation_release_1_4.Clientset) *watch.FakeWatcher { func RegisterWatch(resource string, client *core.Fake) *watch.FakeWatcher {
watcher := watch.NewFake() watcher := watch.NewFake()
client.AddWatchReactor(resource, func(action core.Action) (bool, watch.Interface, error) { return true, watcher, nil }) client.AddWatchReactor(resource, func(action core.Action) (bool, watch.Interface, error) { return true, watcher, nil })
return watcher return watcher
} }
func RegisterList(resource string, client *fake_federation_release_1_4.Clientset, obj runtime.Object) { func RegisterList(resource string, client *core.Fake, obj runtime.Object) {
client.AddReactor("list", resource, func(action core.Action) (bool, runtime.Object, error) { client.AddReactor("list", resource, func(action core.Action) (bool, runtime.Object, error) {
return true, obj, nil return true, obj, nil
}) })
} }
func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.Clientset, watcher *watch.FakeWatcher) chan runtime.Object { func RegisterCopyOnCreate(resource string, client *core.Fake, watcher *watch.FakeWatcher) chan runtime.Object {
objChan := make(chan runtime.Object, 100) objChan := make(chan runtime.Object, 100)
client.AddReactor("create", resource, func(action core.Action) (bool, runtime.Object, error) { client.AddReactor("create", resource, func(action core.Action) (bool, runtime.Object, error) {
createAction := action.(core.CreateAction) createAction := action.(core.CreateAction)
...@@ -162,7 +163,7 @@ func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.C ...@@ -162,7 +163,7 @@ func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.C
return objChan return objChan
} }
func RegisterCopyOnUpdate(resource string, client *fake_federation_release_1_4.Clientset, watcher *watch.FakeWatcher) chan runtime.Object { func RegisterCopyOnUpdate(resource string, client *core.Fake, watcher *watch.FakeWatcher) chan runtime.Object {
objChan := make(chan runtime.Object, 100) objChan := make(chan runtime.Object, 100)
client.AddReactor("update", resource, func(action core.Action) (bool, runtime.Object, error) { client.AddReactor("update", resource, func(action core.Action) (bool, runtime.Object, error) {
updateAction := action.(core.UpdateAction) updateAction := action.(core.UpdateAction)
......
federation/pkg/federation-controller/util/federated_informer.go
View file @ eaa2b6f5
...@@ -27,6 +27,7 @@ import ( ...@@ -27,6 +27,7 @@ import (
api "k8s.io/kubernetes/pkg/api" api "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1" api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache" "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/client/restclient"
"k8s.io/kubernetes/pkg/controller/framework" "k8s.io/kubernetes/pkg/controller/framework"
pkg_runtime "k8s.io/kubernetes/pkg/runtime" pkg_runtime "k8s.io/kubernetes/pkg/runtime"
...@@ -72,7 +73,7 @@ type FederatedReadOnlyStore interface { ...@@ -72,7 +73,7 @@ type FederatedReadOnlyStore interface {
// An interface to access federation members and clients. // An interface to access federation members and clients.
type FederationView interface { type FederationView interface {
// GetClientsetForCluster returns a clientset for the cluster, if present. // 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. // GetReadyClusers returns all clusters for which the sub-informers are run.
GetReadyClusters() ([]*federation_api.Cluster, error) GetReadyClusters() ([]*federation_api.Cluster, error)
...@@ -106,12 +107,12 @@ type FederatedInformer interface { ...@@ -106,12 +107,12 @@ type FederatedInformer interface {
type FederatedInformerForTestOnly interface { type FederatedInformerForTestOnly interface {
FederatedInformer 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 // A function that should be used to create an informer on the target object. Store should use
// framework.DeletionHandlingMetaNamespaceKeyFunc as a keying function. // 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) // 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) // when it is created in federated etcd and ready. Cluster becomes unavailable (and ClusterUnavailable is fired)
...@@ -133,10 +134,10 @@ func NewFederatedInformer( ...@@ -133,10 +134,10 @@ func NewFederatedInformer(
federatedInformer := &federatedInformerImpl{ federatedInformer := &federatedInformerImpl{
targetInformerFactory: targetInformerFactory, 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) clusterConfig, err := BuildClusterConfig(cluster)
if err == nil && clusterConfig != nil { 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 clientset, nil
} }
return nil, err return nil, err
...@@ -249,7 +250,7 @@ type federatedInformerImpl struct { ...@@ -249,7 +250,7 @@ type federatedInformerImpl struct {
targetInformers map[string]informer targetInformers map[string]informer
// A function to build clients. // 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 { type federatedStoreImpl struct {
...@@ -274,7 +275,7 @@ func (f *federatedInformerImpl) Start() { ...@@ -274,7 +275,7 @@ func (f *federatedInformerImpl) Start() {
go f.clusterInformer.controller.Run(f.clusterInformer.stopChan) 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() f.Lock()
defer f.Unlock() defer f.Unlock()
...@@ -282,13 +283,13 @@ func (f *federatedInformerImpl) SetClientFactory(clientFactory func(*federation_ ...@@ -282,13 +283,13 @@ func (f *federatedInformerImpl) SetClientFactory(clientFactory func(*federation_
} }
// GetClientsetForCluster returns a clientset for the cluster, if present. // 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() f.Lock()
defer f.Unlock() defer f.Unlock()
return f.getClientsetForClusterUnlocked(clusterName) 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. // No locking needed. Will happen in f.GetCluster.
if cluster, found, err := f.getReadyClusterUnlocked(clusterName); found && err == nil { if cluster, found, err := f.getReadyClusterUnlocked(clusterName); found && err == nil {
return f.clientFactory(cluster) return f.clientFactory(cluster)
......
federation/pkg/federation-controller/util/federated_informer_test.go
View file @ eaa2b6f5
...@@ -21,11 +21,12 @@ import ( ...@@ -21,11 +21,12 @@ import (
"time" "time"
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1" 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" fake_federation_release_1_4 "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4/fake"
api "k8s.io/kubernetes/pkg/api" api "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1" api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache" "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/client/testing/core"
"k8s.io/kubernetes/pkg/controller/framework" "k8s.io/kubernetes/pkg/controller/framework"
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
...@@ -37,7 +38,7 @@ import ( ...@@ -37,7 +38,7 @@ import (
// Basic test for Federated Informer. Checks whether the subinformer are added and deleted // Basic test for Federated Informer. Checks whether the subinformer are added and deleted
// when the corresponding cluster entries appear and dissapear from etcd. // when the corresponding cluster entries appear and dissapear from etcd.
func TestFederatedInformer(t *testing.T) { 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. // Add a single cluster to federation and remove it when needed.
cluster := federation_api.Cluster{ cluster := federation_api.Cluster{
...@@ -50,11 +51,11 @@ func TestFederatedInformer(t *testing.T) { ...@@ -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 return true, &federation_api.ClusterList{Items: []federation_api.Cluster{cluster}}, nil
}) })
deleteChan := make(chan struct{}) 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() fakeWatch := watch.NewFake()
go func() { go func() {
<-deleteChan <-deleteChan
...@@ -63,6 +64,7 @@ func TestFederatedInformer(t *testing.T) { ...@@ -63,6 +64,7 @@ func TestFederatedInformer(t *testing.T) {
return true, fakeWatch, nil return true, fakeWatch, nil
}) })
fakeKubeClient := &fake_kube_release_1_4.Clientset{}
// There is a single service ns1/s1 in cluster mycluster. // There is a single service ns1/s1 in cluster mycluster.
service := api_v1.Service{ service := api_v1.Service{
ObjectMeta: api_v1.ObjectMeta{ ObjectMeta: api_v1.ObjectMeta{
...@@ -70,14 +72,14 @@ func TestFederatedInformer(t *testing.T) { ...@@ -70,14 +72,14 @@ func TestFederatedInformer(t *testing.T) {
Name: "s1", 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 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 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( return framework.NewInformer(
&cache.ListWatch{ &cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) { ListFunc: func(options api.ListOptions) (runtime.Object, error) {
...@@ -105,9 +107,9 @@ func TestFederatedInformer(t *testing.T) { ...@@ -105,9 +107,9 @@ func TestFederatedInformer(t *testing.T) {
}, },
} }
informer := NewFederatedInformer(fakeClient, targetInformerFactory, &lifecycle).(*federatedInformerImpl) informer := NewFederatedInformer(fakeFederationClient, targetInformerFactory, &lifecycle).(*federatedInformerImpl)
informer.clientFactory = func(cluster *federation_api.Cluster) (federation_release_1_4.Interface, error) { informer.clientFactory = func(cluster *federation_api.Cluster) (kube_release_1_4.Interface, error) {
return fakeClient, nil return fakeKubeClient, nil
} }
assert.NotNil(t, informer) assert.NotNil(t, informer)
informer.Start() informer.Start()
......
federation/pkg/federation-controller/util/federated_updater.go
View file @ eaa2b6f5
...@@ -20,7 +20,7 @@ import ( ...@@ -20,7 +20,7 @@ import (
"fmt" "fmt"
"time" "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" pkg_runtime "k8s.io/kubernetes/pkg/runtime"
) )
...@@ -50,7 +50,7 @@ type FederatedUpdater interface { ...@@ -50,7 +50,7 @@ type FederatedUpdater interface {
} }
// A function that executes some operation using the passed client and object. // 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 { type federatedUpdaterImpl struct {
federation FederationView federation FederationView
......
federation/pkg/federation-controller/util/federated_updater_test.go
View file @ eaa2b6f5
...@@ -22,9 +22,9 @@ import ( ...@@ -22,9 +22,9 @@ import (
"time" "time"
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1" 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" 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" pkg_runtime "k8s.io/kubernetes/pkg/runtime"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
...@@ -34,8 +34,8 @@ import ( ...@@ -34,8 +34,8 @@ import (
type fakeFederationView struct { type fakeFederationView struct {
} }
func (f fakeFederationView) GetClientsetForCluster(clusterName string) (federation_release_1_4.Interface, error) { func (f fakeFederationView) GetClientsetForCluster(clusterName string) (kube_release_1_4.Interface, error) {
return &fake_federation_release_1_4.Clientset{}, nil return &fake_kube_release_1_4.Clientset{}, nil
} }
func (f *fakeFederationView) GetReadyClusters() ([]*federation_api.Cluster, error) { func (f *fakeFederationView) GetReadyClusters() ([]*federation_api.Cluster, error) {
...@@ -55,12 +55,12 @@ func TestFederatedUpdaterOK(t *testing.T) { ...@@ -55,12 +55,12 @@ func TestFederatedUpdaterOK(t *testing.T) {
updateChan := make(chan string, 5) updateChan := make(chan string, 5)
updater := NewFederatedUpdater(&fakeFederationView{}, 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) service := obj.(*api_v1.Service)
addChan <- service.Name addChan <- service.Name
return nil 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) service := obj.(*api_v1.Service)
updateChan <- service.Name updateChan <- service.Name
return nil return nil
...@@ -86,7 +86,7 @@ func TestFederatedUpdaterOK(t *testing.T) { ...@@ -86,7 +86,7 @@ func TestFederatedUpdaterOK(t *testing.T) {
func TestFederatedUpdaterError(t *testing.T) { func TestFederatedUpdaterError(t *testing.T) {
updater := NewFederatedUpdater(&fakeFederationView{}, 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") return fmt.Errorf("boom")
}, noop, noop) }, noop, noop)
...@@ -106,7 +106,7 @@ func TestFederatedUpdaterError(t *testing.T) { ...@@ -106,7 +106,7 @@ func TestFederatedUpdaterError(t *testing.T) {
func TestFederatedUpdaterTimeout(t *testing.T) { func TestFederatedUpdaterTimeout(t *testing.T) {
start := time.Now() start := time.Now()
updater := NewFederatedUpdater(&fakeFederationView{}, 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) time.Sleep(time.Minute)
return nil return nil
}, },
...@@ -136,6 +136,6 @@ func makeService(cluster, name string) *api_v1.Service { ...@@ -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 return nil
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment