plugin/forward: cap default connect attempts (#8365)

Default to two connect attempts per configured upstream so fast failures cannot spin until the request deadline. Track whether max_connect_attempts was explicitly configured so zero still opts into the legacy unbounded behavior.

Fixes #7723

Signed-off-by: houyuwushang <liuluoqianqiu@outlook.com>

plugin/forward/README.md

@@ -75,8 +75,9 @@ forward FROM TO... {
   an upstream to be down. If 0, the upstream will never be marked as down (nor health checked).
   Default is 2.
 * `max_connect_attempts` caps the total number of upstream connect attempts
-  performed for a single incoming DNS request. Default value of 0 means no per-request
+  performed for a single incoming DNS request. The default cap is twice the number of
-  cap.
+  configured upstreams, allowing two complete passes when all upstreams are healthy.
+  Set this to 0 to disable the per-request cap.
 * `expire` **DURATION**, expire (cached) connections after this time, the default is 10s.
 * `doh_method` **GET|POST**, whether to use GET or POST http method for DoH requests (defaults to POST).
 * `max_idle_conns` **INTEGER**, maximum number of idle connections to cache per upstream for reuse.

plugin/forward/forward.go

@@ -30,9 +30,10 @@ import (
 var log = clog.NewWithPlugin("forward")
 
 const (
-	defaultExpire      = 10 * time.Second
+	defaultExpire                     = 10 * time.Second
-	defaultReadTimeout = 2 * time.Second
+	defaultReadTimeout                = 2 * time.Second
-	hcInterval         = 500 * time.Millisecond
+	hcInterval                        = 500 * time.Millisecond
+	defaultConnectAttemptsPerUpstream = 2
 )
 
 // Forward represents a plugin instance that can proxy requests to another (DNS) server. It has a list
@@ -62,6 +63,7 @@ type Forward struct {
 	failfastUnhealthyUpstreams bool
 	failoverRcodes             []int
 	maxConnectAttempts         uint32
+	maxConnectAttemptsSet      bool
 	sourceAddress              net.IP
 
 	// Hostname resolution fields
@@ -135,9 +137,13 @@ func (f *Forward) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg
 	list := f.List()
 	deadline := time.Now().Add(defaultTimeout)
 	start := time.Now()
-	connectAttempts := uint32(0)
+	maxConnectAttempts := uint64(f.maxConnectAttempts)
+	if !f.maxConnectAttemptsSet {
+		maxConnectAttempts = uint64(defaultConnectAttemptsPerUpstream) * uint64(len(list))
+	}
+	connectAttempts := uint64(0)
 
-	for time.Now().Before(deadline) && ctx.Err() == nil && (f.maxConnectAttempts == 0 || connectAttempts < f.maxConnectAttempts) {
+	for time.Now().Before(deadline) && ctx.Err() == nil && (maxConnectAttempts == 0 || connectAttempts < maxConnectAttempts) {
 		if i >= len(list) {
 			// reached the end of list, reset to begin
 			i = 0
@@ -215,11 +221,9 @@ func (f *Forward) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg
 				proxy.Healthcheck()
 			}
 
-			// If a per-request connect-attempt cap is configured, count this
+			if maxConnectAttempts > 0 {
-			// failed connect attempt and stop retrying when the cap is hit.
-			if f.maxConnectAttempts > 0 {
 				connectAttempts++
-				if connectAttempts >= f.maxConnectAttempts {
+				if connectAttempts >= maxConnectAttempts {
 					break
 				}
 			}

plugin/forward/forward_test.go

@@ -99,8 +99,8 @@ func (m *mockResponseWriter) Hijack()                      {}
 
 // TestForward_Regression_NoBusyLoop ensures that ServeDNS does not perform
 // an unbounded number of upstream connect attempts for a single request when
-// maxConnectAttempts is configured, and that maxConnectAttempts=0 keeps the
+// maxConnectAttempts is configured, and that an explicit maxConnectAttempts=0
-// legacy behaviour (no per-request cap).
+// keeps the legacy behaviour (no per-request cap).
 func TestForward_Regression_NoBusyLoop(t *testing.T) {
 	tests := []struct {
 		name        string
@@ -122,6 +122,7 @@ func TestForward_Regression_NoBusyLoop(t *testing.T) {
 
 			// Set maxConnectAttempts to the number of attempts we want to test.
 			f.maxConnectAttempts = tc.maxAttempts
+			f.maxConnectAttemptsSet = true
 
 			// Assume nothing is listening on this port, so the connection will be refused.
 			p := proxy.NewProxy("forward", "127.0.0.1:54321", "tcp")
@@ -162,6 +163,74 @@ func TestForward_Regression_NoBusyLoop(t *testing.T) {
 	}
 }
 
+func TestForward_DefaultConnectAttemptCap(t *testing.T) {
+	for _, proxyCount := range []int{1, 3} {
+		t.Run(fmt.Sprintf("%d upstreams", proxyCount), func(t *testing.T) {
+			f := New()
+			f.opts.ForceTCP = true
+			f.maxfails = 0
+
+			listeners := make([]net.Listener, 0, proxyCount)
+			t.Cleanup(func() {
+				for _, listener := range listeners {
+					_ = listener.Close()
+				}
+			})
+			for range proxyCount {
+				listener, err := net.Listen("tcp", "127.0.0.1:0")
+				if err != nil {
+					t.Fatalf("failed to allocate upstream address: %v", err)
+				}
+				listeners = append(listeners, listener)
+			}
+
+			upstreams := make([]string, 0, proxyCount)
+			for _, listener := range listeners {
+				upstream := listener.Addr().String()
+				if err := listener.Close(); err != nil {
+					t.Fatalf("failed to close upstream listener: %v", err)
+				}
+				upstreams = append(upstreams, upstream)
+				f.SetProxy(proxy.NewProxy("forward", upstream, "tcp"))
+			}
+
+			tracer := mocktracer.New()
+			span := tracer.StartSpan("test")
+			ctx := opentracing.ContextWithSpan(context.Background(), span)
+			ctx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
+			defer cancel()
+
+			req := new(dns.Msg)
+			req.SetQuestion("example.com.", dns.TypeA)
+
+			_, err := f.ServeDNS(ctx, &mockResponseWriter{}, req)
+			if err == nil {
+				t.Fatal("expected connection refused error")
+			}
+
+			want := defaultConnectAttemptsPerUpstream * proxyCount
+			spans := tracer.FinishedSpans()
+			if got := len(spans); got != want {
+				t.Fatalf("expected %d connect attempts, got %d", want, got)
+			}
+
+			attemptsByUpstream := make(map[string]int, proxyCount)
+			for _, span := range spans {
+				upstream, ok := span.Tags()["peer.address"].(string)
+				if !ok {
+					t.Fatal("connect attempt is missing peer.address")
+				}
+				attemptsByUpstream[upstream]++
+			}
+			for _, upstream := range upstreams {
+				if got := attemptsByUpstream[upstream]; got != defaultConnectAttemptsPerUpstream {
+					t.Errorf("expected %d attempts to %s, got %d", defaultConnectAttemptsPerUpstream, upstream, got)
+				}
+			}
+		})
+	}
+}
+
 func TestForward_NextOnNodata(t *testing.T) {
 	tests := []struct {
 		name         string
@@ -323,6 +392,7 @@ func TestForwardDoesNotRetryLocalPackError(t *testing.T) {
 	f := New()
 	f.maxfails = 0
 	f.maxConnectAttempts = 2
+	f.maxConnectAttemptsSet = true
 	f.opts.ForceTCP = true
 	f.proxies = []*proxy.Proxy{
 		proxy.NewProxy("forward", "127.0.0.1:1", transport.DNS),

plugin/forward/setup.go

@@ -293,6 +293,7 @@ func parseBlock(c *caddy.Controller, f *Forward) error {
 			return err
 		}
 		f.maxConnectAttempts = uint32(n)
+		f.maxConnectAttemptsSet = true
 	case "health_check":
 		if !c.NextArg() {
 			return c.ArgErr()

plugin/forward/setup_test.go

@@ -405,13 +405,15 @@ func TestSetupMaxConnectAttempts(t *testing.T) {
 		input       string
 		shouldErr   bool
 		expectedVal uint32
+		expectedSet bool
 		expectedErr string
 	}{
 
-		{"forward . 127.0.0.1 {\n}\n", false, 0, ""},
+		{"forward . 127.0.0.1 {\n}\n", false, 0, false, ""},
-		{"forward . 127.0.0.1 {\nmax_connect_attempts 5\n}\n", false, 5, ""},
+		{"forward . 127.0.0.1 {\nmax_connect_attempts 0\n}\n", false, 0, true, ""},
-		{"forward . 127.0.0.1 {\nmax_connect_attempts many\n}\n", true, 0, "invalid"},
+		{"forward . 127.0.0.1 {\nmax_connect_attempts 5\n}\n", false, 5, true, ""},
-		{"forward . 127.0.0.1 {\nmax_connect_attempts -4\n}\n", true, 0, "invalid"},
+		{"forward . 127.0.0.1 {\nmax_connect_attempts many\n}\n", true, 0, false, "invalid"},
+		{"forward . 127.0.0.1 {\nmax_connect_attempts -4\n}\n", true, 0, false, "invalid"},
 	}
 
 	for i, test := range tests {
@@ -437,6 +439,9 @@ func TestSetupMaxConnectAttempts(t *testing.T) {
 			if f.maxConnectAttempts != test.expectedVal {
 				t.Errorf("Test %d: expected: %d, got: %d", i, test.expectedVal, f.maxConnectAttempts)
 			}
+			if f.maxConnectAttemptsSet != test.expectedSet {
+				t.Errorf("Test %d: expected configured state %t, got %t", i, test.expectedSet, f.maxConnectAttemptsSet)
+			}
 		}
 	}
 }