traefik/provider/kubernetes/client.go

265 lines
8.8 KiB
Go
Raw Normal View History

package kubernetes
import (
"errors"
"fmt"
"io/ioutil"
"time"
2018-03-22 09:14:04 +00:00
"github.com/containous/traefik/log"
2018-02-14 08:56:04 +00:00
corev1 "k8s.io/api/core/v1"
extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
2018-03-22 09:14:04 +00:00
"k8s.io/client-go/informers"
2017-04-07 10:49:53 +00:00
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/cache"
)
const resyncPeriod = 10 * time.Minute
type resourceEventHandler struct {
ev chan<- interface{}
}
func (reh *resourceEventHandler) OnAdd(obj interface{}) {
eventHandlerFunc(reh.ev, obj)
}
func (reh *resourceEventHandler) OnUpdate(oldObj, newObj interface{}) {
eventHandlerFunc(reh.ev, newObj)
}
func (reh *resourceEventHandler) OnDelete(obj interface{}) {
eventHandlerFunc(reh.ev, obj)
}
// Client is a client for the Provider master.
// WatchAll starts the watch of the Provider resources and updates the stores.
// The stores can then be accessed via the Get* functions.
type Client interface {
WatchAll(namespaces Namespaces, stopCh <-chan struct{}) (<-chan interface{}, error)
2018-02-14 08:56:04 +00:00
GetIngresses() []*extensionsv1beta1.Ingress
GetService(namespace, name string) (*corev1.Service, bool, error)
GetSecret(namespace, name string) (*corev1.Secret, bool, error)
GetEndpoints(namespace, name string) (*corev1.Endpoints, bool, error)
2018-05-18 12:12:03 +00:00
UpdateIngressStatus(namespace, name, ip, hostname string) error
}
type clientImpl struct {
clientset *kubernetes.Clientset
factories map[string]informers.SharedInformerFactory
ingressLabelSelector labels.Selector
isNamespaceAll bool
}
func newClientImpl(clientset *kubernetes.Clientset) *clientImpl {
return &clientImpl{
clientset: clientset,
2018-03-22 09:14:04 +00:00
factories: make(map[string]informers.SharedInformerFactory),
}
}
// newInClusterClient returns a new Provider client that is expected to run
// inside the cluster.
func newInClusterClient(endpoint string) (*clientImpl, error) {
config, err := rest.InClusterConfig()
if err != nil {
return nil, fmt.Errorf("failed to create in-cluster configuration: %s", err)
}
if endpoint != "" {
config.Host = endpoint
}
return createClientFromConfig(config)
}
// newExternalClusterClient returns a new Provider client that may run outside
// of the cluster.
// The endpoint parameter must not be empty.
func newExternalClusterClient(endpoint, token, caFilePath string) (*clientImpl, error) {
if endpoint == "" {
return nil, errors.New("endpoint missing for external cluster client")
}
config := &rest.Config{
Host: endpoint,
BearerToken: token,
}
if caFilePath != "" {
caData, err := ioutil.ReadFile(caFilePath)
if err != nil {
return nil, fmt.Errorf("failed to read CA file %s: %s", caFilePath, err)
}
config.TLSClientConfig = rest.TLSClientConfig{CAData: caData}
}
return createClientFromConfig(config)
}
func createClientFromConfig(c *rest.Config) (*clientImpl, error) {
clientset, err := kubernetes.NewForConfig(c)
if err != nil {
return nil, err
}
return newClientImpl(clientset), nil
}
// WatchAll starts namespace-specific controllers for all relevant kinds.
func (c *clientImpl) WatchAll(namespaces Namespaces, stopCh <-chan struct{}) (<-chan interface{}, error) {
eventCh := make(chan interface{}, 1)
if len(namespaces) == 0 {
2018-02-14 08:56:04 +00:00
namespaces = Namespaces{metav1.NamespaceAll}
c.isNamespaceAll = true
}
eventHandler := c.newResourceEventHandler(eventCh)
for _, ns := range namespaces {
factory := informers.NewFilteredSharedInformerFactory(c.clientset, resyncPeriod, ns, nil)
2018-03-22 09:14:04 +00:00
factory.Extensions().V1beta1().Ingresses().Informer().AddEventHandler(eventHandler)
factory.Core().V1().Services().Informer().AddEventHandler(eventHandler)
factory.Core().V1().Endpoints().Informer().AddEventHandler(eventHandler)
c.factories[ns] = factory
}
2018-03-22 09:14:04 +00:00
for _, ns := range namespaces {
c.factories[ns].Start(stopCh)
}
2018-03-22 09:14:04 +00:00
for _, ns := range namespaces {
for t, ok := range c.factories[ns].WaitForCacheSync(stopCh) {
if !ok {
return nil, fmt.Errorf("timed out waiting for controller caches to sync %s in namespace %q", t.String(), ns)
}
}
2018-02-14 08:56:04 +00:00
}
2018-03-22 09:14:04 +00:00
// Do not wait for the Secrets store to get synced since we cannot rely on
// users having granted RBAC permissions for this object.
// https://github.com/containous/traefik/issues/1784 should improve the
// situation here in the future.
for _, ns := range namespaces {
c.factories[ns].Core().V1().Secrets().Informer().AddEventHandler(eventHandler)
c.factories[ns].Start(stopCh)
}
2018-03-22 09:14:04 +00:00
return eventCh, nil
}
// GetIngresses returns all Ingresses for observed namespaces in the cluster.
2018-02-14 08:56:04 +00:00
func (c *clientImpl) GetIngresses() []*extensionsv1beta1.Ingress {
var result []*extensionsv1beta1.Ingress
2018-03-22 09:14:04 +00:00
for ns, factory := range c.factories {
ings, err := factory.Extensions().V1beta1().Ingresses().Lister().List(c.ingressLabelSelector)
2018-03-22 09:14:04 +00:00
if err != nil {
log.Errorf("Failed to list ingresses in namespace %s: %s", ns, err)
}
for _, ing := range ings {
result = append(result, ing)
}
}
return result
}
2018-05-18 12:12:03 +00:00
// UpdateIngressStatus updates an Ingress with a provided status.
func (c *clientImpl) UpdateIngressStatus(namespace, name, ip, hostname string) error {
keyName := namespace + "/" + name
item, exists, err := c.factories[c.lookupNamespace(namespace)].Extensions().V1beta1().Ingresses().Informer().GetStore().GetByKey(keyName)
if err != nil {
return fmt.Errorf("failed to get ingress %s with error: %v", keyName, err)
}
if !exists {
return fmt.Errorf("failed to update ingress %s because it does not exist", keyName)
}
ing := item.(*extensionsv1beta1.Ingress)
if ing.Status.LoadBalancer.Ingress[0].Hostname == hostname && ing.Status.LoadBalancer.Ingress[0].IP == ip {
// If status is already set, skip update
log.Debugf("Skipping status update on ingress %s/%s", ing.Namespace, ing.Name)
} else {
ingCopy := ing.DeepCopy()
ingCopy.Status = extensionsv1beta1.IngressStatus{LoadBalancer: corev1.LoadBalancerStatus{Ingress: []corev1.LoadBalancerIngress{{IP: ip, Hostname: hostname}}}}
_, err := c.clientset.ExtensionsV1beta1().Ingresses(ingCopy.Namespace).UpdateStatus(ingCopy)
if err != nil {
return fmt.Errorf("failed to update ingress status %s with error: %v", keyName, err)
}
log.Infof("Updated status on ingress %s", keyName)
}
return nil
}
// GetService returns the named service from the given namespace.
2018-02-14 08:56:04 +00:00
func (c *clientImpl) GetService(namespace, name string) (*corev1.Service, bool, error) {
var service *corev1.Service
2018-03-22 09:14:04 +00:00
item, exists, err := c.factories[c.lookupNamespace(namespace)].Core().V1().Services().Informer().GetStore().GetByKey(namespace + "/" + name)
if item != nil {
2018-02-14 08:56:04 +00:00
service = item.(*corev1.Service)
}
return service, exists, err
}
// GetEndpoints returns the named endpoints from the given namespace.
2018-02-14 08:56:04 +00:00
func (c *clientImpl) GetEndpoints(namespace, name string) (*corev1.Endpoints, bool, error) {
var endpoint *corev1.Endpoints
2018-03-22 09:14:04 +00:00
item, exists, err := c.factories[c.lookupNamespace(namespace)].Core().V1().Endpoints().Informer().GetStore().GetByKey(namespace + "/" + name)
if item != nil {
2018-02-14 08:56:04 +00:00
endpoint = item.(*corev1.Endpoints)
}
return endpoint, exists, err
}
// GetSecret returns the named secret from the given namespace.
2018-02-14 08:56:04 +00:00
func (c *clientImpl) GetSecret(namespace, name string) (*corev1.Secret, bool, error) {
var secret *corev1.Secret
2018-03-22 09:14:04 +00:00
item, exists, err := c.factories[c.lookupNamespace(namespace)].Core().V1().Secrets().Informer().GetStore().GetByKey(namespace + "/" + name)
if err == nil && item != nil {
2018-02-14 08:56:04 +00:00
secret = item.(*corev1.Secret)
}
return secret, exists, err
}
// lookupNamespace returns the lookup namespace key for the given namespace.
// When listening on all namespaces, it returns the client-go identifier ("")
// for all-namespaces. Otherwise, it returns the given namespace.
// The distinction is necessary because we index all informers on the special
// identifier iff all-namespaces are requested but receive specific namespace
// identifiers from the Kubernetes API, so we have to bridge this gap.
func (c *clientImpl) lookupNamespace(ns string) string {
if c.isNamespaceAll {
2018-02-14 08:56:04 +00:00
return metav1.NamespaceAll
}
return ns
}
func (c *clientImpl) newResourceEventHandler(events chan<- interface{}) cache.ResourceEventHandler {
return &cache.FilteringResourceEventHandler{
FilterFunc: func(obj interface{}) bool {
// Ignore Ingresses that do not match our custom label selector.
if ing, ok := obj.(*extensionsv1beta1.Ingress); ok {
lbls := labels.Set(ing.GetLabels())
return c.ingressLabelSelector.Matches(lbls)
}
return true
},
Handler: &resourceEventHandler{events},
}
}
// eventHandlerFunc will pass the obj on to the events channel or drop it.
// This is so passing the events along won't block in the case of high volume.
// The events are only used for signalling anyway so dropping a few is ok.
func eventHandlerFunc(events chan<- interface{}, obj interface{}) {
select {
case events <- obj:
default:
}
}