traefik/provider/k8s/client.go

package k8s

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	log "github.com/Sirupsen/logrus"
	"github.com/parnurzeal/gorequest"
	"net/http"
	"net/url"
	"strings"
)

const (
	// APIEndpoint defines the base path for kubernetes API resources.
	APIEndpoint        = "/api/v1"
	extentionsEndpoint = "/apis/extensions/v1beta1"
	defaultIngress     = "/ingresses"
	namespaces         = "/namespaces/"
)

// Client is a client for the Kubernetes master.
type Client interface {
	GetIngresses(labelSelector string, predicate func(Ingress) bool) ([]Ingress, error)
	GetService(name, namespace string) (Service, error)
	GetEndpoints(name, namespace string) (Endpoints, error)
	WatchAll(labelSelector string, stopCh <-chan bool) (chan interface{}, chan error, error)
}

type clientImpl struct {
	endpointURL string
	tls         *tls.Config
	token       string
	caCert      []byte
}

// NewClient returns a new Kubernetes client.
// The provided host is an url (scheme://hostname[:port]) of a
// Kubernetes master without any path.
// The provided client is an authorized http.Client used to perform requests to the Kubernetes API master.
func NewClient(baseURL string, caCert []byte, token string) (Client, error) {
	validURL, err := url.Parse(baseURL)
	if err != nil {
		return nil, fmt.Errorf("failed to parse URL %q: %v", baseURL, err)
	}
	return &clientImpl{
		endpointURL: strings.TrimSuffix(validURL.String(), "/"),
		token:       token,
		caCert:      caCert,
	}, nil
}

func makeQueryString(baseParams map[string]string, labelSelector string) (string, error) {
	if labelSelector != "" {
		baseParams["labelSelector"] = labelSelector
	}
	queryData, err := json.Marshal(baseParams)
	if err != nil {
		return "", err
	}
	return string(queryData), nil
}

// GetIngresses returns all ingresses in the cluster
func (c *clientImpl) GetIngresses(labelSelector string, predicate func(Ingress) bool) ([]Ingress, error) {
	getURL := c.endpointURL + extentionsEndpoint + defaultIngress
	queryParams := map[string]string{}
	queryData, err := makeQueryString(queryParams, labelSelector)
	if err != nil {
		return nil, fmt.Errorf("Had problems constructing query string %s : %v", queryParams, err)
	}
	body, err := c.do(c.request(getURL, queryData))
	if err != nil {
		return nil, fmt.Errorf("failed to create ingresses request: GET %q : %v", getURL, err)
	}

	var ingressList IngressList
	if err := json.Unmarshal(body, &ingressList); err != nil {
		return nil, fmt.Errorf("failed to decode list of ingress resources: %v", err)
	}
	ingresses := ingressList.Items[:0]
	for _, ingress := range ingressList.Items {
		if predicate(ingress) {
			ingresses = append(ingresses, ingress)
		}
	}
	return ingresses, nil
}

// WatchIngresses returns all ingresses in the cluster
func (c *clientImpl) WatchIngresses(labelSelector string, stopCh <-chan bool) (chan interface{}, chan error, error) {
	getURL := c.endpointURL + extentionsEndpoint + defaultIngress
	return c.watch(getURL, labelSelector, stopCh)
}

// GetService returns the named service from the named namespace
func (c *clientImpl) GetService(name, namespace string) (Service, error) {
	getURL := c.endpointURL + APIEndpoint + namespaces + namespace + "/services/" + name

	body, err := c.do(c.request(getURL, ""))
	if err != nil {
		return Service{}, fmt.Errorf("failed to create services request: GET %q : %v", getURL, err)
	}

	var service Service
	if err := json.Unmarshal(body, &service); err != nil {
		return Service{}, fmt.Errorf("failed to decode service resource: %v", err)
	}
	return service, nil
}

// WatchServices returns all services in the cluster
func (c *clientImpl) WatchServices(labelSelector string, stopCh <-chan bool) (chan interface{}, chan error, error) {
	getURL := c.endpointURL + APIEndpoint + "/services"
	return c.watch(getURL, labelSelector, stopCh)
}

// GetEndpoints returns the named Endpoints
// Endpoints have the same name as the coresponding service
func (c *clientImpl) GetEndpoints(name, namespace string) (Endpoints, error) {
	getURL := c.endpointURL + APIEndpoint + namespaces + namespace + "/endpoints/" + name

	body, err := c.do(c.request(getURL, ""))
	if err != nil {
		return Endpoints{}, fmt.Errorf("failed to create endpoints request: GET %q : %v", getURL, err)
	}

	var endpoints Endpoints
	if err := json.Unmarshal(body, &endpoints); err != nil {
		return Endpoints{}, fmt.Errorf("failed to decode endpoints resources: %v", err)
	}
	return endpoints, nil
}

// WatchEndpoints returns endpoints in the cluster
func (c *clientImpl) WatchEndpoints(labelSelector string, stopCh <-chan bool) (chan interface{}, chan error, error) {
	getURL := c.endpointURL + APIEndpoint + "/endpoints"
	return c.watch(getURL, labelSelector, stopCh)
}

