Merge pull request #918 from Juliens/httpchk

(WIP) feat: HealthCheck

healthcheck/healthcheck.go

@@ -0,0 +1,116 @@
+package healthcheck
+
+import (
+	"context"
+	"net/http"
+	"net/url"
+	"sync"
+	"time"
+
+	"github.com/containous/traefik/log"
+	"github.com/containous/traefik/safe"
+	"github.com/vulcand/oxy/roundrobin"
+)
+
+var singleton *HealthCheck
+var once sync.Once
+
+// GetHealthCheck Get HealtchCheck Singleton
+func GetHealthCheck() *HealthCheck {
+	once.Do(func() {
+		singleton = newHealthCheck()
+	})
+	return singleton
+}
+
+// BackendHealthCheck HealthCheck configuration for a backend
+type BackendHealthCheck struct {
+	URL          string
+	DisabledURLs []*url.URL
+	lb           loadBalancer
+}
+
+var launch = false
+
+//HealthCheck struct
+type HealthCheck struct {
+	Backends map[string]*BackendHealthCheck
+	cancel   context.CancelFunc
+}
+
+type loadBalancer interface {
+	RemoveServer(u *url.URL) error
+	UpsertServer(u *url.URL, options ...roundrobin.ServerOption) error
+	Servers() []*url.URL
+}
+
+func newHealthCheck() *HealthCheck {
+	return &HealthCheck{make(map[string]*BackendHealthCheck), nil}
+}
+
+// NewBackendHealthCheck Instantiate a new BackendHealthCheck
+func NewBackendHealthCheck(URL string, lb loadBalancer) *BackendHealthCheck {
+	return &BackendHealthCheck{URL, nil, lb}
+}
+
+//SetBackendsConfiguration set backends configuration
+func (hc *HealthCheck) SetBackendsConfiguration(parentCtx context.Context, backends map[string]*BackendHealthCheck) {
+	hc.Backends = backends
+	if hc.cancel != nil {
+		hc.cancel()
+	}
+	ctx, cancel := context.WithCancel(parentCtx)
+	hc.cancel = cancel
+	hc.execute(ctx)
+}
+
+func (hc *HealthCheck) execute(ctx context.Context) {
+	for backendID, backend := range hc.Backends {
+		currentBackend := backend
+		currentBackendID := backendID
+		safe.Go(func() {
+			for {
+				ticker := time.NewTicker(time.Second * 30)
+				select {
+				case <-ctx.Done():
+					log.Debugf("Stopping all current Healthcheck goroutines")
+					return
+				case <-ticker.C:
+					log.Debugf("Refreshing Healthcheck for currentBackend %s ", currentBackendID)
+					enabledURLs := currentBackend.lb.Servers()
+					var newDisabledURLs []*url.URL
+					for _, url := range currentBackend.DisabledURLs {
+						if checkHealth(url, currentBackend.URL) {
+							log.Debugf("HealthCheck is up [%s]: Upsert in server list", url.String())
+							currentBackend.lb.UpsertServer(url, roundrobin.Weight(1))
+						} else {
+							newDisabledURLs = append(newDisabledURLs, url)
+						}
+					}
+					currentBackend.DisabledURLs = newDisabledURLs
+
+					for _, url := range enabledURLs {
+						if !checkHealth(url, currentBackend.URL) {
+							log.Debugf("HealthCheck has failed [%s]: Remove from server list", url.String())
+							currentBackend.lb.RemoveServer(url)
+							currentBackend.DisabledURLs = append(currentBackend.DisabledURLs, url)
+						}
+					}
+
+				}
+			}
+		})
+	}
+}
+
+func checkHealth(serverURL *url.URL, checkURL string) bool {
+	timeout := time.Duration(5 * time.Second)
+	client := http.Client{
+		Timeout: timeout,
+	}
+	resp, err := client.Get(serverURL.String() + checkURL)
+	if err != nil || resp.StatusCode != 200 {
+		return false
+	}
+	return true
+}

server.go

@@ -23,6 +23,7 @@ import (
 	"github.com/codegangsta/negroni"
 	"github.com/containous/mux"
 	"github.com/containous/traefik/cluster"
+	"github.com/containous/traefik/healthcheck"
 	"github.com/containous/traefik/log"
 	"github.com/containous/traefik/middlewares"
 	"github.com/containous/traefik/provider"
@@ -551,6 +552,9 @@ func (server *Server) loadConfig(configurations configs, globalConfiguration Glo
 	redirectHandlers := make(map[string]http.Handler)
 
 	backends := map[string]http.Handler{}
+
+	backendsHealthcheck := map[string]*healthcheck.BackendHealthCheck{}
+
 	backend2FrontendMap := map[string]string{}
 	for _, configuration := range configurations {
 		frontendNames := sortedFrontendNamesForConfig(configuration)
@@ -650,6 +654,9 @@ func (server *Server) loadConfig(configurations configs, globalConfiguration Glo
 									log.Errorf("Skipping frontend %s...", frontendName)
 									continue frontend
 								}
+								if configuration.Backends[frontend.Backend].HealthCheck != nil {
+									backendsHealthcheck[frontend.Backend] = healthcheck.NewBackendHealthCheck(configuration.Backends[frontend.Backend].HealthCheck.URL, rebalancer)
+								}
 							}
 						case types.Wrr:
 							log.Debugf("Creating load-balancer wrr")
@@ -673,6 +680,9 @@ func (server *Server) loadConfig(configurations configs, globalConfiguration Glo
 									continue frontend
 								}
 							}
+							if configuration.Backends[frontend.Backend].HealthCheck != nil {
+								backendsHealthcheck[frontend.Backend] = healthcheck.NewBackendHealthCheck(configuration.Backends[frontend.Backend].HealthCheck.URL, rr)
+							}
 						}
 						maxConns := configuration.Backends[frontend.Backend].MaxConn
 						if maxConns != nil && maxConns.Amount != 0 {
@@ -735,6 +745,7 @@ func (server *Server) loadConfig(configurations configs, globalConfiguration Glo
 			}
 		}
 	}
+	healthcheck.GetHealthCheck().SetBackendsConfiguration(server.routinesPool.Ctx(), backendsHealthcheck)
 	middlewares.SetBackend2FrontendMap(&backend2FrontendMap)
 	//sort routes
 	for _, serverEntryPoint := range serverEntryPoints {

types/types.go

@@ -4,10 +4,11 @@ import (
 	"encoding"
 	"errors"
 	"fmt"
-	"github.com/docker/libkv/store"
-	"github.com/ryanuber/go-glob"
 	"strconv"
 	"strings"
+
+	"github.com/docker/libkv/store"
+	"github.com/ryanuber/go-glob"
 )
 
 // Backend holds backend configuration.
@@ -16,6 +17,7 @@ type Backend struct {
 	CircuitBreaker *CircuitBreaker   `json:"circuitBreaker,omitempty"`
 	LoadBalancer   *LoadBalancer     `json:"loadBalancer,omitempty"`
 	MaxConn        *MaxConn          `json:"maxConn,omitempty"`
+	HealthCheck    *HealthCheck      `json:"healthCheck,omitempty"`
 }
 
 // MaxConn holds maximum connection configuration
@@ -35,6 +37,11 @@ type CircuitBreaker struct {
 	Expression string `json:"expression,omitempty"`
 }
 
+// HealthCheck holds HealthCheck configuration
+type HealthCheck struct {
+	URL string `json:"url,omitempty"`
+}
+
 // Server holds server configuration.
 type Server struct {
 	URL    string `json:"url,omitempty"`