diff options
| -rw-r--r-- | modules/caddyhttp/reverseproxy/healthchecks.go | 12 | ||||
| -rw-r--r-- | modules/caddyhttp/reverseproxy/hosts.go | 161 | ||||
| -rw-r--r-- | modules/caddyhttp/reverseproxy/httptransport.go | 21 | ||||
| -rw-r--r-- | modules/caddyhttp/reverseproxy/reverseproxy.go | 355 | ||||
| -rw-r--r-- | modules/caddyhttp/reverseproxy/selectionpolicies.go | 18 | ||||
| -rw-r--r-- | modules/caddyhttp/reverseproxy/streaming.go | 223 |
6 files changed, 428 insertions, 362 deletions
diff --git a/modules/caddyhttp/reverseproxy/healthchecks.go b/modules/caddyhttp/reverseproxy/healthchecks.go index 96649a4..0b46d04 100644 --- a/modules/caddyhttp/reverseproxy/healthchecks.go +++ b/modules/caddyhttp/reverseproxy/healthchecks.go @@ -30,11 +30,15 @@ import ( "github.com/caddyserver/caddy/v2/modules/caddyhttp" ) +// HealthChecks holds configuration related to health checking. type HealthChecks struct { Active *ActiveHealthChecks `json:"active,omitempty"` Passive *PassiveHealthChecks `json:"passive,omitempty"` } +// ActiveHealthChecks holds configuration related to active +// health checks (that is, health checks which occur in a +// background goroutine independently). type ActiveHealthChecks struct { Path string `json:"path,omitempty"` Port int `json:"port,omitempty"` @@ -49,6 +53,9 @@ type ActiveHealthChecks struct { bodyRegexp *regexp.Regexp } +// PassiveHealthChecks holds configuration related to passive +// health checks (that is, health checks which occur during +// the normal flow of request proxying). type PassiveHealthChecks struct { MaxFails int `json:"max_fails,omitempty"` FailDuration caddy.Duration `json:"fail_duration,omitempty"` @@ -57,6 +64,9 @@ type PassiveHealthChecks struct { UnhealthyLatency caddy.Duration `json:"unhealthy_latency,omitempty"` } +// activeHealthChecker runs active health checks on a +// regular basis and blocks until +// h.HealthChecks.Active.stopChan is closed. func (h *Handler) activeHealthChecker() { ticker := time.NewTicker(time.Duration(h.HealthChecks.Active.Interval)) h.doActiveHealthChecksForAllHosts() @@ -71,6 +81,8 @@ func (h *Handler) activeHealthChecker() { } } +// doActiveHealthChecksForAllHosts immediately performs a +// health checks for all hosts in the global repository. func (h *Handler) doActiveHealthChecksForAllHosts() { hosts.Range(func(key, value interface{}) bool { addr := key.(string) diff --git a/modules/caddyhttp/reverseproxy/hosts.go b/modules/caddyhttp/reverseproxy/hosts.go new file mode 100644 index 0000000..5100936 --- /dev/null +++ b/modules/caddyhttp/reverseproxy/hosts.go @@ -0,0 +1,161 @@ +// Copyright 2015 Matthew Holt and The Caddy Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package reverseproxy + +import ( + "fmt" + "net/url" + "sync/atomic" + + "github.com/caddyserver/caddy/v2" +) + +// Host represents a remote host which can be proxied to. +// Its methods must be safe for concurrent use. +type Host interface { + // NumRequests returns the numnber of requests + // currently in process with the host. + NumRequests() int + + // Fails returns the count of recent failures. + Fails() int + + // Unhealthy returns true if the backend is unhealthy. + Unhealthy() bool + + // CountRequest counts the given number of requests + // as currently in process with the host. The count + // should not go below 0. + CountRequest(int) error + + // CountFail counts the given number of failures + // with the host. The count should not go below 0. + CountFail(int) error + + // SetHealthy marks the host as either healthy (true) + // or unhealthy (false). If the given status is the + // same, this should be a no-op. It returns true if + // the given status was different, false otherwise. + SetHealthy(bool) (bool, error) +} + +// UpstreamPool is a collection of upstreams. +type UpstreamPool []*Upstream + +// Upstream bridges this proxy's configuration to the +// state of the backend host it is correlated with. +type Upstream struct { + Host `json:"-"` + + Address string `json:"address,omitempty"` + MaxRequests int `json:"max_requests,omitempty"` + + // TODO: This could be really useful, to bind requests + // with certain properties to specific backends + // HeaderAffinity string + // IPAffinity string + + healthCheckPolicy *PassiveHealthChecks + hostURL *url.URL +} + +// Available returns true if the remote host +// is available to receive requests. +func (u *Upstream) Available() bool { + return u.Healthy() && !u.Full() +} + +// Healthy returns true if the remote host +// is currently known to be healthy or "up". +func (u *Upstream) Healthy() bool { + healthy := !u.Host.Unhealthy() + if healthy && u.healthCheckPolicy != nil { + healthy = u.Host.Fails() < u.healthCheckPolicy.MaxFails + } + return healthy +} + +// Full returns true if the remote host +// cannot receive more requests at this time. +func (u *Upstream) Full() bool { + return u.MaxRequests > 0 && u.Host.NumRequests() >= u.MaxRequests +} + +// URL returns the upstream host's endpoint URL. +func (u *Upstream) URL() *url.URL { + return u.hostURL +} + +// upstreamHost is the basic, in-memory representation +// of the state of a remote host. It implements the +// Host interface. +type upstreamHost struct { + 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) + fails int64 + unhealthy int32 +} + +// NumRequests returns the number of active requests to the upstream. +func (uh *upstreamHost) NumRequests() int { + return int(atomic.LoadInt64(&uh.numRequests)) +} + +// Fails returns the number of recent failures with the upstream. +func (uh *upstreamHost) Fails() int { + return int(atomic.LoadInt64(&uh.fails)) +} + +// Unhealthy returns whether the upstream is healthy. +func (uh *upstreamHost) Unhealthy() bool { + return atomic.LoadInt32(&uh.unhealthy) == 1 +} + +// CountRequest mutates the active request count by +// delta. It returns an error if the adjustment fails. +func (uh *upstreamHost) CountRequest(delta int) error { + result := atomic.AddInt64(&uh.numRequests, int64(delta)) + if result < 0 { + return fmt.Errorf("count below 0: %d", result) + } + return nil +} + +// CountFail mutates the recent failures count by +// delta. It returns an error if the adjustment fails. +func (uh *upstreamHost) CountFail(delta int) error { + result := atomic.AddInt64(&uh.fails, int64(delta)) + if result < 0 { + return fmt.Errorf("count below 0: %d", result) + } + return nil +} + +// SetHealthy sets the upstream has healthy or unhealthy +// and returns true if the value was different from before, +// or an error if the adjustment failed. +func (uh *upstreamHost) SetHealthy(healthy bool) (bool, error) { + var unhealthy, compare int32 = 1, 0 + if healthy { + unhealthy, compare = 0, 1 + } + swapped := atomic.CompareAndSwapInt32(&uh.unhealthy, compare, unhealthy) + return swapped, nil +} + +// hosts is the global repository for hosts that are +// currently in use by active configuration(s). This +// allows the state of remote hosts to be preserved +// through config reloads. +var hosts = caddy.NewUsagePool() diff --git a/modules/caddyhttp/reverseproxy/httptransport.go b/modules/caddyhttp/reverseproxy/httptransport.go index 999a352..d9dc457 100644 --- a/modules/caddyhttp/reverseproxy/httptransport.go +++ b/modules/caddyhttp/reverseproxy/httptransport.go @@ -31,14 +31,13 @@ func init() { caddy.RegisterModule(HTTPTransport{}) } -// TODO: This is the default transport, basically just http.Transport, but we define JSON struct tags... +// HTTPTransport is essentially a configuration wrapper for http.Transport. +// It defines a JSON structure useful when configuring the HTTP transport +// for Caddy's reverse proxy. type HTTPTransport struct { - // TODO: Actually this is where the TLS config should go, technically... - // as well as keepalives and dial timeouts... // TODO: It's possible that other transports (like fastcgi) might be // able to borrow/use at least some of these config fields; if so, // move them into a type called CommonTransport and embed it - TLS *TLSConfig `json:"tls,omitempty"` KeepAlive *KeepAlive `json:"keep_alive,omitempty"` Compression *bool `json:"compression,omitempty"` @@ -50,7 +49,6 @@ type HTTPTransport struct { MaxResponseHeaderSize int64 `json:"max_response_header_size,omitempty"` WriteBufferSize int `json:"write_buffer_size,omitempty"` ReadBufferSize int `json:"read_buffer_size,omitempty"` - // TODO: ProxyConnectHeader? RoundTripper http.RoundTripper `json:"-"` } @@ -63,6 +61,8 @@ func (HTTPTransport) CaddyModule() caddy.ModuleInfo { } } +// Provision sets up h.RoundTripper with a http.Transport +// that is ready to use. func (h *HTTPTransport) Provision(ctx caddy.Context) error { dialer := &net.Dialer{ Timeout: time.Duration(h.DialTimeout), @@ -109,16 +109,13 @@ func (h *HTTPTransport) Provision(ctx caddy.Context) error { return nil } +// RoundTrip implements http.RoundTripper with h.RoundTripper. func (h HTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) { return h.RoundTripper.RoundTrip(req) } -func defaultTLSConfig() *tls.Config { - return &tls.Config{ - NextProtos: []string{"h2", "http/1.1"}, // TODO: ensure this makes HTTP/2 work - } -} - +// TLSConfig holds configuration related to the +// TLS configuration for the transport/client. type TLSConfig struct { RootCAPool []string `json:"root_ca_pool,omitempty"` // TODO: Should the client cert+key config use caddytls.CertificateLoader modules? @@ -186,6 +183,7 @@ func decodeBase64DERCert(certStr string) (*x509.Certificate, error) { return x509.ParseCertificate(derBytes) } +// KeepAlive holds configuration pertaining to HTTP Keep-Alive. type KeepAlive struct { Enabled *bool `json:"enabled,omitempty"` ProbeInterval caddy.Duration `json:"probe_interval,omitempty"` @@ -200,7 +198,6 @@ var ( KeepAlive: 30 * time.Second, } - // TODO: does this need to be configured to enable HTTP/2? defaultTransport = &http.Transport{ DialContext: defaultDialer.DialContext, TLSHandshakeTimeout: 5 * time.Second, diff --git a/modules/caddyhttp/reverseproxy/reverseproxy.go b/modules/caddyhttp/reverseproxy/reverseproxy.go index ebf6ac1..ca54741 100644 --- a/modules/caddyhttp/reverseproxy/reverseproxy.go +++ b/modules/caddyhttp/reverseproxy/reverseproxy.go @@ -18,14 +18,11 @@ import ( "context" "encoding/json" "fmt" - "io" "net" "net/http" "net/url" "regexp" "strings" - "sync" - "sync/atomic" "time" "github.com/caddyserver/caddy/v2" @@ -37,14 +34,14 @@ func init() { caddy.RegisterModule(Handler{}) } +// Handler implements a highly configurable and production-ready reverse proxy. type Handler struct { TransportRaw json.RawMessage `json:"transport,omitempty"` LoadBalancing *LoadBalancing `json:"load_balancing,omitempty"` HealthChecks *HealthChecks `json:"health_checks,omitempty"` - // UpstreamStorageRaw json.RawMessage `json:"upstream_storage,omitempty"` // TODO: - Upstreams HostPool `json:"upstreams,omitempty"` + Upstreams UpstreamPool `json:"upstreams,omitempty"` + FlushInterval caddy.Duration `json:"flush_interval,omitempty"` - // UpstreamProvider UpstreamProvider `json:"-"` // TODO: Transport http.RoundTripper `json:"-"` } @@ -56,6 +53,7 @@ func (Handler) CaddyModule() caddy.ModuleInfo { } } +// Provision ensures that h is set up properly before use. func (h *Handler) Provision(ctx caddy.Context) error { if h.TransportRaw != nil { 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 // This assumes that no mutations of the request are performed // by h during or after proxying. func (h Handler) prepareRequest(req *http.Request) error { - // ctx := req.Context() - // TODO: do we need to support CloseNotifier? It was deprecated years ago. - // All this does is wrap CloseNotify with context cancel, for those responsewriters - // which didn't support context, but all the ones we'd use should nowadays, right? - // if cn, ok := rw.(http.CloseNotifier); ok { - // var cancel context.CancelFunc - // ctx, cancel = context.WithCancel(ctx) - // defer cancel() - // notifyChan := cn.CloseNotify() - // go func() { - // select { - // case <-notifyChan: - // cancel() - // case <-ctx.Done(): - // } - // }() - // } - - // TODO: do we need to call WithContext, since we won't be changing req.Context() above if we remove the CloseNotifier stuff? - // TODO: (This is where references to req were originally "outreq", a shallow clone, which I think is unnecessary in our case) - // req = req.WithContext(ctx) // includes shallow copies of maps, but okay if req.ContentLength == 0 { req.Body = nil // Issue golang/go#16036: nil Body for http.Transport retries } - // TODO: is this needed? - // req.Header = cloneHeader(req.Header) - req.Close = false // if User-Agent is not set by client, then explicitly @@ -315,10 +289,12 @@ func (h Handler) prepareRequest(req *http.Request) error { return nil } -// TODO: -// this code is the entry point to what was borrowed from the net/http/httputil package in the standard library. +// reverseProxy performs a round-trip to the given backend and processes the response with the client. +// (This method is mostly the beginning of what was borrowed from the net/http/httputil package in the +// Go standard library which was used as the foundation.) func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, upstream *Upstream) error { - // TODO: count this active request + upstream.Host.CountRequest(1) + defer upstream.Host.CountRequest(-1) // point the request to this upstream h.directRequest(req, upstream) @@ -448,202 +424,6 @@ func (h Handler) directRequest(req *http.Request, upstream *Upstream) { } } -func (h Handler) handleUpgradeResponse(rw http.ResponseWriter, req *http.Request, res *http.Response) { - reqUpType := upgradeType(req.Header) - resUpType := upgradeType(res.Header) - if reqUpType != resUpType { - // p.getErrorHandler()(rw, req, fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType)) - return - } - - copyHeader(res.Header, rw.Header()) - - hj, ok := rw.(http.Hijacker) - if !ok { - // p.getErrorHandler()(rw, req, fmt.Errorf("can't switch protocols using non-Hijacker ResponseWriter type %T", rw)) - return - } - backConn, ok := res.Body.(io.ReadWriteCloser) - if !ok { - // p.getErrorHandler()(rw, req, fmt.Errorf("internal error: 101 switching protocols response with non-writable body")) - return - } - defer backConn.Close() - conn, brw, err := hj.Hijack() - if err != nil { - // p.getErrorHandler()(rw, req, fmt.Errorf("Hijack failed on protocol switch: %v", err)) - return - } - defer conn.Close() - res.Body = nil // so res.Write only writes the headers; we have res.Body in backConn above - if err := res.Write(brw); err != nil { - // p.getErrorHandler()(rw, req, fmt.Errorf("response write: %v", err)) - return - } - if err := brw.Flush(); err != nil { - // p.getErrorHandler()(rw, req, fmt.Errorf("response flush: %v", err)) - return - } - errc := make(chan error, 1) - spc := switchProtocolCopier{user: conn, backend: backConn} - go spc.copyToBackend(errc) - go spc.copyFromBackend(errc) - <-errc - return -} - -// flushInterval returns the p.FlushInterval value, conditionally -// overriding its value for a specific request/response. -func (h Handler) flushInterval(req *http.Request, res *http.Response) time.Duration { - resCT := res.Header.Get("Content-Type") - - // For Server-Sent Events responses, flush immediately. - // The MIME type is defined in https://www.w3.org/TR/eventsource/#text-event-stream - if resCT == "text/event-stream" { - return -1 // negative means immediately - } - - // TODO: more specific cases? e.g. res.ContentLength == -1? - // return h.FlushInterval - return 0 -} - -func (h Handler) copyResponse(dst io.Writer, src io.Reader, flushInterval time.Duration) error { - if flushInterval != 0 { - if wf, ok := dst.(writeFlusher); ok { - mlw := &maxLatencyWriter{ - dst: wf, - latency: flushInterval, - } - defer mlw.stop() - dst = mlw - } - } - - // TODO: Figure out how we want to do this... using custom buffer pool type seems unnecessary - // or maybe it is, depending on how we want to handle errors, - // see: https://github.com/golang/go/issues/21814 - // buf := bufPool.Get().(*bytes.Buffer) - // buf.Reset() - // defer bufPool.Put(buf) - // _, err := io.CopyBuffer(dst, src, ) - var buf []byte - // if h.BufferPool != nil { - // buf = h.BufferPool.Get() - // defer h.BufferPool.Put(buf) - // } - _, err := h.copyBuffer(dst, src, buf) - return err -} - -// copyBuffer returns any write errors or non-EOF read errors, and the amount -// of bytes written. -func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, error) { - if len(buf) == 0 { - buf = make([]byte, 32*1024) - } - var written int64 - for { - nr, rerr := src.Read(buf) - if rerr != nil && rerr != io.EOF && rerr != context.Canceled { - // TODO: this could be useful to know (indeed, it revealed an error in our - // fastcgi PoC earlier; but it's this single error report here that necessitates - // a function separate from io.CopyBuffer, since io.CopyBuffer does not distinguish - // between read or write errors; in a reverse proxy situation, write errors are not - // something we need to report to the client, but read errors are a problem on our - // end for sure. so we need to decide what we want.) - // p.logf("copyBuffer: ReverseProxy read error during body copy: %v", rerr) - } - if nr > 0 { - nw, werr := dst.Write(buf[:nr]) - if nw > 0 { - written += int64(nw) - } - if werr != nil { - return written, werr - } - if nr != nw { - return written, io.ErrShortWrite - } - } - if rerr != nil { - if rerr == io.EOF { - rerr = nil - } - return written, rerr - } - } -} - -type writeFlusher interface { - io.Writer - http.Flusher -} - -type maxLatencyWriter struct { - dst writeFlusher - latency time.Duration // non-zero; negative means to flush immediately - - mu sync.Mutex // protects t, flushPending, and dst.Flush - t *time.Timer - flushPending bool -} - -func (m *maxLatencyWriter) Write(p []byte) (n int, err error) { - m.mu.Lock() - defer m.mu.Unlock() - n, err = m.dst.Write(p) - if m.latency < 0 { - m.dst.Flush() - return - } - if m.flushPending { - return - } - if m.t == nil { - m.t = time.AfterFunc(m.latency, m.delayedFlush) - } else { - m.t.Reset(m.latency) - } - m.flushPending = true - return -} - -func (m *maxLatencyWriter) delayedFlush() { - m.mu.Lock() - defer m.mu.Unlock() - if !m.flushPending { // if stop was called but AfterFunc already started this goroutine - return - } - m.dst.Flush() - m.flushPending = false -} - -func (m *maxLatencyWriter) stop() { - m.mu.Lock() - defer m.mu.Unlock() - m.flushPending = false - if m.t != nil { - m.t.Stop() - } -} - -// switchProtocolCopier exists so goroutines proxying data back and -// forth have nice names in stacks. -type switchProtocolCopier struct { - user, backend io.ReadWriter -} - -func (c switchProtocolCopier) copyFromBackend(errc chan<- error) { - _, err := io.Copy(c.user, c.backend) - errc <- err -} - -func (c switchProtocolCopier) copyToBackend(errc chan<- error) { - _, err := io.Copy(c.backend, c.user) - errc <- err -} - // shouldPanicOnCopyError reports whether the reverse proxy should // panic with http.ErrAbortHandler. This is the right thing to do by // default, but Go 1.10 and earlier did not, so existing unit tests @@ -714,6 +494,7 @@ func removeConnectionHeaders(h http.Header) { } } +// LoadBalancing has parameters related to load balancing. type LoadBalancing struct { SelectionPolicyRaw json.RawMessage `json:"selection_policy,omitempty"` TryDuration caddy.Duration `json:"try_duration,omitempty"` @@ -722,8 +503,9 @@ type LoadBalancing struct { SelectionPolicy Selector `json:"-"` } +// Selector selects an available upstream from the pool. type Selector interface { - Select(HostPool, *http.Request) *Upstream + Select(UpstreamPool, *http.Request) *Upstream } // Hop-by-hop headers. These are removed when sent to the backend. @@ -743,117 +525,6 @@ var hopHeaders = []string{ "Upgrade", } -// Host represents a remote host which can be proxied to. -// Its methods must be safe for concurrent use. -type Host interface { - // NumRequests returns the numnber of requests - // currently in process with the host. - NumRequests() int - - // Fails returns the count of recent failures. - Fails() int - - // Unhealthy returns true if the backend is unhealthy. - Unhealthy() bool - - // CountRequest counts the given number of requests - // as currently in process with the host. The count - // should not go below 0. - CountRequest(int) error - - // CountFail counts the given number of failures - // with the host. The count should not go below 0. - CountFail(int) error - - // SetHealthy marks the host as either healthy (true) - // or unhealthy (false). If the given status is the - // same, this should be a no-op. It returns true if - // the given status was different, false otherwise. - SetHealthy(bool) (bool, error) -} - -type HostPool []*Upstream - -type upstreamHost struct { - 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) - fails int64 - unhealthy int32 -} - -func (uh *upstreamHost) NumRequests() int { - return int(atomic.LoadInt64(&uh.numRequests)) -} -func (uh *upstreamHost) Fails() int { - return int(atomic.LoadInt64(&uh.fails)) -} -func (uh *upstreamHost) Unhealthy() bool { - return atomic.LoadInt32(&uh.unhealthy) == 1 -} -func (uh *upstreamHost) CountRequest(delta int) error { - result := atomic.AddInt64(&uh.numRequests, int64(delta)) - if result < 0 { - return fmt.Errorf("count below 0: %d", result) - } - return nil -} -func (uh *upstreamHost) CountFail(delta int) error { - result := atomic.AddInt64(&uh.fails, int64(delta)) - if result < 0 { - return fmt.Errorf("count below 0: %d", result) - } - return nil -} -func (uh *upstreamHost) SetHealthy(healthy bool) (bool, error) { - var unhealthy, compare int32 = 1, 0 - if healthy { - unhealthy, compare = 0, 1 - } - swapped := atomic.CompareAndSwapInt32(&uh.unhealthy, compare, unhealthy) - return swapped, nil -} - -type Upstream struct { - Host `json:"-"` - - Address string `json:"address,omitempty"` - MaxRequests int `json:"max_requests,omitempty"` - - // TODO: This could be really cool, to say that requests with - // certain headers or from certain IPs always go to this upstream - // HeaderAffinity string - // IPAffinity string - - healthCheckPolicy *PassiveHealthChecks - - hostURL *url.URL -} - -func (u Upstream) Available() bool { - return u.Healthy() && !u.Full() -} - -func (u Upstream) Healthy() bool { - healthy := !u.Host.Unhealthy() - if healthy && u.healthCheckPolicy != nil { - healthy = u.Host.Fails() < u.healthCheckPolicy.MaxFails - } - return healthy -} - -func (u Upstream) Full() bool { - return u.MaxRequests > 0 && u.Host.NumRequests() >= u.MaxRequests -} - -func (u Upstream) URL() *url.URL { - return u.hostURL -} - -var hosts = caddy.NewUsagePool() - -// TODO: ... -type UpstreamProvider interface { -} - // TODO: see if we can use this // var bufPool = sync.Pool{ // New: func() interface{} { @@ -863,7 +534,7 @@ type UpstreamProvider interface { // Interface guards var ( - _ caddyhttp.MiddlewareHandler = (*Handler)(nil) _ caddy.Provisioner = (*Handler)(nil) _ caddy.CleanerUpper = (*Handler)(nil) + _ caddyhttp.MiddlewareHandler = (*Handler)(nil) ) diff --git a/modules/caddyhttp/reverseproxy/selectionpolicies.go b/modules/caddyhttp/reverseproxy/selectionpolicies.go index e0518c9..9680583 100644 --- a/modules/caddyhttp/reverseproxy/selectionpolicies.go +++ b/modules/caddyhttp/reverseproxy/selectionpolicies.go @@ -52,7 +52,7 @@ func (RandomSelection) CaddyModule() caddy.ModuleInfo { } // Select returns an available host, if any. -func (r RandomSelection) Select(pool HostPool, request *http.Request) *Upstream { +func (r RandomSelection) Select(pool UpstreamPool, request *http.Request) *Upstream { // use reservoir sampling because the number of available // hosts isn't known: https://en.wikipedia.org/wiki/Reservoir_sampling var randomHost *Upstream @@ -87,6 +87,7 @@ func (RandomChoiceSelection) CaddyModule() caddy.ModuleInfo { } } +// Provision sets up r. func (r *RandomChoiceSelection) Provision(ctx caddy.Context) error { if r.Choose == 0 { r.Choose = 2 @@ -94,6 +95,7 @@ func (r *RandomChoiceSelection) Provision(ctx caddy.Context) error { return nil } +// Validate ensures that r's configuration is valid. func (r RandomChoiceSelection) Validate() error { if r.Choose < 2 { return fmt.Errorf("choose must be at least 2") @@ -102,7 +104,7 @@ func (r RandomChoiceSelection) Validate() error { } // Select returns an available host, if any. -func (r RandomChoiceSelection) Select(pool HostPool, _ *http.Request) *Upstream { +func (r RandomChoiceSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream { k := r.Choose if k > len(pool) { k = len(pool) @@ -142,7 +144,7 @@ func (LeastConnSelection) CaddyModule() caddy.ModuleInfo { // Select selects the up host with the least number of connections in the // pool. If more than one host has the same least number of connections, // one of the hosts is chosen at random. -func (LeastConnSelection) Select(pool HostPool, _ *http.Request) *Upstream { +func (LeastConnSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream { var bestHost *Upstream var count int var leastReqs int @@ -185,7 +187,7 @@ func (RoundRobinSelection) CaddyModule() caddy.ModuleInfo { } // Select returns an available host, if any. -func (r *RoundRobinSelection) Select(pool HostPool, _ *http.Request) *Upstream { +func (r *RoundRobinSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream { n := uint32(len(pool)) if n == 0 { return nil @@ -213,7 +215,7 @@ func (FirstSelection) CaddyModule() caddy.ModuleInfo { } // Select returns an available host, if any. -func (FirstSelection) Select(pool HostPool, _ *http.Request) *Upstream { +func (FirstSelection) Select(pool UpstreamPool, _ *http.Request) *Upstream { for _, host := range pool { if host.Available() { return host @@ -235,7 +237,7 @@ func (IPHashSelection) CaddyModule() caddy.ModuleInfo { } // Select returns an available host, if any. -func (IPHashSelection) Select(pool HostPool, req *http.Request) *Upstream { +func (IPHashSelection) Select(pool UpstreamPool, req *http.Request) *Upstream { clientIP, _, err := net.SplitHostPort(req.RemoteAddr) if err != nil { clientIP = req.RemoteAddr @@ -256,7 +258,7 @@ func (URIHashSelection) CaddyModule() caddy.ModuleInfo { } // Select returns an available host, if any. -func (URIHashSelection) Select(pool HostPool, req *http.Request) *Upstream { +func (URIHashSelection) Select(pool UpstreamPool, req *http.Request) *Upstream { return hostByHashing(pool, req.RequestURI) } @@ -275,7 +277,7 @@ func (HeaderHashSelection) CaddyModule() caddy.ModuleInfo { } // Select returns an available host, if any. -func (s HeaderHashSelection) Select(pool HostPool, req *http.Request) *Upstream { +func (s HeaderHashSelection) Select(pool UpstreamPool, req *http.Request) *Upstream { if s.Field == "" { return nil } diff --git a/modules/caddyhttp/reverseproxy/streaming.go b/modules/caddyhttp/reverseproxy/streaming.go new file mode 100644 index 0000000..a3b711b --- /dev/null +++ b/modules/caddyhttp/reverseproxy/streaming.go @@ -0,0 +1,223 @@ +// Copyright 2015 Matthew Holt and The Caddy Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Most of the code in this file was initially borrowed from the Go +// standard library, which has this copyright notice: +// Copyright 2011 The Go Authors. + +package reverseproxy + +import ( + "context" + "io" + "net/http" + "sync" + "time" +) + +func (h Handler) handleUpgradeResponse(rw http.ResponseWriter, req *http.Request, res *http.Response) { + reqUpType := upgradeType(req.Header) + resUpType := upgradeType(res.Header) + if reqUpType != resUpType { + // TODO: figure out our own error handling + // p.getErrorHandler()(rw, req, fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType)) + return + } + + copyHeader(res.Header, rw.Header()) + + hj, ok := rw.(http.Hijacker) + if !ok { + // p.getErrorHandler()(rw, req, fmt.Errorf("can't switch protocols using non-Hijacker ResponseWriter type %T", rw)) + return + } + backConn, ok := res.Body.(io.ReadWriteCloser) + if !ok { + // p.getErrorHandler()(rw, req, fmt.Errorf("internal error: 101 switching protocols response with non-writable body")) + return + } + defer backConn.Close() + conn, brw, err := hj.Hijack() + if err != nil { + // p.getErrorHandler()(rw, req, fmt.Errorf("Hijack failed on protocol switch: %v", err)) + return + } + defer conn.Close() + res.Body = nil // so res.Write only writes the headers; we have res.Body in backConn above + if err := res.Write(brw); err != nil { + // p.getErrorHandler()(rw, req, fmt.Errorf("response write: %v", err)) + return + } + if err := brw.Flush(); err != nil { + // p.getErrorHandler()(rw, req, fmt.Errorf("response flush: %v", err)) + return + } + errc := make(chan error, 1) + spc := switchProtocolCopier{user: conn, backend: backConn} + go spc.copyToBackend(errc) + go spc.copyFromBackend(errc) + <-errc + return +} + +// flushInterval returns the p.FlushInterval value, conditionally +// overriding its value for a specific request/response. +func (h Handler) flushInterval(req *http.Request, res *http.Response) time.Duration { + resCT := res.Header.Get("Content-Type") + + // For Server-Sent Events responses, flush immediately. + // The MIME type is defined in https://www.w3.org/TR/eventsource/#text-event-stream + if resCT == "text/event-stream" { + return -1 // negative means immediately + } + + // TODO: more specific cases? e.g. res.ContentLength == -1? (this TODO is from the std lib) + return time.Duration(h.FlushInterval) +} + +func (h Handler) copyResponse(dst io.Writer, src io.Reader, flushInterval time.Duration) error { + if flushInterval != 0 { + if wf, ok := dst.(writeFlusher); ok { + mlw := &maxLatencyWriter{ + dst: wf, + latency: flushInterval, + } + defer mlw.stop() + dst = mlw + } + } + + // TODO: Figure out how we want to do this... using custom buffer pool type seems unnecessary + // or maybe it is, depending on how we want to handle errors, + // see: https://github.com/golang/go/issues/21814 + // buf := bufPool.Get().(*bytes.Buffer) + // buf.Reset() + // defer bufPool.Put(buf) + // _, err := io.CopyBuffer(dst, src, ) + var buf []byte + // if h.BufferPool != nil { + // buf = h.BufferPool.Get() + // defer h.BufferPool.Put(buf) + // } + _, err := h.copyBuffer(dst, src, buf) + return err +} + +// copyBuffer returns any write errors or non-EOF read errors, and the amount +// of bytes written. +func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, error) { + if len(buf) == 0 { + buf = make([]byte, 32*1024) + } + var written int64 + for { + nr, rerr := src.Read(buf) + if rerr != nil && rerr != io.EOF && rerr != context.Canceled { + // TODO: this could be useful to know (indeed, it revealed an error in our + // fastcgi PoC earlier; but it's this single error report here that necessitates + // a function separate from io.CopyBuffer, since io.CopyBuffer does not distinguish + // between read or write errors; in a reverse proxy situation, write errors are not + // something we need to report to the client, but read errors are a problem on our + // end for sure. so we need to decide what we want.) + // p.logf("copyBuffer: ReverseProxy read error during body copy: %v", rerr) + } + if nr > 0 { + nw, werr := dst.Write(buf[:nr]) + if nw > 0 { + written += int64(nw) + } + if werr != nil { + return written, werr + } + if nr != nw { + return written, io.ErrShortWrite + } + } + if rerr != nil { + if rerr == io.EOF { + rerr = nil + } + return written, rerr + } + } +} + +type writeFlusher interface { + io.Writer + http.Flusher +} + +type maxLatencyWriter struct { + dst writeFlusher + latency time.Duration // non-zero; negative means to flush immediately + + mu sync.Mutex // protects t, flushPending, and dst.Flush + t *time.Timer + flushPending bool +} + +func (m *maxLatencyWriter) Write(p []byte) (n int, err error) { + m.mu.Lock() + defer m.mu.Unlock() + n, err = m.dst.Write(p) + if m.latency < 0 { + m.dst.Flush() + return + } + if m.flushPending { + return + } + if m.t == nil { + m.t = time.AfterFunc(m.latency, m.delayedFlush) + } else { + m.t.Reset(m.latency) + } + m.flushPending = true + return +} + +func (m *maxLatencyWriter) delayedFlush() { + m.mu.Lock() + defer m.mu.Unlock() + if !m.flushPending { // if stop was called but AfterFunc already started this goroutine + return + } + m.dst.Flush() + m.flushPending = false +} + +func (m *maxLatencyWriter) stop() { + m.mu.Lock() + defer m.mu.Unlock() + m.flushPending = false + if m.t != nil { + m.t.Stop() + } +} + +// switchProtocolCopier exists so goroutines proxying data back and +// forth have nice names in stacks. +type switchProtocolCopier struct { + user, backend io.ReadWriter +} + +func (c switchProtocolCopier) copyFromBackend(errc chan<- error) { + _, err := io.Copy(c.user, c.backend) + errc <- err +} + +func (c switchProtocolCopier) copyToBackend(errc chan<- error) { + _, err := io.Copy(c.backend, c.user) + errc <- err +} |