plugin/forward: add query timeout (#1665)

2 files changed, 62 insertions, 3 deletions

diff --git a/plugin/forward/forward.go b/plugin/forward/forward.go
index 6d06f79f2..ba251d9ea 100644
--- a/plugin/forward/forward.go
+++ b/plugin/forward/forward.go
@@ -67,8 +67,19 @@ func (f *Forward) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg
 	var span, child ot.Span
 	var upstreamErr error
 	span = ot.SpanFromContext(ctx)
+	i := 0
+	list := f.list()
+	deadline := time.Now().Add(defaultTimeout)
+
+	for time.Now().Before(deadline) {
+		if i >= len(list) {
+			// reached the end of list, reset to begin
+			i = 0
+			fails = 0
+		}
 
-	for _, proxy := range f.list() {
+		proxy := list[i]
+		i++
 		if proxy.Down(f.maxfails) {
 			fails++
 			if fails < len(f.proxies) {
@@ -183,3 +194,5 @@ const (
 	randomPolicy policy = iota
 	roundRobinPolicy
 )
+
+const defaultTimeout = 5 * time.Second
diff --git a/plugin/forward/health_test.go b/plugin/forward/health_test.go
index 2698d13e2..6d286a496 100644
--- a/plugin/forward/health_test.go
+++ b/plugin/forward/health_test.go
@@ -45,6 +45,7 @@ func TestHealth(t *testing.T) {
 func TestHealthTimeout(t *testing.T) {
 	const expected = 1
 	i := uint32(0)
+	q := uint32(0)
 	s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
 		if r.Question[0].Name == "." {
 			// health check, answer
@@ -52,8 +53,15 @@ func TestHealthTimeout(t *testing.T) {
 			ret := new(dns.Msg)
 			ret.SetReply(r)
 			w.WriteMsg(ret)
+			return
 		}
-		// not a health check, do a timeout
+		if atomic.LoadUint32(&q) == 0 { //drop only first query
+			atomic.AddUint32(&q, 1)
+			return
+		}
+		ret := new(dns.Msg)
+		ret.SetReply(r)
+		w.WriteMsg(ret)
 	})
 	defer s.Close()
 
@@ -77,6 +85,7 @@ func TestHealthTimeout(t *testing.T) {
 func TestHealthFailTwice(t *testing.T) {
 	const expected = 2
 	i := uint32(0)
+	q := uint32(0)
 	s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
 		if r.Question[0].Name == "." {
 			atomic.AddUint32(&i, 1)
@@ -88,7 +97,15 @@ func TestHealthFailTwice(t *testing.T) {
 			ret := new(dns.Msg)
 			ret.SetReply(r)
 			w.WriteMsg(ret)
+			return
+		}
+		if atomic.LoadUint32(&q) == 0 { //drop only first query
+			atomic.AddUint32(&q, 1)
+			return
 		}
+		ret := new(dns.Msg)
+		ret.SetReply(r)
+		w.WriteMsg(ret)
 	})
 	defer s.Close()
 
@@ -110,10 +127,39 @@ func TestHealthFailTwice(t *testing.T) {
 }
 
 func TestHealthMaxFails(t *testing.T) {
+	s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
+		// timeout
+	})
+	defer s.Close()
+
+	p := NewProxy(s.Addr, nil /* no TLS */)
+	f := New()
+	f.maxfails = 2
+	f.SetProxy(p)
+	defer f.Close()
+
+	req := new(dns.Msg)
+	req.SetQuestion("example.org.", dns.TypeA)
+
+	f.ServeDNS(context.TODO(), &test.ResponseWriter{}, req)
+
+	time.Sleep(1 * time.Second)
+	if !p.Down(f.maxfails) {
+		t.Errorf("Expected Proxy fails to be greater than %d, got %d", f.maxfails, p.fails)
+	}
+}
+
+func TestHealthNoMaxFails(t *testing.T) {
 	const expected = 0
 	i := uint32(0)
 	s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
-		// timeout
+		if r.Question[0].Name == "." {
+			// health check, answer
+			atomic.AddUint32(&i, 1)
+			ret := new(dns.Msg)
+			ret.SetReply(r)
+			w.WriteMsg(ret)
+		}
 	})
 	defer s.Close()