Merge pull request #408 from errm/k8s-endpoints

Build backend config using the K8S endpoint resource.

Makefile

@@ -18,6 +18,7 @@ REPONAME := $(shell echo $(REPO) | tr '[:upper:]' '[:lower:]')
 TRAEFIK_IMAGE := $(if $(REPONAME),$(REPONAME),"containous/traefik")
 INTEGRATION_OPTS := $(if $(MAKE_DOCKER_HOST),-e "DOCKER_HOST=$(MAKE_DOCKER_HOST)", -v "/var/run/docker.sock:/var/run/docker.sock")
 
+DOCKER_BUILD_ARGS := $(if $(DOCKER_VERSION), "--build-arg=DOCKER_VERSION=$(DOCKER_VERSION)",)
 DOCKER_RUN_TRAEFIK := docker run $(INTEGRATION_OPTS) -it $(TRAEFIK_ENVS) $(TRAEFIK_MOUNT) "$(TRAEFIK_DEV_IMAGE)"
 
 print-%: ; @echo $*=$($*)
@@ -46,7 +47,7 @@ validate: build  ## validate gofmt, golint and go vet
 	$(DOCKER_RUN_TRAEFIK) ./script/make.sh  validate-gofmt validate-govet validate-golint 
 
 build: dist
-	docker build --build-arg=DOCKER_VERSION=${DOCKER_VERSION} -t "$(TRAEFIK_DEV_IMAGE)" -f build.Dockerfile .
+	docker build $(DOCKER_BUILD_ARGS) -t "$(TRAEFIK_DEV_IMAGE)" -f build.Dockerfile .
 
 build-webui:
 	docker build -t traefik-webui -f webui/Dockerfile webui

cmd.go

