2016-02-08 20:57:32 +00:00
|
|
|
package k8s
|
|
|
|
|
|
|
|
import (
|
2016-11-13 21:11:58 +00:00
|
|
|
"time"
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
"k8s.io/client-go/1.5/kubernetes"
|
|
|
|
"k8s.io/client-go/1.5/pkg/api"
|
|
|
|
"k8s.io/client-go/1.5/pkg/api/v1"
|
|
|
|
"k8s.io/client-go/1.5/pkg/apis/extensions/v1beta1"
|
|
|
|
"k8s.io/client-go/1.5/pkg/fields"
|
|
|
|
"k8s.io/client-go/1.5/pkg/labels"
|
|
|
|
"k8s.io/client-go/1.5/pkg/runtime"
|
|
|
|
"k8s.io/client-go/1.5/pkg/watch"
|
|
|
|
"k8s.io/client-go/1.5/rest"
|
|
|
|
"k8s.io/client-go/1.5/tools/cache"
|
2016-02-08 20:57:32 +00:00
|
|
|
)
|
|
|
|
|
2016-11-13 21:11:58 +00:00
|
|
|
const resyncPeriod = time.Minute * 5
|
|
|
|
|
2016-02-08 20:57:32 +00:00
|
|
|
// Client is a client for the Kubernetes master.
|
2016-04-20 11:26:51 +00:00
|
|
|
type Client interface {
|
2016-11-11 22:50:20 +00:00
|
|
|
GetIngresses(namespaces Namespaces) []*v1beta1.Ingress
|
|
|
|
GetService(namespace, name string) (*v1.Service, bool, error)
|
|
|
|
GetEndpoints(namespace, name string) (*v1.Endpoints, bool, error)
|
|
|
|
WatchAll(labelSelector string, stopCh <-chan bool) (chan interface{}, error)
|
2016-04-20 11:26:51 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
type clientImpl struct {
|
2016-11-11 22:50:20 +00:00
|
|
|
ingController *cache.Controller
|
|
|
|
svcController *cache.Controller
|
|
|
|
epController *cache.Controller
|
|
|
|
|
|
|
|
ingStore cache.Store
|
|
|
|
svcStore cache.Store
|
|
|
|
epStore cache.Store
|
|
|
|
|
|
|
|
clientset *kubernetes.Clientset
|
2016-02-08 20:57:32 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// NewInClusterClient returns a new Kubernetes client.
|
|
|
|
// WatchAll starts the watch of the Kubernetes ressources and updates the stores.
|
|
|
|
// The stores can be accessed via the Get* functions.
|
|
|
|
func NewInClusterClient() (Client, error) {
|
|
|
|
config, err := rest.InClusterConfig()
|
2016-02-08 20:57:32 +00:00
|
|
|
if err != nil {
|
2016-11-11 22:50:20 +00:00
|
|
|
return nil, err
|
2016-02-08 20:57:32 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
clientset, err := kubernetes.NewForConfig(config)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2016-04-20 11:26:51 +00:00
|
|
|
return &clientImpl{
|
2016-11-11 22:50:20 +00:00
|
|
|
clientset: clientset,
|
2016-02-08 20:57:32 +00:00
|
|
|
}, nil
|
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// NewInClusterClientWithEndpoint is the same as NewInClusterClient but uses the provided endpoint URL
|
|
|
|
func NewInClusterClientWithEndpoint(endpoint string) (Client, error) {
|
|
|
|
config, err := rest.InClusterConfig()
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2016-07-12 05:25:01 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
|
|
|
|
config.Host = endpoint
|
|
|
|
|
|
|
|
clientset, err := kubernetes.NewForConfig(config)
|
2016-07-12 05:25:01 +00:00
|
|
|
if err != nil {
|
2016-11-11 22:50:20 +00:00
|
|
|
return nil, err
|
2016-07-12 05:25:01 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
|
|
|
|
return &clientImpl{
|
|
|
|
clientset: clientset,
|
|
|
|
}, nil
|
2016-07-12 05:25:01 +00:00
|
|
|
}
|
|
|
|
|
2016-04-28 00:23:55 +00:00
|
|
|
// GetIngresses returns all ingresses in the cluster
|
2016-11-11 22:50:20 +00:00
|
|
|
func (c *clientImpl) GetIngresses(namespaces Namespaces) []*v1beta1.Ingress {
|
|
|
|
ingList := c.ingStore.List()
|
|
|
|
result := make([]*v1beta1.Ingress, 0, len(ingList))
|
2016-02-08 20:57:32 +00:00
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
for _, obj := range ingList {
|
|
|
|
ingress := obj.(*v1beta1.Ingress)
|
|
|
|
if HasNamespace(ingress, namespaces) {
|
|
|
|
result = append(result, ingress)
|
2016-02-08 20:57:32 +00:00
|
|
|
}
|
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
|
|
|
|
return result
|
2016-02-08 20:57:32 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// WatchIngresses starts the watch of Kubernetes Ingresses resources and updates the corresponding store
|
|
|
|
func (c *clientImpl) WatchIngresses(labelSelector labels.Selector, stopCh <-chan struct{}) chan interface{} {
|
|
|
|
watchCh := make(chan interface{}, 10)
|
|
|
|
|
|
|
|
source := NewListWatchFromClient(
|
|
|
|
c.clientset.ExtensionsClient,
|
|
|
|
"ingresses",
|
|
|
|
api.NamespaceAll,
|
|
|
|
fields.Everything(),
|
|
|
|
labelSelector)
|
|
|
|
|
|
|
|
c.ingStore, c.ingController = cache.NewInformer(
|
|
|
|
source,
|
|
|
|
&v1beta1.Ingress{},
|
2016-11-13 21:11:58 +00:00
|
|
|
resyncPeriod,
|
2016-11-11 22:50:20 +00:00
|
|
|
newResourceEventHandlerFuncs(watchCh))
|
|
|
|
go c.ingController.Run(stopCh)
|
|
|
|
|
|
|
|
return watchCh
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
func newResourceEventHandlerFuncs(events chan interface{}) cache.ResourceEventHandlerFuncs {
|
2016-04-25 14:56:06 +00:00
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
return cache.ResourceEventHandlerFuncs{
|
|
|
|
AddFunc: func(obj interface{}) { events <- obj },
|
|
|
|
UpdateFunc: func(old, new interface{}) { events <- new },
|
|
|
|
DeleteFunc: func(obj interface{}) { events <- obj },
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// GetService returns the named service from the named namespace
|
|
|
|
func (c *clientImpl) GetService(namespace, name string) (*v1.Service, bool, error) {
|
|
|
|
var service *v1.Service
|
|
|
|
item, exists, err := c.svcStore.GetByKey(namespace + "/" + name)
|
|
|
|
if item != nil {
|
|
|
|
service = item.(*v1.Service)
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
|
|
|
|
return service, exists, err
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// WatchServices starts the watch of Kubernetes Service resources and updates the corresponding store
|
|
|
|
func (c *clientImpl) WatchServices(stopCh <-chan struct{}) chan interface{} {
|
|
|
|
watchCh := make(chan interface{}, 10)
|
|
|
|
|
|
|
|
source := cache.NewListWatchFromClient(
|
|
|
|
c.clientset.CoreClient,
|
|
|
|
"services",
|
|
|
|
api.NamespaceAll,
|
|
|
|
fields.Everything())
|
|
|
|
|
|
|
|
c.svcStore, c.svcController = cache.NewInformer(
|
|
|
|
source,
|
|
|
|
&v1.Service{},
|
2016-11-13 21:11:58 +00:00
|
|
|
resyncPeriod,
|
2016-11-11 22:50:20 +00:00
|
|
|
newResourceEventHandlerFuncs(watchCh))
|
|
|
|
go c.svcController.Run(stopCh)
|
|
|
|
|
|
|
|
return watchCh
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-05-20 16:34:57 +00:00
|
|
|
// GetEndpoints returns the named Endpoints
|
|
|
|
// Endpoints have the same name as the coresponding service
|
2016-11-11 22:50:20 +00:00
|
|
|
func (c *clientImpl) GetEndpoints(namespace, name string) (*v1.Endpoints, bool, error) {
|
|
|
|
var endpoint *v1.Endpoints
|
|
|
|
item, exists, err := c.epStore.GetByKey(namespace + "/" + name)
|
2016-04-25 14:56:06 +00:00
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
if item != nil {
|
|
|
|
endpoint = item.(*v1.Endpoints)
|
2016-05-20 16:34:57 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
return endpoint, exists, err
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// WatchEndpoints starts the watch of Kubernetes Endpoints resources and updates the corresponding store
|
|
|
|
func (c *clientImpl) WatchEndpoints(stopCh <-chan struct{}) chan interface{} {
|
|
|
|
watchCh := make(chan interface{}, 10)
|
|
|
|
|
|
|
|
source := cache.NewListWatchFromClient(
|
|
|
|
c.clientset.CoreClient,
|
|
|
|
"endpoints",
|
|
|
|
api.NamespaceAll,
|
|
|
|
fields.Everything())
|
|
|
|
|
|
|
|
c.epStore, c.epController = cache.NewInformer(
|
|
|
|
source,
|
|
|
|
&v1.Endpoints{},
|
2016-11-13 21:11:58 +00:00
|
|
|
resyncPeriod,
|
2016-11-11 22:50:20 +00:00
|
|
|
newResourceEventHandlerFuncs(watchCh))
|
|
|
|
go c.epController.Run(stopCh)
|
|
|
|
|
|
|
|
return watchCh
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// WatchAll returns events in the cluster and updates the stores via informer
|
|
|
|
// Filters ingresses by labelSelector
|
|
|
|
func (c *clientImpl) WatchAll(labelSelector string, stopCh <-chan bool) (chan interface{}, error) {
|
2016-11-22 21:02:51 +00:00
|
|
|
watchCh := make(chan interface{}, 100)
|
|
|
|
stopWatchCh := make(chan struct{}, 1)
|
2016-02-08 20:57:32 +00:00
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
kubeLabelSelector, err := labels.Parse(labelSelector)
|
2016-04-25 14:56:06 +00:00
|
|
|
if err != nil {
|
2016-11-11 22:50:20 +00:00
|
|
|
return nil, err
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
|
2016-11-22 21:02:51 +00:00
|
|
|
chanIngresses := c.WatchIngresses(kubeLabelSelector, stopWatchCh)
|
|
|
|
chanServices := c.WatchServices(stopWatchCh)
|
|
|
|
chanEndpoints := c.WatchEndpoints(stopWatchCh)
|
2016-11-11 22:50:20 +00:00
|
|
|
|
2016-04-25 14:56:06 +00:00
|
|
|
go func() {
|
2016-11-22 21:02:51 +00:00
|
|
|
defer close(stopWatchCh)
|
2016-04-25 14:56:06 +00:00
|
|
|
defer close(watchCh)
|
2016-02-08 20:57:32 +00:00
|
|
|
|
2016-04-25 14:56:06 +00:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-stopCh:
|
2016-05-19 18:09:01 +00:00
|
|
|
return
|
2016-04-25 14:56:06 +00:00
|
|
|
case event := <-chanIngresses:
|
2016-11-11 22:50:20 +00:00
|
|
|
c.fireEvent(event, watchCh)
|
2016-04-25 14:56:06 +00:00
|
|
|
case event := <-chanServices:
|
2016-11-11 22:50:20 +00:00
|
|
|
c.fireEvent(event, watchCh)
|
2016-05-20 16:34:57 +00:00
|
|
|
case event := <-chanEndpoints:
|
2016-11-11 22:50:20 +00:00
|
|
|
c.fireEvent(event, watchCh)
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
return watchCh, nil
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// fireEvent checks if all controllers have synced before firing
|
|
|
|
// Used after startup or a reconnect
|
|
|
|
func (c *clientImpl) fireEvent(event interface{}, watchCh chan interface{}) {
|
|
|
|
if c.ingController.HasSynced() && c.svcController.HasSynced() && c.epController.HasSynced() {
|
|
|
|
watchCh <- event
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// HasNamespace checks if the ingress is in one of the namespaces
|
|
|
|
func HasNamespace(ingress *v1beta1.Ingress, namespaces Namespaces) bool {
|
|
|
|
if len(namespaces) == 0 {
|
|
|
|
return true
|
2016-05-19 08:52:17 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
for _, n := range namespaces {
|
|
|
|
if ingress.ObjectMeta.Namespace == n {
|
|
|
|
return true
|
|
|
|
}
|
2016-02-08 20:57:32 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
return false
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
|
|
|
|
2016-11-11 22:50:20 +00:00
|
|
|
// NewListWatchFromClient creates a new ListWatch from the specified client, resource, namespace, field selector and label selector.
|
|
|
|
// Extends cache.NewListWatchFromClient to support labelSelector
|
|
|
|
func NewListWatchFromClient(c cache.Getter, resource string, namespace string, fieldSelector fields.Selector, labelSelector labels.Selector) *cache.ListWatch {
|
|
|
|
listFunc := func(options api.ListOptions) (runtime.Object, error) {
|
|
|
|
return c.Get().
|
|
|
|
Namespace(namespace).
|
|
|
|
Resource(resource).
|
|
|
|
VersionedParams(&options, api.ParameterCodec).
|
|
|
|
FieldsSelectorParam(fieldSelector).
|
|
|
|
LabelsSelectorParam(labelSelector).
|
|
|
|
Do().
|
|
|
|
Get()
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
watchFunc := func(options api.ListOptions) (watch.Interface, error) {
|
|
|
|
return c.Get().
|
|
|
|
Prefix("watch").
|
|
|
|
Namespace(namespace).
|
|
|
|
Resource(resource).
|
|
|
|
VersionedParams(&options, api.ParameterCodec).
|
|
|
|
FieldsSelectorParam(fieldSelector).
|
|
|
|
LabelsSelectorParam(labelSelector).
|
|
|
|
Watch()
|
2016-04-25 14:56:06 +00:00
|
|
|
}
|
2016-11-11 22:50:20 +00:00
|
|
|
return &cache.ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
|
2016-02-08 20:57:32 +00:00
|
|
|
}
|