github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-18 08:09:58 +00:00
Code Issues Projects Releases Wiki Activity

federation

Browse Source
This commit is contained in:
Chao Xu 2016-11-18 12:59:18 -08:00
parent bb675d395f
commit eca157588d
17 changed files with 118 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
federation/cmd/federation-apiserver/app/server.go
View File

@ -143,7 +143,7 @@ func Run(s *options.ServerRunOptions) error {
if err != nil {
glog.Errorf("Failed to create clientset: %v", err)
}
sharedInformers := informers.NewSharedInformerFactory(client, 10*time.Minute)
sharedInformers := informers.NewSharedInformerFactory(nil, client, 10*time.Minute)
authorizationConfig := authorizer.AuthorizationConfig{
PolicyFile: s.GenericServerRunOptions.AuthorizationPolicyFile,

12
federation/pkg/federation-controller/cluster/clustercontroller.go
View File

@ -24,8 +24,6 @@ import (
federation_v1beta1 "k8s.io/kubernetes/federation/apis/federation/v1beta1"
cluster_cache "k8s.io/kubernetes/federation/client/cache"
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5"
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache"
"k8s.io/kubernetes/pkg/controller"
@ -66,13 +64,11 @@ func NewclusterController(federationClient federationclientset.Interface, cluste
}
cc.clusterStore.Store, cc.clusterController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return cc.federationClient.Federation().Clusters().List(versionedOptions)
ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
return cc.federationClient.Federation().Clusters().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return cc.federationClient.Federation().Clusters().Watch(versionedOptions)
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
return cc.federationClient.Federation().Clusters().Watch(options)
},
},
&federation_v1beta1.Cluster{},

22
federation/pkg/federation-controller/configmap/configmap_controller.go
View File

@ -79,7 +79,7 @@ type ConfigMapController struct {
func NewConfigMapController(client federationclientset.Interface) *ConfigMapController {
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-configmaps-controller"})
recorder := broadcaster.NewRecorder(api_v1.EventSource{Component: "federated-configmaps-controller"})
configmapcontroller := &ConfigMapController{
federatedApiClient: client,
@ -98,13 +98,11 @@ func NewConfigMapController(client federationclientset.Interface) *ConfigMapCont
// Start informer on federated API servers on configmaps that should be federated.
configmapcontroller.configmapInformerStore, configmapcontroller.configmapInformerController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Core().ConfigMaps(api_v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (pkg_runtime.Object, error) {
return client.Core().ConfigMaps(api_v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Core().ConfigMaps(api_v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return client.Core().ConfigMaps(api_v1.NamespaceAll).Watch(options)
},
},
&api_v1.ConfigMap{},
@ -117,13 +115,11 @@ func NewConfigMapController(client federationclientset.Interface) *ConfigMapCont
func(cluster *federation_api.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().ConfigMaps(api_v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (pkg_runtime.Object, error) {
return targetClient.Core().ConfigMaps(api_v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().ConfigMaps(api_v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return targetClient.Core().ConfigMaps(api_v1.NamespaceAll).Watch(options)
},
},
&api_v1.ConfigMap{},

22
federation/pkg/federation-controller/daemonset/daemonset_controller.go
View File

@ -87,7 +87,7 @@ type DaemonSetController struct {
func NewDaemonSetController(client federationclientset.Interface) *DaemonSetController {
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-daemonset-controller"})
recorder := broadcaster.NewRecorder(api_v1.EventSource{Component: "federated-daemonset-controller"})
daemonsetcontroller := &DaemonSetController{
federatedApiClient: client,
@ -106,13 +106,11 @@ func NewDaemonSetController(client federationclientset.Interface) *DaemonSetCont
// Start informer in federated API servers on daemonsets that should be federated.
daemonsetcontroller.daemonsetInformerStore, daemonsetcontroller.daemonsetInformerController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Extensions().DaemonSets(api_v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (pkg_runtime.Object, error) {
return client.Extensions().DaemonSets(api_v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Extensions().DaemonSets(api_v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return client.Extensions().DaemonSets(api_v1.NamespaceAll).Watch(options)
},
},
&extensionsv1.DaemonSet{},
@ -125,13 +123,11 @@ func NewDaemonSetController(client federationclientset.Interface) *DaemonSetCont
func(cluster *federation_api.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Extensions().DaemonSets(api_v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (pkg_runtime.Object, error) {
return targetClient.Extensions().DaemonSets(api_v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Extensions().DaemonSets(api_v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return targetClient.Extensions().DaemonSets(api_v1.NamespaceAll).Watch(options)
},
},
&extensionsv1.DaemonSet{},

32
federation/pkg/federation-controller/deployment/deploymentcontroller.go
View File

@ -104,7 +104,7 @@ type DeploymentController struct {
func NewDeploymentController(federationClient fedclientset.Interface) *DeploymentController {
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(federationClient))
recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-deployment-controller"})
recorder := broadcaster.NewRecorder(apiv1.EventSource{Component: "federated-deployment-controller"})
fdc := &DeploymentController{
fedClient: federationClient,
@ -123,13 +123,11 @@ func NewDeploymentController(federationClient fedclientset.Interface) *Deploymen
deploymentFedInformerFactory := func(cluster *fedv1.Cluster, clientset kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Extensions().Deployments(apiv1.NamespaceAll).List(versionedOptions)
ListFunc: func(options apiv1.ListOptions) (runtime.Object, error) {
return clientset.Extensions().Deployments(apiv1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Extensions().Deployments(apiv1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options apiv1.ListOptions) (watch.Interface, error) {
return clientset.Extensions().Deployments(apiv1.NamespaceAll).Watch(options)
},
},
&extensionsv1.Deployment{},
@ -152,13 +150,11 @@ func NewDeploymentController(federationClient fedclientset.Interface) *Deploymen
podFedInformerFactory := func(cluster *fedv1.Cluster, clientset kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Core().Pods(apiv1.NamespaceAll).List(versionedOptions)
ListFunc: func(options apiv1.ListOptions) (runtime.Object, error) {
return clientset.Core().Pods(apiv1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Core().Pods(apiv1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options apiv1.ListOptions) (watch.Interface, error) {
return clientset.Core().Pods(apiv1.NamespaceAll).Watch(options)
},
},
&apiv1.Pod{},
@ -174,13 +170,11 @@ func NewDeploymentController(federationClient fedclientset.Interface) *Deploymen
fdc.deploymentStore, fdc.deploymentController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return fdc.fedClient.Extensions().Deployments(apiv1.NamespaceAll).List(versionedOptions)
ListFunc: func(options apiv1.ListOptions) (runtime.Object, error) {
return fdc.fedClient.Extensions().Deployments(apiv1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return fdc.fedClient.Extensions().Deployments(apiv1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options apiv1.ListOptions) (watch.Interface, error) {
return fdc.fedClient.Extensions().Deployments(apiv1.NamespaceAll).Watch(options)
},
},
&extensionsv1.Deployment{},

32
federation/pkg/federation-controller/ingress/ingress_controller.go
View File

@ -108,7 +108,7 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
glog.V(4).Infof("->NewIngressController V(4)")
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-ingress-controller"})
recorder := broadcaster.NewRecorder(v1.EventSource{Component: "federated-ingress-controller"})
ic := &IngressController{
federatedApiClient: client,
ingressReviewDelay: time.Second * 10,
@ -129,13 +129,11 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
// Start informer in federated API servers on ingresses that should be federated.
ic.ingressInformerStore, ic.ingressInformerController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Extensions().Ingresses(api.NamespaceAll).List(versionedOptions)
ListFunc: func(options v1.ListOptions) (pkg_runtime.Object, error) {
return client.Extensions().Ingresses(api.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Extensions().Ingresses(api.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
return client.Extensions().Ingresses(api.NamespaceAll).Watch(options)
},
},
&extensions_v1beta1.Ingress{},
@ -152,13 +150,11 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
func(cluster *federation_api.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Extensions().Ingresses(api.NamespaceAll).List(versionedOptions)
ListFunc: func(options v1.ListOptions) (pkg_runtime.Object, error) {
return targetClient.Extensions().Ingresses(api.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Extensions().Ingresses(api.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
return targetClient.Extensions().Ingresses(api.NamespaceAll).Watch(options)
},
},
&extensions_v1beta1.Ingress{},
@ -187,19 +183,17 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
glog.V(4).Infof("Returning new informer for cluster %q", cluster.Name)
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
ListFunc: func(options v1.ListOptions) (pkg_runtime.Object, error) {
if targetClient == nil {
glog.Errorf("Internal error: targetClient is nil")
}
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().ConfigMaps(uidConfigMapNamespace).List(versionedOptions) // we only want to list one by name - unfortunately Kubernetes don't have a selector for that.
return targetClient.Core().ConfigMaps(uidConfigMapNamespace).List(options) // we only want to list one by name - unfortunately Kubernetes don't have a selector for that.
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
if targetClient == nil {
glog.Errorf("Internal error: targetClient is nil")
}
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().ConfigMaps(uidConfigMapNamespace).Watch(versionedOptions) // as above
return targetClient.Core().ConfigMaps(uidConfigMapNamespace).Watch(options) // as above
},
},
&v1.ConfigMap{},

22
federation/pkg/federation-controller/namespace/namespace_controller.go
View File

@ -84,7 +84,7 @@ type NamespaceController struct {
func NewNamespaceController(client federationclientset.Interface) *NamespaceController {
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-namespace-controller"})
recorder := broadcaster.NewRecorder(api_v1.EventSource{Component: "federated-namespace-controller"})
nc := &NamespaceController{
federatedApiClient: client,
@ -103,13 +103,11 @@ func NewNamespaceController(client federationclientset.Interface) *NamespaceCont
// Start informer in federated API servers on namespaces that should be federated.
nc.namespaceInformerStore, nc.namespaceInformerController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Core().Namespaces().List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (runtime.Object, error) {
return client.Core().Namespaces().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Core().Namespaces().Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return client.Core().Namespaces().Watch(options)
},
},
&api_v1.Namespace{},
@ -122,13 +120,11 @@ func NewNamespaceController(client federationclientset.Interface) *NamespaceCont
func(cluster *federation_api.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().Namespaces().List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (runtime.Object, error) {
return targetClient.Core().Namespaces().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().Namespaces().Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return targetClient.Core().Namespaces().Watch(options)
},
},
&api_v1.Namespace{},

32
federation/pkg/federation-controller/replicaset/replicasetcontroller.go
View File

@ -106,7 +106,7 @@ type ReplicaSetController struct {
func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSetController {
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(federationClient))
recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-replicaset-controller"})
recorder := broadcaster.NewRecorder(apiv1.EventSource{Component: "federated-replicaset-controller"})
frsc := &ReplicaSetController{
fedClient: federationClient,
@ -125,13 +125,11 @@ func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSe
replicaSetFedInformerFactory := func(cluster *fedv1.Cluster, clientset kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Extensions().ReplicaSets(apiv1.NamespaceAll).List(versionedOptions)
ListFunc: func(options apiv1.ListOptions) (runtime.Object, error) {
return clientset.Extensions().ReplicaSets(apiv1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Extensions().ReplicaSets(apiv1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options apiv1.ListOptions) (watch.Interface, error) {
return clientset.Extensions().ReplicaSets(apiv1.NamespaceAll).Watch(options)
},
},
&extensionsv1.ReplicaSet{},
@ -154,13 +152,11 @@ func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSe
podFedInformerFactory := func(cluster *fedv1.Cluster, clientset kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Core().Pods(apiv1.NamespaceAll).List(versionedOptions)
ListFunc: func(options apiv1.ListOptions) (runtime.Object, error) {
return clientset.Core().Pods(apiv1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return clientset.Core().Pods(apiv1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options apiv1.ListOptions) (watch.Interface, error) {
return clientset.Core().Pods(apiv1.NamespaceAll).Watch(options)
},
},
&apiv1.Pod{},
@ -176,13 +172,11 @@ func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSe
frsc.replicaSetStore.Indexer, frsc.replicaSetController = cache.NewIndexerInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return frsc.fedClient.Extensions().ReplicaSets(apiv1.NamespaceAll).List(versionedOptions)
ListFunc: func(options apiv1.ListOptions) (runtime.Object, error) {
return frsc.fedClient.Extensions().ReplicaSets(apiv1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := fedutil.VersionizeV1ListOptions(options)
return frsc.fedClient.Extensions().ReplicaSets(apiv1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options apiv1.ListOptions) (watch.Interface, error) {
return frsc.fedClient.Extensions().ReplicaSets(apiv1.NamespaceAll).Watch(options)
},
},
&extensionsv1.ReplicaSet{},

22
federation/pkg/federation-controller/secret/secret_controller.go
View File

@ -85,7 +85,7 @@ type SecretController struct {
func NewSecretController(client federationclientset.Interface) *SecretController {
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-secrets-controller"})
recorder := broadcaster.NewRecorder(api_v1.EventSource{Component: "federated-secrets-controller"})
secretcontroller := &SecretController{
federatedApiClient: client,
@ -104,13 +104,11 @@ func NewSecretController(client federationclientset.Interface) *SecretController
// Start informer in federated API servers on secrets that should be federated.
secretcontroller.secretInformerStore, secretcontroller.secretInformerController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Core().Secrets(api_v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (pkg_runtime.Object, error) {
return client.Core().Secrets(api_v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return client.Core().Secrets(api_v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return client.Core().Secrets(api_v1.NamespaceAll).Watch(options)
},
},
&api_v1.Secret{},
@ -123,13 +121,11 @@ func NewSecretController(client federationclientset.Interface) *SecretController
func(cluster *federation_api.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().Secrets(api_v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (pkg_runtime.Object, error) {
return targetClient.Core().Secrets(api_v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return targetClient.Core().Secrets(api_v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return targetClient.Core().Secrets(api_v1.NamespaceAll).Watch(options)
},
},
&api_v1.Secret{},

21
federation/pkg/federation-controller/service/cluster_helper.go
View File

@ -20,7 +20,6 @@ import (
"sync"
v1beta1 "k8s.io/kubernetes/federation/apis/federation/v1beta1"
"k8s.io/kubernetes/pkg/api"
v1 "k8s.io/kubernetes/pkg/api/v1"
cache "k8s.io/kubernetes/pkg/client/cache"
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
@ -92,13 +91,11 @@ func (cc *clusterClientCache) startClusterLW(cluster *v1beta1.Cluster, clusterNa
}
cachedClusterClient.endpointStore.Store, cachedClusterClient.endpointController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return clientset.Core().Endpoints(v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options v1.ListOptions) (pkg_runtime.Object, error) {
return clientset.Core().Endpoints(v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return clientset.Core().Endpoints(v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
return clientset.Core().Endpoints(v1.NamespaceAll).Watch(options)
},
},
&v1.Endpoints{},
@ -118,13 +115,11 @@ func (cc *clusterClientCache) startClusterLW(cluster *v1beta1.Cluster, clusterNa
cachedClusterClient.serviceStore.Indexer, cachedClusterClient.serviceController = cache.NewIndexerInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return clientset.Core().Services(v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options v1.ListOptions) (pkg_runtime.Object, error) {
return clientset.Core().Services(v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return clientset.Core().Services(v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
return clientset.Core().Services(v1.NamespaceAll).Watch(options)
},
},
&v1.Service{},

3
federation/pkg/federation-controller/service/dns.go
View File

@ -22,9 +22,10 @@ import (
"github.com/golang/glog"
"strings"
"k8s.io/kubernetes/federation/pkg/dnsprovider"
"k8s.io/kubernetes/federation/pkg/dnsprovider/rrstype"
"strings"
)
const (

5
federation/pkg/federation-controller/service/dns_test.go
View File

@ -21,12 +21,13 @@ import (
"testing"
"fmt"
"reflect"
"sort"
"k8s.io/kubernetes/federation/apis/federation/v1beta1"
"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns" // Only for unit testing purposes.
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/util/sets"
"reflect"
"sort"
)
func TestServiceController_ensureDnsRecords(t *testing.T) {

24
federation/pkg/federation-controller/service/servicecontroller.go
View File

@ -28,8 +28,6 @@ import (
federationcache "k8s.io/kubernetes/federation/client/cache"
fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5"
"k8s.io/kubernetes/federation/pkg/dnsprovider"
"k8s.io/kubernetes/federation/pkg/federation-controller/util"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/errors"
v1 "k8s.io/kubernetes/pkg/api/v1"
cache "k8s.io/kubernetes/pkg/client/cache"
@ -145,7 +143,7 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface,
broadcaster := record.NewBroadcaster()
// federationClient event is not supported yet
// broadcaster.StartRecordingToSink(&unversioned_core.EventSinkImpl{Interface: kubeClient.Core().Events("")})
recorder := broadcaster.NewRecorder(api.EventSource{Component: UserAgentName})
recorder := broadcaster.NewRecorder(v1.EventSource{Component: UserAgentName})
s := &ServiceController{
dns: dns,
@ -166,13 +164,11 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface,
}
s.serviceStore.Indexer, s.serviceController = cache.NewIndexerInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return s.federationClient.Core().Services(v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options v1.ListOptions) (pkg_runtime.Object, error) {
return s.federationClient.Core().Services(v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return s.federationClient.Core().Services(v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
return s.federationClient.Core().Services(v1.NamespaceAll).Watch(options)
},
},
&v1.Service{},
@ -191,13 +187,11 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface,
)
s.clusterStore.Store, s.clusterController = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return s.federationClient.Federation().Clusters().List(versionedOptions)
ListFunc: func(options v1.ListOptions) (pkg_runtime.Object, error) {
return s.federationClient.Federation().Clusters().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := util.VersionizeV1ListOptions(options)
return s.federationClient.Federation().Clusters().Watch(versionedOptions)
WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
return s.federationClient.Federation().Clusters().Watch(options)
},
},
&v1beta1.Cluster{},

32
federation/pkg/federation-controller/util/eventsink/eventsink.go
View File

@ -37,36 +37,14 @@ func NewFederatedEventSink(clientset fedclientset.Interface) *FederatedEventSink
}
}
func (fes *FederatedEventSink) Create(event *api.Event) (*api.Event, error) {
return fes.executeOperation(event, func(eventV1 *api_v1.Event) (*api_v1.Event, error) {
return fes.clientset.Core().Events(event.Namespace).Create(eventV1)
})
func (fes *FederatedEventSink) Create(event *api_v1.Event) (*api_v1.Event, error) {
return fes.clientset.Core().Events(event.Namespace).Create(event)
}
func (fes *FederatedEventSink) Update(event *api.Event) (*api.Event, error) {
return fes.executeOperation(event, func(eventV1 *api_v1.Event) (*api_v1.Event, error) {
return fes.clientset.Core().Events(event.Namespace).Update(eventV1)
})
func (fes *FederatedEventSink) Update(event *api_v1.Event) (*api_v1.Event, error) {
return fes.clientset.Core().Events(event.Namespace).Update(event)
}
func (fes *FederatedEventSink) Patch(event *api.Event, data []byte) (*api.Event, error) {
return fes.executeOperation(event, func(eventV1 *api_v1.Event) (*api_v1.Event, error) {
func (fes *FederatedEventSink) Patch(event *api_v1.Event, data []byte) (*api_v1.Event, error) {
return fes.clientset.Core().Events(event.Namespace).Patch(event.Name, api.StrategicMergePatchType, data)
})
}
func (fes *FederatedEventSink) executeOperation(event *api.Event, operation func(*api_v1.Event) (*api_v1.Event, error)) (*api.Event, error) {
var versionedEvent api_v1.Event
if err := api.Scheme.Convert(event, &versionedEvent, nil); err != nil {
return nil, err
}
versionedEventPtr, err := operation(&versionedEvent)
if err != nil {
return nil, err
}
var unversionedEvent api.Event
if err := api.Scheme.Convert(versionedEventPtr, &unversionedEvent, nil); err != nil {
return nil, err
}
return &unversionedEvent, nil
}

5
federation/pkg/federation-controller/util/eventsink/eventsink_test.go
View File

@ -21,7 +21,6 @@ import (
fake_fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5/fake"
. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
api "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/testing/core"
"k8s.io/kubernetes/pkg/runtime"
@ -46,8 +45,8 @@ func TestEventSink(t *testing.T) {
return true, obj, nil
})
event := api.Event{
ObjectMeta: api.ObjectMeta{
event := api_v1.Event{
ObjectMeta: api_v1.ObjectMeta{
Name: "bzium",
Namespace: "ns",
},

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

@ -24,7 +24,6 @@ import (
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1"
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5"
api "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache"
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
@ -161,13 +160,11 @@ func NewFederatedInformer(
federatedInformer.clusterInformer.store, federatedInformer.clusterInformer.controller = cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
versionedOptions := VersionizeV1ListOptions(options)
return federationClient.Federation().Clusters().List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (pkg_runtime.Object, error) {
return federationClient.Federation().Clusters().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := VersionizeV1ListOptions(options)
return federationClient.Federation().Clusters().Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return federationClient.Federation().Clusters().Watch(options)
},
},
&federation_api.Cluster{},

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

@ -22,7 +22,6 @@ import (
federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1"
fakefederationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5/fake"
api "k8s.io/kubernetes/pkg/api"
api_v1 "k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/cache"
kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
@ -81,13 +80,11 @@ func TestFederatedInformer(t *testing.T) {
targetInformerFactory := func(cluster *federation_api.Cluster, clientset kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
return cache.NewInformer(
&cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
versionedOptions := VersionizeV1ListOptions(options)
return clientset.Core().Services(api_v1.NamespaceAll).List(versionedOptions)
ListFunc: func(options api_v1.ListOptions) (runtime.Object, error) {
return clientset.Core().Services(api_v1.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
versionedOptions := VersionizeV1ListOptions(options)
return clientset.Core().Services(api_v1.NamespaceAll).Watch(versionedOptions)
WatchFunc: func(options api_v1.ListOptions) (watch.Interface, error) {
return clientset.Core().Services(api_v1.NamespaceAll).Watch(options)
},
},
&api_v1.Service{},