forked from holygeek/traefik
-
Notifications
You must be signed in to change notification settings - Fork 0
/
healthcheck.go
212 lines (181 loc) · 5.79 KB
/
healthcheck.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
package healthcheck
import (
"context"
"fmt"
"net"
"net/http"
"net/url"
"strconv"
"sync"
"time"
"github.com/containous/traefik/log"
"github.com/containous/traefik/safe"
"github.com/go-kit/kit/metrics"
"github.com/vulcand/oxy/roundrobin"
)
var singleton *HealthCheck
var once sync.Once
// BalancerHandler includes functionality for load-balancing management.
type BalancerHandler interface {
ServeHTTP(w http.ResponseWriter, req *http.Request)
Servers() []*url.URL
RemoveServer(u *url.URL) error
UpsertServer(u *url.URL, options ...roundrobin.ServerOption) error
}
// metricsRegistry is a local interface in the health check package, exposing only the required metrics
// necessary for the health check package. This makes it easier for the tests.
type metricsRegistry interface {
BackendServerUpGauge() metrics.Gauge
}
// Options are the public health check options.
type Options struct {
Headers map[string]string
Hostname string
Scheme string
Path string
Port int
Transport http.RoundTripper
Interval time.Duration
LB BalancerHandler
}
func (opt Options) String() string {
return fmt.Sprintf("[Hostname: %s Headers: %v Path: %s Port: %d Interval: %s]", opt.Hostname, opt.Headers, opt.Path, opt.Port, opt.Interval)
}
// BackendConfig HealthCheck configuration for a backend
type BackendConfig struct {
Options
name string
disabledURLs []*url.URL
requestTimeout time.Duration
}
func (b *BackendConfig) newRequest(serverURL *url.URL) (*http.Request, error) {
u := &url.URL{}
*u = *serverURL
if len(b.Scheme) > 0 {
u.Scheme = b.Scheme
}
if b.Port != 0 {
u.Host = net.JoinHostPort(u.Hostname(), strconv.Itoa(b.Port))
}
u.Path += b.Path
return http.NewRequest(http.MethodGet, u.String(), nil)
}
// this function adds additional http headers and hostname to http.request
func (b *BackendConfig) addHeadersAndHost(req *http.Request) *http.Request {
if b.Options.Hostname != "" {
req.Host = b.Options.Hostname
}
for k, v := range b.Options.Headers {
req.Header.Set(k, v)
}
return req
}
// HealthCheck struct
type HealthCheck struct {
Backends map[string]*BackendConfig
metrics metricsRegistry
cancel context.CancelFunc
}
// SetBackendsConfiguration set backends configuration
func (hc *HealthCheck) SetBackendsConfiguration(parentCtx context.Context, backends map[string]*BackendConfig) {
hc.Backends = backends
if hc.cancel != nil {
hc.cancel()
}
ctx, cancel := context.WithCancel(parentCtx)
hc.cancel = cancel
for _, backend := range backends {
currentBackend := backend
safe.Go(func() {
hc.execute(ctx, currentBackend)
})
}
}
func (hc *HealthCheck) execute(ctx context.Context, backend *BackendConfig) {
log.Debugf("Initial health check for backend: %q", backend.name)
hc.checkBackend(backend)
ticker := time.NewTicker(backend.Interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
log.Debugf("Stopping current health check goroutines of backend: %s", backend.name)
return
case <-ticker.C:
log.Debugf("Refreshing health check for backend: %s", backend.name)
hc.checkBackend(backend)
}
}
}
func (hc *HealthCheck) checkBackend(backend *BackendConfig) {
enabledURLs := backend.LB.Servers()
var newDisabledURLs []*url.URL
for _, url := range backend.disabledURLs {
serverUpMetricValue := float64(0)
if err := checkHealth(url, backend); err == nil {
log.Warnf("Health check up: Returning to server list. Backend: %q URL: %q", backend.name, url.String())
backend.LB.UpsertServer(url, roundrobin.Weight(1))
serverUpMetricValue = 1
} else {
log.Warnf("Health check still failing. Backend: %q URL: %q Reason: %s", backend.name, url.String(), err)
newDisabledURLs = append(newDisabledURLs, url)
}
labelValues := []string{"backend", backend.name, "url", url.String()}
hc.metrics.BackendServerUpGauge().With(labelValues...).Set(serverUpMetricValue)
}
backend.disabledURLs = newDisabledURLs
for _, url := range enabledURLs {
serverUpMetricValue := float64(1)
if err := checkHealth(url, backend); err != nil {
log.Warnf("Health check failed: Remove from server list. Backend: %q URL: %q Reason: %s", backend.name, url.String(), err)
backend.LB.RemoveServer(url)
backend.disabledURLs = append(backend.disabledURLs, url)
serverUpMetricValue = 0
}
labelValues := []string{"backend", backend.name, "url", url.String()}
hc.metrics.BackendServerUpGauge().With(labelValues...).Set(serverUpMetricValue)
}
}
// GetHealthCheck returns the health check which is guaranteed to be a singleton.
func GetHealthCheck(metrics metricsRegistry) *HealthCheck {
once.Do(func() {
singleton = newHealthCheck(metrics)
})
return singleton
}
func newHealthCheck(metrics metricsRegistry) *HealthCheck {
return &HealthCheck{
Backends: make(map[string]*BackendConfig),
metrics: metrics,
}
}
// NewBackendConfig Instantiate a new BackendConfig
func NewBackendConfig(options Options, backendName string) *BackendConfig {
return &BackendConfig{
Options: options,
name: backendName,
requestTimeout: 5 * time.Second,
}
}
// 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.
func checkHealth(serverURL *url.URL, backend *BackendConfig) error {
req, err := backend.newRequest(serverURL)
if err != nil {
return fmt.Errorf("failed to create HTTP request: %s", err)
}
req = backend.addHeadersAndHost(req)
client := http.Client{
Timeout: backend.requestTimeout,
Transport: backend.Options.Transport,
}
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("HTTP request failed: %s", 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
}