l4proxy: add dynamic upstreams via DNS (SRV and A/AAAA)

docs/handlers/proxy.md

@@ -26,6 +26,17 @@ The handler has the following optional fields:
 - `upstreams` may contain a list of `l4proxy.Upstream` structures (valid for JSON). In a Caddyfile, multiple `upstream`
   options or blocks are unmarshalled into a list of such structures.
 
+- `dynamic_upstreams` may contain an upstream-source module that discovers the upstreams at runtime instead of listing
+  them statically, so the backend set need not be restated in config when DNS already publishes it. In a Caddyfile it
+  is `dynamic <source> { ... }`. Two DNS sources are provided:
+  - `srv` resolves SRV records. Options: `service`, `proto`, `name` (or just `name` for the full domain), `refresh`
+    (default `1m`), `grace_period` (serve stale results for this long on lookup failure), `dial_network`.
+  - `a` resolves A/AAAA records for a `name` on a configured `port`. Options: `name`, `port`, `refresh`,
+    `grace_period`, `dial_network`.
+
+  When `dynamic_upstreams` is configured, the static `upstreams` list may be empty. Note: active health checks run on
+  statically-configured upstreams only.
+
 **Active health checks** occur independently in a background goroutine. They run in the background on a timer.
 To minimally enable active health checks, set `active` field equal to an empty structure inside `health_checks` in
 a JSON configuration or include any active health check option into a Caddyfile.

integration/caddyfile_adapt/gd_handler_proxy_dynamic_srv.caddytest

@@ -0,0 +1,46 @@
+{
+	layer4 {
+		:5432 {
+			route {
+				proxy {
+					dynamic srv {
+						service postgres
+						proto tcp
+						name db.internal
+						refresh 30s
+					}
+				}
+			}
+		}
+	}
+}
+----------
+{
+	"apps": {
+		"layer4": {
+			"servers": {
+				"srv0": {
+					"listen": [
+						":5432"
+					],
+					"routes": [
+						{
+							"handle": [
+								{
+									"dynamic_upstreams": {
+										"name": "db.internal",
+										"proto": "tcp",
+										"refresh": 30000000000,
+										"service": "postgres",
+										"source": "srv"
+									},
+									"handler": "proxy"
+								}
+							]
+						}
+					]
+				}
+			}
+		}
+	}
+}

modules/l4proxy/proxy.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"fmt"
 	"io"
 	"log"
@@ -46,9 +47,14 @@ func init() {
 
 // Handler is a handler that can proxy connections.
 type Handler struct {
-	// Upstreams is the list of backends to proxy to.
+	// Upstreams is the static list of backends to proxy to.
 	Upstreams UpstreamPool `json:"upstreams,omitempty"`
 
+	// DynamicUpstreamsRaw is a module that discovers upstreams dynamically (per
+	// connection) instead of listing them statically — e.g. from DNS SRV
+	// records, so the backend set need not be restated in config.
+	DynamicUpstreamsRaw json.RawMessage `json:"dynamic_upstreams,omitempty" caddy:"namespace=layer4.proxy.upstreams inline_key=source"`
+
 	// Health checks update the status of backends, whether they are
 	// up or down. Down backends will not be proxied to.
 	HealthChecks *HealthChecks `json:"health_checks,omitempty"`
@@ -62,6 +68,8 @@ type Handler struct {
 
 	proxyProtocolVersion uint8
 
+	dynamicUpstreams UpstreamSource
+
 	ctx    caddy.Context
 	logger *zap.Logger
 }
@@ -98,8 +106,17 @@ func (h *Handler) Provision(ctx caddy.Context) error {
 		return fmt.Errorf("proxy_protocol: \"%s\" should be empty, or one of \"v1\" \"v2\"", proxyProtocol)
 	}
 
+	// load the dynamic upstreams source module, if configured
+	if h.DynamicUpstreamsRaw != nil {
+		mod, err := ctx.LoadModule(h, "DynamicUpstreamsRaw")
+		if err != nil {
+			return fmt.Errorf("loading dynamic upstreams source module: %v", err)
+		}
+		h.dynamicUpstreams = mod.(UpstreamSource)
+	}
+
 	// prepare upstreams
-	if len(h.Upstreams) == 0 {
+	if len(h.Upstreams) == 0 && h.dynamicUpstreams == nil {
 		return fmt.Errorf("no upstreams defined")
 	}
 	for i, ups := range h.Upstreams {
@@ -160,9 +177,20 @@ func (h *Handler) Handle(down *layer4.Connection, _ layer4.Handler) error {
 	var upConns []net.Conn
 	var proxyErr error
 
+	// determine the pool: dynamically discovered (per connection) or static
+	pool := h.Upstreams
+	if h.dynamicUpstreams != nil {
+		dynUpstreams, err := h.dynamicUpstreams.GetUpstreams(repl)
+		if err != nil {
+			h.logger.Error("getting dynamic upstreams", zap.Error(err))
+		} else {
+			pool = dynUpstreams
+		}
+	}
+
 	for {
 		// choose an available upstream
-		upstream := h.LoadBalancing.SelectionPolicy.Select(h.Upstreams, down)
+		upstream := h.LoadBalancing.SelectionPolicy.Select(pool, down)
 		if upstream == nil {
 			if proxyErr == nil {
 				proxyErr = fmt.Errorf("no upstreams available")
@@ -502,6 +530,11 @@ func (h *Handler) Cleanup() error {
 //
 //		proxy_protocol <v1|v2>
 //
+//		# discover upstreams dynamically instead of listing them
+//		dynamic <source> [<args...>] {
+//			...
+//		}
+//
 //		# multiple upstream options are supported
 //		upstream [<args...>] {
 //			...
@@ -697,6 +730,28 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 				return d.Errf("duplicate %s option '%s'", wrapper, optionName)
 			}
 			_, h.ProxyProtocol, hasProxyProtocol = d.NextArg(), d.Val(), true
+		case "dynamic":
+			if h.DynamicUpstreamsRaw != nil {
+				return d.Errf("duplicate %s option '%s'", wrapper, optionName)
+			}
+			if !d.NextArg() {
+				return d.ArgErr()
+			}
+			sourceName := d.Val()
+			unm, err := caddyfile.UnmarshalModule(d, "layer4.proxy.upstreams."+sourceName)
+			if err != nil {
+				return err
+			}
+			source, ok := unm.(UpstreamSource)
+			if !ok {
+				return d.Errf("module '%s' is not an upstream source", sourceName)
+			}
+			sourceRaw := caddyconfig.JSON(source, nil)
+			sourceRaw, err = layer4.SetModuleNameInline("source", sourceName, sourceRaw)
+			if err != nil {
+				return d.Errf("re-encoding module '%s' configuration: %v", sourceName, err)
+			}
+			h.DynamicUpstreamsRaw = sourceRaw
 		case "upstream":
 			u := &Upstream{}
 			if err := u.UnmarshalCaddyfile(d.NewFromNextSegment()); err != nil {