// WatchAll returns events in the cluster
func (c *clientImpl) WatchAll(labelSelector string, stopCh <-chan bool) (chan interface{}, chan error, error) {
	watchCh := make(chan interface{}, 10)
	errCh := make(chan error, 10)

	stopIngresses := make(chan bool)
	chanIngresses, chanIngressesErr, err := c.WatchIngresses(labelSelector, stopIngresses)
	if err != nil {
		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
	}
	stopServices := make(chan bool)
	chanServices, chanServicesErr, err := c.WatchServices(labelSelector, stopServices)
	if err != nil {
		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
	}
	stopEndpoints := make(chan bool)
	chanEndpoints, chanEndpointsErr, err := c.WatchEndpoints(labelSelector, stopEndpoints)
	if err != nil {
		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
	}
	go func() {
		defer close(watchCh)
		defer close(errCh)
		defer close(stopIngresses)
		defer close(stopServices)
		defer close(stopEndpoints)

		for {
			select {
			case <-stopCh:
				stopIngresses <- true
				stopServices <- true
				stopEndpoints <- true
				return
			case err := <-chanIngressesErr:
				errCh <- err
			case err := <-chanServicesErr:
				errCh <- err
			case err := <-chanEndpointsErr:
				errCh <- err
			case event := <-chanIngresses:
				watchCh <- event
			case event := <-chanServices:
				watchCh <- event
			case event := <-chanEndpoints:
				watchCh <- event
			}
		}
	}()

	return watchCh, errCh, nil
}

func (c *clientImpl) do(request *gorequest.SuperAgent) ([]byte, error) {
	res, body, errs := request.EndBytes()
	if errs != nil {
		return nil, fmt.Errorf("failed to create request: GET %q : %v", request.Url, errs)
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("http error %d GET %q: %q", res.StatusCode, request.Url, string(body))
	}
	return body, nil
}

func (c *clientImpl) request(reqURL string, queryContent interface{}) *gorequest.SuperAgent {
	// Make request to Kubernetes API
	parsedURL, parseErr := url.Parse(reqURL)
	if parseErr != nil {
		log.Errorf("Had issues parsing url %s. Trying anyway.", reqURL)
	}
	request := gorequest.New().Get(reqURL)
	request.Transport.DisableKeepAlives = true

	if parsedURL.Scheme == "https" {
		pool := x509.NewCertPool()
		pool.AppendCertsFromPEM(c.caCert)
		c.tls = &tls.Config{RootCAs: pool}
		request.TLSClientConfig(c.tls)
	}
	if len(c.token) > 0 {
		request.Header["Authorization"] = "Bearer " + c.token
	}
	request.Query(queryContent)
	return request
}

// GenericObject generic object
type GenericObject struct {
	TypeMeta `json:",inline"`
	ListMeta `json:"metadata,omitempty"`
}

func (c *clientImpl) watch(url string, labelSelector string, stopCh <-chan bool) (chan interface{}, chan error, error) {
	watchCh := make(chan interface{}, 10)
	errCh := make(chan error, 10)

	// get version
	body, err := c.do(c.request(url, ""))
	if err != nil {
		return watchCh, errCh, fmt.Errorf("failed to do version request: GET %q : %v", url, err)
	}

	var generic GenericObject
	if err := json.Unmarshal(body, &generic); err != nil {
		return watchCh, errCh, fmt.Errorf("failed to decode version %v", err)
	}
	resourceVersion := generic.ResourceVersion
	queryParams := map[string]string{"watch": "", "resourceVersion": resourceVersion}
	queryData, err := makeQueryString(queryParams, labelSelector)
	if err != nil {
		return watchCh, errCh, fmt.Errorf("Unable to construct query args")
	}
	request := c.request(url, queryData)
	req, err := request.MakeRequest()
	if err != nil {
		return watchCh, errCh, fmt.Errorf("failed to make watch request: GET %q : %v", url, err)
	}
	request.Client.Transport = request.Transport

	res, err := request.Client.Do(req)
	if err != nil {
		return watchCh, errCh, fmt.Errorf("failed to do watch request: GET %q: %v", url, err)
	}

	go func() {
		finishCh := make(chan bool)
		defer close(finishCh)
		defer close(watchCh)
		defer close(errCh)
		go func() {
			defer res.Body.Close()
			for {
				var eventList interface{}
				if err := json.NewDecoder(res.Body).Decode(&eventList); err != nil {
					if !strings.Contains(err.Error(), "net/http: request canceled") {
						errCh <- fmt.Errorf("failed to decode watch event: GET %q : %v", url, err)
					}
					finishCh <- true
					return
				}
				watchCh <- eventList
			}
		}()
		select {
		case <-stopCh:
			go func() {
				request.Transport.CancelRequest(req)
			}()
			<-finishCh
			return
		}
	}()
	return watchCh, errCh, nil
}