Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS
Mirrored from https://github.com/Slazee/caddy.git
Synced Nov 14, 05:15 UTC
reverse_proxy: Implement active health checks
Matthew Holt committed
Sep 3, 2019 at 12:10 UTC
ccfb12347b1d2f65b279352116527df430e0fba6
3 files changed
+309
-74
modules/caddyhttp/reverseproxy/healthchecks.go
new
+220
@@ -0,0 +1,220 @@
1
+// Copyright 2015 Matthew Holt and The Caddy Authors
2
+//
3
+// Licensed under the Apache License, Version 2.0 (the "License");
4
+// you may not use this file except in compliance with the License.
5
+// You may obtain a copy of the License at
6
+//
7
+// http://www.apache.org/licenses/LICENSE-2.0
8
+//
9
+// Unless required by applicable law or agreed to in writing, software
10
+// distributed under the License is distributed on an "AS IS" BASIS,
11
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
+// See the License for the specific language governing permissions and
13
+// limitations under the License.
14
+
15
+package reverseproxy
16
+
17
+import (
18
+ "fmt"
19
+ "io"
20
+ "io/ioutil"
21
+ "log"
22
+ "net"
23
+ "net/http"
24
+ "net/url"
25
+ "regexp"
26
+ "strconv"
27
+ "time"
28
+
29
+ "github.com/caddyserver/caddy/v2"
30
+ "github.com/caddyserver/caddy/v2/modules/caddyhttp"
31
+)
32
+
33
+type HealthChecks struct {
34
+ Active *ActiveHealthChecks `json:"active,omitempty"`
35
+ Passive *PassiveHealthChecks `json:"passive,omitempty"`
36
+}
37
+
38
+type ActiveHealthChecks struct {
39
+ Path string `json:"path,omitempty"`
40
+ Port int `json:"port,omitempty"`
41
+ Interval caddy.Duration `json:"interval,omitempty"`
42
+ Timeout caddy.Duration `json:"timeout,omitempty"`
43
+ MaxSize int64 `json:"max_size,omitempty"`
44
+ ExpectStatus int `json:"expect_status,omitempty"`
45
+ ExpectBody string `json:"expect_body,omitempty"`
46
+
47
+ stopChan chan struct{}
48
+ httpClient *http.Client
49
+ bodyRegexp *regexp.Regexp
50
+}
51
+
52
+type PassiveHealthChecks struct {
53
+ MaxFails int `json:"max_fails,omitempty"`
54
+ FailDuration caddy.Duration `json:"fail_duration,omitempty"`
55
+ UnhealthyRequestCount int `json:"unhealthy_request_count,omitempty"`
56
+ UnhealthyStatus []int `json:"unhealthy_status,omitempty"`
57
+ UnhealthyLatency caddy.Duration `json:"unhealthy_latency,omitempty"`
58
+}
59
+
60
+func (h *Handler) activeHealthChecker() {
61
+ ticker := time.NewTicker(time.Duration(h.HealthChecks.Active.Interval))
62
+ h.doActiveHealthChecksForAllHosts()
63
+ for {
64
+ select {
65
+ case <-ticker.C:
66
+ h.doActiveHealthChecksForAllHosts()
67
+ case <-h.HealthChecks.Active.stopChan:
68
+ ticker.Stop()
69
+ return
70
+ }
71
+ }
72
+}
73
+
74
+func (h *Handler) doActiveHealthChecksForAllHosts() {
75
+ hosts.Range(func(key, value interface{}) bool {
76
+ addr := key.(string)
77
+ host := value.(Host)
78
+
79
+ go func(addr string, host Host) {
80
+ err := h.doActiveHealthCheck(addr, host)
81
+ if err != nil {
82
+ log.Printf("[ERROR] reverse_proxy: active health check for host %s: %v", addr, err)
83
+ }
84
+ }(addr, host)
85
+
86
+ // continue to iterate all hosts
87
+ return true
88
+ })
89
+}
90
+
91
+// doActiveHealthCheck performs a health check to host which
92
+// can be reached at address hostAddr. The actual address for
93
+// the request will be built according to active health checker
94
+// config. The health status of the host will be updated
95
+// according to whether it passes the health check. An error is
96
+// returned only if the health check fails to occur or if marking
97
+// the host's health status fails.
98
+func (h *Handler) doActiveHealthCheck(hostAddr string, host Host) error {
99
+ // create the URL for the health check
100
+ u, err := url.Parse(hostAddr)
101
+ if err != nil {
102
+ return err
103
+ }
104
+ if h.HealthChecks.Active.Path != "" {
105
+ u.Path = h.HealthChecks.Active.Path
106
+ }
107
+ if h.HealthChecks.Active.Port != 0 {
108
+ portStr := strconv.Itoa(h.HealthChecks.Active.Port)
109
+ u.Host = net.JoinHostPort(u.Hostname(), portStr)
110
+ }
111
+
112
+ req, err := http.NewRequest(http.MethodGet, u.String(), nil)
113
+ if err != nil {
114
+ return err
115
+ }
116
+
117
+ // do the request, careful to tame the response body
118
+ resp, err := h.HealthChecks.Active.httpClient.Do(req)
119
+ if err != nil {
120
+ log.Printf("[INFO] reverse_proxy: active health check: %s is down (HTTP request failed: %v)", hostAddr, err)
121
+ _, err2 := host.SetHealthy(false)
122
+ if err2 != nil {
123
+ return fmt.Errorf("marking unhealthy: %v", err2)
124
+ }
125
+ return nil
126
+ }
127
+ var body io.Reader = resp.Body
128
+ if h.HealthChecks.Active.MaxSize > 0 {
129
+ body = io.LimitReader(body, h.HealthChecks.Active.MaxSize)
130
+ }
131
+ defer func() {
132
+ // drain any remaining body so connection can be re-used
133
+ io.Copy(ioutil.Discard, body)
134
+ resp.Body.Close()
135
+ }()
136
+
137
+ // if status code is outside criteria, mark down
138
+ if h.HealthChecks.Active.ExpectStatus > 0 {
139
+ if !caddyhttp.StatusCodeMatches(resp.StatusCode, h.HealthChecks.Active.ExpectStatus) {
140
+ log.Printf("[INFO] reverse_proxy: active health check: %s is down (status code %d unexpected)", hostAddr, resp.StatusCode)
141
+ _, err := host.SetHealthy(false)
142
+ if err != nil {
143
+ return fmt.Errorf("marking unhealthy: %v", err)
144
+ }
145
+ return nil
146
+ }
147
+ } else if resp.StatusCode < 200 || resp.StatusCode >= 400 {
148
+ log.Printf("[INFO] reverse_proxy: active health check: %s is down (status code %d out of tolerances)", hostAddr, resp.StatusCode)
149
+ _, err := host.SetHealthy(false)
150
+ if err != nil {
151
+ return fmt.Errorf("marking unhealthy: %v", err)
152
+ }
153
+ return nil
154
+ }
155
+
156
+ // if body does not match regex, mark down
157
+ if h.HealthChecks.Active.bodyRegexp != nil {
158
+ bodyBytes, err := ioutil.ReadAll(body)
159
+ if err != nil {
160
+ log.Printf("[INFO] reverse_proxy: active health check: %s is down (failed to read response body)", hostAddr)
161
+ _, err := host.SetHealthy(false)
162
+ if err != nil {
163
+ return fmt.Errorf("marking unhealthy: %v", err)
164
+ }
165
+ return nil
166
+ }
167
+ if !h.HealthChecks.Active.bodyRegexp.Match(bodyBytes) {
168
+ log.Printf("[INFO] reverse_proxy: active health check: %s is down (response body failed expectations)", hostAddr)
169
+ _, err := host.SetHealthy(false)
170
+ if err != nil {
171
+ return fmt.Errorf("marking unhealthy: %v", err)
172
+ }
173
+ return nil
174
+ }
175
+ }
176
+
177
+ // passed health check parameters, so mark as healthy
178
+ swapped, err := host.SetHealthy(true)
179
+ if swapped {
180
+ log.Printf("[INFO] reverse_proxy: active health check: %s is back up", hostAddr)
181
+ }
182
+ if err != nil {
183
+ return fmt.Errorf("marking healthy: %v", err)
184
+ }
185
+
186
+ return nil
187
+}
188
+
189
+// countFailure is used with passive health checks. It
190
+// remembers 1 failure for upstream for the configured
191
+// duration. If passive health checks are disabled or
192
+// failure expiry is 0, this is a no-op.
193
+func (h Handler) countFailure(upstream *Upstream) {
194
+ // only count failures if passive health checking is enabled
195
+ // and if failures are configured have a non-zero expiry
196
+ if h.HealthChecks == nil || h.HealthChecks.Passive == nil {
197
+ return
198
+ }
199
+ failDuration := time.Duration(h.HealthChecks.Passive.FailDuration)
200
+ if failDuration == 0 {
201
+ return
202
+ }
203
+
204
+ // count failure immediately
205
+ err := upstream.Host.CountFail(1)
206
+ if err != nil {
207
+ log.Printf("[ERROR] proxy: upstream %s: counting failure: %v",
208
+ upstream.hostURL, err)
209
+ }
210
+
211
+ // forget it later
212
+ go func(host Host, failDuration time.Duration) {
213
+ time.Sleep(failDuration)
214
+ err := host.CountFail(-1)
215
+ if err != nil {
216
+ log.Printf("[ERROR] proxy: upstream %s: expiring failure: %v",
217
+ upstream.hostURL, err)
218
+ }
219
+ }(upstream.Host, failDuration)
220
+}
modules/caddyhttp/reverseproxy/reverseproxy.go
+81
-74
@@ -15,15 +15,14 @@
15
package reverseproxy
16
17
import (
18
- "bytes"
18
"context"
19
"encoding/json"
20
"fmt"
21
"io"
23
- "log"
22
"net"
23
"net/http"
24
"net/url"
25
+ "regexp"
26
"strings"
27
"sync"
28
"sync/atomic"
@@ -90,11 +89,41 @@ func (h *Handler) Provision(ctx caddy.Context) error {
89
if h.LoadBalancing.TryDuration > 0 && h.LoadBalancing.TryInterval == 0 {
90
// a non-zero try_duration with a zero try_interval
91
// will always spin the CPU for try_duration if the
93
- // upstream is local or low-latency; default to some
94
- // sane waiting period before try attempts
92
+ // upstream is local or low-latency; avoid that by
93
+ // defaulting to a sane wait period between attempts
94
h.LoadBalancing.TryInterval = caddy.Duration(250 * time.Millisecond)
95
}
96
97
+ // if active health checks are enabled, configure them and start a worker
98
+ if h.HealthChecks != nil &&
99
+ h.HealthChecks.Active != nil &&
100
+ (h.HealthChecks.Active.Path != "" || h.HealthChecks.Active.Port != 0) {
101
+ timeout := time.Duration(h.HealthChecks.Active.Timeout)
102
+ if timeout == 0 {
103
+ timeout = 10 * time.Second
104
+ }
105
+
106
+ h.HealthChecks.Active.stopChan = make(chan struct{})
107
+ h.HealthChecks.Active.httpClient = &http.Client{
108
+ Timeout: timeout,
109
+ Transport: h.Transport,
110
+ }
111
+
112
+ if h.HealthChecks.Active.Interval == 0 {
113
+ h.HealthChecks.Active.Interval = caddy.Duration(30 * time.Second)
114
+ }
115
+
116
+ if h.HealthChecks.Active.ExpectBody != "" {
117
+ var err error
118
+ h.HealthChecks.Active.bodyRegexp, err = regexp.Compile(h.HealthChecks.Active.ExpectBody)
119
+ if err != nil {
120
+ return fmt.Errorf("expect_body: compiling regular expression: %v", err)
121
+ }
122
+ }
123
+
124
+ go h.activeHealthChecker()
125
+ }
126
+
127
for _, upstream := range h.Upstreams {
128
// url parser requires a scheme
129
if !strings.Contains(upstream.Address, "://") {
@@ -130,8 +159,6 @@ func (h *Handler) Provision(ctx caddy.Context) error {
159
upstream.MaxRequests = h.HealthChecks.Passive.UnhealthyRequestCount
160
}
161
133
- // TODO: active health checks
134
-
162
if h.HealthChecks != nil {
163
// upstreams need independent access to the passive
164
// health check policy so they can, you know, passively
@@ -143,11 +170,20 @@ func (h *Handler) Provision(ctx caddy.Context) error {
170
return nil
171
}
172
173
+// Cleanup cleans up the resources made by h during provisioning.
174
func (h *Handler) Cleanup() error {
147
- // TODO: finish this up, make sure it takes care of any active health checkers or whatever
175
+ // stop the active health checker
176
+ if h.HealthChecks != nil &&
177
+ h.HealthChecks.Active != nil &&
178
+ h.HealthChecks.Active.stopChan != nil {
179
+ close(h.HealthChecks.Active.stopChan)
180
+ }
181
+
182
+ // remove hosts from our config from the pool
183
for _, upstream := range h.Upstreams {
184
hosts.Delete(upstream.hostURL.String())
185
}
186
+
187
return nil
188
}
189
@@ -539,38 +575,6 @@ func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, er
575
}
576
}
577
542
-// countFailure remembers 1 failure for upstream for the
543
-// configured duration. If passive health checks are
544
-// disabled or failure expiry is 0, this is a no-op.
545
-func (h Handler) countFailure(upstream *Upstream) {
546
- // only count failures if passive health checking is enabled
547
- // and if failures are configured have a non-zero expiry
548
- if h.HealthChecks == nil || h.HealthChecks.Passive == nil {
549
- return
550
- }
551
- failDuration := time.Duration(h.HealthChecks.Passive.FailDuration)
552
- if failDuration == 0 {
553
- return
554
- }
555
-
556
- // count failure immediately
557
- err := upstream.Host.CountFail(1)
558
- if err != nil {
559
- log.Printf("[ERROR] proxy: upstream %s: counting failure: %v",
560
- upstream.hostURL, err)
561
- }
562
-
563
- // forget it later
564
- go func(host Host, failDuration time.Duration) {
565
- time.Sleep(failDuration)
566
- err := host.CountFail(-1)
567
- if err != nil {
568
- log.Printf("[ERROR] proxy: upstream %s: expiring failure: %v",
569
- upstream.hostURL, err)
570
- }
571
- }(upstream.Host, failDuration)
572
-}
573
-
578
type writeFlusher interface {
579
io.Writer
580
http.Flusher
@@ -722,29 +726,6 @@ type Selector interface {
726
Select(HostPool, *http.Request) *Upstream
727
}
728
725
-type HealthChecks struct {
726
- Active *ActiveHealthChecks `json:"active,omitempty"`
727
- Passive *PassiveHealthChecks `json:"passive,omitempty"`
728
-}
729
-
730
-type ActiveHealthChecks struct {
731
- Path string `json:"path,omitempty"`
732
- Port int `json:"port,omitempty"`
733
- Interval caddy.Duration `json:"interval,omitempty"`
734
- Timeout caddy.Duration `json:"timeout,omitempty"`
735
- MaxSize int `json:"max_size,omitempty"`
736
- ExpectStatus int `json:"expect_status,omitempty"`
737
- ExpectBody string `json:"expect_body,omitempty"`
738
-}
739
-
740
-type PassiveHealthChecks struct {
741
- MaxFails int `json:"max_fails,omitempty"`
742
- FailDuration caddy.Duration `json:"fail_duration,omitempty"`
743
- UnhealthyRequestCount int `json:"unhealthy_request_count,omitempty"`
744
- UnhealthyStatus []int `json:"unhealthy_status,omitempty"`
745
- UnhealthyLatency caddy.Duration `json:"unhealthy_latency,omitempty"`
746
-}
747
-
729
// Hop-by-hop headers. These are removed when sent to the backend.
730
// As of RFC 7230, hop-by-hop headers are required to appear in the
731
// Connection header field. These are the headers defined by the
@@ -762,22 +743,33 @@ var hopHeaders = []string{
743
"Upgrade",
744
}
745
765
-var bufPool = sync.Pool{
766
- New: func() interface{} {
767
- return new(bytes.Buffer)
768
- },
769
-}
770
-
771
-//////////////////////////////////
772
-// TODO:
773
-
746
+// Host represents a remote host which can be proxied to.
747
+// Its methods must be safe for concurrent use.
748
type Host interface {
749
+ // NumRequests returns the numnber of requests
750
+ // currently in process with the host.
751
NumRequests() int
752
+
753
+ // Fails returns the count of recent failures.
754
Fails() int
755
+
756
+ // Unhealthy returns true if the backend is unhealthy.
757
Unhealthy() bool
758
759
+ // CountRequest counts the given number of requests
760
+ // as currently in process with the host. The count
761
+ // should not go below 0.
762
CountRequest(int) error
763
+
764
+ // CountFail counts the given number of failures
765
+ // with the host. The count should not go below 0.
766
CountFail(int) error
767
+
768
+ // SetHealthy marks the host as either healthy (true)
769
+ // or unhealthy (false). If the given status is the
770
+ // same, this should be a no-op. It returns true if
771
+ // the given status was different, false otherwise.
772
+ SetHealthy(bool) (bool, error)
773
}
774
775
type HostPool []*Upstream
@@ -788,13 +780,13 @@ type upstreamHost struct {
780
unhealthy int32
781
}
782
791
-func (uh upstreamHost) NumRequests() int {
783
+func (uh *upstreamHost) NumRequests() int {
784
return int(atomic.LoadInt64(&uh.numRequests))
785
}
794
-func (uh upstreamHost) Fails() int {
786
+func (uh *upstreamHost) Fails() int {
787
return int(atomic.LoadInt64(&uh.fails))
788
}
797
-func (uh upstreamHost) Unhealthy() bool {
789
+func (uh *upstreamHost) Unhealthy() bool {
790
return atomic.LoadInt32(&uh.unhealthy) == 1
791
}
792
func (uh *upstreamHost) CountRequest(delta int) error {
@@ -811,6 +803,14 @@ func (uh *upstreamHost) CountFail(delta int) error {
803
}
804
return nil
805
}
806
+func (uh *upstreamHost) SetHealthy(healthy bool) (bool, error) {
807
+ var unhealthy, compare int32 = 1, 0
808
+ if healthy {
809
+ unhealthy, compare = 0, 1
810
+ }
811
+ swapped := atomic.CompareAndSwapInt32(&uh.unhealthy, compare, unhealthy)
812
+ return swapped, nil
813
+}
814
815
type Upstream struct {
816
Host `json:"-"`
@@ -854,6 +854,13 @@ var hosts = caddy.NewUsagePool()
854
type UpstreamProvider interface {
855
}
856
857
+// TODO: see if we can use this
858
+// var bufPool = sync.Pool{
859
+// New: func() interface{} {
860
+// return new(bytes.Buffer)
861
+// },
862
+// }
863
+
864
// Interface guards
865
var (
866
_ caddyhttp.MiddlewareHandler = (*Handler)(nil)
usagepool.go
+8
@@ -80,6 +80,14 @@ func (up *UsagePool) LoadOrStore(key, val interface{}) (actual interface{}, load
80
return
81
}
82
83
+// Range iterates the pool the same way sync.Map.Range does.
84
+// This does not affect usage counts.
85
+func (up *UsagePool) Range(f func(key, value interface{}) bool) {
86
+ up.pool.Range(func(key, value interface{}) bool {
87
+ return f(key, value.(*usagePoolVal).value)
88
+ })
89
+}
90
+
91
type usagePoolVal struct {
92
usage int32 // accessed atomically; must be 64-bit aligned for 32-bit systems
93
value interface{}