Merge pull request #1320 from containous/configure-healthcheck-via-marathon-label

Make Traefik health checks label-configurable with Marathon.

docs/toml.md

@@ -989,6 +989,8 @@ Labels can be used on containers to override default behaviour:
 - `traefik.backend.loadbalancer.method=drr`: override the default `wrr` load balancer algorithm
 - `traefik.backend.loadbalancer.sticky=true`: enable backend sticky sessions
 - `traefik.backend.circuitbreaker.expression=NetworkErrorRatio() > 0.5`: create a [circuit breaker](/basics/#backends) to be used against the backend
+- `traefik.backend.healthcheck.path=/health`: set the Traefik health check path [default: no health checks]
+- `traefik.backend.healthcheck.interval=5s`: sets a custom health check interval in Go-parseable (`time.ParseDuration`) format [default: 30s]
 - `traefik.portIndex=1`: register port by index in the application's ports array. Useful when the application exposes multiple ports.
 - `traefik.port=80`: register the explicit application port value. Cannot be used alongside `traefik.portIndex`.
 - `traefik.protocol=https`: override the default `http` protocol

healthcheck/healthcheck.go

@@ -12,6 +12,9 @@ import (
 	"github.com/vulcand/oxy/roundrobin"
 )
 
+// DefaultInterval is the default health check interval.
+const DefaultInterval = 30 * time.Second
+
 var singleton *HealthCheck
 var once sync.Once
 
@@ -23,16 +26,19 @@ func GetHealthCheck() *HealthCheck {
 	return singleton
 }
 
-// BackendHealthCheck HealthCheck configuration for a backend
-type BackendHealthCheck struct {
+// Options are the public health check options.
+type Options struct {
 	Path     string
 	Interval time.Duration
-	DisabledURLs   []*url.URL
-	requestTimeout time.Duration
-	lb             loadBalancer
+	LB       LoadBalancer
 }
 
-var launch = false
+// BackendHealthCheck HealthCheck configuration for a backend
+type BackendHealthCheck struct {
+	Options
+	disabledURLs   []*url.URL
+	requestTimeout time.Duration
+}
 
 //HealthCheck struct
 type HealthCheck struct {
@@ -40,7 +46,8 @@ type HealthCheck struct {
 	cancel   context.CancelFunc
 }
 
-type loadBalancer interface {
+// LoadBalancer includes functionality for load-balancing management.
+type LoadBalancer interface {
 	RemoveServer(u *url.URL) error
 	UpsertServer(u *url.URL, options ...roundrobin.ServerOption) error
 	Servers() []*url.URL
@@ -53,12 +60,10 @@ func newHealthCheck() *HealthCheck {
 }
 
 // NewBackendHealthCheck Instantiate a new BackendHealthCheck
-func NewBackendHealthCheck(Path string, interval time.Duration, lb loadBalancer) *BackendHealthCheck {
+func NewBackendHealthCheck(options Options) *BackendHealthCheck {
 	return &BackendHealthCheck{
-		Path:           Path,
-		Interval:       interval,
+		Options:        options,
 		requestTimeout: 5 * time.Second,
-		lb:             lb,
 	}
 }
 
@@ -98,24 +103,24 @@ func (hc *HealthCheck) execute(ctx context.Context, backendID string, backend *B
 }
 
 func checkBackend(currentBackend *BackendHealthCheck) {
-	enabledURLs := currentBackend.lb.Servers()
+	enabledURLs := currentBackend.LB.Servers()
 	var newDisabledURLs []*url.URL
-	for _, url := range currentBackend.DisabledURLs {
+	for _, url := range currentBackend.disabledURLs {
 		if checkHealth(url, currentBackend) {
 			log.Debugf("HealthCheck is up [%s]: Upsert in server list", url.String())
-			currentBackend.lb.UpsertServer(url, roundrobin.Weight(1))
+			currentBackend.LB.UpsertServer(url, roundrobin.Weight(1))
 		} else {
 			log.Warnf("HealthCheck is still failing [%s]", url.String())
 			newDisabledURLs = append(newDisabledURLs, url)
 		}
 	}
-	currentBackend.DisabledURLs = newDisabledURLs
+	currentBackend.disabledURLs = newDisabledURLs
 
 	for _, url := range enabledURLs {
 		if !checkHealth(url, currentBackend) {
 			log.Warnf("HealthCheck has failed [%s]: Remove from server list", url.String())
-			currentBackend.lb.RemoveServer(url)
-			currentBackend.DisabledURLs = append(currentBackend.DisabledURLs, url)
+			currentBackend.LB.RemoveServer(url)
+			currentBackend.disabledURLs = append(currentBackend.disabledURLs, url)
 		}
 	}
 }

healthcheck/healthcheck_test.go

@@ -148,12 +148,16 @@ func TestSetBackendsConfiguration(t *testing.T) {
 			defer ts.Close()
 
 			lb := &testLoadBalancer{RWMutex: &sync.RWMutex{}}
-			backend := NewBackendHealthCheck("/path", healthCheckInterval, lb)
+			backend := NewBackendHealthCheck(Options{
+				Path:     "/path",
+				Interval: healthCheckInterval,
+				LB:       lb,
+			})
 			serverURL := MustParseURL(ts.URL)
 			if test.startHealthy {
 				lb.servers = append(lb.servers, serverURL)
 			} else {
-				backend.DisabledURLs = append(backend.DisabledURLs, serverURL)
+				backend.disabledURLs = append(backend.disabledURLs, serverURL)
 			}
 
 			healthCheck := HealthCheck{

integration/fixtures/healthcheck/simple.toml

@@ -13,7 +13,7 @@ logLevel = "DEBUG"
 [backends]
   [backends.backend1]
     [backends.backend1.healthcheck]
-    url = "/health"
+    path = "/health"
     interval = "1s"
     [backends.backend1.servers.server1]
     url = "http://{{.Server1}}:80"

integration/healthcheck_test.go

@@ -16,16 +16,16 @@ import (
 	checker "github.com/vdemeester/shakers"
 )
 
-// HealchCheck test suites (using libcompose)
-type HealchCheckSuite struct{ BaseSuite }
+// HealthCheck test suites (using libcompose)
+type HealthCheckSuite struct{ BaseSuite }
 
-func (s *HealchCheckSuite) SetUpSuite(c *check.C) {
+func (s *HealthCheckSuite) SetUpSuite(c *check.C) {
 	s.createComposeProject(c, "healthcheck")
 	s.composeProject.Start(c)
 
 }
 
-func (s *HealchCheckSuite) TestSimpleConfiguration(c *check.C) {
+func (s *HealthCheckSuite) TestSimpleConfiguration(c *check.C) {
 
 	whoami1Host := s.composeProject.Container(c, "whoami1").NetworkSettings.IPAddress
 	whoami2Host := s.composeProject.Container(c, "whoami2").NetworkSettings.IPAddress

integration/integration_test.go

@@ -27,6 +27,7 @@ func init() {
 	check.Suite(&AccessLogSuite{})
 	check.Suite(&HTTPSSuite{})
 	check.Suite(&FileSuite{})
+	check.Suite(&HealthCheckSuite{})
 	check.Suite(&DockerSuite{})
 	check.Suite(&ConsulSuite{})
 	check.Suite(&ConsulCatalogSuite{})

provider/marathon/marathon.go

@@ -26,6 +26,8 @@ import (
 const (
 	labelPort                       = "traefik.port"
 	labelPortIndex                  = "traefik.portIndex"
+	labelBackendHealthCheckPath     = "traefik.backend.healthcheck.path"
+	labelBackendHealthCheckInterval = "traefik.backend.healthcheck.interval"
 )
 
 var _ provider.Provider = (*Provider)(nil)
@@ -157,6 +159,9 @@ func (p *Provider) loadMarathonConfig() *types.Configuration {
 		"getLoadBalancerMethod":       p.getLoadBalancerMethod,
 		"getCircuitBreakerExpression": p.getCircuitBreakerExpression,
 		"getSticky":                   p.getSticky,
+		"hasHealthCheckLabels":        p.hasHealthCheckLabels,
+		"getHealthCheckPath":          p.getHealthCheckPath,
+		"getHealthCheckInterval":      p.getHealthCheckInterval,
 	}
 
 	applications, err := p.marathonClient.Applications(nil)
@@ -461,6 +466,24 @@ func (p *Provider) getCircuitBreakerExpression(application marathon.Application)
 	return "NetworkErrorRatio() > 1"
 }
 
+func (p *Provider) hasHealthCheckLabels(application marathon.Application) bool {
+	return p.getHealthCheckPath(application) != ""
+}
+
+func (p *Provider) getHealthCheckPath(application marathon.Application) string {
+	if label, ok := p.getLabel(application, labelBackendHealthCheckPath); ok {
+		return label
+	}
+	return ""
+}
+
+func (p *Provider) getHealthCheckInterval(application marathon.Application) string {
+	if label, ok := p.getLabel(application, labelBackendHealthCheckInterval); ok {
+		return label
+	}
+	return ""
+}
+
 func processPorts(application marathon.Application, task marathon.Task) (int, error) {
 	if portLabel, ok := (*application.Labels)[labelPort]; ok {
 		port, err := strconv.Atoi(portLabel)

provider/marathon/marathon_test.go

@@ -361,10 +361,10 @@ func TestMarathonLoadConfig(t *testing.T) {
 			} else {
 				// Compare backends
 				if !reflect.DeepEqual(actualConfig.Backends, c.expectedBackends) {
-					t.Errorf("got backend %v, want %v", spew.Sdump(actualConfig.Backends), spew.Sdump(c.expectedBackends))
+					t.Errorf("got backend\n%v\nwant\n\n%v", spew.Sdump(actualConfig.Backends), spew.Sdump(c.expectedBackends))
 				}
 				if !reflect.DeepEqual(actualConfig.Frontends, c.expectedFrontends) {
-					t.Errorf("got frontend %v, want %v", spew.Sdump(actualConfig.Frontends), spew.Sdump(c.expectedFrontends))
+					t.Errorf("got frontend\n%v\nwant\n\n%v", spew.Sdump(actualConfig.Frontends), spew.Sdump(c.expectedFrontends))
 				}
 			}
 		})
@@ -1430,6 +1430,125 @@ func TestMarathonGetSubDomain(t *testing.T) {
 	}
 }
 
+func TestMarathonHasHealthCheckLabels(t *testing.T) {
+	tests := []struct {
+		desc  string
+		value *string
+		want  bool
+	}{
+		{
+			desc:  "label missing",
+			value: nil,
+			want:  false,
+		},
+		{
+			desc:  "empty path",
+			value: stringp(""),
+			want:  false,
+		},
+		{
+			desc:  "non-empty path",
+			value: stringp("/path"),
+			want:  true,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.desc, func(t *testing.T) {
+			t.Parallel()
+			app := marathon.Application{
+				Labels: &map[string]string{},
+			}
+			if test.value != nil {
+				app.AddLabel(labelBackendHealthCheckPath, *test.value)
+			}
+			prov := &Provider{}
+			got := prov.hasHealthCheckLabels(app)
+			if got != test.want {
+				t.Errorf("got %t, want %t", got, test.want)
+			}
+		})
+	}
+}
+
+func TestMarathonGetHealthCheckPath(t *testing.T) {
+	tests := []struct {
+		desc  string
+		value *string
+		want  string
+	}{
+		{
+			desc:  "label missing",
+			value: nil,
+			want:  "",
+		},
+		{
+			desc:  "path existing",
+			value: stringp("/path"),
+			want:  "/path",
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.desc, func(t *testing.T) {
+			t.Parallel()
+			app := marathon.Application{}
+			app.EmptyLabels()
+			if test.value != nil {
+				app.AddLabel(labelBackendHealthCheckPath, *test.value)
+			}
+			prov := &Provider{}
+			got := prov.getHealthCheckPath(app)
+			if got != test.want {
+				t.Errorf("got %s, want %s", got, test.want)
+			}
+		})
+	}
+}
+
+func TestMarathonGetHealthCheckInterval(t *testing.T) {
+	tests := []struct {
+		desc  string
+		value *string
+		want  string
+	}{
+		{
+			desc:  "label missing",
+			value: nil,
+			want:  "",
+		},
+		{
+			desc:  "interval existing",
+			value: stringp("5m"),
+			want:  "5m",
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.desc, func(t *testing.T) {
+			t.Parallel()
+			app := marathon.Application{
+				Labels: &map[string]string{},
+			}
+			if test.value != nil {
+				app.AddLabel(labelBackendHealthCheckInterval, *test.value)
+			}
+			prov := &Provider{}
+			got := prov.getHealthCheckInterval(app)
+			if got != test.want {
+				t.Errorf("got %s, want %s", got, test.want)
+			}
+		})
+	}
+}
+
+func stringp(s string) *string {
+	return &s
+}
+
 func TestGetBackendServer(t *testing.T) {
 	appID := "appId"
 	host := "host"

server/server.go

@@ -659,16 +659,9 @@ func (server *Server) loadConfig(configurations configs, globalConfiguration Glo
 									log.Errorf("Skipping frontend %s...", frontendName)
 									continue frontend
 								}
-								if configuration.Backends[frontend.Backend].HealthCheck != nil {
-									var interval time.Duration
-									if configuration.Backends[frontend.Backend].HealthCheck.Interval != "" {
-										interval, err = time.ParseDuration(configuration.Backends[frontend.Backend].HealthCheck.Interval)
-										if err != nil {
-											log.Errorf("Wrong healthcheck interval: %s", err)
-											interval = time.Second * 30
-										}
-									}
-									backendsHealthcheck[frontend.Backend] = healthcheck.NewBackendHealthCheck(configuration.Backends[frontend.Backend].HealthCheck.Path, interval, rebalancer)
+								hcOpts := parseHealthCheckOptions(rebalancer, frontend.Backend, configuration.Backends[frontend.Backend].HealthCheck)
+								if hcOpts != nil {
+									backendsHealthcheck[frontend.Backend] = healthcheck.NewBackendHealthCheck(*hcOpts)
 								}
 							}
 						case types.Wrr:
@@ -692,16 +685,9 @@ func (server *Server) loadConfig(configurations configs, globalConfiguration Glo
 									continue frontend
 								}
 							}
-							if configuration.Backends[frontend.Backend].HealthCheck != nil {
-								var interval time.Duration
-								if configuration.Backends[frontend.Backend].HealthCheck.Interval != "" {
-									interval, err = time.ParseDuration(configuration.Backends[frontend.Backend].HealthCheck.Interval)
-									if err != nil {
-										log.Errorf("Wrong healthcheck interval: %s", err)
-										interval = time.Second * 30
-									}
-								}
-								backendsHealthcheck[frontend.Backend] = healthcheck.NewBackendHealthCheck(configuration.Backends[frontend.Backend].HealthCheck.Path, interval, rr)
+							hcOpts := parseHealthCheckOptions(rr, frontend.Backend, configuration.Backends[frontend.Backend].HealthCheck)
+							if hcOpts != nil {
+								backendsHealthcheck[frontend.Backend] = healthcheck.NewBackendHealthCheck(*hcOpts)
 							}
 						}
 						maxConns := configuration.Backends[frontend.Backend].MaxConn
@@ -861,6 +847,31 @@ func (server *Server) buildDefaultHTTPRouter() *mux.Router {
 	return router
 }
 
+func parseHealthCheckOptions(lb healthcheck.LoadBalancer, backend string, hc *types.HealthCheck) *healthcheck.Options {
+	if hc == nil || hc.Path == "" {
+		return nil
+	}
+
+	interval := healthcheck.DefaultInterval
+	if hc.Interval != "" {
+		intervalOverride, err := time.ParseDuration(hc.Interval)
+		switch {
+		case err != nil:
+			log.Errorf("Illegal healthcheck interval for backend '%s': %s", backend, err)
+		case intervalOverride <= 0:
+			log.Errorf("Healthcheck interval smaller than zero for backend '%s', backend", backend)
+		default:
+			interval = intervalOverride
+		}
+	}
+
+	return &healthcheck.Options{
+		Path:     hc.Path,
+		Interval: interval,
+		LB:       lb,
+	}
+}
+
 func getRoute(serverRoute *serverRoute, route *types.Route) error {
 	rules := Rules{route: serverRoute}
 	newRoute, err := rules.Parse(route.Rule)

server/server_test.go

@@ -0,0 +1,157 @@
+package server
+
+import (
+	"fmt"
+	"net/url"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/containous/traefik/healthcheck"
+	"github.com/containous/traefik/types"
+	"github.com/vulcand/oxy/roundrobin"
+)
+
+type testLoadBalancer struct{}
+
+func (lb *testLoadBalancer) RemoveServer(u *url.URL) error {
+	return nil
+}
+
+func (lb *testLoadBalancer) UpsertServer(u *url.URL, options ...roundrobin.ServerOption) error {
+	return nil
+}
+
+func (lb *testLoadBalancer) Servers() []*url.URL {
+	return []*url.URL{}
+}
+
+func TestServerLoadConfigHealthCheckOptions(t *testing.T) {
+	healthChecks := []*types.HealthCheck{
+		nil,
+		{
+			Path: "/path",
+		},
+	}
+
+	for _, lbMethod := range []string{"Wrr", "Drr"} {
+		for _, healthCheck := range healthChecks {
+			t.Run(fmt.Sprintf("%s/hc=%t", lbMethod, healthCheck != nil), func(t *testing.T) {
+				globalConfig := GlobalConfiguration{
+					EntryPoints: EntryPoints{
+						"http": &EntryPoint{},
+					},
+				}
+
+				dynamicConfigs := configs{
+					"config": &types.Configuration{
+						Frontends: map[string]*types.Frontend{
+							"frontend": {
+								EntryPoints: []string{"http"},
+								Backend:     "backend",
+							},
+						},
+						Backends: map[string]*types.Backend{
+							"backend": {
+								Servers: map[string]types.Server{
+									"server": {
+										URL: "http://localhost",
+									},
+								},
+								LoadBalancer: &types.LoadBalancer{
+									Method: lbMethod,
+								},
+								HealthCheck: healthCheck,
+							},
+						},
+					},
+				}
+
+				srv := NewServer(globalConfig)
+				if _, err := srv.loadConfig(dynamicConfigs, globalConfig); err != nil {
+					t.Fatalf("got error: %s", err)
+				}
+
+				wantNumHealthCheckBackends := 0
+				if healthCheck != nil {
+					wantNumHealthCheckBackends = 1
+				}
+				gotNumHealthCheckBackends := len(healthcheck.GetHealthCheck().Backends)
+				if gotNumHealthCheckBackends != wantNumHealthCheckBackends {
+					t.Errorf("got %d health check backends, want %d", gotNumHealthCheckBackends, wantNumHealthCheckBackends)
+				}
+			})
+		}
+	}
+}
+
+func TestServerParseHealthCheckOptions(t *testing.T) {
+	lb := &testLoadBalancer{}
+
+	tests := []struct {
+		desc     string
+		hc       *types.HealthCheck
+		wantOpts *healthcheck.Options
+	}{
+		{
+			desc:     "nil health check",
+			hc:       nil,
+			wantOpts: nil,
+		},
+		{
+			desc: "empty path",
+			hc: &types.HealthCheck{
+				Path: "",
+			},
+			wantOpts: nil,
+		},
+		{
+			desc: "unparseable interval",
+			hc: &types.HealthCheck{
+				Path:     "/path",
+				Interval: "unparseable",
+			},
+			wantOpts: &healthcheck.Options{
+				Path:     "/path",
+				Interval: healthcheck.DefaultInterval,
+				LB:       lb,
+			},
+		},
+		{
+			desc: "sub-zero interval",
+			hc: &types.HealthCheck{
+				Path:     "/path",
+				Interval: "-15s",
+			},
+			wantOpts: &healthcheck.Options{
+				Path:     "/path",
+				Interval: healthcheck.DefaultInterval,
+				LB:       lb,
+			},
+		},
+		{
+			desc: "parseable interval",
+			hc: &types.HealthCheck{
+				Path:     "/path",
+				Interval: "5m",
+			},
+			wantOpts: &healthcheck.Options{
+				Path:     "/path",
+				Interval: 5 * time.Minute,
+				LB:       lb,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.desc, func(t *testing.T) {
+			t.Parallel()
+
+			gotOpts := parseHealthCheckOptions(lb, "backend", test.hc)
+			if !reflect.DeepEqual(gotOpts, test.wantOpts) {
+				t.Errorf("got health check options %+v, want %+v", gotOpts, test.wantOpts)
+			}
+		})
+	}
+}

templates/marathon.tmpl

@@ -20,6 +20,11 @@
       [backends."backend{{getFrontendBackend . }}".circuitbreaker]
         expression = "{{getCircuitBreakerExpression . }}"
 {{end}}
+{{ if hasHealthCheckLabels . }}
+      [backends."backend{{getFrontendBackend . }}".healthcheck]
+        path = "{{getHealthCheckPath . }}"
+        interval = "{{getHealthCheckInterval . }}"
+{{end}}
 {{end}}
 
 [frontends]{{range .Applications}}