Merge pull request #315 from containous/add-backoff-marathon

Add backoff to marathon provider

adapters.go

@@ -23,9 +23,9 @@ func (oxylogger *OxyLogger) Warningf(format string, args ...interface{}) {
 	log.Warningf(format, args...)
 }
 
-// Errorf logs specified string as Error level in logrus.
+// Errorf logs specified string as Warningf level in logrus.
 func (oxylogger *OxyLogger) Errorf(format string, args ...interface{}) {
-	log.Errorf(format, args...)
+	log.Warningf(format, args...)
 }
 
 func notFoundHandler(w http.ResponseWriter, r *http.Request) {

build.Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.6.0-alpine
+FROM golang:1.6.1-alpine
 
 RUN apk update && apk add git bash gcc musl-dev \
 && go get github.com/Masterminds/glide \

docs/basics.md

@@ -70,7 +70,7 @@ Frontends can be defined using the following rules:
 - `PathPrefixStrip`: Same as `PathPrefix` but strip the given prefix from the request URL's Path.
 
 You can optionally enable `passHostHeader` to forward client `Host` header to the backend.
- 
+
 Here is an example of frontends definition:
 
 ```toml
@@ -107,7 +107,7 @@ A circuit breaker can also be applied to a backend, preventing high loads on fai
 Initial state is Standby. CB observes the statistics and does not modify the request.
 In case if condition matches, CB enters Tripped state, where it responds with predefines code or redirects to another frontend.
 Once Tripped timer expires, CB enters Recovering state and resets all stats.
-In case if the condition does not match and recovery timer expries, CB enters Standby state.
+In case if the condition does not match and recovery timer expires, CB enters Standby state.
 
 It can be configured using:
 
@@ -120,6 +120,26 @@ For example:
 - `LatencyAtQuantileMS(50.0) > 50`:  watch latency at quantile in milliseconds.
 - `ResponseCodeRatio(500, 600, 0, 600) > 0.5`: ratio of response codes in range [500-600) to  [0-600)
 
+To proactively prevent backends from being overwhelmed with high load, a maximum connection limit can
+also be applied to each backend.
+
+Maximum connections can be configured by specifying an integer value for `maxconn.amount` and
+`maxconn.extractorfunc` which is a strategy used to determine how to categorize requests in order to
+evaluate the maximum connections.
+
+For example:
+```toml
+[backends]
+  [backends.backend1]
+    [backends.backend1.maxconn]
+       amount = 10
+       extractorfunc = "request.host"
+```
+
+- `backend1` will return `HTTP code 429 Too Many Requests` if there are already 10 requests in progress for the same Host header.
+- Another possible value for `extractorfunc` is `client.ip` which will categorize requests based on client source ip.
+- Lastly `extractorfunc` can take the value of `request.header.ANY_HEADER` which will categorize requests based on `ANY_HEADER` that you provide.
+
 ## Servers
 
 Servers are simply defined using a `URL`. You can also apply a custom `weight` to each server (this will be used by load-balacning).

docs/benchmarks.md

@@ -207,6 +207,7 @@ Traefik is obviously slower than Nginx, but not so much: Traefik can serve 28392
 Not bad for young project :) !
 
 Some areas of possible improvements:
+
 - Use [GO_REUSEPORT](https://github.com/kavu/go_reuseport) listener
 - Run a separate server instance per CPU core with `GOMAXPROCS=1` (it appears during benchmarks that there is a lot more context switches with traefik than with nginx)
 

docs/css/traefik.css

@@ -40,4 +40,22 @@ h1, h2, h3, H4 {
 
 blockquote p {
     font-size: 14px;
+}
+
+.navbar-default .navbar-nav>.open>a, .navbar-default .navbar-nav>.open>a:hover, .navbar-default .navbar-nav>.open>a:focus {
+    color: #fff;
+    background-color: #25606F;
+}
+
+.dropdown-menu>li>a:hover, .dropdown-menu>li>a:focus {
+    color: #fff;
+    text-decoration: none;
+    background-color: #25606F;
+}
+
+.dropdown-menu>.active>a, .dropdown-menu>.active>a:hover, .dropdown-menu>.active>a:focus {
+    color: #fff;
+    text-decoration: none;
+    background-color: #25606F;
+    outline: 0;
 }

docs/toml.md

@@ -89,6 +89,10 @@
 #     [entryPoints.http.redirect]
 #       regex = "^http://localhost/(.*)"
 #       replacement = "http://mydomain/$1"
+
+entryPoints]
+  [entryPoints.http]
+  address = ":80"
 ```
 
 ## Retry configuration
@@ -98,7 +102,7 @@
 #
 # Optional
 #
-# [retry]
+[retry]
 
 # Number of attempts
 #
@@ -122,27 +126,27 @@
 #
 # Optional
 #
-# [acme]
+[acme]
 
 # Email address used for registration
 #
 # Required
 #
-# email = "test@traefik.io"
+email = "test@traefik.io"
 
 # File used for certificates storage.
 # WARNING, if you use Traefik in Docker, don't forget to mount this file as a volume.
 #
 # Required
 #
-# storageFile = "acme.json"
+storageFile = "acme.json"
 
 # Entrypoint to proxy acme challenge to.
-# WARNING, must point to an entrypoint on port 443 
+# WARNING, must point to an entrypoint on port 443
 #
 # Required
 #
-# entryPoint = "https"
+entryPoint = "https"
 
 # Enable on demand certificate. This will request a certificate from Let's Encrypt during the first TLS handshake for a hostname that does not yet have a certificate.
 # WARNING, TLS handshakes will be slow when requesting a hostname certificate for the first time, this can leads to DoS attacks.
@@ -175,6 +179,13 @@
 #   main = "local3.com"
 # [[acme.domains]]
 #   main = "local4.com"
+[[acme.domains]]
+   main = "local1.com"
+   sans = ["test1.local1.com", "test2.local1.com"]
+[[acme.domains]]
+   main = "local3.com"
+[[acme.domains]]
+   main = "local4.com"
 ```
 
 # Configuration backends
@@ -218,6 +229,9 @@ defaultEntryPoints = ["http", "https"]
     url = "http://172.17.0.3:80"
     weight = 1
   [backends.backend2]
+    [backends.backend1.maxconn]
+      amount = 10
+      extractorfunc = "request.host"
     [backends.backend2.LoadBalancer]
       method = "drr"
     [backends.backend2.servers.server1]
@@ -281,6 +295,9 @@ filename = "rules.toml"
     url = "http://172.17.0.3:80"
     weight = 1
   [backends.backend2]
+    [backends.backend1.maxconn]
+      amount = 10
+      extractorfunc = "request.host"
     [backends.backend2.LoadBalancer]
       method = "drr"
     [backends.backend2.servers.server1]
@@ -512,7 +529,7 @@ Labels can be used on containers to override default behaviour:
 - `traefik.protocol=https`: override the default `http` protocol
 - `traefik.weight=10`: assign this weight to the container
 - `traefik.enable=false`: disable this container in Træfɪk
-- `traefik.frontend.rule=Host:test.traefik.io`: override the default frontend rule (Default: `Host:{containerName}.{domain}`). See [frontends](#frontends).
+- `traefik.frontend.rule=Host:test.traefik.io`: override the default frontend rule (Default: `Host:{containerName}.{domain}`).
 - `traefik.frontend.passHostHeader=true`: forward client `Host` header to the backend.
 - `traefik.frontend.entryPoints=http,https`: assign this frontend to entry points `http` and `https`. Overrides `defaultEntryPoints`.
 * `traefik.domain=traefik.localhost`: override the default domain
@@ -592,7 +609,7 @@ Labels can be used on containers to override default behaviour:
 - `traefik.protocol=https`: override the default `http` protocol
 - `traefik.weight=10`: assign this weight to the application
 - `traefik.enable=false`: disable this application in Træfɪk
-- `traefik.frontend.rule=Host:test.traefik.io`: override the default frontend rule (Default: `Host:{containerName}.{domain}`). See [frontends](#frontends).
+- `traefik.frontend.rule=Host:test.traefik.io`: override the default frontend rule (Default: `Host:{containerName}.{domain}`).
 - `traefik.frontend.passHostHeader=true`: forward client `Host` header to the backend.
 - `traefik.frontend.entryPoints=http,https`: assign this frontend to entry points `http` and `https`. Overrides `defaultEntryPoints`.
 * `traefik.domain=traefik.localhost`: override the default domain
@@ -687,12 +704,13 @@ This backend will create routes matching on hostname based on the service name
 used in consul.
 
 Additional settings can be defined using Consul Catalog tags:
+
 - ```traefik.enable=false```: disable this container in Træfɪk
 - ```traefik.protocol=https```: override the default `http` protocol
 - ```traefik.backend.weight=10```: assign this weight to the container
 - ```traefik.backend.circuitbreaker=NetworkErrorRatio() > 0.5```
 - ```traefik.backend.loadbalancer=drr```: override the default load balancing mode
-- ```traefik.frontend.rule=Host:test.traefik.io```: override the default frontend rule (Default: `Host:{containerName}.{domain}`). See [frontends](#frontends).
+- ```traefik.frontend.rule=Host:test.traefik.io```: override the default frontend rule (Default: `Host:{containerName}.{domain}`).
 - ```traefik.frontend.passHostHeader=true```: forward client `Host` header to the backend.
 - ```traefik.frontend.entryPoints=http,https```: assign this frontend to entry points `http` and `https`. Overrides `defaultEntryPoints`.
 
@@ -709,25 +727,25 @@ Træfɪk can be configured to use Etcd as a backend configuration:
 #
 # Optional
 #
-# [etcd]
+[etcd]
 
 # Etcd server endpoint
 #
 # Required
 #
-# endpoint = "127.0.0.1:4001"
+endpoint = "127.0.0.1:4001"
 
 # Enable watch Etcd changes
 #
 # Optional
 #
-# watch = true
+watch = true
 
 # Prefix used for KV store.
 #
 # Optional
 #
-# prefix = "/traefik"
+prefix = "/traefik"
 
 # Override default configuration template. For advanced users :)
 #
@@ -762,25 +780,25 @@ Træfɪk can be configured to use Zookeeper as a backend configuration:
 #
 # Optional
 #
-# [zookeeper]
+[zookeeper]
 
 # Zookeeper server endpoint
 #
 # Required
 #
-# endpoint = "127.0.0.1:2181"
+endpoint = "127.0.0.1:2181"
 
 # Enable watch Zookeeper changes
 #
 # Optional
 #
-# watch = true
+watch = true
 
 # Prefix used for KV store.
 #
 # Optional
 #
-# prefix = "/traefik"
+prefix = "/traefik"
 
 # Override default configuration template. For advanced users :)
 #
@@ -804,25 +822,25 @@ Træfɪk can be configured to use BoltDB as a backend configuration:
 #
 # Optional
 #
-# [boltdb]
+[boltdb]
 
 # BoltDB file
 #
 # Required
 #
-# endpoint = "/my.db"
+endpoint = "/my.db"
 
 # Enable watch BoltDB changes
 #
 # Optional
 #
-# watch = true
+watch = true
 
 # Prefix used for KV store.
 #
 # Optional
 #
-# prefix = "/traefik"
+prefix = "/traefik"
 
 # Override default configuration template. For advanced users :)
 #
@@ -851,6 +869,8 @@ The Keys-Values structure should look (using `prefix = "/traefik"`):
 
 | Key                                                 | Value                  |
 |-----------------------------------------------------|------------------------|
+| `/traefik/backends/backend2/maxconn/amount`         | `10`                   |
+| `/traefik/backends/backend2/maxconn/extractorfunc`  | `request.host`         |
 | `/traefik/backends/backend2/loadbalancer/method`    | `drr`                  |
 | `/traefik/backends/backend2/servers/server1/url`    | `http://172.17.0.4:80` |
 | `/traefik/backends/backend2/servers/server1/weight` | `1`                    |
@@ -911,99 +931,3 @@ Once the `/traefik/alias` key is updated, the new `/traefik_configurations/2` co
 
 Note that Træfɪk *will not watch for key changes in the `/traefik_configurations` prefix*. It will only watch for changes in the `/traefik` prefix. Further, if the `/traefik/alias` key is set, all other sibling keys with the `/traefik` prefix are ignored.
 
-
-# Examples
-
-## HTTP only
-
-```
-defaultEntryPoints = ["http"]
-[entryPoints]
-  [entryPoints.http]
-  address = ":80"
-```
-
-## HTTP + HTTPS (with SNI)
-
-```
-defaultEntryPoints = ["http", "https"]
-[entryPoints]
-  [entryPoints.http]
-  address = ":80"
-  [entryPoints.https]
-  address = ":443"
-    [entryPoints.https.tls]
-      [[entryPoints.https.tls.certificates]]
-      CertFile = "integration/fixtures/https/snitest.com.cert"
-      KeyFile = "integration/fixtures/https/snitest.com.key"
-      [[entryPoints.https.tls.certificates]]
-      CertFile = "integration/fixtures/https/snitest.org.cert"
-      KeyFile = "integration/fixtures/https/snitest.org.key"
-```
-
-## HTTP redirect on HTTPS
-
-```
-defaultEntryPoints = ["http", "https"]
-[entryPoints]
-  [entryPoints.http]
-  address = ":80"
-    [entryPoints.http.redirect]
-    entryPoint = "https"
-  [entryPoints.https]
-  address = ":443"
-    [entryPoints.https.tls]
-      [[entryPoints.https.tls.certificates]]
-      certFile = "tests/traefik.crt"
-      keyFile = "tests/traefik.key"
-```
-
-## Let's Encrypt support
-
-```
-[entryPoints]
-  [entryPoints.https]
-  address = ":443"
-    [entryPoints.https.tls]
-      # certs used as default certs
-      [[entryPoints.https.tls.certificates]]
-      certFile = "tests/traefik.crt"
-      keyFile = "tests/traefik.key"
-[acme]
-email = "test@traefik.io"
-storageFile = "acme.json"
-onDemand = true
-caServer = "http://172.18.0.1:4000/directory"
-entryPoint = "https"
-
-[[acme.domains]]
-  main = "local1.com"
-  sans = ["test1.local1.com", "test2.local1.com"]
-[[acme.domains]]
-  main = "local2.com"
-  sans = ["test1.local2.com", "test2x.local2.com"]
-[[acme.domains]]
-  main = "local3.com"
-[[acme.domains]]
-  main = "local4.com"
-```
-
-## Override entrypoints in frontends
-
-```
-[frontends]
-  [frontends.frontend1]
-  backend = "backend2"
-    [frontends.frontend1.routes.test_1]
-    rule = "Host:test.localhost"
-  [frontends.frontend2]
-  backend = "backend1"
-  passHostHeader = true
-  entrypoints = ["https"] # overrides defaultEntryPoints
-    [frontends.frontend2.routes.test_1]
-    rule = "Host:{subdomain:[a-z]+}.localhost"
-  [frontends.frontend3]
-  entrypoints = ["http", "https"] # overrides defaultEntryPoints
-  backend = "backend2"
-    rule = "Path:/test"
-```

docs/user-guide/examples.md

@@ -0,0 +1,98 @@
+
+# Examples
+
+You will find here some configuration examples of Træfɪk.
+
+## HTTP only
+
+```
+defaultEntryPoints = ["http"]
+[entryPoints]
+  [entryPoints.http]
+  address = ":80"
+```
+
+## HTTP + HTTPS (with SNI)
+
+```
+defaultEntryPoints = ["http", "https"]
+[entryPoints]
+  [entryPoints.http]
+  address = ":80"
+  [entryPoints.https]
+  address = ":443"
+    [entryPoints.https.tls]
+      [[entryPoints.https.tls.certificates]]
+      CertFile = "integration/fixtures/https/snitest.com.cert"
+      KeyFile = "integration/fixtures/https/snitest.com.key"
+      [[entryPoints.https.tls.certificates]]
+      CertFile = "integration/fixtures/https/snitest.org.cert"
+      KeyFile = "integration/fixtures/https/snitest.org.key"
+```
+
+## HTTP redirect on HTTPS
+
+```
+defaultEntryPoints = ["http", "https"]
+[entryPoints]
+  [entryPoints.http]
+  address = ":80"
+    [entryPoints.http.redirect]
+    entryPoint = "https"
+  [entryPoints.https]
+  address = ":443"
+    [entryPoints.https.tls]
+      [[entryPoints.https.tls.certificates]]
+      certFile = "tests/traefik.crt"
+      keyFile = "tests/traefik.key"
+```
+
+## Let's Encrypt support
+
+```
+[entryPoints]
+  [entryPoints.https]
+  address = ":443"
+    [entryPoints.https.tls]
+      # certs used as default certs
+      [[entryPoints.https.tls.certificates]]
+      certFile = "tests/traefik.crt"
+      keyFile = "tests/traefik.key"
+[acme]
+email = "test@traefik.io"
+storageFile = "acme.json"
+onDemand = true
+caServer = "http://172.18.0.1:4000/directory"
+entryPoint = "https"
+
+[[acme.domains]]
+  main = "local1.com"
+  sans = ["test1.local1.com", "test2.local1.com"]
+[[acme.domains]]
+  main = "local2.com"
+  sans = ["test1.local2.com", "test2x.local2.com"]
+[[acme.domains]]
+  main = "local3.com"
+[[acme.domains]]
+  main = "local4.com"
+```
+
+## Override entrypoints in frontends
+
+```
+[frontends]
+  [frontends.frontend1]
+  backend = "backend2"
+    [frontends.frontend1.routes.test_1]
+    rule = "Host:test.localhost"
+  [frontends.frontend2]
+  backend = "backend1"
+  passHostHeader = true
+  entrypoints = ["https"] # overrides defaultEntryPoints
+    [frontends.frontend2.routes.test_1]
+    rule = "Host:{subdomain:[a-z]+}.localhost"
+  [frontends.frontend3]
+  entrypoints = ["http", "https"] # overrides defaultEntryPoints
+  backend = "backend2"
+    rule = "Path:/test"
+```

docs/user-guide/swarm.md

@@ -0,0 +1,170 @@
+# Swarm cluster
+
+This section explains how to create a multi-host [swarm](https://docs.docker.com/swarm) cluster using [docker-machine](https://docs.docker.com/machine/) and how to deploy Træfɪk on it.
+The cluster will be made of:
+
+- 2 servers
+- 1 swarm master
+- 2 swarm nodes
+- 1 [overlay](https://docs.docker.com/engine/userguide/networking/dockernetworks/#an-overlay-network) network (multi-host networking)
+
+## Prerequisites
+
+1. You will need to install [docker-machine](https://docs.docker.com/machine/)
+2. You will need the latest [VirtualBox](https://www.virtualbox.org/wiki/Downloads)
+
+## Cluster provisioning
+
+We will first follow [this guide](https://docs.docker.com/engine/userguide/networking/get-started-overlay/) to create the cluster.
+
+### Create machine `mh-keystore`
+
+This machine will be the service registry of our cluster.
+
+```sh
+docker-machine create -d virtualbox mh-keystore
+```
+
+Then we install the service registry [Consul](https://consul.io) on this machine:
+
+```sh
+eval "$(docker-machine env mh-keystore)"
+docker run -d \
+    -p "8500:8500" \
+    -h "consul" \
+    progrium/consul -server -bootstrap
+```
+
+### Create machine `mhs-demo0`
+
+This machine will have a swarm master and a swarm agent on it.
+
+```sh
+docker-machine create -d virtualbox \
+    --swarm --swarm-master \
+    --swarm-discovery="consul://$(docker-machine ip mh-keystore):8500" \
+    --engine-opt="cluster-store=consul://$(docker-machine ip mh-keystore):8500" \
+    --engine-opt="cluster-advertise=eth1:2376" \
+    mhs-demo0
+```
+
+### Create machine `mhs-demo1`
+
+This machine will have a swarm agent on it.
+
+```sh
+docker-machine create -d virtualbox \
+    --swarm \
+    --swarm-discovery="consul://$(docker-machine ip mh-keystore):8500" \
+    --engine-opt="cluster-store=consul://$(docker-machine ip mh-keystore):8500" \
+    --engine-opt="cluster-advertise=eth1:2376" \
+    mhs-demo1
+```
+
+### Create the overlay Network
+
+Create the overlay network on the swarm master:
+
+```sh
+eval $(docker-machine env --swarm mhs-demo0)
+docker network create --driver overlay --subnet=10.0.9.0/24 my-net
+```
+
+## Deploy Træfɪk
+
+Deploy Træfɪk:
+
+```sh
+docker $(docker-machine config mhs-demo0) run \
+    -d \
+    -p 80:80 -p 8080:8080 \
+    --net=my-net \
+    -v /var/lib/boot2docker/:/ssl \
+    traefik \
+    -l DEBUG \
+    -c /dev/null \
+    --docker \
+    --docker.domain traefik \
+    --docker.endpoint tcp://$(docker-machine ip mhs-demo0):3376 \
+    --docker.tls \
+    --docker.tls.ca /ssl/ca.pem \
+    --docker.tls.cert /ssl/server.pem \
+    --docker.tls.key /ssl/server-key.pem \
+    --docker.tls.insecureSkipVerify \
+    --docker.watch  \
+    --web
+```
+
+Let's explain this command:
+
+- `-p 80:80 -p 8080:8080`: we bind ports 80 and 8080
+- `--net=my-net`: run the container on the network my-net
+- `-v /var/lib/boot2docker/:/ssl`: mount the ssl keys generated by docker-machine
+- `-c /dev/null`: empty config file
+- `--docker`: enable docker backend
+- `--docker.endpoint tcp://172.18.0.1:3376`: connect to the swarm master using the docker_gwbridge network
+- `--docker.tls`: enable TLS using the docker-machine keys
+- `--web`: activate the webUI on port 8080
+
+## Deploy your apps
+
+We can now deploy our app on the cluster, here [whoami](https://github.com/emilevauge/whoami), a simple web server in GO, on the network `my-net`:
+
+```sh
+eval $(docker-machine env --swarm mhs-demo0)
+docker run -d --name=whoami0 --net=my-net --env="constraint:node==mhs-demo0" emilevauge/whoami
+docker run -d --name=whoami1 --net=my-net --env="constraint:node==mhs-demo1" emilevauge/whoami
+```
+
+Check that everything is started:
+
+```sh
+docker ps
+CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                                                      NAMES
+ba2c21488299        emilevauge/whoami   "/whoamI"                8 seconds ago       Up 9 seconds        80/tcp                                                     mhs-demo1/whoami1
+8147a7746e7a        emilevauge/whoami   "/whoamI"                19 seconds ago      Up 20 seconds       80/tcp                                                     mhs-demo0/whoami0
+8fbc39271b4c        traefik             "/traefik -l DEBUG -c"   36 seconds ago      Up 37 seconds       192.168.99.101:80->80/tcp, 192.168.99.101:8080->8080/tcp   mhs-demo0/serene_bhabha
+```
+
+## Access to your apps through Træfɪk
+
+```sh
+curl -H Host:whoami0.traefik http://$(docker-machine ip mhs-demo0)
+Hostname: 8147a7746e7a
+IP: 127.0.0.1
+IP: ::1
+IP: 10.0.9.3
+IP: fe80::42:aff:fe00:903
+IP: 172.18.0.3
+IP: fe80::42:acff:fe12:3
+GET / HTTP/1.1
+Host: 10.0.9.3:80
+User-Agent: curl/7.35.0
+Accept: */*
+Accept-Encoding: gzip
+X-Forwarded-For: 192.168.99.1
+X-Forwarded-Host: 10.0.9.3:80
+X-Forwarded-Proto: http
+X-Forwarded-Server: 8fbc39271b4c
+
+curl -H Host:whoami1.traefik http://$(docker-machine ip mhs-demo0)
+Hostname: ba2c21488299
+IP: 127.0.0.1
+IP: ::1
+IP: 10.0.9.4
+IP: fe80::42:aff:fe00:904
+IP: 172.18.0.2
+IP: fe80::42:acff:fe12:2
+GET / HTTP/1.1
+Host: 10.0.9.4:80
+User-Agent: curl/7.35.0
+Accept: */*
+Accept-Encoding: gzip
+X-Forwarded-For: 192.168.99.1
+X-Forwarded-Host: 10.0.9.4:80
+X-Forwarded-Proto: http
+X-Forwarded-Server: 8fbc39271b4c
+```
+
+![](http://i.giphy.com/ujUdrdpX7Ok5W.gif)
+

examples/compose-marathon.yml

@@ -6,7 +6,7 @@ zk:
     ZK_ID: 1
 
 master:
-  image: mesosphere/mesos-master:0.26.0-0.2.145.ubuntu1404
+  image: mesosphere/mesos-master:0.28.1-2.0.20.ubuntu1404
   net: host
   environment:
     MESOS_ZK: zk://127.0.0.1:2181/mesos
@@ -17,7 +17,7 @@ master:
     MESOS_WORK_DIR: /var/lib/mesos
 
 slave:
-  image: mesosphere/mesos-slave:0.26.0-0.2.145.ubuntu1404
+  image: mesosphere/mesos-slave:0.28.1-2.0.20.ubuntu1404
   net: host
   pid: host
   privileged: true
@@ -34,10 +34,19 @@ slave:
     - /lib/x86_64-linux-gnu/libsystemd-journal.so.0:/lib/x86_64-linux-gnu/libsystemd-journal.so.0
 
 marathon:
-  image: mesosphere/marathon:v0.13.0
+  image: mesosphere/marathon:v1.1.1
   net: host
   environment:
     MARATHON_MASTER: zk://127.0.0.1:2181/mesos
     MARATHON_ZK: zk://127.0.0.1:2181/marathon
     MARATHON_HOSTNAME: 127.0.0.1
   command: --event_subscriber http_callback
+
+traefik:
+  image: traefik
+  command: -c /dev/null --web --logLevel=DEBUG --marathon --marathon.domain marathon.localhost --marathon.endpoint http://172.17.0.1:8080 --marathon.watch
+  ports:
+    - "8000:80"
+    - "8081:8080"
+  volumes:
+    - /var/run/docker.sock:/var/run/docker.sock

examples/compose-traefik.yml

@@ -1,12 +1,11 @@
 traefik:
   image: traefik
-  command: --web --docker --docker.domain=docker.localhost --logLevel=DEBUG
+  command: -c /dev/null --web --docker --docker.domain=docker.localhost --logLevel=DEBUG
   ports:
     - "80:80"
     - "8080:8080"
   volumes:
     - /var/run/docker.sock:/var/run/docker.sock
-    - /dev/null:/traefik.toml
 
 whoami1:
   image: emilevauge/whoami

examples/whoami.json

@@ -26,6 +26,6 @@
   "labels": {
       "traefik.weight": "1",
       "traefik.protocole": "http",
-      "traefik.frontend.rule" : "Headers:Host,test.localhost"
+      "traefik.frontend.rule" : "Host:test.marathon.localhost"
   }
 }

glide.lock

@@ -1,5 +1,5 @@
-hash: 79b6eb2a613b5e2ce5c57150eec41ac04def3f232a3613fd8b5a88b5e1041b38
-updated: 2016-04-02T15:42:37.505896092+02:00
+hash: fffa87220825895f7e3a6ceed3b13ecbf6bc934ab072fc9be3d00e3eef411ecb
+updated: 2016-04-13T14:05:41.300658168+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: 0b5b371bce661385d35439204298fa6fb5db5463
+  version: 021f82bd8260ba15f5862a9fe62018437720dff5
   subpackages:
   - cbreaker
   - forward
@@ -42,6 +42,7 @@ imports:
   version: ff6f38ccb69afa96214c7ee955359465d1fc767a
   subpackages:
   - reference
+  - digest
 - name: github.com/docker/docker
   version: f39987afe8d611407887b3094c03d6ba6a766a67
   subpackages:
@@ -89,11 +90,14 @@ imports:
   - types/container
   - types/filters
   - types/strslice
+  - types/events
   - client/transport
   - client/transport/cancellable
   - types/network
+  - types/reference
   - types/registry
   - types/time
+  - types/versions
   - types/blkiodev
 - name: github.com/docker/go-connections
   version: f549a9393d05688dff0992ef3efd8bbe6c628aeb
@@ -128,7 +132,7 @@ imports:
 - name: github.com/golang/glog
   version: fca8c8854093a154ff1eb580aae10276ad6b1b5f
 - name: github.com/google/go-querystring
-  version: 6bb77fe6f42b85397288d4f6f67ac72f8f400ee7
+  version: 9235644dd9e52eeae6fa48efd539fdc351a0af53
   subpackages:
   - query
 - name: github.com/gorilla/context
@@ -171,9 +175,9 @@ imports:
 - name: github.com/mailgun/timetools
   version: fd192d755b00c968d312d23f521eb0cdc6f66bd0
 - name: github.com/Microsoft/go-winio
-  version: 9e2895e5f6c3f16473b91d37fae6e89990a4520c
+  version: 862b6557927a5c5c81e411c12aa6de7e566cbb7a
 - name: github.com/miekg/dns
-  version: 7e024ce8ce18b21b475ac6baf8fa3c42536bf2fa
+  version: dd83d5cbcfd986f334b2747feeb907e281318fdf
 - name: github.com/mitchellh/mapstructure
   version: d2dd0262208475919e1a362f675cfc0e7c10e905
 - name: github.com/opencontainers/runc
@@ -193,19 +197,21 @@ imports:
 - name: github.com/spf13/cast
   version: ee7b3e0353166ab1f3a605294ac8cd2b77953778
 - name: github.com/spf13/cobra
-  version: 2ccf9e982a3e3eb21eba9c9ad8e546529fd74c71
+  version: 4c05eb1145f16d0e6bb4a3e1b6d769f4713cb41f
   subpackages:
   - cobra
 - name: github.com/spf13/jwalterweatherman
   version: 33c24e77fb80341fe7130ee7c594256ff08ccc46
 - name: github.com/spf13/pflag
-  version: 7f60f83a2c81bc3c3c0d5297f61ddfa68da9d3b7
+  version: 1f296710f879815ad9e6d39d947c828c3e4b4c3d
 - name: github.com/spf13/viper
   version: a212099cbe6fbe8d07476bfda8d2d39b6ff8f325
+- name: github.com/streamrail/concurrent-map
+  version: 788b276dc7eabf20890ea3fa280956664d58b329
 - name: github.com/stretchr/objx
   version: cbeaeb16a013161a98496fad62933b1d21786672
 - name: github.com/stretchr/testify
-  version: 6fe211e493929a8aac0469b93f28b1d0688a9a3a
+  version: bcd9e3389dd03b0b668d11f4d462a6af6c2dfd60
   subpackages:
   - mock
   - assert
@@ -214,13 +220,13 @@ imports:
 - name: github.com/unrolled/render
   version: 26b4e3aac686940fe29521545afad9966ddfc80c
 - name: github.com/vdemeester/docker-events
-  version: bd72e1848b08db4b5ed1a2e9277621b9f5e5d1f3
+  version: 6ea3f28df37f29a47498bc8b32b36ad8491dbd37
 - name: github.com/vdemeester/libkermit
   version: 7e4e689a6fa9281e0fb9b7b9c297e22d5342a5ec
 - name: github.com/vdemeester/shakers
   version: 24d7f1d6a71aa5d9cbe7390e4afb66b7eef9e1b3
 - name: github.com/vulcand/oxy
-  version: 8aaf36279137ac04ace3792a4f86098631b27d5a
+  version: 11677428db34c4a05354d66d028174d0e3c6e905
   subpackages:
   - memmetrics
   - utils
@@ -237,11 +243,11 @@ imports:
 - name: github.com/wendal/errors
   version: f66c77a7882b399795a8987ebf87ef64a427417e
 - name: github.com/xenolf/lego
-  version: ca19a90028e242e878585941c2a27c8f3b3efc25
+  version: 23e88185c255e95a106835d80e76e5a3a66d7c54
   subpackages:
   - acme
 - name: golang.org/x/crypto
-  version: 9e7f5dc375abeb9619ea3c5c58502c428f457aa2
+  version: d68c3ecb62c850b645dc072a8d78006286bf81ca
   subpackages:
   - ocsp
 - name: golang.org/x/net

glide.yaml

@@ -7,7 +7,7 @@ import:
   - package: github.com/mailgun/log
     ref:     44874009257d4d47ba9806f1b7f72a32a015e4d8
   - package: github.com/containous/oxy
-    ref:     0b5b371bce661385d35439204298fa6fb5db5463
+    ref:     021f82bd8260ba15f5862a9fe62018437720dff5
     subpackages:
       - cbreaker
       - forward
@@ -174,3 +174,4 @@ import:
     - tlsconfig
   - package: github.com/docker/go-units
   - package: github.com/mailgun/multibuf
+  - package: github.com/streamrail/concurrent-map

middlewares/handlerSwitcher.go

@@ -1,40 +1,35 @@
 package middlewares
 
 import (
+	"github.com/containous/traefik/safe"
 	"github.com/gorilla/mux"
 	"net/http"
-	"sync"
 )
 
 // HandlerSwitcher allows hot switching of http.ServeMux
 type HandlerSwitcher struct {
-	handler     *mux.Router
-	handlerLock *sync.Mutex
+	handler *safe.Safe
 }
 
 // NewHandlerSwitcher builds a new instance of HandlerSwitcher
 func NewHandlerSwitcher(newHandler *mux.Router) (hs *HandlerSwitcher) {
 	return &HandlerSwitcher{
-		handler:     newHandler,
-		handlerLock: &sync.Mutex{},
+		handler: safe.New(newHandler),
 	}
 }
 
 func (hs *HandlerSwitcher) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
-	hs.handlerLock.Lock()
-	handlerBackup := hs.handler
-	hs.handlerLock.Unlock()
+	handlerBackup := hs.handler.Get().(*mux.Router)
 	handlerBackup.ServeHTTP(rw, r)
 }
 
 // GetHandler returns the current http.ServeMux
 func (hs *HandlerSwitcher) GetHandler() (newHandler *mux.Router) {
-	return hs.handler
+	handler := hs.handler.Get().(*mux.Router)
+	return handler
 }
 
 // UpdateHandler safely updates the current http.ServeMux with a new one
 func (hs *HandlerSwitcher) UpdateHandler(newHandler *mux.Router) {
-	hs.handlerLock.Lock()
-	hs.handler = newHandler
-	defer hs.handlerLock.Unlock()
+	hs.handler.Set(newHandler)
 }

middlewares/logger.go

@@ -35,5 +35,7 @@ func (l *Logger) ServeHTTP(rw http.ResponseWriter, r *http.Request, next http.Ha
 
 // Close closes the logger (i.e. the file).
 func (l *Logger) Close() {
-	l.file.Close()
+	if l.file != nil {
+		l.file.Close()
+	}
 }

mkdocs.yml

@@ -46,4 +46,7 @@ pages:
   - Getting Started: index.md
   - Basics: basics.md
   - traefik.toml: toml.md
+  - User Guide:
+      - 'Configuration examples': 'user-guide/examples.md'
+      - 'Swarm cluster': 'user-guide/swarm.md'
   - Benchmarks: benchmarks.md

provider/boltdb.go

@@ -1,6 +1,7 @@
 package provider
 
 import (
+	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"github.com/docker/libkv/store"
 	"github.com/docker/libkv/store/boltdb"
@@ -13,8 +14,8 @@ type BoltDb struct {
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *BoltDb) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *BoltDb) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	provider.storeType = store.BOLTDB
 	boltdb.Register()
-	return provider.provide(configurationChan)
+	return provider.provide(configurationChan, pool)
 }

provider/consul.go

@@ -1,6 +1,7 @@
 package provider
 
 import (
+	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"github.com/docker/libkv/store"
 	"github.com/docker/libkv/store/consul"
@@ -13,8 +14,8 @@ type Consul struct {
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *Consul) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *Consul) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	provider.storeType = store.CONSUL
 	consul.Register()
-	return provider.provide(configurationChan)
+	return provider.provide(configurationChan, pool)
 }

provider/consul_catalog.go

@@ -189,7 +189,7 @@ func (provider *ConsulCatalog) getNodes(index map[string][]string) ([]catalogUpd
 	return nodes, nil
 }
 
-func (provider *ConsulCatalog) watch(configurationChan chan<- types.ConfigMessage) error {
+func (provider *ConsulCatalog) watch(configurationChan chan<- types.ConfigMessage, stop chan bool) error {
 	stopCh := make(chan struct{})
 	serviceCatalog := provider.watchServices(stopCh)
 
@@ -197,6 +197,8 @@ func (provider *ConsulCatalog) watch(configurationChan chan<- types.ConfigMessag
 
 	for {
 		select {
+		case <-stop:
+			return nil
 		case index, ok := <-serviceCatalog:
 			if !ok {
 				return errors.New("Consul service list nil")
@@ -217,7 +219,7 @@ func (provider *ConsulCatalog) watch(configurationChan chan<- types.ConfigMessag
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *ConsulCatalog) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *ConsulCatalog) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	config := api.DefaultConfig()
 	config.Address = provider.Endpoint
 	client, err := api.NewClient(config)
@@ -226,12 +228,12 @@ func (provider *ConsulCatalog) Provide(configurationChan chan<- types.ConfigMess
 	}
 	provider.client = client
 
-	safe.Go(func() {
+	pool.Go(func(stop chan bool) {
 		notify := func(err error, time time.Duration) {
 			log.Errorf("Consul connection error %+v, retrying in %s", err, time)
 		}
 		worker := func() error {
-			return provider.watch(configurationChan)
+			return provider.watch(configurationChan, stop)
 		}
 		err := backoff.RetryNotify(worker, backoff.NewExponentialBackOff(), notify)
 		if err != nil {

provider/docker.go

@@ -79,7 +79,8 @@ func (provider *Docker) createClient() (client.APIClient, error) {
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *Docker) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *Docker) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
+	// TODO register this routine in pool, and watch for stop channel
 	safe.Go(func() {
 		operation := func() error {
 			var err error
@@ -127,7 +128,17 @@ func (provider *Docker) Provide(configurationChan chan<- types.ConfigMessage) er
 				}
 				eventHandler.Handle("start", startStopHandle)
 				eventHandler.Handle("die", startStopHandle)
+
 				errChan := events.MonitorWithHandler(ctx, dockerClient, options, eventHandler)
+				pool.Go(func(stop chan bool) {
+					for {
+						select {
+						case <-stop:
+							cancel()
+							return
+						}
+					}
+				})
 				if err := <-errChan; err != nil {
 					return err
 				}

provider/etcd.go

@@ -1,6 +1,7 @@
 package provider
 
 import (
+	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"github.com/docker/libkv/store"
 	"github.com/docker/libkv/store/etcd"
@@ -13,8 +14,8 @@ type Etcd struct {
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *Etcd) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *Etcd) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	provider.storeType = store.ETCD
 	etcd.Register()
-	return provider.provide(configurationChan)
+	return provider.provide(configurationChan, pool)
 }

provider/file.go

@@ -19,7 +19,7 @@ type File struct {
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *File) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *File) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
 		log.Error("Error creating file watcher", err)
@@ -35,10 +35,12 @@ func (provider *File) Provide(configurationChan chan<- types.ConfigMessage) erro
 
 	if provider.Watch {
 		// Process events
-		safe.Go(func() {
+		pool.Go(func(stop chan bool) {
 			defer watcher.Close()
 			for {
 				select {
+				case <-stop:
+					return
 				case event := <-watcher.Events:
 					if strings.Contains(event.Name, file.Name()) {
 						log.Debug("File event:", event)

provider/kv.go

@@ -36,15 +36,17 @@ type KvTLS struct {
 	InsecureSkipVerify bool
 }
 
-func (provider *Kv) watchKv(configurationChan chan<- types.ConfigMessage, prefix string) {
+func (provider *Kv) watchKv(configurationChan chan<- types.ConfigMessage, prefix string, stop chan bool) {
 	for {
-		chanKeys, err := provider.kvclient.WatchTree(provider.Prefix, make(chan struct{}) /* stop chan */)
+		events, err := provider.kvclient.WatchTree(provider.Prefix, make(chan struct{}) /* stop chan */)
 		if err != nil {
 			log.Errorf("Failed to WatchTree %s", err)
 			continue
 		}
-
-		for range chanKeys {
+		select {
+		case <-stop:
+			return
+		case <-events:
 			configuration := provider.loadConfig()
 			if configuration != nil {
 				configurationChan <- types.ConfigMessage{
@@ -53,11 +55,10 @@ func (provider *Kv) watchKv(configurationChan chan<- types.ConfigMessage, prefix
 				}
 			}
 		}
-		log.Warnf("Intermittent failure to WatchTree KV. Retrying.")
 	}
 }
 
-func (provider *Kv) provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *Kv) provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	storeConfig := &store.Config{
 		ConnectionTimeout: 30 * time.Second,
 		Bucket:            "traefik",
@@ -102,8 +103,8 @@ func (provider *Kv) provide(configurationChan chan<- types.ConfigMessage) error
 	}
 	provider.kvclient = kv
 	if provider.Watch {
-		safe.Go(func() {
-			provider.watchKv(configurationChan, provider.Prefix)
+		pool.Go(func(stop chan bool) {
+			provider.watchKv(configurationChan, provider.Prefix, stop)
 		})
 	}
 	configuration := provider.loadConfig()

provider/kv_test.go

@@ -258,7 +258,7 @@ func TestKvWatchTree(t *testing.T) {
 
 	configChan := make(chan types.ConfigMessage)
 	safe.Go(func() {
-		provider.watchKv(configChan, "prefix")
+		provider.watchKv(configChan, "prefix", make(chan bool, 1))
 	})
 
 	select {

provider/marathon.go

@@ -10,10 +10,12 @@ import (
 	"crypto/tls"
 	"github.com/BurntSushi/ty/fun"
 	log "github.com/Sirupsen/logrus"
+	"github.com/cenkalti/backoff"
 	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"github.com/gambol99/go-marathon"
 	"net/http"
+	"time"
 )
 
 // Marathon holds configuration of the Marathon provider.
@@ -40,50 +42,59 @@ type lightMarathonClient interface {
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *Marathon) Provide(configurationChan chan<- types.ConfigMessage) error {
-	config := marathon.NewDefaultConfig()
-	config.URL = provider.Endpoint
-	config.EventsTransport = marathon.EventsTransportSSE
-	if provider.Basic != nil {
-		config.HTTPBasicAuthUser = provider.Basic.HTTPBasicAuthUser
-		config.HTTPBasicPassword = provider.Basic.HTTPBasicPassword
-	}
-	config.HTTPClient = &http.Client{
-		Transport: &http.Transport{
-			TLSClientConfig: provider.TLS,
-		},
-	}
-	client, err := marathon.NewClient(config)
-	if err != nil {
-		log.Errorf("Failed to create a client for marathon, error: %s", err)
-		return err
-	}
-	provider.marathonClient = client
-	update := make(marathon.EventsChannel, 5)
-	if provider.Watch {
-		if err := client.AddEventsListener(update, marathon.EVENTS_APPLICATIONS); err != nil {
-			log.Errorf("Failed to register for events, %s", err)
-		} else {
-			safe.Go(func() {
+func (provider *Marathon) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
+	operation := func() error {
+		config := marathon.NewDefaultConfig()
+		config.URL = provider.Endpoint
+		config.EventsTransport = marathon.EventsTransportSSE
+		if provider.Basic != nil {
+			config.HTTPBasicAuthUser = provider.Basic.HTTPBasicAuthUser
+			config.HTTPBasicPassword = provider.Basic.HTTPBasicPassword
+		}
+		config.HTTPClient = &http.Client{
+			Transport: &http.Transport{
+				TLSClientConfig: provider.TLS,
+			},
+		}
+		client, err := marathon.NewClient(config)
+		if err != nil {
+			log.Errorf("Failed to create a client for marathon, error: %s", err)
+			return err
+		}
+		provider.marathonClient = client
+		update := make(marathon.EventsChannel, 5)
+		if provider.Watch {
+			if err := client.AddEventsListener(update, marathon.EVENTS_APPLICATIONS); err != nil {
+				log.Errorf("Failed to register for events, %s", err)
+				return err
+			}
+			pool.Go(func(stop chan bool) {
 				for {
-					event := <-update
-					log.Debug("Marathon event receveived", event)
-					configuration := provider.loadMarathonConfig()
-					if configuration != nil {
-						configurationChan <- types.ConfigMessage{
-							ProviderName:  "marathon",
-							Configuration: configuration,
+					select {
+					case <-stop:
+						return
+					case event := <-update:
+						log.Debug("Marathon event receveived", event)
+						configuration := provider.loadMarathonConfig()
+						if configuration != nil {
+							configurationChan <- types.ConfigMessage{
+								ProviderName:  "marathon",
+								Configuration: configuration,
+							}
 						}
 					}
 				}
 			})
 		}
+		return nil
 	}
 
-	configuration := provider.loadMarathonConfig()
-	configurationChan <- types.ConfigMessage{
-		ProviderName:  "marathon",
-		Configuration: configuration,
+	notify := func(err error, time time.Duration) {
+		log.Errorf("Marathon connection error %+v, retrying in %s", err, time)
+	}
+	err := backoff.RetryNotify(operation, backoff.NewExponentialBackOff(), notify)
+	if err != nil {
+		log.Fatalf("Cannot connect to Marathon server %+v", err)
 	}
 	return nil
 }

provider/provider.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/BurntSushi/toml"
 	"github.com/containous/traefik/autogen"
+	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"unicode"
 )
@@ -16,7 +17,7 @@ import (
 type Provider interface {
 	// Provide allows the provider to provide configurations to traefik
 	// using the given configuration channel.
-	Provide(configurationChan chan<- types.ConfigMessage) error
+	Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error
 }
 
 // BaseProvider should be inherited by providers

provider/provider_test.go

@@ -168,3 +168,41 @@ func TestReplace(t *testing.T) {
 		}
 	}
 }
+
+func TestGetConfigurationReturnsCorrectMaxConnConfiguration(t *testing.T) {
+	templateFile, err := ioutil.TempFile("", "provider-configuration")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(templateFile.Name())
+	data := []byte(`[backends]
+  [backends.backend1]
+    [backends.backend1.maxconn]
+      amount = 10
+      extractorFunc = "request.host"`)
+	err = ioutil.WriteFile(templateFile.Name(), data, 0700)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	provider := &myProvider{
+		BaseProvider{
+			Filename: templateFile.Name(),
+		},
+	}
+	configuration, err := provider.getConfiguration(templateFile.Name(), nil, nil)
+	if err != nil {
+		t.Fatalf("Shouldn't have error out, got %v", err)
+	}
+	if configuration == nil {
+		t.Fatalf("Configuration should not be nil, but was")
+	}
+
+	if configuration.Backends["backend1"].MaxConn.Amount != 10 {
+		t.Fatalf("Configuration did not parse MaxConn.Amount properly")
+	}
+
+	if configuration.Backends["backend1"].MaxConn.ExtractorFunc != "request.host" {
+		t.Fatalf("Configuration did not parse MaxConn.ExtractorFunc properly")
+	}
+}

provider/zk.go

@@ -1,6 +1,7 @@
 package provider
 
 import (
+	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"github.com/docker/libkv/store"
 	"github.com/docker/libkv/store/zookeeper"
@@ -13,8 +14,8 @@ type Zookepper struct {
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *Zookepper) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *Zookepper) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	provider.storeType = store.ZK
 	zookeeper.Register()
-	return provider.provide(configurationChan)
+	return provider.provide(configurationChan, pool)
 }

safe/routine.go

@@ -0,0 +1,70 @@
+package safe
+
+import (
+	"log"
+	"runtime/debug"
+	"sync"
+)
+
+type routine struct {
+	goroutine func(chan bool)
+	stop      chan bool
+}
+
+// Pool creates a pool of go routines
+type Pool struct {
+	routines  []routine
+	waitGroup sync.WaitGroup
+	lock      sync.Mutex
+}
+
+// Go starts a recoverable goroutine, and can be stopped with stop chan
+func (p *Pool) Go(goroutine func(stop chan bool)) {
+	p.lock.Lock()
+	newRoutine := routine{
+		goroutine: goroutine,
+		stop:      make(chan bool, 1),
+	}
+	p.routines = append(p.routines, newRoutine)
+	p.waitGroup.Add(1)
+	Go(func() {
+		goroutine(newRoutine.stop)
+		p.waitGroup.Done()
+	})
+	p.lock.Unlock()
+}
+
+// Stop stops all started routines, waiting for their termination
+func (p *Pool) Stop() {
+	p.lock.Lock()
+	for _, routine := range p.routines {
+		routine.stop <- true
+	}
+	p.waitGroup.Wait()
+	for _, routine := range p.routines {
+		close(routine.stop)
+	}
+	p.lock.Unlock()
+}
+
+// Go starts a recoverable goroutine
+func Go(goroutine func()) {
+	GoWithRecover(goroutine, defaultRecoverGoroutine)
+}
+
+// GoWithRecover starts a recoverable goroutine using given customRecover() function
+func GoWithRecover(goroutine func(), customRecover func(err interface{})) {
+	go func() {
+		defer func() {
+			if err := recover(); err != nil {
+				customRecover(err)
+			}
+		}()
+		goroutine()
+	}()
+}
+
+func defaultRecoverGoroutine(err interface{}) {
+	log.Println(err)
+	debug.PrintStack()
+}

safe/safe.go

@@ -1,28 +1,30 @@
 package safe
 
 import (
-	"log"
-	"runtime/debug"
+	"sync"
 )
 
-// Go starts a recoverable goroutine
-func Go(goroutine func()) {
-	GoWithRecover(goroutine, defaultRecoverGoroutine)
+// Safe contains a thread-safe value
+type Safe struct {
+	value interface{}
+	lock  sync.RWMutex
 }
 
-// GoWithRecover starts a recoverable goroutine using given customRecover() function
-func GoWithRecover(goroutine func(), customRecover func(err interface{})) {
-	go func() {
-		defer func() {
-			if err := recover(); err != nil {
-				customRecover(err)
-			}
-		}()
-		goroutine()
-	}()
+// New create a new Safe instance given a value
+func New(value interface{}) *Safe {
+	return &Safe{value: value, lock: sync.RWMutex{}}
 }
 
-func defaultRecoverGoroutine(err interface{}) {
-	log.Println(err)
-	debug.PrintStack()
+// Get returns the value
+func (s *Safe) Get() interface{} {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.value
+}
+
+// Set sets a new value
+func (s *Safe) Set(value interface{}) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.value = value
 }

server.go

@@ -15,22 +15,24 @@ import (
 	"regexp"
 	"sort"
 	"strconv"
-	"sync"
 	"syscall"
 	"time"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/codegangsta/negroni"
 	"github.com/containous/oxy/cbreaker"
+	"github.com/containous/oxy/connlimit"
 	"github.com/containous/oxy/forward"
 	"github.com/containous/oxy/roundrobin"
 	"github.com/containous/oxy/stream"
+	"github.com/containous/oxy/utils"
 	"github.com/containous/traefik/middlewares"
 	"github.com/containous/traefik/provider"
 	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"github.com/gorilla/mux"
 	"github.com/mailgun/manners"
+	"github.com/streamrail/concurrent-map"
 )
 
 var oxyLogger = &OxyLogger{}
@@ -43,10 +45,10 @@ type Server struct {
 	signals                    chan os.Signal
 	stopChan                   chan bool
 	providers                  []provider.Provider
-	serverLock                 sync.Mutex
-	currentConfigurations      configs
+	currentConfigurations      safe.Safe
 	globalConfiguration        GlobalConfiguration
 	loggerMiddleware           *middlewares.Logger
+	routinesPool               safe.Pool
 }
 
 type serverEntryPoints map[string]*serverEntryPoint
@@ -69,10 +71,11 @@ func NewServer(globalConfiguration GlobalConfiguration) *Server {
 	server.configurationChan = make(chan types.ConfigMessage, 10)
 	server.configurationValidatedChan = make(chan types.ConfigMessage, 10)
 	server.signals = make(chan os.Signal, 1)
-	server.stopChan = make(chan bool)
+	server.stopChan = make(chan bool, 1)
 	server.providers = []provider.Provider{}
 	signal.Notify(server.signals, syscall.SIGINT, syscall.SIGTERM)
-	server.currentConfigurations = make(configs)
+	currentConfigurations := make(configs)
+	server.currentConfigurations.Set(currentConfigurations)
 	server.globalConfiguration = globalConfiguration
 	server.loggerMiddleware = middlewares.NewLogger(globalConfiguration.AccessLogsFile)
 
@@ -82,11 +85,11 @@ func NewServer(globalConfiguration GlobalConfiguration) *Server {
 // Start starts the server and blocks until server is shutted down.
 func (server *Server) Start() {
 	server.startHTTPServers()
-	safe.Go(func() {
-		server.listenProviders()
+	server.routinesPool.Go(func(stop chan bool) {
+		server.listenProviders(stop)
 	})
-	safe.Go(func() {
-		server.listenConfigurations()
+	server.routinesPool.Go(func(stop chan bool) {
+		server.listenConfigurations(stop)
 	})
 	server.configureProviders()
 	server.startProviders()
@@ -104,6 +107,7 @@ func (server *Server) Stop() {
 
 // Close destroys the server
 func (server *Server) Close() {
+	server.routinesPool.Stop()
 	close(server.configurationChan)
 	close(server.configurationValidatedChan)
 	close(server.signals)
@@ -124,58 +128,79 @@ func (server *Server) startHTTPServers() {
 	}
 }
 
-func (server *Server) listenProviders() {
-	lastReceivedConfiguration := time.Unix(0, 0)
-	lastConfigs := make(map[string]*types.ConfigMessage)
+func (server *Server) listenProviders(stop chan bool) {
+	lastReceivedConfiguration := safe.New(time.Unix(0, 0))
+	lastConfigs := cmap.New()
 	for {
-		configMsg := <-server.configurationChan
-		jsonConf, _ := json.Marshal(configMsg.Configuration)
-		log.Debugf("Configuration received from provider %s: %s", configMsg.ProviderName, string(jsonConf))
-		lastConfigs[configMsg.ProviderName] = &configMsg
-		if time.Now().After(lastReceivedConfiguration.Add(time.Duration(server.globalConfiguration.ProvidersThrottleDuration))) {
-			log.Debugf("Last %s config received more than %s, OK", configMsg.ProviderName, server.globalConfiguration.ProvidersThrottleDuration)
-			// last config received more than n s ago
-			server.configurationValidatedChan <- configMsg
-		} else {
-			log.Debugf("Last %s config received less than %s, waiting...", configMsg.ProviderName, server.globalConfiguration.ProvidersThrottleDuration)
-			safe.Go(func() {
-				<-time.After(server.globalConfiguration.ProvidersThrottleDuration)
-				if time.Now().After(lastReceivedConfiguration.Add(time.Duration(server.globalConfiguration.ProvidersThrottleDuration))) {
-					log.Debugf("Waited for %s config, OK", configMsg.ProviderName)
-					server.configurationValidatedChan <- *lastConfigs[configMsg.ProviderName]
-				}
-			})
+		select {
+		case <-stop:
+			return
+		case configMsg, ok := <-server.configurationChan:
+			if !ok {
+				return
+			}
+			jsonConf, _ := json.Marshal(configMsg.Configuration)
+			log.Debugf("Configuration received from provider %s: %s", configMsg.ProviderName, string(jsonConf))
+			lastConfigs.Set(configMsg.ProviderName, &configMsg)
+			lastReceivedConfigurationValue := lastReceivedConfiguration.Get().(time.Time)
+			if time.Now().After(lastReceivedConfigurationValue.Add(time.Duration(server.globalConfiguration.ProvidersThrottleDuration))) {
+				log.Debugf("Last %s config received more than %s, OK", configMsg.ProviderName, server.globalConfiguration.ProvidersThrottleDuration)
+				// last config received more than n s ago
+				server.configurationValidatedChan <- configMsg
+			} else {
+				log.Debugf("Last %s config received less than %s, waiting...", configMsg.ProviderName, server.globalConfiguration.ProvidersThrottleDuration)
+				server.routinesPool.Go(func(stop chan bool) {
+					select {
+					case <-stop:
+						return
+					case <-time.After(server.globalConfiguration.ProvidersThrottleDuration):
+						lastReceivedConfigurationValue := lastReceivedConfiguration.Get().(time.Time)
+						if time.Now().After(lastReceivedConfigurationValue.Add(time.Duration(server.globalConfiguration.ProvidersThrottleDuration))) {
+							log.Debugf("Waited for %s config, OK", configMsg.ProviderName)
+							if lastConfig, ok := lastConfigs.Get(configMsg.ProviderName); ok {
+								server.configurationValidatedChan <- *lastConfig.(*types.ConfigMessage)
+							}
+						}
+					}
+				})
+			}
+			lastReceivedConfiguration.Set(time.Now())
 		}
-		lastReceivedConfiguration = time.Now()
 	}
 }
 
-func (server *Server) listenConfigurations() {
+func (server *Server) listenConfigurations(stop chan bool) {
 	for {
-		configMsg := <-server.configurationValidatedChan
-		if configMsg.Configuration == nil {
-			log.Info("Skipping empty Configuration")
-		} else if reflect.DeepEqual(server.currentConfigurations[configMsg.ProviderName], configMsg.Configuration) {
-			log.Info("Skipping same configuration")
-		} else {
-			// Copy configurations to new map so we don't change current if LoadConfig fails
-			newConfigurations := make(configs)
-			for k, v := range server.currentConfigurations {
-				newConfigurations[k] = v
+		select {
+		case <-stop:
+			return
+		case configMsg, ok := <-server.configurationValidatedChan:
+			if !ok {
+				return
 			}
-			newConfigurations[configMsg.ProviderName] = configMsg.Configuration
-
-			newServerEntryPoints, err := server.loadConfig(newConfigurations, server.globalConfiguration)
-			if err == nil {
-				server.serverLock.Lock()
-				for newServerEntryPointName, newServerEntryPoint := range newServerEntryPoints {
-					server.serverEntryPoints[newServerEntryPointName].httpRouter.UpdateHandler(newServerEntryPoint.httpRouter.GetHandler())
-					log.Infof("Server configuration reloaded on %s", server.serverEntryPoints[newServerEntryPointName].httpServer.Addr)
-				}
-				server.currentConfigurations = newConfigurations
-				server.serverLock.Unlock()
+			currentConfigurations := server.currentConfigurations.Get().(configs)
+			if configMsg.Configuration == nil {
+				log.Info("Skipping empty Configuration")
+			} else if reflect.DeepEqual(currentConfigurations[configMsg.ProviderName], configMsg.Configuration) {
+				log.Info("Skipping same configuration")
 			} else {
-				log.Error("Error loading new configuration, aborted ", err)
+				// Copy configurations to new map so we don't change current if LoadConfig fails
+				newConfigurations := make(configs)
+				for k, v := range currentConfigurations {
+					newConfigurations[k] = v
+				}
+				newConfigurations[configMsg.ProviderName] = configMsg.Configuration
+
+				newServerEntryPoints, err := server.loadConfig(newConfigurations, server.globalConfiguration)
+				if err == nil {
+					for newServerEntryPointName, newServerEntryPoint := range newServerEntryPoints {
+						server.serverEntryPoints[newServerEntryPointName].httpRouter.UpdateHandler(newServerEntryPoint.httpRouter.GetHandler())
+						log.Infof("Server configuration reloaded on %s", server.serverEntryPoints[newServerEntryPointName].httpServer.Addr)
+					}
+					server.currentConfigurations.Set(newConfigurations)
+				} else {
+					log.Error("Error loading new configuration, aborted ", err)
+				}
 			}
 		}
 	}
@@ -220,7 +245,7 @@ func (server *Server) startProviders() {
 		log.Infof("Starting provider %v %s", reflect.TypeOf(provider), jsonConf)
 		currentProvider := provider
 		safe.Go(func() {
-			err := currentProvider.Provide(server.configurationChan)
+			err := currentProvider.Provide(server.configurationChan, &server.routinesPool)
 			if err != nil {
 				log.Errorf("Error starting provider %s", err)
 			}
@@ -423,6 +448,18 @@ func (server *Server) loadConfig(configurations configs, globalConfiguration Glo
 								}
 							}
 						}
+						maxConns := configuration.Backends[frontend.Backend].MaxConn
+						if maxConns != nil && maxConns.Amount != 0 {
+							extractFunc, err := utils.NewExtractor(maxConns.ExtractorFunc)
+							if err != nil {
+								return nil, err
+							}
+							log.Debugf("Creating loadd-balancer connlimit")
+							lb, err = connlimit.New(lb, extractFunc, maxConns.Amount, connlimit.Logger(oxyLogger))
+							if err != nil {
+								return nil, err
+							}
+						}
 						// retry ?
 						if globalConfiguration.Retry != nil {
 							retries := len(configuration.Backends[frontend.Backend].Servers)

templates/kv.tmpl

@@ -17,6 +17,16 @@
     method = "{{$loadBalancer}}"
 {{end}}
 
+{{$maxConnAmt := Get "" . "/maxconn/" "amount"}}
+{{$maxConnExtractorFunc := Get "" . "/maxconn/" "extractorfunc"}}
+{{with $maxConnAmt}}
+{{with $maxConnExtractorFunc}}
+[backends.{{Last $backend}}.maxConn]
+    amount = {{$maxConnAmt}}
+    extractorFunc = "{{$maxConnExtractorFunc}}"
+{{end}}
+{{end}}
+
 {{range $servers}}
 [backends.{{Last $backend}}.servers.{{Last .}}]
     url = "{{Get "" . "/url"}}"

types/types.go

@@ -10,6 +10,13 @@ type Backend struct {
 	Servers        map[string]Server `json:"servers,omitempty"`
 	CircuitBreaker *CircuitBreaker   `json:"circuitBreaker,omitempty"`
 	LoadBalancer   *LoadBalancer     `json:"loadBalancer,omitempty"`
+	MaxConn        *MaxConn          `json:"maxConn,omitempty"`
+}
+
+// MaxConn holds maximum connection configuraiton
+type MaxConn struct {
+	Amount        int64  `json:"amount,omitempty"`
+	ExtractorFunc string `json:"extractorFunc,omitempty"`
 }
 
 // LoadBalancer holds load balancing configuration.

web.go

@@ -8,6 +8,7 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/containous/traefik/autogen"
+	"github.com/containous/traefik/safe"
 	"github.com/containous/traefik/types"
 	"github.com/elazarl/go-bindata-assetfs"
 	"github.com/gorilla/mux"
@@ -34,7 +35,7 @@ var (
 
 // Provide allows the provider to provide configurations to traefik
 // using the given configuration channel.
-func (provider *WebProvider) Provide(configurationChan chan<- types.ConfigMessage) error {
+func (provider *WebProvider) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
 	systemRouter := mux.NewRouter()
 
 	// health route
@@ -104,13 +105,15 @@ func (provider *WebProvider) getHealthHandler(response http.ResponseWriter, requ
 }
 
 func (provider *WebProvider) getConfigHandler(response http.ResponseWriter, request *http.Request) {
-	templatesRenderer.JSON(response, http.StatusOK, provider.server.currentConfigurations)
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	templatesRenderer.JSON(response, http.StatusOK, currentConfigurations)
 }
 
 func (provider *WebProvider) getProviderHandler(response http.ResponseWriter, request *http.Request) {
 	vars := mux.Vars(request)
 	providerID := vars["provider"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		templatesRenderer.JSON(response, http.StatusOK, provider)
 	} else {
 		http.NotFound(response, request)
@@ -120,7 +123,8 @@ func (provider *WebProvider) getProviderHandler(response http.ResponseWriter, re
 func (provider *WebProvider) getBackendsHandler(response http.ResponseWriter, request *http.Request) {
 	vars := mux.Vars(request)
 	providerID := vars["provider"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		templatesRenderer.JSON(response, http.StatusOK, provider.Backends)
 	} else {
 		http.NotFound(response, request)
@@ -131,7 +135,8 @@ func (provider *WebProvider) getBackendHandler(response http.ResponseWriter, req
 	vars := mux.Vars(request)
 	providerID := vars["provider"]
 	backendID := vars["backend"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		if backend, ok := provider.Backends[backendID]; ok {
 			templatesRenderer.JSON(response, http.StatusOK, backend)
 			return
@@ -144,7 +149,8 @@ func (provider *WebProvider) getServersHandler(response http.ResponseWriter, req
 	vars := mux.Vars(request)
 	providerID := vars["provider"]
 	backendID := vars["backend"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		if backend, ok := provider.Backends[backendID]; ok {
 			templatesRenderer.JSON(response, http.StatusOK, backend.Servers)
 			return
@@ -158,7 +164,8 @@ func (provider *WebProvider) getServerHandler(response http.ResponseWriter, requ
 	providerID := vars["provider"]
 	backendID := vars["backend"]
 	serverID := vars["server"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		if backend, ok := provider.Backends[backendID]; ok {
 			if server, ok := backend.Servers[serverID]; ok {
 				templatesRenderer.JSON(response, http.StatusOK, server)
@@ -172,7 +179,8 @@ func (provider *WebProvider) getServerHandler(response http.ResponseWriter, requ
 func (provider *WebProvider) getFrontendsHandler(response http.ResponseWriter, request *http.Request) {
 	vars := mux.Vars(request)
 	providerID := vars["provider"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		templatesRenderer.JSON(response, http.StatusOK, provider.Frontends)
 	} else {
 		http.NotFound(response, request)
@@ -183,7 +191,8 @@ func (provider *WebProvider) getFrontendHandler(response http.ResponseWriter, re
 	vars := mux.Vars(request)
 	providerID := vars["provider"]
 	frontendID := vars["frontend"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		if frontend, ok := provider.Frontends[frontendID]; ok {
 			templatesRenderer.JSON(response, http.StatusOK, frontend)
 			return
@@ -196,7 +205,8 @@ func (provider *WebProvider) getRoutesHandler(response http.ResponseWriter, requ
 	vars := mux.Vars(request)
 	providerID := vars["provider"]
 	frontendID := vars["frontend"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		if frontend, ok := provider.Frontends[frontendID]; ok {
 			templatesRenderer.JSON(response, http.StatusOK, frontend.Routes)
 			return
@@ -210,7 +220,8 @@ func (provider *WebProvider) getRouteHandler(response http.ResponseWriter, reque
 	providerID := vars["provider"]
 	frontendID := vars["frontend"]
 	routeID := vars["route"]
-	if provider, ok := provider.server.currentConfigurations[providerID]; ok {
+	currentConfigurations := provider.server.currentConfigurations.Get().(configs)
+	if provider, ok := currentConfigurations[providerID]; ok {
 		if frontend, ok := provider.Frontends[frontendID]; ok {
 			if route, ok := frontend.Routes[routeID]; ok {
 				templatesRenderer.JSON(response, http.StatusOK, route)