Skip to content

K
k3s
Commit 47f317ee authored by jianhuiz's avatar jianhuiz
Browse files
Options

update federation controllers to use kube clientset

parent 65cb1765
Hide whitespace changes
Inline Side-by-side
Showing with 54 additions and 50 deletions
federation/pkg/federation-controller/namespace/namespace_controller.go
View file @ 47f317ee
......@@ -26,6 +26,7 @@ import (
"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/controller"
"k8s.io/kubernetes/pkg/controller/framework"
pkg_runtime "k8s.io/kubernetes/pkg/runtime"
......@@ -102,7 +103,7 @@ func NewNamespaceController(client federation_release_1_4.Interface) *NamespaceC
// Federated informer on namespaces in members of federation.
nc.namespaceFederatedInformer = util.NewFederatedInformer(
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(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
......@@ -131,17 +132,17 @@ func NewNamespaceController(client federation_release_1_4.Interface) *NamespaceC
// Federated updeater along with Create/Update/Delete operations.
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)
_, err := client.Core().Namespaces().Create(namespace)
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)
_, err := client.Core().Namespaces().Update(namespace)
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)
err := client.Core().Namespaces().Delete(namespace.Name, &api.DeleteOptions{})
return err
......
federation/pkg/federation-controller/namespace/namespace_controller_test.go
View file @ 47f317ee
......@@ -22,10 +22,11 @@ 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"
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
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/runtime"
"k8s.io/kubernetes/pkg/watch"
......@@ -38,25 +39,25 @@ func TestNamespaceController(t *testing.T) {
cluster2 := mkCluster("cluster2", api_v1.ConditionTrue)
fakeClient := &fake_federation_release_1_4.Clientset{}
RegisterList("clusters", fakeClient, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}})
RegisterList("namespaces", fakeClient, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
namespaceWatch := RegisterWatch("namespaces", fakeClient)
clusterWatch := RegisterWatch("clusters", fakeClient)
cluster1Client := &fake_federation_release_1_4.Clientset{}
cluster1Watch := RegisterWatch("namespaces", cluster1Client)
RegisterList("namespaces", cluster1Client, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
cluster1CreateChan := RegisterCopyOnCreate("namespaces", cluster1Client, cluster1Watch)
cluster1UpdateChan := RegisterCopyOnUpdate("namespaces", cluster1Client, cluster1Watch)
cluster2Client := &fake_federation_release_1_4.Clientset{}
cluster2Watch := RegisterWatch("namespaces", cluster2Client)
RegisterList("namespaces", cluster2Client, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
cluster2CreateChan := RegisterCopyOnCreate("namespaces", cluster2Client, cluster2Watch)
RegisterList("clusters", &fakeClient.Fake, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}})
RegisterList("namespaces", &fakeClient.Fake, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
namespaceWatch := RegisterWatch("namespaces", &fakeClient.Fake)
clusterWatch := RegisterWatch("clusters", &fakeClient.Fake)
cluster1Client := &fake_kube_release_1_4.Clientset{}
cluster1Watch := RegisterWatch("namespaces", &cluster1Client.Fake)
RegisterList("namespaces", &cluster1Client.Fake, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
cluster1CreateChan := RegisterCopyOnCreate("namespaces", &cluster1Client.Fake, cluster1Watch)
cluster1UpdateChan := RegisterCopyOnUpdate("namespaces", &cluster1Client.Fake, cluster1Watch)
cluster2Client := &fake_kube_release_1_4.Clientset{}
cluster2Watch := RegisterWatch("namespaces", &cluster2Client.Fake)
RegisterList("namespaces", &cluster2Client.Fake, &api_v1.NamespaceList{Items: []api_v1.Namespace{}})
cluster2CreateChan := RegisterCopyOnCreate("namespaces", &cluster2Client.Fake, cluster2Watch)
namespaceController := NewNamespaceController(fakeClient)
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 {
case cluster1.Name:
return cluster1Client, nil
......@@ -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()
client.AddWatchReactor(resource, func(action core.Action) (bool, watch.Interface, error) { return true, watcher, nil })
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) {
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)
client.AddReactor("create", resource, func(action core.Action) (bool, runtime.Object, error) {
createAction := action.(core.CreateAction)
......@@ -150,7 +151,7 @@ func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.C
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)
client.AddReactor("update", resource, func(action core.Action) (bool, runtime.Object, error) {
updateAction := action.(core.UpdateAction)
......
federation/pkg/federation-controller/secret/secret_controller.go
View file @ 47f317ee
......@@ -26,6 +26,7 @@ import (
"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/controller"
"k8s.io/kubernetes/pkg/controller/framework"
pkg_runtime "k8s.io/kubernetes/pkg/runtime"
......@@ -102,7 +103,7 @@ func NewSecretController(client federation_release_1_4.Interface) *SecretControl
// Federated informer on secrets in members of federation.
secretcontroller.secretFederatedInformer = util.NewFederatedInformer(
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(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
......@@ -133,17 +134,17 @@ func NewSecretController(client federation_release_1_4.Interface) *SecretControl
// Federated updeater along with Create/Update/Delete operations.
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)
_, err := client.Core().Secrets(secret.Namespace).Create(secret)
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)
_, err := client.Core().Secrets(secret.Namespace).Update(secret)
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)
err := client.Core().Secrets(secret.Namespace).Delete(secret.Name, &api.DeleteOptions{})
return err
......
federation/pkg/federation-controller/secret/secret_controller_test.go
View file @ 47f317ee
......@@ -22,10 +22,11 @@ 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"
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
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/runtime"
"k8s.io/kubernetes/pkg/watch"
......@@ -38,25 +39,25 @@ func TestSecretController(t *testing.T) {
cluster2 := mkCluster("cluster2", api_v1.ConditionTrue)
fakeClient := &fake_federation_release_1_4.Clientset{}
RegisterList("clusters", fakeClient, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}})
RegisterList("secrets", fakeClient, &api_v1.SecretList{Items: []api_v1.Secret{}})
secretWatch := RegisterWatch("secrets", fakeClient)
clusterWatch := RegisterWatch("clusters", fakeClient)
cluster1Client := &fake_federation_release_1_4.Clientset{}
cluster1Watch := RegisterWatch("secrets", cluster1Client)
RegisterList("secrets", cluster1Client, &api_v1.SecretList{Items: []api_v1.Secret{}})
cluster1CreateChan := RegisterCopyOnCreate("secrets", cluster1Client, cluster1Watch)
cluster1UpdateChan := RegisterCopyOnUpdate("secrets", cluster1Client, cluster1Watch)
cluster2Client := &fake_federation_release_1_4.Clientset{}
cluster2Watch := RegisterWatch("secrets", cluster2Client)
RegisterList("secrets", cluster2Client, &api_v1.SecretList{Items: []api_v1.Secret{}})
cluster2CreateChan := RegisterCopyOnCreate("secrets", cluster2Client, cluster2Watch)
RegisterList("clusters", &fakeClient.Fake, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}})
RegisterList("secrets", &fakeClient.Fake, &api_v1.SecretList{Items: []api_v1.Secret{}})
secretWatch := RegisterWatch("secrets", &fakeClient.Fake)
clusterWatch := RegisterWatch("clusters", &fakeClient.Fake)
cluster1Client := &fake_kube_release_1_4.Clientset{}
cluster1Watch := RegisterWatch("secrets", &cluster1Client.Fake)
RegisterList("secrets", &cluster1Client.Fake, &api_v1.SecretList{Items: []api_v1.Secret{}})
cluster1CreateChan := RegisterCopyOnCreate("secrets", &cluster1Client.Fake, cluster1Watch)
cluster1UpdateChan := RegisterCopyOnUpdate("secrets", &cluster1Client.Fake, cluster1Watch)
cluster2Client := &fake_kube_release_1_4.Clientset{}
cluster2Watch := RegisterWatch("secrets", &cluster2Client.Fake)
RegisterList("secrets", &cluster2Client.Fake, &api_v1.SecretList{Items: []api_v1.Secret{}})
cluster2CreateChan := RegisterCopyOnCreate("secrets", &cluster2Client.Fake, cluster2Watch)
secretController := NewSecretController(fakeClient)
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 {
case cluster1.Name:
return cluster1Client, nil
......@@ -123,19 +124,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()
client.AddWatchReactor(resource, func(action core.Action) (bool, watch.Interface, error) { return true, watcher, nil })
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) {
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)
client.AddReactor("create", resource, func(action core.Action) (bool, runtime.Object, error) {
createAction := action.(core.CreateAction)
......@@ -149,7 +150,7 @@ func RegisterCopyOnCreate(resource string, client *fake_federation_release_1_4.C
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)
client.AddReactor("update", resource, func(action core.Action) (bool, runtime.Object, error) {
updateAction := action.(core.UpdateAction)
......
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