@@ -10,12 +10,13 @@ import (
 	"strings"
 	"time"
 
+	"net/http"
+
 	log "github.com/Sirupsen/logrus"
 	"github.com/containous/traefik/middlewares"
 	"github.com/containous/traefik/provider"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
-	"net/http"
 )
 
 var traefikCmd = &cobra.Command{
@@ -93,7 +94,8 @@ var arguments = struct {
 
 func init() {
 	traefikCmd.AddCommand(versionCmd)
-	traefikCmd.PersistentFlags().StringP("configFile", "c", "", "Configuration file to use (TOML, JSON, YAML, HCL).")
+	traefikCmd.PersistentFlags().StringP("configFile", "c", "", "Configuration file to use (TOML).")
+	traefikCmd.PersistentFlags().BoolVarP(&arguments.Debug, "debug", "d", false, "Enable debug mode")
 	traefikCmd.PersistentFlags().StringP("graceTimeOut", "g", "10", "Timeout in seconds. Duration to give active requests a chance to finish during hot-reloads")
 	traefikCmd.PersistentFlags().String("accessLogsFile", "log/access.log", "Access logs file")
 	traefikCmd.PersistentFlags().String("traefikLogsFile", "log/traefik.log", "Traefik logs file")
@@ -171,12 +173,13 @@ func init() {
 	traefikCmd.PersistentFlags().StringVar(&arguments.Boltdb.Prefix, "boltdb.prefix", "/traefik", "Prefix used for KV store")
 
 	traefikCmd.PersistentFlags().BoolVar(&arguments.kubernetes, "kubernetes", false, "Enable Kubernetes backend")
-	traefikCmd.PersistentFlags().StringVar(&arguments.Kubernetes.Endpoint, "kubernetes.endpoint", "127.0.0.1:8080", "Kubernetes server endpoint")
+	traefikCmd.PersistentFlags().StringVar(&arguments.Kubernetes.Endpoint, "kubernetes.endpoint", "http://127.0.0.1:8080", "Kubernetes server endpoint")
 	traefikCmd.PersistentFlags().StringSliceVar(&arguments.Kubernetes.Namespaces, "kubernetes.namespaces", []string{}, "Kubernetes namespaces")
 
 	_ = viper.BindPFlag("configFile", traefikCmd.PersistentFlags().Lookup("configFile"))
 	_ = viper.BindPFlag("graceTimeOut", traefikCmd.PersistentFlags().Lookup("graceTimeOut"))
 	_ = viper.BindPFlag("logLevel", traefikCmd.PersistentFlags().Lookup("logLevel"))
+	_ = viper.BindPFlag("debug", traefikCmd.PersistentFlags().Lookup("debug"))
 	// TODO: wait for this issue to be corrected: https://github.com/spf13/viper/issues/105
 	_ = viper.BindPFlag("providersThrottleDuration", traefikCmd.PersistentFlags().Lookup("providersThrottleDuration"))
 	_ = viper.BindPFlag("maxIdleConnsPerHost", traefikCmd.PersistentFlags().Lookup("maxIdleConnsPerHost"))

configuration.go

@@ -19,6 +19,7 @@ import (
 // It's populated from the traefik configuration file passed as an argument to the binary.
 type GlobalConfiguration struct {
 	GraceTimeOut              int64
+	Debug                     bool
 	AccessLogsFile            string
 	TraefikLogsFile           string
 	LogLevel                  string

docs/benchmarks.md

@@ -146,7 +146,7 @@ defaultEntryPoints = ["http"]
 
 ### whoami:
 ```
-wrk -t8 -c1000 -d60s -H "Host: test.traefik" --latency  http://IP-whoami:80/bench
+wrk -t20 -c1000 -d60s -H "Host: test.traefik" --latency  http://IP-whoami:80/bench
 Running 1m test @ http://IP-whoami:80/bench
   20 threads and 1000 connections
   Thread Stats   Avg      Stdev     Max   +/- Stdev
@@ -184,7 +184,7 @@ Transfer/sec:      4.97MB
 
 ### traefik:
 ```
-wrk -t8 -c1000 -d60s -H "Host: test.traefik" --latency  http://IP-traefik:8000/bench
+wrk -t20 -c1000 -d60s -H "Host: test.traefik" --latency  http://IP-traefik:8000/bench
 Running 1m test @ http://IP-traefik:8000/bench
   20 threads and 1000 connections
   Thread Stats   Avg      Stdev     Max   +/- Stdev

docs/toml.md

@@ -934,7 +934,7 @@ The Keys-Values structure should look (using `prefix = "/traefik"`):
 | `/traefik/frontends/frontend2/backend`             | `backend1`   |
 | `/traefik/frontends/frontend2/passHostHeader`      | `true`       |
 | `/traefik/frontends/frontend2/entrypoints`         | `http,https` |
-| `/traefik/frontends/frontend2/routes/test_2/rule`  | `Path:/test` |
+| `/traefik/frontends/frontend2/routes/test_2/rule`  | `PathPrefix:/test` |
 
 ## Atomic configuration changes
 

examples/compose-k8s.yaml

@@ -1,8 +1,3 @@
-# etcd:
-#   image: gcr.io/google_containers/etcd:2.2.1
-#   net: host
-#   command: ['/usr/local/bin/etcd', '--addr=127.0.0.1:4001', '--bind-addr=0.0.0.0:4001', '--data-dir=/var/etcd/data']
-
 kubelet:
   image: gcr.io/google_containers/hyperkube-amd64:v1.2.2
   privileged: true

glide.lock

@@ -1,5 +1,5 @@
-hash: da7239dce8bda69f6e10b2f2bfae57dd4fd95b817055dca1379a72af42939b97
-updated: 2016-05-12T11:48:22.158455011+02:00
+hash: 68bc4f87206f9a486e1455f1dfcad737369c359a803566271432fcb85de3a12c
+updated: 2016-05-23T13:57:35.191541555+02:00
 imports:
 - name: github.com/alecthomas/template
   version: b867cc6ab45cece8143cfcc6fc9c77cf3f2c23c0
@@ -22,7 +22,7 @@ imports:
 - name: github.com/codegangsta/negroni
   version: c7477ad8e330bef55bf1ebe300cf8aa67c492d1b
 - name: github.com/containous/oxy
-  version: 021f82bd8260ba15f5862a9fe62018437720dff5
+  version: 183212964e13e7b8afe01a08b193d04300554a68
   subpackages:
   - cbreaker
   - forward
@@ -86,7 +86,7 @@ imports:
   - utils
   - volume
 - name: github.com/docker/engine-api
-  version: fd7f99d354831e7e809386087e7ec3129fdb1520
+  version: 3d3d0b6c9d2651aac27f416a6da0224c1875b3eb
   subpackages:
   - client
   - types
@@ -103,7 +103,7 @@ imports:
   - types/versions
   - types/blkiodev
 - name: github.com/docker/go-connections
-  version: 5b7154ba2efe13ff86ae8830a9e7cb120b080d6e
+  version: c7838b258fbfa3fe88eecfb2a0e08ea0dbd6a646
   subpackages:
   - nat
   - sockets
@@ -168,7 +168,7 @@ imports:
 - name: github.com/kr/text
   version: 7cafcd837844e784b526369c9bce262804aebc60
 - name: github.com/libkermit/docker
-  version: 9f5a90f8eb3c49bf56e81621f98b3cd86fe4139f
+  version: 3b5eb2973efff7af33cfb65141deaf4ed25c6d02
 - name: github.com/libkermit/docker-check
   version: bb75a86b169c6c5d22c0ee98278124036f272d7b
 - name: github.com/magiconair/properties
@@ -184,19 +184,19 @@ imports:
 - name: github.com/mattn/go-shellwords
   version: 525bedee691b5a8df547cb5cf9f86b7fb1883e24
 - name: github.com/Microsoft/go-winio
-  version: 3b8b3c98b207f95fe0cd6c7c311a9ac497ba7c0f
+  version: 4f1a71750d95a5a8a46c40a67ffbed8129c2f138
 - name: github.com/miekg/dns
   version: 48ab6605c66ac797e07f615101c3e9e10e932b66
 - name: github.com/mitchellh/mapstructure
   version: d2dd0262208475919e1a362f675cfc0e7c10e905
 - name: github.com/moul/http2curl
-  version: 1812aee76a1ce98d604a44200c6a23c689b17a89
+  version: b1479103caacaa39319f75e7f57fc545287fca0d
 - name: github.com/opencontainers/runc
   version: 2441732d6fcc0fb0a542671a4372e0c7bc99c19e
   subpackages:
   - libcontainer/user
 - name: github.com/parnurzeal/gorequest
-  version: a39a2f8d0463091df7344dbf586a9986e9f7184f
+  version: 2169dfca686cfcbefc983a98a25e9c22a2815be4
 - name: github.com/pmezard/go-difflib
   version: d8ed2627bdf02c080bf22230dbb337003b7aba2d
   subpackages:
@@ -210,7 +210,7 @@ imports:
 - name: github.com/spf13/cast
   version: ee7b3e0353166ab1f3a605294ac8cd2b77953778
 - name: github.com/spf13/cobra
-  version: 0f866a6211e33cde2091d9290c08f6afd6c9ebbc
+  version: f368244301305f414206f889b1735a54cfc8bde8
   subpackages:
   - cobra
 - name: github.com/spf13/jwalterweatherman
@@ -237,7 +237,7 @@ imports:
 - name: github.com/unrolled/render
   version: 26b4e3aac686940fe29521545afad9966ddfc80c
 - name: github.com/vdemeester/docker-events
-  version: ce5347b72aafad4e3bebd966f15e4183839d5172
+  version: b308d2e8d639d928c882913bcb4f85b3a84c7a07
 - name: github.com/vdemeester/shakers
   version: 24d7f1d6a71aa5d9cbe7390e4afb66b7eef9e1b3
 - name: github.com/vulcand/oxy
@@ -258,11 +258,11 @@ imports:
 - name: github.com/wendal/errors
   version: f66c77a7882b399795a8987ebf87ef64a427417e
 - name: github.com/xenolf/lego
-  version: 948483535f53c34d144419869ecbed86251a30f6
+  version: b119bc45fbd1cc71348003541aac9d3a7da63654
   subpackages:
   - acme
 - name: golang.org/x/crypto
-  version: b76c864ef1dca1d8f271f917c290cddcce3d9e0d
+  version: 5bcd134fee4dd1475da17714aac19c0aa0142e2f
   subpackages:
   - ocsp
 - name: golang.org/x/net
@@ -275,6 +275,7 @@ imports:
   version: eb2c74142fd19a79b3f237334c7384d5167b1b46
   subpackages:
   - unix
+  - windows
 - name: gopkg.in/alecthomas/kingpin.v2
   version: 639879d6110b1b0409410c7b737ef0bb18325038
 - name: gopkg.in/fsnotify.v1

glide.yaml

@@ -7,7 +7,7 @@ import:
 - package: github.com/mailgun/log
   version: 44874009257d4d47ba9806f1b7f72a32a015e4d8
 - package: github.com/containous/oxy
-  version: 021f82bd8260ba15f5862a9fe62018437720dff5
+  version: 183212964e13e7b8afe01a08b193d04300554a68
   subpackages:
   - cbreaker
   - forward
@@ -160,7 +160,7 @@ import:
 - package: github.com/libkermit/docker-check
   version: bb75a86b169c6c5d22c0ee98278124036f272d7b
 - package: github.com/libkermit/docker
-  version: 9f5a90f8eb3c49bf56e81621f98b3cd86fe4139f
+  version: 3b5eb2973efff7af33cfb65141deaf4ed25c6d02
 - package: github.com/docker/libcompose
   version: 8ee7bcc364f7b8194581a3c6bd9fa019467c7873
 - package: github.com/docker/distribution
@@ -168,7 +168,7 @@ import:
   subpackages:
   - reference
 - package: github.com/docker/engine-api
-  version: fd7f99d354831e7e809386087e7ec3129fdb1520
+  version: 3d3d0b6c9d2651aac27f416a6da0224c1875b3eb
   subpackages:
   - client
   - types

provider/k8s/client.go

@@ -5,7 +5,6 @@ import (
 	"crypto/x509"
 	"encoding/json"
 	"fmt"
-	"github.com/containous/traefik/safe"
 	"github.com/parnurzeal/gorequest"
 	"net/http"
 	"net/url"
@@ -17,12 +16,14 @@ const (
 	APIEndpoint        = "/api/v1"
 	extentionsEndpoint = "/apis/extensions/v1beta1"
 	defaultIngress     = "/ingresses"
+	namespaces         = "/namespaces/"
 )
 
 // Client is a client for the Kubernetes master.
 type Client interface {
 	GetIngresses(predicate func(Ingress) bool) ([]Ingress, error)
-	GetServices(predicate func(Service) bool) ([]Service, error)
+	GetService(name, namespace string) (Service, error)
+	GetEndpoints(name, namespace string) (Endpoints, error)
 	WatchAll(stopCh <-chan bool) (chan interface{}, chan error, error)
 }
 
@@ -77,26 +78,20 @@ func (c *clientImpl) WatchIngresses(stopCh <-chan bool) (chan interface{}, chan
 	return c.watch(getURL, stopCh)
 }
 
-// GetServices returns all services in the cluster
-func (c *clientImpl) GetServices(predicate func(Service) bool) ([]Service, error) {
-	getURL := c.endpointURL + APIEndpoint + "/services"
+// 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 nil, fmt.Errorf("failed to create services request: GET %q : %v", getURL, err)
+		return Service{}, fmt.Errorf("failed to create services request: GET %q : %v", getURL, err)
 	}
 
-	var serviceList ServiceList
-	if err := json.Unmarshal(body, &serviceList); err != nil {
-		return nil, fmt.Errorf("failed to decode list of services resources: %v", err)
+	var service Service
+	if err := json.Unmarshal(body, &service); err != nil {
+		return Service{}, fmt.Errorf("failed to decode service resource: %v", err)
 	}
-	services := serviceList.Items[:0]
-	for _, service := range serviceList.Items {
-		if predicate(service) {
-			services = append(services, service)
-		}
-	}
-	return services, nil
+	return service, nil
 }
 
 // WatchServices returns all services in the cluster
@@ -105,28 +100,33 @@ func (c *clientImpl) WatchServices(stopCh <-chan bool) (chan interface{}, chan e
 	return c.watch(getURL, stopCh)
 }
 
-// WatchEvents returns events in the cluster
-func (c *clientImpl) WatchEvents(stopCh <-chan bool) (chan interface{}, chan error, error) {
-	getURL := c.endpointURL + APIEndpoint + "/events"
-	return c.watch(getURL, 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
 }
 
-// WatchPods returns pods in the cluster
-func (c *clientImpl) WatchPods(stopCh <-chan bool) (chan interface{}, chan error, error) {
-	getURL := c.endpointURL + APIEndpoint + "/pods"
-	return c.watch(getURL, stopCh)
-}
-
-// WatchReplicationControllers returns ReplicationControllers in the cluster
-func (c *clientImpl) WatchReplicationControllers(stopCh <-chan bool) (chan interface{}, chan error, error) {
-	getURL := c.endpointURL + APIEndpoint + "/replicationcontrollers"
+// WatchEndpoints returns endpoints in the cluster
+func (c *clientImpl) WatchEndpoints(stopCh <-chan bool) (chan interface{}, chan error, error) {
+	getURL := c.endpointURL + APIEndpoint + "/endpoints"
 	return c.watch(getURL, stopCh)
 }
 
 // WatchAll returns events in the cluster
 func (c *clientImpl) WatchAll(stopCh <-chan bool) (chan interface{}, chan error, error) {
-	watchCh := make(chan interface{})
-	errCh := make(chan error)
+	watchCh := make(chan interface{}, 10)
+	errCh := make(chan error, 10)
 
 	stopIngresses := make(chan bool)
 	chanIngresses, chanIngressesErr, err := c.WatchIngresses(stopIngresses)
@@ -138,13 +138,8 @@ func (c *clientImpl) WatchAll(stopCh <-chan bool) (chan interface{}, chan error,
 	if err != nil {
 		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
 	}
-	stopPods := make(chan bool)
-	chanPods, chanPodsErr, err := c.WatchPods(stopPods)
-	if err != nil {
-		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
-	}
-	stopReplicationControllers := make(chan bool)
-	chanReplicationControllers, chanReplicationControllersErr, err := c.WatchReplicationControllers(stopReplicationControllers)
+	stopEndpoints := make(chan bool)
+	chanEndpoints, chanEndpointsErr, err := c.WatchEndpoints(stopEndpoints)
 	if err != nil {
 		return watchCh, errCh, fmt.Errorf("failed to create watch: %v", err)
 	}
@@ -153,32 +148,26 @@ func (c *clientImpl) WatchAll(stopCh <-chan bool) (chan interface{}, chan error,
 		defer close(errCh)
 		defer close(stopIngresses)
 		defer close(stopServices)
-		defer close(stopPods)
-		defer close(stopReplicationControllers)
+		defer close(stopEndpoints)
 
 		for {
 			select {
 			case <-stopCh:
 				stopIngresses <- true
 				stopServices <- true
-				stopPods <- true
-				stopReplicationControllers <- true
-				break
+				stopEndpoints <- true
+				return
 			case err := <-chanIngressesErr:
 				errCh <- err
 			case err := <-chanServicesErr:
 				errCh <- err
-			case err := <-chanPodsErr:
-				errCh <- err
-			case err := <-chanReplicationControllersErr:
+			case err := <-chanEndpointsErr:
 				errCh <- err
 			case event := <-chanIngresses:
 				watchCh <- event
 			case event := <-chanServices:
 				watchCh <- event
-			case event := <-chanPods:
-				watchCh <- event
-			case event := <-chanReplicationControllers:
+			case event := <-chanEndpoints:
 				watchCh <- event
 			}
 		}
@@ -192,6 +181,7 @@ func (c *clientImpl) do(request *gorequest.SuperAgent) ([]byte, error) {
 	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))
 	}
@@ -201,6 +191,12 @@ func (c *clientImpl) do(request *gorequest.SuperAgent) ([]byte, error) {
 func (c *clientImpl) request(url string) *gorequest.SuperAgent {
 	// Make request to Kubernetes API
 	request := gorequest.New().Get(url)
+	request.Transport.DisableKeepAlives = true
+
+	if strings.HasPrefix(url, "http://") {
+		return request
+	}
+
 	if len(c.token) > 0 {
 		request.Header["Authorization"] = "Bearer " + c.token
 		pool := x509.NewCertPool()
@@ -217,8 +213,8 @@ type GenericObject struct {
 }
 
 func (c *clientImpl) watch(url string, stopCh <-chan bool) (chan interface{}, chan error, error) {
-	watchCh := make(chan interface{})
-	errCh := make(chan error)
+	watchCh := make(chan interface{}, 10)
+	errCh := make(chan error, 10)
 
 	// get version
 	body, err := c.do(c.request(url))
@@ -240,34 +236,38 @@ func (c *clientImpl) watch(url string, stopCh <-chan bool) (chan interface{}, ch
 		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)
 	}
 
-	shouldStop := safe.New(false)
-
-	go func() {
-		select {
-		case <-stopCh:
-			shouldStop.Set(true)
-			res.Body.Close()
-			return
-		}
-	}()
-
 	go func() {
+		finishCh := make(chan bool)
+		defer close(finishCh)
 		defer close(watchCh)
 		defer close(errCh)
-		for {
-			var eventList interface{}
-			if err := json.NewDecoder(res.Body).Decode(&eventList); err != nil {
-				if !shouldStop.Get().(bool) {
-					errCh <- fmt.Errorf("failed to decode watch event: %v", err)
+		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
 				}
-				return
+				watchCh <- eventList
 			}
-			watchCh <- eventList
+		}()
+		select {
+		case <-stopCh:
+			go func() {
+				request.Transport.CancelRequest(req)
+			}()
+			<-finishCh
+			return
 		}
 	}()
 	return watchCh, errCh, nil

provider/k8s/endpoints.go

@@ -0,0 +1,84 @@
+package k8s
+
+// Endpoints is a collection of endpoints that implement the actual service.  Example:
+//   Name: "mysvc",
+//   Subsets: [
+//     {
+//       Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//       Ports: [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//     },
+//     {
+//       Addresses: [{"ip": "10.10.3.3"}],
+//       Ports: [{"name": "a", "port": 93}, {"name": "b", "port": 76}]
+//     },
+//  ]
+type Endpoints struct {
+	TypeMeta   `json:",inline"`
+	ObjectMeta `json:"metadata,omitempty"`
+
+	// The set of all endpoints is the union of all subsets.
+	Subsets []EndpointSubset
+}
+
+// EndpointSubset is a group of addresses with a common set of ports.  The
+// expanded set of endpoints is the Cartesian product of Addresses x Ports.
+// For example, given:
+//   {
+//     Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//     Ports:     [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//   }
+// The resulting set of endpoints can be viewed as:
+//     a: [ 10.10.1.1:8675, 10.10.2.2:8675 ],
+//     b: [ 10.10.1.1:309, 10.10.2.2:309 ]
+type EndpointSubset struct {
+	Addresses         []EndpointAddress
+	NotReadyAddresses []EndpointAddress
+	Ports             []EndpointPort
+}
+
+// EndpointAddress is a tuple that describes single IP address.
+type EndpointAddress struct {
+	// The IP of this endpoint.
+	// IPv6 is also accepted but not fully supported on all platforms. Also, certain
+	// kubernetes components, like kube-proxy, are not IPv6 ready.
+	// TODO: This should allow hostname or IP, see #4447.
+	IP string
+	// Optional: Hostname of this endpoint
+	// Meant to be used by DNS servers etc.
+	Hostname string `json:"hostname,omitempty"`
+	// Optional: The kubernetes object related to the entry point.
+	TargetRef *ObjectReference
+}
+
+// EndpointPort is a tuple that describes a single port.
+type EndpointPort struct {
+	// The name of this port (corresponds to ServicePort.Name).  Optional
+	// if only one port is defined.  Must be a DNS_LABEL.
+	Name string
+
+	// The port number.
+	Port int32
+
+	// The IP protocol for this port.
+	Protocol Protocol
+}
+
+// ObjectReference contains enough information to let you inspect or modify the referred object.
+type ObjectReference struct {
+	Kind            string `json:"kind,omitempty"`
+	Namespace       string `json:"namespace,omitempty"`
+	Name            string `json:"name,omitempty"`
+	UID             UID    `json:"uid,omitempty"`
+	APIVersion      string `json:"apiVersion,omitempty"`
+	ResourceVersion string `json:"resourceVersion,omitempty"`
+
+	// Optional. If referring to a piece of an object instead of an entire object, this string
+	// should contain information to identify the sub-object. For example, if the object
+	// reference is to a container within a pod, this would take on a value like:
+	// "spec.containers{name}" (where "name" refers to the name of the container that triggered
+	// the event) or if no container name is specified "spec.containers[2]" (container with
+	// index 2 in this pod). This syntax is chosen only to have some well-defined way of
+	// referencing a part of an object.
+	// TODO: this design is not final and this field is subject to change in the future.
+	FieldPath string `json:"fieldPath,omitempty"`
+}

provider/kubernetes.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"io/ioutil"
 	"os"
+	"strconv"
 	"strings"
 	"text/template"
 	"time"
@@ -61,19 +62,20 @@ func (provider *Kubernetes) Provide(configurationChan chan<- types.ConfigMessage
 	backOff := backoff.NewExponentialBackOff()
 
 	pool.Go(func(stop chan bool) {
-		stopWatch := make(chan bool)
-		defer close(stopWatch)
 		operation := func() error {
-			select {
-			case <-stop:
-				return nil
-			default:
-			}
 			for {
+				stopWatch := make(chan bool, 5)
+				defer close(stopWatch)
 				eventsChan, errEventsChan, err := k8sClient.WatchAll(stopWatch)
 				if err != nil {
 					log.Errorf("Error watching kubernetes events: %v", err)
-					return err
+					timer := time.NewTimer(1 * time.Second)
+					select {
+					case <-timer.C:
+						return err
+					case <-stop:
+						return nil
+					}
 				}
 			Watch:
 				for {
@@ -81,14 +83,15 @@ func (provider *Kubernetes) Provide(configurationChan chan<- types.ConfigMessage
 					case <-stop:
 						stopWatch <- true
 						return nil
-					case err := <-errEventsChan:
-						if strings.Contains(err.Error(), io.EOF.Error()) {
+					case err, ok := <-errEventsChan:
+						stopWatch <- true
+						if ok && strings.Contains(err.Error(), io.EOF.Error()) {
 							// edge case, kubernetes long-polling disconnection
 							break Watch
 						}
 						return err
 					case event := <-eventsChan:
-						log.Debugf("Received event from kubenetes %+v", event)
+						log.Debugf("Received event from kubernetes %+v", event)
 						templateObjects, err := provider.loadIngresses(k8sClient)
 						if err != nil {
 							return err
@@ -159,9 +162,11 @@ func (provider *Kubernetes) loadIngresses(k8sClient k8s.Client) (*types.Configur
 						Routes:         make(map[string]types.Route),
 					}
 				}
-				if _, exists := templateObjects.Frontends[r.Host+pa.Path].Routes[r.Host]; !exists {
-					templateObjects.Frontends[r.Host+pa.Path].Routes[r.Host] = types.Route{
-						Rule: "Host:" + r.Host,
+				if len(r.Host) > 0 {
+					if _, exists := templateObjects.Frontends[r.Host+pa.Path].Routes[r.Host]; !exists {
+						templateObjects.Frontends[r.Host+pa.Path].Routes[r.Host] = types.Route{
+							Rule: "Host:" + r.Host,
+						}
 					}
 				}
 				if len(pa.Path) > 0 {
@@ -185,31 +190,42 @@ func (provider *Kubernetes) loadIngresses(k8sClient k8s.Client) (*types.Configur
 						Rule: ruleType + ":" + pa.Path,
 					}
 				}
-				services, err := k8sClient.GetServices(func(service k8s.Service) bool {
-					return service.Name == pa.Backend.ServiceName
-				})
+				service, err := k8sClient.GetService(pa.Backend.ServiceName, i.ObjectMeta.Namespace)
 				if err != nil {
-					log.Errorf("Error retrieving services: %v", err)
+					log.Warnf("Error retrieving services: %v", err)
+					delete(templateObjects.Frontends, r.Host+pa.Path)
+					log.Warnf("Error retrieving services %s", pa.Backend.ServiceName)
 					continue
 				}
-				if len(services) == 0 {
-					// no backends found, delete frontend...
-					delete(templateObjects.Frontends, r.Host+pa.Path)
-					log.Errorf("Error retrieving services %s", pa.Backend.ServiceName)
-				}
-				for _, service := range services {
-					protocol := "http"
-					for _, port := range service.Spec.Ports {
-						if port.Port == pa.Backend.ServicePort.IntValue() {
-							if port.Port == 443 {
-								protocol = "https"
-							}
+				protocol := "http"
+				for _, port := range service.Spec.Ports {
+					if equalPorts(port, pa.Backend.ServicePort) {
+						if port.Port == 443 {
+							protocol = "https"
+						}
+						endpoints, err := k8sClient.GetEndpoints(service.ObjectMeta.Name, service.ObjectMeta.Namespace)
+						if err != nil {
+							log.Errorf("Error retrieving endpoints: %v", err)
+							continue
+						}
+						if len(endpoints.Subsets) == 0 {
+							log.Warnf("Endpoints not found for %s/%s, falling back to Service ClusterIP", service.ObjectMeta.Namespace, service.ObjectMeta.Name)
 							templateObjects.Backends[r.Host+pa.Path].Servers[string(service.UID)] = types.Server{
-								URL:    protocol + "://" + service.Spec.ClusterIP + ":" + pa.Backend.ServicePort.String(),
+								URL:    protocol + "://" + service.Spec.ClusterIP + ":" + strconv.Itoa(port.Port),
 								Weight: 1,
 							}
-							break
+						} else {
+							for _, subset := range endpoints.Subsets {
+								for _, address := range subset.Addresses {
+									url := protocol + "://" + address.IP + ":" + strconv.Itoa(endpointPortNumber(port, subset.Ports))
+									templateObjects.Backends[r.Host+pa.Path].Servers[url] = types.Server{
+										URL:    url,
+										Weight: 1,
+									}
+								}
+							}
 						}
+						break
 					}
 				}
 			}
@@ -218,6 +234,30 @@ func (provider *Kubernetes) loadIngresses(k8sClient k8s.Client) (*types.Configur
 	return &templateObjects, nil
 }
 
+func endpointPortNumber(servicePort k8s.ServicePort, endpointPorts []k8s.EndpointPort) int {
+	if len(endpointPorts) > 0 {
+		//name is optional if there is only one port
+		port := endpointPorts[0]
+		for _, endpointPort := range endpointPorts {
+			if servicePort.Name == endpointPort.Name {
+				port = endpointPort
+			}
+		}
+		return int(port.Port)
+	}
+	return servicePort.Port
+}
+
+func equalPorts(servicePort k8s.ServicePort, ingressPort k8s.IntOrString) bool {
+	if servicePort.Port == ingressPort.IntValue() {
+		return true
+	}
+	if servicePort.Name != "" && servicePort.Name == ingressPort.String() {
+		return true
+	}
+	return false
+}
+
 func (provider *Kubernetes) getPassHostHeader() bool {
 	if provider.disablePassHostHeaders {
 		return false

provider/kubernetes_test.go

@@ -10,6 +10,9 @@ import (
 
 func TestLoadIngresses(t *testing.T) {
 	ingresses := []k8s.Ingress{{
+		ObjectMeta: k8s.ObjectMeta{
+			Namespace: "testing",
+		},
 		Spec: k8s.IngressSpec{
 			Rules: []k8s.IngressRule{
 				{
@@ -21,7 +24,7 @@ func TestLoadIngresses(t *testing.T) {
 									Path: "/bar",
 									Backend: k8s.IngressBackend{
 										ServiceName: "service1",
-										ServicePort: k8s.FromInt(801),
+										ServicePort: k8s.FromInt(80),
 									},
 								},
 							},
@@ -36,7 +39,7 @@ func TestLoadIngresses(t *testing.T) {
 								{
 									Backend: k8s.IngressBackend{
 										ServiceName: "service3",
-										ServicePort: k8s.FromInt(443),
+										ServicePort: k8s.FromString("https"),
 									},
 								},
 								{
@@ -55,23 +58,24 @@ func TestLoadIngresses(t *testing.T) {
 	services := []k8s.Service{
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service1",
-				UID:  "1",
+				Name:      "service1",
+				UID:       "1",
+				Namespace: "testing",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.1",
 				Ports: []k8s.ServicePort{
 					{
-						Name: "http",
-						Port: 801,
+						Port: 80,
 					},
 				},
 			},
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service2",
-				UID:  "2",
+				Name:      "service2",
+				UID:       "2",
+				Namespace: "testing",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.2",
@@ -84,24 +88,108 @@ func TestLoadIngresses(t *testing.T) {
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service3",
-				UID:  "3",
+				Name:      "service3",
+				UID:       "3",
+				Namespace: "testing",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.3",
 				Ports: []k8s.ServicePort{
 					{
 						Name: "http",
+						Port: 80,
+					},
+					{
+						Name: "https",
 						Port: 443,
 					},
 				},
 			},
 		},
 	}
+	endpoints := []k8s.Endpoints{
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Name:      "service1",
+				UID:       "1",
+				Namespace: "testing",
+			},
+			Subsets: []k8s.EndpointSubset{
+				{
+					Addresses: []k8s.EndpointAddress{
+						{
+							IP: "10.10.0.1",
+						},
+					},
+					Ports: []k8s.EndpointPort{
+						{
+							Port: 8080,
+						},
+					},
+				},
+				{
+					Addresses: []k8s.EndpointAddress{
+						{
+							IP: "10.21.0.1",
+						},
+					},
+					Ports: []k8s.EndpointPort{
+						{
+							Port: 8080,
+						},
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Name:      "service3",
+				UID:       "3",
+				Namespace: "testing",
+			},
+			Subsets: []k8s.EndpointSubset{
+				{
+					Addresses: []k8s.EndpointAddress{
+						{
+							IP: "10.15.0.1",
+						},
+					},
+					Ports: []k8s.EndpointPort{
+						{
+							Name: "http",
+							Port: 8080,
+						},
+						{
+							Name: "https",
+							Port: 8443,
+						},
+					},
+				},
+				{
+					Addresses: []k8s.EndpointAddress{
+						{
+							IP: "10.15.0.2",
+						},
+					},
+					Ports: []k8s.EndpointPort{
+						{
+							Name: "http",
+							Port: 9080,
+						},
+						{
+							Name: "https",
+							Port: 9443,
+						},
+					},
+				},
+			},
+		},
+	}
 	watchChan := make(chan interface{})
 	client := clientMock{
 		ingresses: ingresses,
 		services:  services,
+		endpoints: endpoints,
 		watchChan: watchChan,
 	}
 	provider := Kubernetes{}
@@ -114,8 +202,12 @@ func TestLoadIngresses(t *testing.T) {
 		Backends: map[string]*types.Backend{
 			"foo/bar": {
 				Servers: map[string]types.Server{
-					"1": {
-						URL:    "http://10.0.0.1:801",
+					"http://10.10.0.1:8080": {
+						URL:    "http://10.10.0.1:8080",
+						Weight: 1,
+					},
+					"http://10.21.0.1:8080": {
+						URL:    "http://10.21.0.1:8080",
 						Weight: 1,
 					},
 				},
@@ -128,8 +220,12 @@ func TestLoadIngresses(t *testing.T) {
 						URL:    "http://10.0.0.2:802",
 						Weight: 1,
 					},
-					"3": {
-						URL:    "https://10.0.0.3:443",
+					"https://10.15.0.1:8443": {
+						URL:    "https://10.15.0.1:8443",
+						Weight: 1,
+					},
+					"https://10.15.0.2:9443": {
+						URL:    "https://10.15.0.2:9443",
 						Weight: 1,
 					},
 				},
@@ -394,6 +490,9 @@ func TestRuleType(t *testing.T) {
 
 func TestGetPassHostHeader(t *testing.T) {
 	ingresses := []k8s.Ingress{{
+		ObjectMeta: k8s.ObjectMeta{
+			Namespace: "awesome",
+		},
 		Spec: k8s.IngressSpec{
 			Rules: []k8s.IngressRule{
 				{
@@ -418,8 +517,9 @@ func TestGetPassHostHeader(t *testing.T) {
 	services := []k8s.Service{
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service1",
-				UID:  "1",
+				Name:      "service1",
+				Namespace: "awesome",
+				UID:       "1",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.1",
@@ -479,6 +579,112 @@ func TestGetPassHostHeader(t *testing.T) {
 	}
 }
 
+func TestOnlyReferencesServicesFromOwnNamespace(t *testing.T) {
+	ingresses := []k8s.Ingress{
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Namespace: "awesome",
+			},
+			Spec: k8s.IngressSpec{
+				Rules: []k8s.IngressRule{
+					{
+						Host: "foo",
+						IngressRuleValue: k8s.IngressRuleValue{
+							HTTP: &k8s.HTTPIngressRuleValue{
+								Paths: []k8s.HTTPIngressPath{
+									{
+										Backend: k8s.IngressBackend{
+											ServiceName: "service",
+											ServicePort: k8s.FromInt(80),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	services := []k8s.Service{
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Name:      "service",
+				UID:       "1",
+				Namespace: "awesome",
+			},
+			Spec: k8s.ServiceSpec{
+				ClusterIP: "10.0.0.1",
+				Ports: []k8s.ServicePort{
+					{
+						Name: "http",
+						Port: 80,
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Name:      "service",
+				UID:       "2",
+				Namespace: "not-awesome",
+			},
+			Spec: k8s.ServiceSpec{
+				ClusterIP: "10.0.0.2",
+				Ports: []k8s.ServicePort{
+					{
+						Name: "http",
+						Port: 80,
+					},
+				},
+			},
+		},
+	}
+	watchChan := make(chan interface{})
+	client := clientMock{
+		ingresses: ingresses,
+		services:  services,
+		watchChan: watchChan,
+	}
+	provider := Kubernetes{}
+	actual, err := provider.loadIngresses(client)
+	if err != nil {
+		t.Fatalf("error %+v", err)
+	}
+
+	expected := &types.Configuration{
+		Backends: map[string]*types.Backend{
+			"foo": {
+				Servers: map[string]types.Server{
+					"1": {
+						URL:    "http://10.0.0.1:80",
+						Weight: 1,
+					},
+				},
+				CircuitBreaker: nil,
+				LoadBalancer:   nil,
+			},
+		},
+		Frontends: map[string]*types.Frontend{
+			"foo": {
+				Backend:        "foo",
+				PassHostHeader: true,
+				Routes: map[string]types.Route{
+					"foo": {
+						Rule: "Host:foo",
+					},
+				},
+			},
+		},
+	}
+	actualJSON, _ := json.Marshal(actual)
+	expectedJSON, _ := json.Marshal(expected)
+
+	if !reflect.DeepEqual(actual, expected) {
+		t.Fatalf("expected %+v, got %+v", string(expectedJSON), string(actualJSON))
+	}
+}
+
 func TestLoadNamespacedIngresses(t *testing.T) {
 	ingresses := []k8s.Ingress{
 		{
@@ -556,8 +762,9 @@ func TestLoadNamespacedIngresses(t *testing.T) {
 	services := []k8s.Service{
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service1",
-				UID:  "1",
+				Namespace: "awesome",
+				Name:      "service1",
+				UID:       "1",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.1",
@@ -571,8 +778,25 @@ func TestLoadNamespacedIngresses(t *testing.T) {
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service2",
-				UID:  "2",
+				Name:      "service1",
+				Namespace: "not-awesome",
+				UID:       "1",
+			},
+			Spec: k8s.ServiceSpec{
+				ClusterIP: "10.0.0.1",
+				Ports: []k8s.ServicePort{
+					{
+						Name: "http",
+						Port: 801,
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Name:      "service2",
+				Namespace: "awesome",
+				UID:       "2",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.2",
@@ -585,8 +809,9 @@ func TestLoadNamespacedIngresses(t *testing.T) {
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service3",
-				UID:  "3",
+				Name:      "service3",
+				Namespace: "awesome",
+				UID:       "3",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.3",
@@ -774,8 +999,9 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 	services := []k8s.Service{
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service1",
-				UID:  "1",
+				Name:      "service1",
+				Namespace: "awesome",
+				UID:       "1",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.1",
@@ -789,8 +1015,25 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service2",
-				UID:  "2",
+				Namespace: "somewhat-awesome",
+				Name:      "service1",
+				UID:       "17",
+			},
+			Spec: k8s.ServiceSpec{
+				ClusterIP: "10.0.0.4",
+				Ports: []k8s.ServicePort{
+					{
+						Name: "http",
+						Port: 801,
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Namespace: "awesome",
+				Name:      "service2",
+				UID:       "2",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.2",
@@ -803,8 +1046,9 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 		},
 		{
 			ObjectMeta: k8s.ObjectMeta{
-				Name: "service3",
-				UID:  "3",
+				Namespace: "awesome",
+				Name:      "service3",
+				UID:       "3",
 			},
 			Spec: k8s.ServiceSpec{
 				ClusterIP: "10.0.0.3",
@@ -859,8 +1103,8 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 			},
 			"awesome/quix": {
 				Servers: map[string]types.Server{
-					"1": {
-						URL:    "http://10.0.0.1:801",
+					"17": {
+						URL:    "http://10.0.0.4:801",
 						Weight: 1,
 					},
 				},
@@ -912,9 +1156,97 @@ func TestLoadMultipleNamespacedIngresses(t *testing.T) {
 	}
 }
 
+func TestHostlessIngress(t *testing.T) {
+	ingresses := []k8s.Ingress{{
+		ObjectMeta: k8s.ObjectMeta{
+			Namespace: "awesome",
+		},
+		Spec: k8s.IngressSpec{
+			Rules: []k8s.IngressRule{
+				{
+					IngressRuleValue: k8s.IngressRuleValue{
+						HTTP: &k8s.HTTPIngressRuleValue{
+							Paths: []k8s.HTTPIngressPath{
+								{
+									Path: "/bar",
+									Backend: k8s.IngressBackend{
+										ServiceName: "service1",
+										ServicePort: k8s.FromInt(801),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}}
+	services := []k8s.Service{
+		{
+			ObjectMeta: k8s.ObjectMeta{
+				Name:      "service1",
+				Namespace: "awesome",
+				UID:       "1",
+			},
+			Spec: k8s.ServiceSpec{
+				ClusterIP: "10.0.0.1",
+				Ports: []k8s.ServicePort{
+					{
+						Name: "http",
+						Port: 801,
+					},
+				},
+			},
+		},
+	}
+	watchChan := make(chan interface{})
+	client := clientMock{
+		ingresses: ingresses,
+		services:  services,
+		watchChan: watchChan,
+	}
+	provider := Kubernetes{disablePassHostHeaders: true}
+	actual, err := provider.loadIngresses(client)
+	if err != nil {
+		t.Fatalf("error %+v", err)
+	}
+
+	expected := &types.Configuration{
+		Backends: map[string]*types.Backend{
+			"/bar": {
+				Servers: map[string]types.Server{
+					"1": {
+						URL:    "http://10.0.0.1:801",
+						Weight: 1,
+					},
+				},
+				CircuitBreaker: nil,
+				LoadBalancer:   nil,
+			},
+		},
+		Frontends: map[string]*types.Frontend{
+			"/bar": {
+				Backend: "/bar",
+				Routes: map[string]types.Route{
+					"/bar": {
+						Rule: "PathPrefix:/bar",
+					},
+				},
+			},
+		},
+	}
+	actualJSON, _ := json.Marshal(actual)
+	expectedJSON, _ := json.Marshal(expected)
+
+	if !reflect.DeepEqual(actual, expected) {
+		t.Fatalf("expected %+v, got %+v", string(expectedJSON), string(actualJSON))
+	}
+}
+
 type clientMock struct {
 	ingresses []k8s.Ingress
 	services  []k8s.Service
+	endpoints []k8s.Endpoints
 	watchChan chan interface{}
 }
 
@@ -930,9 +1262,24 @@ func (c clientMock) GetIngresses(predicate func(k8s.Ingress) bool) ([]k8s.Ingres
 func (c clientMock) WatchIngresses(predicate func(k8s.Ingress) bool, stopCh <-chan bool) (chan interface{}, chan error, error) {
 	return c.watchChan, make(chan error), nil
 }
-func (c clientMock) GetServices(predicate func(k8s.Service) bool) ([]k8s.Service, error) {
-	return c.services, nil
+func (c clientMock) GetService(name, namespace string) (k8s.Service, error) {
+	for _, service := range c.services {
+		if service.Namespace == namespace && service.Name == name {
+			return service, nil
+		}
+	}
+	return k8s.Service{}, nil
 }
+
+func (c clientMock) GetEndpoints(name, namespace string) (k8s.Endpoints, error) {
+	for _, endpoints := range c.endpoints {
+		if endpoints.Namespace == namespace && endpoints.Name == name {
+			return endpoints, nil
+		}
+	}
+	return k8s.Endpoints{}, nil
+}
+
 func (c clientMock) WatchAll(stopCh <-chan bool) (chan interface{}, chan error, error) {
 	return c.watchChan, make(chan error), nil
 }

server.go

@@ -68,8 +68,8 @@ func NewServer(globalConfiguration GlobalConfiguration) *Server {
 	server := new(Server)
 
 	server.serverEntryPoints = make(map[string]*serverEntryPoint)
-	server.configurationChan = make(chan types.ConfigMessage, 10)
-	server.configurationValidatedChan = make(chan types.ConfigMessage, 10)
+	server.configurationChan = make(chan types.ConfigMessage, 100)
+	server.configurationValidatedChan = make(chan types.ConfigMessage, 100)
 	server.signals = make(chan os.Signal, 1)
 	server.stopChan = make(chan bool, 1)
 	server.providers = []provider.Provider{}

web.go

@@ -2,9 +2,11 @@ package main
 
 import (
 	"encoding/json"
+	"expvar"
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"runtime"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/containous/traefik/autogen"
@@ -33,6 +35,14 @@ var (
 	})
 )
 
+func init() {
+	expvar.Publish("Goroutines", expvar.Func(goroutines))
+}
+
+func goroutines() interface{} {
+	return runtime.NumGoroutine()
+}
+
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
 func (provider *WebProvider) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
@@ -84,6 +94,11 @@ func (provider *WebProvider) Provide(configurationChan chan<- types.ConfigMessag
 	})
 	systemRouter.Methods("GET").PathPrefix("/dashboard/").Handler(http.StripPrefix("/dashboard/", http.FileServer(&assetfs.AssetFS{Asset: autogen.Asset, AssetDir: autogen.AssetDir, Prefix: "static"})))
 
+	// expvars
+	if provider.server.globalConfiguration.Debug {
+		systemRouter.Methods("GET").Path("/debug/vars").HandlerFunc(expvarHandler)
+	}
+
 	go func() {
 		if len(provider.CertFile) > 0 && len(provider.KeyFile) > 0 {
 			err := http.ListenAndServeTLS(provider.Address, provider.CertFile, provider.KeyFile, systemRouter)
@@ -231,3 +246,17 @@ func (provider *WebProvider) getRouteHandler(response http.ResponseWriter, reque
 	}
 	http.NotFound(response, request)
 }
+
+func expvarHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	fmt.Fprintf(w, "{\n")
+	first := true
+	expvar.Do(func(kv expvar.KeyValue) {
+		if !first {
+			fmt.Fprintf(w, ",\n")
+		}
+		first = false
+		fmt.Fprintf(w, "%q: %s", kv.Key, kv.Value)
+	})
+	fmt.Fprintf(w, "\n}\n")
+}