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
Some cleanup and godoc
Matthew Holt committed
Sep 3, 2019 at 16:56 UTC
652460e03e11a037d9f86b09b3546c9e42733d2d
6 files changed
+428
-362
modules/caddyhttp/reverseproxy/healthchecks.go
+12
@@ -30,11 +30,15 @@ import (
30
"github.com/caddyserver/caddy/v2/modules/caddyhttp"
31
)
32
33
+// HealthChecks holds configuration related to health checking.
34
type HealthChecks struct {
35
Active *ActiveHealthChecks `json:"active,omitempty"`
36
Passive *PassiveHealthChecks `json:"passive,omitempty"`
37
}
38
39
+// ActiveHealthChecks holds configuration related to active
40
+// health checks (that is, health checks which occur in a
41
+// background goroutine independently).
42
type ActiveHealthChecks struct {
43
Path string `json:"path,omitempty"`
44
Port int `json:"port,omitempty"`
@@ -49,6 +53,9 @@ type ActiveHealthChecks struct {
53
bodyRegexp *regexp.Regexp
54
}
55
56
+// PassiveHealthChecks holds configuration related to passive
57
+// health checks (that is, health checks which occur during
58
+// the normal flow of request proxying).
59
type PassiveHealthChecks struct {
60
MaxFails int `json:"max_fails,omitempty"`
61
FailDuration caddy.Duration `json:"fail_duration,omitempty"`
@@ -57,6 +64,9 @@ type PassiveHealthChecks struct {
64
UnhealthyLatency caddy.Duration `json:"unhealthy_latency,omitempty"`
65
}
66
67
+// activeHealthChecker runs active health checks on a
68
+// regular basis and blocks until
69
+// h.HealthChecks.Active.stopChan is closed.
70
func (h *Handler) activeHealthChecker() {
71
ticker := time.NewTicker(time.Duration(h.HealthChecks.Active.Interval))
72
h.doActiveHealthChecksForAllHosts()
@@ -71,6 +81,8 @@ func (h *Handler) activeHealthChecker() {
81
}
82
}
83
84
+// doActiveHealthChecksForAllHosts immediately performs a
85
+// health checks for all hosts in the global repository.
86
func (h *Handler) doActiveHealthChecksForAllHosts() {
87
hosts.Range(func(key, value interface{}) bool {
88
addr := key.(string)
modules/caddyhttp/reverseproxy/hosts.go
new
+161
@@ -0,0 +1,161 @@
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
+ "net/url"
20
+ "sync/atomic"
21
+
22
+ "github.com/caddyserver/caddy/v2"
23
+)
24
+
25
+// Host represents a remote host which can be proxied to.
26
+// Its methods must be safe for concurrent use.
27
+type Host interface {
28
+ // NumRequests returns the numnber of requests
29
+ // currently in process with the host.
30
+ NumRequests() int
31
+
32
+ // Fails returns the count of recent failures.
33
+ Fails() int
34
+
35
+ // Unhealthy returns true if the backend is unhealthy.
36
+ Unhealthy() bool
37
+
38
+ // CountRequest counts the given number of requests
39
+ // as currently in process with the host. The count
40
+ // should not go below 0.
41
+ CountRequest(int) error
42
+
43
+ // CountFail counts the given number of failures
44
+ // with the host. The count should not go below 0.
45
+ CountFail(int) error
46
+
47
+ // SetHealthy marks the host as either healthy (true)
48
+ // or unhealthy (false). If the given status is the
49
+ // same, this should be a no-op. It returns true if
50
+ // the given status was different, false otherwise.
51
+ SetHealthy(bool) (bool, error)
52
+}
53
+
54
+// UpstreamPool is a collection of upstreams.
55
+type UpstreamPool []*Upstream
56
+
57
+// Upstream bridges this proxy's configuration to the
58
+// state of the backend host it is correlated with.
59
+type Upstream struct {
60
+ Host `json:"-"`
61
+
62
+ Address string `json:"address,omitempty"`
63
+ MaxRequests int `json:"max_requests,omitempty"`
64
+
65
+ // TODO: This could be really useful, to bind requests
66
+ // with certain properties to specific backends
67
+ // HeaderAffinity string
68
+ // IPAffinity string
69
+
70
+ healthCheckPolicy *PassiveHealthChecks
71
+ hostURL *url.URL
72
+}
73
+
74
+// Available returns true if the remote host
75
+// is available to receive requests.
76
+func (u *Upstream) Available() bool {
77
+ return u.Healthy() && !u.Full()
78
+}
79
+
80
+// Healthy returns true if the remote host
81
+// is currently known to be healthy or "up".
82
+func (u *Upstream) Healthy() bool {
83
+ healthy := !u.Host.Unhealthy()
84
+ if healthy && u.healthCheckPolicy != nil {
85
+ healthy = u.Host.Fails() < u.healthCheckPolicy.MaxFails
86
+ }
87
+ return healthy
88
+}
89
+
90
+// Full returns true if the remote host
91
+// cannot receive more requests at this time.
92
+func (u *Upstream) Full() bool {
93
+ return u.MaxRequests > 0 && u.Host.NumRequests() >= u.MaxRequests
94
+}
95
+
96
+// URL returns the upstream host's endpoint URL.
97
+func (u *Upstream) URL() *url.URL {
98
+ return u.hostURL
99
+}
100
+
101
+// upstreamHost is the basic, in-memory representation
102
+// of the state of a remote host. It implements the
103
+// Host interface.
104
+type upstreamHost struct {
105
+ numRequests int64 // must be first field to be 64-bit aligned on 32-bit systems (see https://golang.org/pkg/sync/atomic/#pkg-note-BUG)
106
+ fails int64
107
+ unhealthy int32
108
+}
109
+
110
+// NumRequests returns the number of active requests to the upstream.
111
+func (uh *upstreamHost) NumRequests() int {
112
+ return int(atomic.LoadInt64(&uh.numRequests))
113
+}
114
+
115
+// Fails returns the number of recent failures with the upstream.
116
+func (uh *upstreamHost) Fails() int {
117
+ return int(atomic.LoadInt64(&uh.fails))
118
+}
119
+
120
+// Unhealthy returns whether the upstream is healthy.
121
+func (uh *upstreamHost) Unhealthy() bool {
122
+ return atomic.LoadInt32(&uh.unhealthy) == 1
123
+}
124
+
125
+// CountRequest mutates the active request count by
126
+// delta. It returns an error if the adjustment fails.
127
+func (uh *upstreamHost) CountRequest(delta int) error {
128
+ result := atomic.AddInt64(&uh.numRequests, int64(delta))
129
+ if result < 0 {
130
+ return fmt.Errorf("count below 0: %d", result)
131
+ }
132
+ return nil
133
+}
134
+
135
+// CountFail mutates the recent failures count by
136
+// delta. It returns an error if the adjustment fails.
137
+func (uh *upstreamHost) CountFail(delta int) error {
138
+ result := atomic.AddInt64(&uh.fails, int64(delta))
139
+ if result < 0 {
140
+ return fmt.Errorf("count below 0: %d", result)
141
+ }
142
+ return nil
143
+}
144
+
145
+// SetHealthy sets the upstream has healthy or unhealthy
146
+// and returns true if the value was different from before,
147
+// or an error if the adjustment failed.
148
+func (uh *upstreamHost) SetHealthy(healthy bool) (bool, error) {
149
+ var unhealthy, compare int32 = 1, 0
150
+ if healthy {
151
+ unhealthy, compare = 0, 1
152
+ }
153
+ swapped := atomic.CompareAndSwapInt32(&uh.unhealthy, compare, unhealthy)
154
+ return swapped, nil
155
+}
156
+
157
+// hosts is the global repository for hosts that are
158
+// currently in use by active configuration(s). This
159
+// allows the state of remote hosts to be preserved
160
+// through config reloads.
161
+var hosts = caddy.NewUsagePool()
modules/caddyhttp/reverseproxy/httptransport.go
+9
-12
@@ -31,14 +31,13 @@ func init() {
31
caddy.RegisterModule(HTTPTransport{})
32
}
33
34
-// TODO: This is the default transport, basically just http.Transport, but we define JSON struct tags...
34
+// HTTPTransport is essentially a configuration wrapper for http.Transport.
35
+// It defines a JSON structure useful when configuring the HTTP transport
36
+// for Caddy's reverse proxy.
37
type HTTPTransport struct {
36
- // TODO: Actually this is where the TLS config should go, technically...
37
- // as well as keepalives and dial timeouts...
38
// TODO: It's possible that other transports (like fastcgi) might be
39
// able to borrow/use at least some of these config fields; if so,
40
// move them into a type called CommonTransport and embed it
41
-
41
TLS *TLSConfig `json:"tls,omitempty"`
42
KeepAlive *KeepAlive `json:"keep_alive,omitempty"`
43
Compression *bool `json:"compression,omitempty"`
@@ -50,7 +49,6 @@ type HTTPTransport struct {
49
MaxResponseHeaderSize int64 `json:"max_response_header_size,omitempty"`
50
WriteBufferSize int `json:"write_buffer_size,omitempty"`
51
ReadBufferSize int `json:"read_buffer_size,omitempty"`
53
- // TODO: ProxyConnectHeader?
52
53
RoundTripper http.RoundTripper `json:"-"`
54
}
@@ -63,6 +61,8 @@ func (HTTPTransport) CaddyModule() caddy.ModuleInfo {
61
}
62
}
63
64
+// Provision sets up h.RoundTripper with a http.Transport
65
+// that is ready to use.
66
func (h *HTTPTransport) Provision(ctx caddy.Context) error {
67
dialer := &net.Dialer{
68
Timeout: time.Duration(h.DialTimeout),
@@ -109,16 +109,13 @@ func (h *HTTPTransport) Provision(ctx caddy.Context) error {
109
return nil
110
}
111
112
+// RoundTrip implements http.RoundTripper with h.RoundTripper.
113
func (h HTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) {
114
return h.RoundTripper.RoundTrip(req)
115
}
116
116
-func defaultTLSConfig() *tls.Config {
117
- return &tls.Config{
118
- NextProtos: []string{"h2", "http/1.1"}, // TODO: ensure this makes HTTP/2 work
119
- }
120
-}
121
-
117
+// TLSConfig holds configuration related to the
118
+// TLS configuration for the transport/client.
119
type TLSConfig struct {
120
RootCAPool []string `json:"root_ca_pool,omitempty"`
121
// TODO: Should the client cert+key config use caddytls.CertificateLoader modules?
@@ -186,6 +183,7 @@ func decodeBase64DERCert(certStr string) (*x509.Certificate, error) {
183
return x509.ParseCertificate(derBytes)
184
}
185
186
+// KeepAlive holds configuration pertaining to HTTP Keep-Alive.
187
type KeepAlive struct {
188
Enabled *bool `json:"enabled,omitempty"`
189
ProbeInterval caddy.Duration `json:"probe_interval,omitempty"`
@@ -200,7 +198,6 @@ var (
198
KeepAlive: 30 * time.Second,
199
}
200
203
- // TODO: does this need to be configured to enable HTTP/2?
201
defaultTransport = &http.Transport{
202
DialContext: defaultDialer.DialContext,
203
TLSHandshakeTimeout: 5 * time.Second,
modules/caddyhttp/reverseproxy/reverseproxy.go
+13
-342
@@ -18,14 +18,11 @@ import (
18
"context"
19
"encoding/json"
20
"fmt"
21
- "io"
21
"net"
22
"net/http"
23
"net/url"
24
"regexp"
25
"strings"
27
- "sync"
28
- "sync/atomic"
26
"time"
27
28
"github.com/caddyserver/caddy/v2"
@@ -37,14 +34,14 @@ func init() {
34
caddy.RegisterModule(Handler{})
35
}
36
37
+// Handler implements a highly configurable and production-ready reverse proxy.
38
type Handler struct {
39
TransportRaw json.RawMessage `json:"transport,omitempty"`
40
LoadBalancing *LoadBalancing `json:"load_balancing,omitempty"`
41
HealthChecks *HealthChecks `json:"health_checks,omitempty"`
44
- // UpstreamStorageRaw json.RawMessage `json:"upstream_storage,omitempty"` // TODO:
45
- Upstreams HostPool `json:"upstreams,omitempty"`
42
+ Upstreams UpstreamPool `json:"upstreams,omitempty"`
43
+ FlushInterval caddy.Duration `json:"flush_interval,omitempty"`
44
47
- // UpstreamProvider UpstreamProvider `json:"-"` // TODO:
45
Transport http.RoundTripper `json:"-"`
46
}
47
@@ -56,6 +53,7 @@ func (Handler) CaddyModule() caddy.ModuleInfo {
53
}
54
}
55
56
+// Provision ensures that h is set up properly before use.
57
func (h *Handler) Provision(ctx caddy.Context) error {
58
if h.TransportRaw != nil {
59
val, err := ctx.LoadModuleInline("protocol", "http.handlers.reverse_proxy.transport", h.TransportRaw)
@@ -236,34 +234,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyht
234
// This assumes that no mutations of the request are performed
235
// by h during or after proxying.
236
func (h Handler) prepareRequest(req *http.Request) error {
239
- // ctx := req.Context()
240
- // TODO: do we need to support CloseNotifier? It was deprecated years ago.
241
- // All this does is wrap CloseNotify with context cancel, for those responsewriters
242
- // which didn't support context, but all the ones we'd use should nowadays, right?
243
- // if cn, ok := rw.(http.CloseNotifier); ok {
244
- // var cancel context.CancelFunc
245
- // ctx, cancel = context.WithCancel(ctx)
246
- // defer cancel()
247
- // notifyChan := cn.CloseNotify()
248
- // go func() {
249
- // select {
250
- // case <-notifyChan:
251
- // cancel()
252
- // case <-ctx.Done():
253
- // }
254
- // }()
255
- // }
256
-
257
- // TODO: do we need to call WithContext, since we won't be changing req.Context() above if we remove the CloseNotifier stuff?
258
- // TODO: (This is where references to req were originally "outreq", a shallow clone, which I think is unnecessary in our case)
259
- // req = req.WithContext(ctx) // includes shallow copies of maps, but okay
237
if req.ContentLength == 0 {
238
req.Body = nil // Issue golang/go#16036: nil Body for http.Transport retries
239
}
240
264
- // TODO: is this needed?
265
- // req.Header = cloneHeader(req.Header)
266
-
241
req.Close = false
242
243
// if User-Agent is not set by client, then explicitly
@@ -315,10 +289,12 @@ func (h Handler) prepareRequest(req *http.Request) error {
289
return nil
290
}
291
318
-// TODO:
319
-// this code is the entry point to what was borrowed from the net/http/httputil package in the standard library.
292
+// reverseProxy performs a round-trip to the given backend and processes the response with the client.
293
+// (This method is mostly the beginning of what was borrowed from the net/http/httputil package in the
294
+// Go standard library which was used as the foundation.)
295
func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, upstream *Upstream) error {
321
- // TODO: count this active request
296
+ upstream.Host.CountRequest(1)
297
+ defer upstream.Host.CountRequest(-1)
298
299
// point the request to this upstream
300
h.directRequest(req, upstream)
@@ -448,202 +424,6 @@ func (h Handler) directRequest(req *http.Request, upstream *Upstream) {
424
}
425
}
426
451
-func (h Handler) handleUpgradeResponse(rw http.ResponseWriter, req *http.Request, res *http.Response) {
452
- reqUpType := upgradeType(req.Header)
453
- resUpType := upgradeType(res.Header)
454
- if reqUpType != resUpType {
455
- // p.getErrorHandler()(rw, req, fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType))
456
- return
457
- }
458
-
459
- copyHeader(res.Header, rw.Header())
460
-
461
- hj, ok := rw.(http.Hijacker)
462
- if !ok {
463
- // p.getErrorHandler()(rw, req, fmt.Errorf("can't switch protocols using non-Hijacker ResponseWriter type %T", rw))
464
- return
465
- }
466
- backConn, ok := res.Body.(io.ReadWriteCloser)
467
- if !ok {
468
- // p.getErrorHandler()(rw, req, fmt.Errorf("internal error: 101 switching protocols response with non-writable body"))
469
- return
470
- }
471
- defer backConn.Close()
472
- conn, brw, err := hj.Hijack()
473
- if err != nil {
474
- // p.getErrorHandler()(rw, req, fmt.Errorf("Hijack failed on protocol switch: %v", err))
475
- return
476
- }
477
- defer conn.Close()
478
- res.Body = nil // so res.Write only writes the headers; we have res.Body in backConn above
479
- if err := res.Write(brw); err != nil {
480
- // p.getErrorHandler()(rw, req, fmt.Errorf("response write: %v", err))
481
- return
482
- }
483
- if err := brw.Flush(); err != nil {
484
- // p.getErrorHandler()(rw, req, fmt.Errorf("response flush: %v", err))
485
- return
486
- }
487
- errc := make(chan error, 1)
488
- spc := switchProtocolCopier{user: conn, backend: backConn}
489
- go spc.copyToBackend(errc)
490
- go spc.copyFromBackend(errc)
491
- <-errc
492
- return
493
-}
494
-
495
-// flushInterval returns the p.FlushInterval value, conditionally
496
-// overriding its value for a specific request/response.
497
-func (h Handler) flushInterval(req *http.Request, res *http.Response) time.Duration {
498
- resCT := res.Header.Get("Content-Type")
499
-
500
- // For Server-Sent Events responses, flush immediately.
501
- // The MIME type is defined in https://www.w3.org/TR/eventsource/#text-event-stream
502
- if resCT == "text/event-stream" {
503
- return -1 // negative means immediately
504
- }
505
-
506
- // TODO: more specific cases? e.g. res.ContentLength == -1?
507
- // return h.FlushInterval
508
- return 0
509
-}
510
-
511
-func (h Handler) copyResponse(dst io.Writer, src io.Reader, flushInterval time.Duration) error {
512
- if flushInterval != 0 {
513
- if wf, ok := dst.(writeFlusher); ok {
514
- mlw := &maxLatencyWriter{
515
- dst: wf,
516
- latency: flushInterval,
517
- }
518
- defer mlw.stop()
519
- dst = mlw
520
- }
521
- }
522
-
523
- // TODO: Figure out how we want to do this... using custom buffer pool type seems unnecessary
524
- // or maybe it is, depending on how we want to handle errors,
525
- // see: https://github.com/golang/go/issues/21814
526
- // buf := bufPool.Get().(*bytes.Buffer)
527
- // buf.Reset()
528
- // defer bufPool.Put(buf)
529
- // _, err := io.CopyBuffer(dst, src, )
530
- var buf []byte
531
- // if h.BufferPool != nil {
532
- // buf = h.BufferPool.Get()
533
- // defer h.BufferPool.Put(buf)
534
- // }
535
- _, err := h.copyBuffer(dst, src, buf)
536
- return err
537
-}
538
-
539
-// copyBuffer returns any write errors or non-EOF read errors, and the amount
540
-// of bytes written.
541
-func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, error) {
542
- if len(buf) == 0 {
543
- buf = make([]byte, 32*1024)
544
- }
545
- var written int64
546
- for {
547
- nr, rerr := src.Read(buf)
548
- if rerr != nil && rerr != io.EOF && rerr != context.Canceled {
549
- // TODO: this could be useful to know (indeed, it revealed an error in our
550
- // fastcgi PoC earlier; but it's this single error report here that necessitates
551
- // a function separate from io.CopyBuffer, since io.CopyBuffer does not distinguish
552
- // between read or write errors; in a reverse proxy situation, write errors are not
553
- // something we need to report to the client, but read errors are a problem on our
554
- // end for sure. so we need to decide what we want.)
555
- // p.logf("copyBuffer: ReverseProxy read error during body copy: %v", rerr)
556
- }
557
- if nr > 0 {
558
- nw, werr := dst.Write(buf[:nr])
559
- if nw > 0 {
560
- written += int64(nw)
561
- }
562
- if werr != nil {
563
- return written, werr
564
- }
565
- if nr != nw {
566
- return written, io.ErrShortWrite
567
- }
568
- }
569
- if rerr != nil {
570
- if rerr == io.EOF {
571
- rerr = nil
572
- }
573
- return written, rerr
574
- }
575
- }
576
-}
577
-
578
-type writeFlusher interface {
579
- io.Writer
580
- http.Flusher
581
-}
582
-
583
-type maxLatencyWriter struct {
584
- dst writeFlusher
585
- latency time.Duration // non-zero; negative means to flush immediately
586
-
587
- mu sync.Mutex // protects t, flushPending, and dst.Flush
588
- t *time.Timer
589
- flushPending bool
590
-}
591
-
592
-func (m *maxLatencyWriter) Write(p []byte) (n int, err error) {
593
- m.mu.Lock()
594
- defer m.mu.Unlock()
595
- n, err = m.dst.Write(p)
596
- if m.latency < 0 {
597
- m.dst.Flush()
598
- return
599
- }
600
- if m.flushPending {
601
- return
602
- }
603
- if m.t == nil {
604
- m.t = time.AfterFunc(m.latency, m.delayedFlush)
605
- } else {
606
- m.t.Reset(m.latency)
607
- }
608
- m.flushPending = true
609
- return
610
-}
611
-
612
-func (m *maxLatencyWriter) delayedFlush() {
613
- m.mu.Lock()
614
- defer m.mu.Unlock()
615
- if !m.flushPending { // if stop was called but AfterFunc already started this goroutine
616
- return
617
- }
618
- m.dst.Flush()
619
- m.flushPending = false
620
-}
621
-
622
-func (m *maxLatencyWriter) stop() {
623
- m.mu.Lock()
624
- defer m.mu.Unlock()
625
- m.flushPending = false
626
- if m.t != nil {
627
- m.t.Stop()
628
- }
629
-}
630
-
631
-// switchProtocolCopier exists so goroutines proxying data back and
632
-// forth have nice names in stacks.
633
-type switchProtocolCopier struct {
634
- user, backend io.ReadWriter
635
-}
636
-
637
-func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {
638
- _, err := io.Copy(c.user, c.backend)
639
- errc <- err
640
-}
641
-
642
-func (c switchProtocolCopier) copyToBackend(errc chan<- error) {
643
- _, err := io.Copy(c.backend, c.user)
644
- errc <- err
645
-}
646
-
427
// shouldPanicOnCopyError reports whether the reverse proxy should
428
// panic with http.ErrAbortHandler. This is the right thing to do by
429
// default, but Go 1.10 and earlier did not, so existing unit tests
@@ -714,6 +494,7 @@ func removeConnectionHeaders(h http.Header) {
494
}
495
}
496
497
+// LoadBalancing has parameters related to load balancing.
498
type LoadBalancing struct {
499
SelectionPolicyRaw json.RawMessage `json:"selection_policy,omitempty"`
500
TryDuration caddy.Duration `json:"try_duration,omitempty"`
@@ -722,8 +503,9 @@ type LoadBalancing struct {
503
SelectionPolicy Selector `json:"-"`
504
}
505
506
+// Selector selects an available upstream from the pool.
507
type Selector interface {
726
- Select(HostPool, *http.Request) *Upstream
508
+ Select(UpstreamPool, *http.Request) *Upstream
509
}
510
511
// Hop-by-hop headers. These are removed when sent to the backend.
@@ -743,117 +525,6 @@ var hopHeaders = []string{
525
"Upgrade",
526
}
527
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
776
-
777
-type upstreamHost struct {
778
- numRequests int64 // must be first field to be 64-bit aligned on 32-bit systems (see https://golang.org/pkg/sync/atomic/#pkg-note-BUG)
779
- fails int64
780
- unhealthy int32
781
-}
782
-
783
-func (uh *upstreamHost) NumRequests() int {
784
- return int(atomic.LoadInt64(&uh.numRequests))
785
-}
786
-func (uh *upstreamHost) Fails() int {
787
- return int(atomic.LoadInt64(&uh.fails))
788
-}
789
-func (uh *upstreamHost) Unhealthy() bool {
790
- return atomic.LoadInt32(&uh.unhealthy) == 1
791
-}
792
-func (uh *upstreamHost) CountRequest(delta int) error {
793
- result := atomic.AddInt64(&uh.numRequests, int64(delta))
794
- if result < 0 {
795
- return fmt.Errorf("count below 0: %d", result)
796
- }
797
- return nil
798
-}
799
-func (uh *upstreamHost) CountFail(delta int) error {
800
- result := atomic.AddInt64(&uh.fails, int64(delta))
801
- if result < 0 {
802
- return fmt.Errorf("count below 0: %d", result)
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:"-"`
817
-
818
- Address string `json:"address,omitempty"`
819
- MaxRequests int `json:"max_requests,omitempty"`
820
-
821
- // TODO: This could be really cool, to say that requests with
822
- // certain headers or from certain IPs always go to this upstream
823
- // HeaderAffinity string
824
- // IPAffinity string
825
-
826
- healthCheckPolicy *PassiveHealthChecks
827
-
828
- hostURL *url.URL
829
-}
830
-
831
-func (u Upstream) Available() bool {
832
- return u.Healthy() && !u.Full()
833
-}
834
-
835
-func (u Upstream) Healthy() bool {
836
- healthy := !u.Host.Unhealthy()
837
- if healthy && u.healthCheckPolicy != nil {
838
- healthy = u.Host.Fails() < u.healthCheckPolicy.MaxFails
839
- }
840
- return healthy
841
-}
842
-
843
-func (u Upstream) Full() bool {
844
- return u.MaxRequests > 0 && u.Host.NumRequests() >= u.MaxRequests
845
-}
846
-
847
-func (u Upstream) URL() *url.URL {
848
- return u.hostURL
849
-}
850
-
851
-var hosts = caddy.NewUsagePool()
852
-
853
-// TODO: ...
854
-type UpstreamProvider interface {
855
-}
856
-
528
// TODO: see if we can use this
529
// var bufPool = sync.Pool{
530
// New: func() interface{} {
@@ -863,7 +534,7 @@ type UpstreamProvider interface {
534
535
// Interface guards
536
var (
866
- _ caddyhttp.MiddlewareHandler = (*Handler)(nil)
537
_ caddy.Provisioner = (*Handler)(nil)
538
_ caddy.CleanerUpper = (*Handler)(nil)
539
+ _ caddyhttp.MiddlewareHandler = (*Handler)(nil)
540
)
modules/caddyhttp/reverseproxy/selectionpolicies.go
+10
-8
@@ -52,7 +52,7 @@ func (RandomSelection) CaddyModule() caddy.ModuleInfo {
52
}
53
54
// Select returns an available host, if any.
55
-func (r RandomSelection) Select(pool HostPool, request *http.Request) *Upstream {
55
+func (r RandomSelection) Select(pool UpstreamPool, request *http.Request) *Upstream {
56
// use reservoir sampling because the number of available
57
// hosts isn't known: https://en.wikipedia.org/wiki/Reservoir_sampling
58
var randomHost *Upstream
@@ -87,6 +87,7 @@ func (RandomChoiceSelection) CaddyModule() caddy.ModuleInfo {
87
}
88
}
89
90
+// Provision sets up r.
91
func (r *RandomChoiceSelection) Provision(ctx caddy.Context) error {
92
if r.Choose == 0 {
93
r.Choose = 2
@@ -94,6 +95,7 @@ func (r *RandomChoiceSelection) Provision(ctx caddy.Context) error {
95
return nil
96
}
97
98
+// Validate ensures that r's configuration is valid.
99
func (r RandomChoiceSelection) Validate() error {
100
if r.Choose < 2 {
101
return fmt.Errorf("choose must be at least 2")
@@ -102,7 +104,7 @@ func (r RandomChoiceSelection) Validate() error {
104
}
105
106
// Select returns an available host, if any.
105
-func (r RandomChoiceSelection) Select(pool HostPool, _ *http.Request) *Upstream {
107
+func (r RandomChoiceSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream {
108
k := r.Choose
109
if k > len(pool) {
110
k = len(pool)
@@ -142,7 +144,7 @@ func (LeastConnSelection) CaddyModule() caddy.ModuleInfo {
144
// Select selects the up host with the least number of connections in the
145
// pool. If more than one host has the same least number of connections,
146
// one of the hosts is chosen at random.
145
-func (LeastConnSelection) Select(pool HostPool, _ *http.Request) *Upstream {
147
+func (LeastConnSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream {
148
var bestHost *Upstream
149
var count int
150
var leastReqs int
@@ -185,7 +187,7 @@ func (RoundRobinSelection) CaddyModule() caddy.ModuleInfo {
187
}
188
189
// Select returns an available host, if any.
188
-func (r *RoundRobinSelection) Select(pool HostPool, _ *http.Request) *Upstream {
190
+func (r *RoundRobinSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream {
191
n := uint32(len(pool))
192
if n == 0 {
193
return nil
@@ -213,7 +215,7 @@ func (FirstSelection) CaddyModule() caddy.ModuleInfo {
215
}
216
217
// Select returns an available host, if any.
216
-func (FirstSelection) Select(pool HostPool, _ *http.Request) *Upstream {
218
+func (FirstSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream {
219
for _, host := range pool {
220
if host.Available() {
221
return host
@@ -235,7 +237,7 @@ func (IPHashSelection) CaddyModule() caddy.ModuleInfo {
237
}
238
239
// Select returns an available host, if any.
238
-func (IPHashSelection) Select(pool HostPool, req *http.Request) *Upstream {
240
+func (IPHashSelection) Select(pool UpstreamPool, req *http.Request) *Upstream {
241
clientIP, _, err := net.SplitHostPort(req.RemoteAddr)
242
if err != nil {
243
clientIP = req.RemoteAddr
@@ -256,7 +258,7 @@ func (URIHashSelection) CaddyModule() caddy.ModuleInfo {
258
}
259
260
// Select returns an available host, if any.
259
-func (URIHashSelection) Select(pool HostPool, req *http.Request) *Upstream {
261
+func (URIHashSelection) Select(pool UpstreamPool, req *http.Request) *Upstream {
262
return hostByHashing(pool, req.RequestURI)
263
}
264
@@ -275,7 +277,7 @@ func (HeaderHashSelection) CaddyModule() caddy.ModuleInfo {
277
}
278
279
// Select returns an available host, if any.
278
-func (s HeaderHashSelection) Select(pool HostPool, req *http.Request) *Upstream {
280
+func (s HeaderHashSelection) Select(pool UpstreamPool, req *http.Request) *Upstream {
281
if s.Field == "" {
282
return nil
283
}
modules/caddyhttp/reverseproxy/streaming.go
new
+223
@@ -0,0 +1,223 @@
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
+// Most of the code in this file was initially borrowed from the Go
16
+// standard library, which has this copyright notice:
17
+// Copyright 2011 The Go Authors.
18
+
19
+package reverseproxy
20
+
21
+import (
22
+ "context"
23
+ "io"
24
+ "net/http"
25
+ "sync"
26
+ "time"
27
+)
28
+
29
+func (h Handler) handleUpgradeResponse(rw http.ResponseWriter, req *http.Request, res *http.Response) {
30
+ reqUpType := upgradeType(req.Header)
31
+ resUpType := upgradeType(res.Header)
32
+ if reqUpType != resUpType {
33
+ // TODO: figure out our own error handling
34
+ // p.getErrorHandler()(rw, req, fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType))
35
+ return
36
+ }
37
+
38
+ copyHeader(res.Header, rw.Header())
39
+
40
+ hj, ok := rw.(http.Hijacker)
41
+ if !ok {
42
+ // p.getErrorHandler()(rw, req, fmt.Errorf("can't switch protocols using non-Hijacker ResponseWriter type %T", rw))
43
+ return
44
+ }
45
+ backConn, ok := res.Body.(io.ReadWriteCloser)
46
+ if !ok {
47
+ // p.getErrorHandler()(rw, req, fmt.Errorf("internal error: 101 switching protocols response with non-writable body"))
48
+ return
49
+ }
50
+ defer backConn.Close()
51
+ conn, brw, err := hj.Hijack()
52
+ if err != nil {
53
+ // p.getErrorHandler()(rw, req, fmt.Errorf("Hijack failed on protocol switch: %v", err))
54
+ return
55
+ }
56
+ defer conn.Close()
57
+ res.Body = nil // so res.Write only writes the headers; we have res.Body in backConn above
58
+ if err := res.Write(brw); err != nil {
59
+ // p.getErrorHandler()(rw, req, fmt.Errorf("response write: %v", err))
60
+ return
61
+ }
62
+ if err := brw.Flush(); err != nil {
63
+ // p.getErrorHandler()(rw, req, fmt.Errorf("response flush: %v", err))
64
+ return
65
+ }
66
+ errc := make(chan error, 1)
67
+ spc := switchProtocolCopier{user: conn, backend: backConn}
68
+ go spc.copyToBackend(errc)
69
+ go spc.copyFromBackend(errc)
70
+ <-errc
71
+ return
72
+}
73
+
74
+// flushInterval returns the p.FlushInterval value, conditionally
75
+// overriding its value for a specific request/response.
76
+func (h Handler) flushInterval(req *http.Request, res *http.Response) time.Duration {
77
+ resCT := res.Header.Get("Content-Type")
78
+
79
+ // For Server-Sent Events responses, flush immediately.
80
+ // The MIME type is defined in https://www.w3.org/TR/eventsource/#text-event-stream
81
+ if resCT == "text/event-stream" {
82
+ return -1 // negative means immediately
83
+ }
84
+
85
+ // TODO: more specific cases? e.g. res.ContentLength == -1? (this TODO is from the std lib)
86
+ return time.Duration(h.FlushInterval)
87
+}
88
+
89
+func (h Handler) copyResponse(dst io.Writer, src io.Reader, flushInterval time.Duration) error {
90
+ if flushInterval != 0 {
91
+ if wf, ok := dst.(writeFlusher); ok {
92
+ mlw := &maxLatencyWriter{
93
+ dst: wf,
94
+ latency: flushInterval,
95
+ }
96
+ defer mlw.stop()
97
+ dst = mlw
98
+ }
99
+ }
100
+
101
+ // TODO: Figure out how we want to do this... using custom buffer pool type seems unnecessary
102
+ // or maybe it is, depending on how we want to handle errors,
103
+ // see: https://github.com/golang/go/issues/21814
104
+ // buf := bufPool.Get().(*bytes.Buffer)
105
+ // buf.Reset()
106
+ // defer bufPool.Put(buf)
107
+ // _, err := io.CopyBuffer(dst, src, )
108
+ var buf []byte
109
+ // if h.BufferPool != nil {
110
+ // buf = h.BufferPool.Get()
111
+ // defer h.BufferPool.Put(buf)
112
+ // }
113
+ _, err := h.copyBuffer(dst, src, buf)
114
+ return err
115
+}
116
+
117
+// copyBuffer returns any write errors or non-EOF read errors, and the amount
118
+// of bytes written.
119
+func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, error) {
120
+ if len(buf) == 0 {
121
+ buf = make([]byte, 32*1024)
122
+ }
123
+ var written int64
124
+ for {
125
+ nr, rerr := src.Read(buf)
126
+ if rerr != nil && rerr != io.EOF && rerr != context.Canceled {
127
+ // TODO: this could be useful to know (indeed, it revealed an error in our
128
+ // fastcgi PoC earlier; but it's this single error report here that necessitates
129
+ // a function separate from io.CopyBuffer, since io.CopyBuffer does not distinguish
130
+ // between read or write errors; in a reverse proxy situation, write errors are not
131
+ // something we need to report to the client, but read errors are a problem on our
132
+ // end for sure. so we need to decide what we want.)
133
+ // p.logf("copyBuffer: ReverseProxy read error during body copy: %v", rerr)
134
+ }
135
+ if nr > 0 {
136
+ nw, werr := dst.Write(buf[:nr])
137
+ if nw > 0 {
138
+ written += int64(nw)
139
+ }
140
+ if werr != nil {
141
+ return written, werr
142
+ }
143
+ if nr != nw {
144
+ return written, io.ErrShortWrite
145
+ }
146
+ }
147
+ if rerr != nil {
148
+ if rerr == io.EOF {
149
+ rerr = nil
150
+ }
151
+ return written, rerr
152
+ }
153
+ }
154
+}
155
+
156
+type writeFlusher interface {
157
+ io.Writer
158
+ http.Flusher
159
+}
160
+
161
+type maxLatencyWriter struct {
162
+ dst writeFlusher
163
+ latency time.Duration // non-zero; negative means to flush immediately
164
+
165
+ mu sync.Mutex // protects t, flushPending, and dst.Flush
166
+ t *time.Timer
167
+ flushPending bool
168
+}
169
+
170
+func (m *maxLatencyWriter) Write(p []byte) (n int, err error) {
171
+ m.mu.Lock()
172
+ defer m.mu.Unlock()
173
+ n, err = m.dst.Write(p)
174
+ if m.latency < 0 {
175
+ m.dst.Flush()
176
+ return
177
+ }
178
+ if m.flushPending {
179
+ return
180
+ }
181
+ if m.t == nil {
182
+ m.t = time.AfterFunc(m.latency, m.delayedFlush)
183
+ } else {
184
+ m.t.Reset(m.latency)
185
+ }
186
+ m.flushPending = true
187
+ return
188
+}
189
+
190
+func (m *maxLatencyWriter) delayedFlush() {
191
+ m.mu.Lock()
192
+ defer m.mu.Unlock()
193
+ if !m.flushPending { // if stop was called but AfterFunc already started this goroutine
194
+ return
195
+ }
196
+ m.dst.Flush()
197
+ m.flushPending = false
198
+}
199
+
200
+func (m *maxLatencyWriter) stop() {
201
+ m.mu.Lock()
202
+ defer m.mu.Unlock()
203
+ m.flushPending = false
204
+ if m.t != nil {
205
+ m.t.Stop()
206
+ }
207
+}
208
+
209
+// switchProtocolCopier exists so goroutines proxying data back and
210
+// forth have nice names in stacks.
211
+type switchProtocolCopier struct {
212
+ user, backend io.ReadWriter
213
+}
214
+
215
+func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {
216
+ _, err := io.Copy(c.user, c.backend)
217
+ errc <- err
218
+}
219
+
220
+func (c switchProtocolCopier) copyToBackend(errc chan<- error) {
221
+ _, err := io.Copy(c.backend, c.user)
222
+ errc <- err
223
+}