traefik/pkg/healthcheck/healthcheck.go

434 lines
12 KiB
Go
Raw Normal View History

2016-11-26 18:48:49 +00:00
package healthcheck
import (
"context"
"errors"
"fmt"
"net"
2016-11-26 18:48:49 +00:00
"net/http"
"net/url"
"strconv"
"strings"
2016-11-26 18:48:49 +00:00
"sync"
"time"
2017-01-31 21:55:02 +00:00
2020-09-26 11:30:03 +00:00
gokitmetrics "github.com/go-kit/kit/metrics"
"github.com/traefik/traefik/v2/pkg/config/dynamic"
"github.com/traefik/traefik/v2/pkg/config/runtime"
"github.com/traefik/traefik/v2/pkg/log"
2020-09-26 11:30:03 +00:00
"github.com/traefik/traefik/v2/pkg/metrics"
"github.com/traefik/traefik/v2/pkg/safe"
2017-01-31 21:55:02 +00:00
"github.com/vulcand/oxy/roundrobin"
2016-11-26 18:48:49 +00:00
)
const (
serverUp = "UP"
serverDown = "DOWN"
)
2020-07-07 12:42:03 +00:00
var (
singleton *HealthCheck
once sync.Once
)
2016-11-26 18:48:49 +00:00
2020-05-11 10:06:07 +00:00
// Balancer is the set of operations required to manage the list of servers in a load-balancer.
type Balancer interface {
2018-06-11 09:36:03 +00:00
Servers() []*url.URL
RemoveServer(u *url.URL) error
UpsertServer(u *url.URL, options ...roundrobin.ServerOption) error
}
// BalancerHandler includes functionality for load-balancing management.
type BalancerHandler interface {
ServeHTTP(w http.ResponseWriter, req *http.Request)
Balancer
}
// BalancerStatusHandler is an http Handler that does load-balancing,
2022-02-21 11:40:09 +00:00
// and updates its parents of its status.
type BalancerStatusHandler interface {
BalancerHandler
StatusUpdater
}
2020-09-26 11:30:03 +00:00
type metricsHealthcheck struct {
serverUpGauge gokitmetrics.Gauge
2016-11-26 18:48:49 +00:00
}
// Options are the public health check options.
type Options struct {
Headers map[string]string
Hostname string
Scheme string
Path string
Method string
Port int
FollowRedirects bool
Transport http.RoundTripper
Interval time.Duration
Timeout time.Duration
LB Balancer
}
func (opt Options) String() string {
return fmt.Sprintf("[Hostname: %s Headers: %v Path: %s Method: %s Port: %d Interval: %s Timeout: %s FollowRedirects: %v]", opt.Hostname, opt.Headers, opt.Path, opt.Method, opt.Port, opt.Interval, opt.Timeout, opt.FollowRedirects)
}
type backendURL struct {
url *url.URL
weight int
}
2020-05-11 10:06:07 +00:00
// BackendConfig HealthCheck configuration for a backend.
2018-06-11 09:36:03 +00:00
type BackendConfig struct {
Options
name string
disabledURLs []backendURL
2016-11-26 18:48:49 +00:00
}
2018-06-11 09:36:03 +00:00
func (b *BackendConfig) newRequest(serverURL *url.URL) (*http.Request, error) {
2018-11-15 14:50:03 +00:00
u, err := serverURL.Parse(b.Path)
if err != nil {
return nil, err
}
2016-11-26 18:48:49 +00:00
2018-06-11 09:36:03 +00:00
if len(b.Scheme) > 0 {
u.Scheme = b.Scheme
}
2016-11-26 18:48:49 +00:00
2018-06-11 09:36:03 +00:00
if b.Port != 0 {
u.Host = net.JoinHostPort(u.Hostname(), strconv.Itoa(b.Port))
}
2016-11-26 18:48:49 +00:00
2018-10-23 08:10:04 +00:00
return http.NewRequest(http.MethodGet, u.String(), http.NoBody)
}
// setRequestOptions sets all request options present on the BackendConfig.
func (b *BackendConfig) setRequestOptions(req *http.Request) *http.Request {
2018-06-11 09:36:03 +00:00
if b.Options.Hostname != "" {
req.Host = b.Options.Hostname
}
for k, v := range b.Options.Headers {
req.Header.Set(k, v)
}
if b.Options.Method != "" {
req.Method = strings.ToUpper(b.Options.Method)
}
2018-06-11 09:36:03 +00:00
return req
}
2020-05-11 10:06:07 +00:00
// HealthCheck struct.
2018-06-11 09:36:03 +00:00
type HealthCheck struct {
Backends map[string]*BackendConfig
2020-09-26 11:30:03 +00:00
metrics metricsHealthcheck
2018-06-11 09:36:03 +00:00
cancel context.CancelFunc
2016-11-26 18:48:49 +00:00
}
2020-05-11 10:06:07 +00:00
// SetBackendsConfiguration set backends configuration.
2018-06-11 09:36:03 +00:00
func (hc *HealthCheck) SetBackendsConfiguration(parentCtx context.Context, backends map[string]*BackendConfig) {
2016-11-26 18:48:49 +00:00
hc.Backends = backends
if hc.cancel != nil {
hc.cancel()
}
2017-01-31 21:55:02 +00:00
ctx, cancel := context.WithCancel(parentCtx)
hc.cancel = cancel
2016-11-26 18:48:49 +00:00
2018-01-15 16:27:37 +00:00
for _, backend := range backends {
currentBackend := backend
2017-01-31 21:55:02 +00:00
safe.Go(func() {
hc.execute(ctx, currentBackend)
2017-01-31 21:55:02 +00:00
})
}
2016-11-26 18:48:49 +00:00
}
2018-06-11 09:36:03 +00:00
func (hc *HealthCheck) execute(ctx context.Context, backend *BackendConfig) {
2019-09-13 17:28:04 +00:00
logger := log.FromContext(ctx)
2019-09-13 17:28:04 +00:00
logger.Debugf("Initial health check for backend: %q", backend.name)
hc.checkServersLB(ctx, backend)
2019-09-13 17:28:04 +00:00
ticker := time.NewTicker(backend.Interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
2019-09-13 17:28:04 +00:00
logger.Debugf("Stopping current health check goroutines of backend: %s", backend.name)
return
case <-ticker.C:
logger.Debugf("Routine health check refresh for backend: %s", backend.name)
hc.checkServersLB(ctx, backend)
}
}
}
func (hc *HealthCheck) checkServersLB(ctx context.Context, backend *BackendConfig) {
2019-09-13 17:28:04 +00:00
logger := log.FromContext(ctx)
enabledURLs := backend.LB.Servers()
2020-09-26 11:30:03 +00:00
var newDisabledURLs []backendURL
for _, disabledURL := range backend.disabledURLs {
2020-09-26 11:30:03 +00:00
serverUpMetricValue := float64(0)
if err := checkHealth(disabledURL.url, backend); err == nil {
logger.Warnf("Health check up: returning to server list. Backend: %q URL: %q Weight: %d",
backend.name, disabledURL.url.String(), disabledURL.weight)
if err = backend.LB.UpsertServer(disabledURL.url, roundrobin.Weight(disabledURL.weight)); err != nil {
2019-09-13 17:28:04 +00:00
logger.Error(err)
2018-08-06 18:00:03 +00:00
}
2020-09-26 11:30:03 +00:00
serverUpMetricValue = 1
} else {
logger.Warnf("Health check still failing. Backend: %q URL: %q Reason: %s", backend.name, disabledURL.url.String(), err)
newDisabledURLs = append(newDisabledURLs, disabledURL)
}
2020-09-26 11:30:03 +00:00
labelValues := []string{"service", backend.name, "url", disabledURL.url.String()}
hc.metrics.serverUpGauge.With(labelValues...).Set(serverUpMetricValue)
}
2020-09-26 11:30:03 +00:00
backend.disabledURLs = newDisabledURLs
for _, enabledURL := range enabledURLs {
2020-09-26 11:30:03 +00:00
serverUpMetricValue := float64(1)
if err := checkHealth(enabledURL, backend); err != nil {
weight := 1
rr, ok := backend.LB.(*roundrobin.RoundRobin)
if ok {
var gotWeight bool
weight, gotWeight = rr.ServerWeight(enabledURL)
if !gotWeight {
weight = 1
}
}
2020-09-26 11:30:03 +00:00
logger.Warnf("Health check failed, removing from server list. Backend: %q URL: %q Weight: %d Reason: %s",
backend.name, enabledURL.String(), weight, err)
if err := backend.LB.RemoveServer(enabledURL); err != nil {
2019-09-13 17:28:04 +00:00
logger.Error(err)
2018-08-06 18:00:03 +00:00
}
2020-09-26 11:30:03 +00:00
backend.disabledURLs = append(backend.disabledURLs, backendURL{enabledURL, weight})
2020-09-26 11:30:03 +00:00
serverUpMetricValue = 0
}
2020-09-26 11:30:03 +00:00
labelValues := []string{"service", backend.name, "url", enabledURL.String()}
2020-09-26 11:30:03 +00:00
hc.metrics.serverUpGauge.With(labelValues...).Set(serverUpMetricValue)
}
}
2018-06-11 09:36:03 +00:00
// GetHealthCheck returns the health check which is guaranteed to be a singleton.
2020-09-26 11:30:03 +00:00
func GetHealthCheck(registry metrics.Registry) *HealthCheck {
2018-06-11 09:36:03 +00:00
once.Do(func() {
2020-09-26 11:30:03 +00:00
singleton = newHealthCheck(registry)
2018-06-11 09:36:03 +00:00
})
return singleton
}
2020-09-26 11:30:03 +00:00
func newHealthCheck(registry metrics.Registry) *HealthCheck {
2018-06-11 09:36:03 +00:00
return &HealthCheck{
Backends: make(map[string]*BackendConfig),
2020-09-26 11:30:03 +00:00
metrics: metricsHealthcheck{
serverUpGauge: registry.ServiceServerUpGauge(),
},
2018-04-16 09:40:03 +00:00
}
2018-06-11 09:36:03 +00:00
}
2020-05-11 10:06:07 +00:00
// NewBackendConfig Instantiate a new BackendConfig.
2018-06-11 09:36:03 +00:00
func NewBackendConfig(options Options, backendName string) *BackendConfig {
return &BackendConfig{
2018-09-27 18:16:03 +00:00
Options: options,
name: backendName,
2018-04-16 09:40:03 +00:00
}
}
// checkHealth returns a nil error in case it was successful and otherwise
// a non-nil error with a meaningful description why the health check failed.
2018-06-11 09:36:03 +00:00
func checkHealth(serverURL *url.URL, backend *BackendConfig) error {
req, err := backend.newRequest(serverURL)
if err != nil {
2020-05-11 10:06:07 +00:00
return fmt.Errorf("failed to create HTTP request: %w", err)
}
req = backend.setRequestOptions(req)
client := http.Client{
2018-09-27 18:16:03 +00:00
Timeout: backend.Options.Timeout,
Transport: backend.Options.Transport,
2016-11-26 18:48:49 +00:00
}
if !backend.FollowRedirects {
client.CheckRedirect = func(req *http.Request, via []*http.Request) error {
return http.ErrUseLastResponse
}
}
resp, err := client.Do(req)
if err != nil {
2020-05-11 10:06:07 +00:00
return fmt.Errorf("HTTP request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusBadRequest {
return fmt.Errorf("received error status code: %v", resp.StatusCode)
}
return nil
2016-11-26 18:48:49 +00:00
}
// StatusUpdater should be implemented by a service that, when its status
// changes (e.g. all if its children are down), needs to propagate upwards (to
// their parent(s)) that change.
type StatusUpdater interface {
RegisterStatusUpdater(fn func(up bool)) error
}
2020-05-11 10:06:07 +00:00
// NewLBStatusUpdater returns a new LbStatusUpdater.
func NewLBStatusUpdater(bh BalancerHandler, info *runtime.ServiceInfo, hc *dynamic.ServerHealthCheck) *LbStatusUpdater {
return &LbStatusUpdater{
BalancerHandler: bh,
serviceInfo: info,
wantsHealthCheck: hc != nil,
}
}
// LbStatusUpdater wraps a BalancerHandler and a ServiceInfo,
// so it can keep track of the status of a server in the ServiceInfo.
type LbStatusUpdater struct {
BalancerHandler
serviceInfo *runtime.ServiceInfo // can be nil
updaters []func(up bool)
wantsHealthCheck bool
}
// RegisterStatusUpdater adds fn to the list of hooks that are run when the
// status of the Balancer changes.
// Not thread safe.
func (lb *LbStatusUpdater) RegisterStatusUpdater(fn func(up bool)) error {
if !lb.wantsHealthCheck {
return errors.New("healthCheck not enabled in config for this loadbalancer service")
}
lb.updaters = append(lb.updaters, fn)
return nil
}
// RemoveServer removes the given server from the BalancerHandler,
// and updates the status of the server to "DOWN".
func (lb *LbStatusUpdater) RemoveServer(u *url.URL) error {
// TODO(mpl): when we have the freedom to change the signature of RemoveServer
// (kinda stuck because of oxy for now), let's pass around a context to improve
// logging.
ctx := context.TODO()
upBefore := len(lb.BalancerHandler.Servers()) > 0
err := lb.BalancerHandler.RemoveServer(u)
if err != nil {
return err
}
if lb.serviceInfo != nil {
lb.serviceInfo.UpdateServerStatus(u.String(), serverDown)
}
log.FromContext(ctx).Debugf("child %s now %s", u.String(), serverDown)
if !upBefore {
// we were already down, and we still are, no need to propagate.
log.FromContext(ctx).Debugf("Still %s, no need to propagate", serverDown)
return nil
}
if len(lb.BalancerHandler.Servers()) > 0 {
// we were up, and we still are, no need to propagate
log.FromContext(ctx).Debugf("Still %s, no need to propagate", serverUp)
return nil
}
log.FromContext(ctx).Debugf("Propagating new %s status", serverDown)
for _, fn := range lb.updaters {
fn(false)
}
return nil
}
// UpsertServer adds the given server to the BalancerHandler,
// and updates the status of the server to "UP".
func (lb *LbStatusUpdater) UpsertServer(u *url.URL, options ...roundrobin.ServerOption) error {
ctx := context.TODO()
upBefore := len(lb.BalancerHandler.Servers()) > 0
err := lb.BalancerHandler.UpsertServer(u, options...)
if err != nil {
return err
}
if lb.serviceInfo != nil {
lb.serviceInfo.UpdateServerStatus(u.String(), serverUp)
}
log.FromContext(ctx).Debugf("child %s now %s", u.String(), serverUp)
if upBefore {
// we were up, and we still are, no need to propagate
log.FromContext(ctx).Debugf("Still %s, no need to propagate", serverUp)
return nil
}
log.FromContext(ctx).Debugf("Propagating new %s status", serverUp)
for _, fn := range lb.updaters {
fn(true)
}
return nil
}
// Balancers is a list of Balancers(s) that implements the Balancer interface.
type Balancers []Balancer
// Servers returns the deduplicated server URLs from all the Balancer.
// Note that the deduplication is only possible because all the underlying
// balancers are of the same kind (the oxy implementation).
// The comparison property is the same as the one found at:
// https://github.com/vulcand/oxy/blob/fb2728c857b7973a27f8de2f2190729c0f22cf49/roundrobin/rr.go#L347.
func (b Balancers) Servers() []*url.URL {
seen := make(map[string]struct{})
var servers []*url.URL
for _, lb := range b {
for _, server := range lb.Servers() {
key := serverKey(server)
if _, ok := seen[key]; ok {
continue
}
servers = append(servers, server)
seen[key] = struct{}{}
}
}
return servers
}
// RemoveServer removes the given server from all the Balancer,
// and updates the status of the server to "DOWN".
func (b Balancers) RemoveServer(u *url.URL) error {
for _, lb := range b {
if err := lb.RemoveServer(u); err != nil {
return err
}
}
return nil
}
// UpsertServer adds the given server to all the Balancer,
// and updates the status of the server to "UP".
func (b Balancers) UpsertServer(u *url.URL, options ...roundrobin.ServerOption) error {
for _, lb := range b {
if err := lb.UpsertServer(u, options...); err != nil {
return err
}
}
return nil
}
func serverKey(u *url.URL) string {
return u.Path + u.Host + u.Scheme
}