Add *ready* plugin (#2616)

Add a ready plugin that allows plugin to signal when they are ready.
Once a plugin is ready it is not queried again.

This uses same mechanism as the health plugin: each plugin needs to
implement an interface.

Implement readines for the *erratic* plugin to aid in testing.

Add README.md and tests moduled after the health plugin; which will be
relegated to just providing process health. In similar vein to health
this is a process wide setting.

With this Corefile:
~~~
. {
    erratic
    whoami
    ready
}

bla {
    erratic
    whoami
}
~~~

ready will lead to:

~~~ sh
% curl localhost:8181/ready
% dig @localhost -p 1053 mx example.org
% curl localhost:8181/ready
OK%
~~~

Meanwhile CoreDNS logs:

~~~
.:1053
bla.:1053
2019-02-26T20:59:07.137Z [INFO] CoreDNS-1.3.1
2019-02-26T20:59:07.137Z [INFO] linux/amd64, go1.11.4,
CoreDNS-1.3.1
linux/amd64, go1.11.4,
2019-02-26T20:59:11.415Z [INFO] plugin/ready: Still waiting on: "erratic"
2019-02-26T20:59:13.510Z [INFO] plugin/ready: Still waiting on: "erratic"
~~~

*ready* can be used in multiple server blocks and will do the right
thing; query all those plugins from all server blocks for readiness.
This does a similar thing to the prometheus plugin.

Signed-off-by: Miek Gieben <miek@miek.nl>

core/dnsserver/zdirectives.go

@@ -18,6 +18,7 @@ var Directives = []string{
 	"bind",
 	"debug",
 	"trace",
+	"ready",
 	"health",
 	"pprof",
 	"prometheus",

core/plugin/zplugin.go

@@ -29,6 +29,7 @@ import (
 	_ "github.com/coredns/coredns/plugin/metrics"
 	_ "github.com/coredns/coredns/plugin/nsid"
 	_ "github.com/coredns/coredns/plugin/pprof"
+	_ "github.com/coredns/coredns/plugin/ready"
 	_ "github.com/coredns/coredns/plugin/reload"
 	_ "github.com/coredns/coredns/plugin/rewrite"
 	_ "github.com/coredns/coredns/plugin/root"

plugin.cfg

@@ -27,6 +27,7 @@ root:root
 bind:bind
 debug:debug
 trace:trace
+ready:ready
 health:health
 pprof:pprof
 prometheus:metrics

plugin.md

@@ -57,8 +57,13 @@ server.
 
 When exporting metrics the *Namespace* should be `plugin.Namespace` (="coredns"), and the
 *Subsystem* should be the name of the plugin. The README.md for the plugin should then also contain
- a *Metrics* section detailing the metrics. If the plugin supports dynamic health reporting it
- should also have *Health* section detailing on some of its inner workings.
+a *Metrics* section detailing the metrics.
+
+If the plugin supports dynamic health reporting it should also have *Health* section detailing on
+some of its inner workings.
+
+If the plugins supports signalling readiness it should have a *Ready* section detailing how it
+works.
 
 ## Documentation
 

plugin/erratic/README.md

@@ -37,6 +37,10 @@ In case of a zone transfer and truncate the final SOA record *isn't* added to th
 
 This plugin implements dynamic health checking. For every dropped query it turns unhealthy.
 
+## Ready
+
+This plugin reports readiness to the ready plugin.
+
 ## Examples
 
 ~~~ corefile

plugin/erratic/ready.go

@@ -0,0 +1,13 @@
+package erratic
+
+import "sync/atomic"
+
+// Ready returns true if the number of received queries is in the range [3, 5). All other values return false.
+// To aid in testing we want to this flip between ready and not ready.
+func (e *Erratic) Ready() bool {
+	q := atomic.LoadUint64(&e.q)
+	if q >= 3 && q < 5 {
+		return true
+	}
+	return false
+}

plugin/ready/README.md

@@ -0,0 +1,56 @@
+# ready
+
+## Name
+
+*ready* - enables a readiness check HTTP endpoint.
+
+## Description
+
+By enabling *ready* an HTTP endpoint on port 8181 will return 200 OK, when all plugins that are able
+to signal readiness have done so. If some are not ready yet the endpoint will return a 503 with the
+body containing the list of plugins that are not ready. Once a plugin has signaled it is ready it
+will not be queried again.
+
+Each Server Block that enables the *ready* plugin will have the plugins *in that server block*
+report readiness into the /ready endpoint that runs on the same port.
+
+## Syntax
+
+~~~
+ready [ADDRESS]
+~~~
+
+*ready* optionally takes an address; the default is `:8181`. The path is fixed to `/ready`. The
+readiness endpoint returns a 200 response code and the word "OK" when this server is ready. It
+returns a 503 otherwise.
+
+## Plugins
+
+Any plugin wanting to signal readiness will need to implement the `ready.Readiness` interface by
+implementing a method `Ready() bool` that returns true when the plugin is ready and false otherwise.
+
+## Examples
+
+Let *ready* report readiness for both the `.` and `example.org` servers (assuming the *whois*
+plugin also exports readiness):
+
+~~~ txt
+. {
+    ready
+    erratic
+}
+
+example.org {
+    ready
+    whoami
+}
+
+~~~
+
+Run *ready* on a different port.
+
+~~~ txt
+. {
+    ready localhost:8091
+}
+~~~

plugin/ready/list.go

@@ -0,0 +1,48 @@
+package ready
+
+import (
+	"sort"
+	"strings"
+	"sync"
+)
+
+// list is structure that holds the plugins that signals readiness for this server block.
+type list struct {
+	sync.RWMutex
+	rs    []Readiness
+	names []string
+}
+
+// Append adds a new readiness to l.
+func (l *list) Append(r Readiness, name string) {
+	l.Lock()
+	defer l.Unlock()
+	l.rs = append(l.rs, r)
+	l.names = append(l.names, name)
+}
+
+// Ready return true when all plugins ready, if the returned value is false the string
+// contains a comma separated list of plugins that are not ready.
+func (l *list) Ready() (bool, string) {
+	l.RLock()
+	defer l.RUnlock()
+	ok := true
+	s := []string{}
+	for i, r := range l.rs {
+		if r == nil {
+			continue
+		}
+		if !r.Ready() {
+			ok = false
+			s = append(s, l.names[i])
+		} else {
+			// if ok, this plugin is ready and will not be queried anymore.
+			l.rs[i] = nil
+		}
+	}
+	if ok {
+		return true, ""
+	}
+	sort.Strings(s)
+	return false, strings.Join(s, ",")
+}

plugin/ready/readiness.go

@@ -0,0 +1,7 @@
+package ready
+
+// The Readiness interface needs to be implemented by each plugin willing to provide a readiness check.
+type Readiness interface {
+	// Ready is called by ready to see whether the plugin is ready.
+	Ready() bool
+}

plugin/ready/ready.go

@@ -0,0 +1,81 @@
+// Package ready is used to signal readiness of the CoreDNS process. Once all
+// plugins have called in the plugin will signal readiness by returning a 200
+// OK on the HTTP handler (on port 8181). If not ready yet, the handler will
+// return a 503.
+package ready
+
+import (
+	"io"
+	"net"
+	"net/http"
+	"sync"
+
+	clog "github.com/coredns/coredns/plugin/pkg/log"
+	"github.com/coredns/coredns/plugin/pkg/uniq"
+)
+
+var (
+	log      = clog.NewWithPlugin("ready")
+	plugins  = &list{}
+	uniqAddr = uniq.New()
+)
+
+type ready struct {
+	Addr string
+
+	sync.RWMutex
+	ln   net.Listener
+	done bool
+	mux  *http.ServeMux
+}
+
+func (rd *ready) onStartup() error {
+	if rd.Addr == "" {
+		rd.Addr = defAddr
+	}
+
+	ln, err := net.Listen("tcp", rd.Addr)
+	if err != nil {
+		return err
+	}
+
+	rd.Lock()
+	rd.ln = ln
+	rd.mux = http.NewServeMux()
+	rd.done = true
+	rd.Unlock()
+
+	rd.mux.HandleFunc("/ready", func(w http.ResponseWriter, _ *http.Request) {
+		ok, todo := plugins.Ready()
+		if ok {
+			w.WriteHeader(http.StatusOK)
+			io.WriteString(w, "OK")
+			return
+		}
+		log.Infof("Still waiting on: %q", todo)
+		w.WriteHeader(http.StatusServiceUnavailable)
+		io.WriteString(w, todo)
+	})
+
+	go func() { http.Serve(rd.ln, rd.mux) }()
+
+	return nil
+}
+
+func (rd *ready) onRestart() error { return rd.onFinalShutdown() }
+
+func (rd *ready) onFinalShutdown() error {
+	rd.Lock()
+	defer rd.Unlock()
+	if !rd.done {
+		return nil
+	}
+
+	uniqAddr.Unset(rd.Addr)
+
+	rd.ln.Close()
+	rd.done = false
+	return nil
+}
+
+const defAddr = ":8181"

plugin/ready/ready_test.go

@@ -0,0 +1,81 @@
+package ready
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+	"testing"
+
+	"github.com/coredns/coredns/plugin/erratic"
+	clog "github.com/coredns/coredns/plugin/pkg/log"
+	"github.com/coredns/coredns/plugin/test"
+
+	"github.com/miekg/dns"
+)
+
+func init() { clog.Discard() }
+
+func TestReady(t *testing.T) {
+	rd := &ready{Addr: ":0"}
+	e := &erratic.Erratic{}
+	plugins.Append(e, "erratic")
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		if err := rd.onStartup(); err != nil {
+			t.Fatalf("Unable to startup the readiness server: %v", err)
+		}
+		wg.Done()
+	}()
+	wg.Wait()
+
+	defer rd.onFinalShutdown()
+
+	address := fmt.Sprintf("http://%s/ready", rd.ln.Addr().String())
+
+	wg.Add(1)
+	go func() {
+		response, err := http.Get(address)
+		if err != nil {
+			t.Fatalf("Unable to query %s: %v", address, err)
+		}
+		if response.StatusCode != 503 {
+			t.Errorf("Invalid status code: expecting %d, got %d", 503, response.StatusCode)
+		}
+		response.Body.Close()
+		wg.Done()
+	}()
+	wg.Wait()
+
+	// make it ready by giving erratic 3 queries.
+	m := new(dns.Msg)
+	m.SetQuestion("example.org.", dns.TypeA)
+	e.ServeDNS(context.TODO(), &test.ResponseWriter{}, m)
+	e.ServeDNS(context.TODO(), &test.ResponseWriter{}, m)
+	e.ServeDNS(context.TODO(), &test.ResponseWriter{}, m)
+
+	response, err := http.Get(address)
+	if err != nil {
+		t.Fatalf("Unable to query %s: %v", address, err)
+	}
+	if response.StatusCode != 200 {
+		t.Errorf("Invalid status code: expecting %d, got %d", 200, response.StatusCode)
+	}
+	response.Body.Close()
+
+	// make erratic not-ready by giving it more queries, this should not change the process readiness
+	e.ServeDNS(context.TODO(), &test.ResponseWriter{}, m)
+	e.ServeDNS(context.TODO(), &test.ResponseWriter{}, m)
+	e.ServeDNS(context.TODO(), &test.ResponseWriter{}, m)
+
+	response, err = http.Get(address)
+	if err != nil {
+		t.Fatalf("Unable to query %s: %v", address, err)
+	}
+	if response.StatusCode != 200 {
+		t.Errorf("Invalid status code: expecting %d, got %d", 200, response.StatusCode)
+	}
+	response.Body.Close()
+}

plugin/ready/setup.go

@@ -0,0 +1,75 @@
+package ready
+
+import (
+	"net"
+
+	"github.com/coredns/coredns/core/dnsserver"
+	"github.com/coredns/coredns/plugin"
+
+	"github.com/mholt/caddy"
+)
+
+func init() {
+	caddy.RegisterPlugin("ready", caddy.Plugin{
+		ServerType: "dns",
+		Action:     setup,
+	})
+}
+
+func setup(c *caddy.Controller) error {
+	addr, err := parse(c)
+	if err != nil {
+		return plugin.Error("ready", err)
+	}
+
+	rd := &ready{Addr: addr}
+
+	uniqAddr.Set(addr, rd.onStartup, rd)
+
+	c.OncePerServerBlock(func() error {
+		c.OnStartup(func() error {
+			return uniqAddr.ForEach()
+		})
+		return nil
+	})
+
+	c.OnStartup(func() error {
+		// Each plugin in this server block will (if they support it) report readiness.
+		plugs := dnsserver.GetConfig(c).Handlers()
+		for _, p := range plugs {
+			if r, ok := p.(Readiness); ok {
+				plugins.Append(r, p.Name())
+			}
+		}
+		return nil
+	})
+
+	c.OnRestart(rd.onRestart)
+	c.OnFinalShutdown(rd.onFinalShutdown)
+
+	return nil
+}
+
+func parse(c *caddy.Controller) (string, error) {
+	addr := ""
+	i := 0
+	for c.Next() {
+		if i > 0 {
+			return "", plugin.ErrOnce
+		}
+		i++
+		args := c.RemainingArgs()
+
+		switch len(args) {
+		case 0:
+		case 1:
+			addr = args[0]
+			if _, _, e := net.SplitHostPort(addr); e != nil {
+				return "", e
+			}
+		default:
+			return "", c.ArgErr()
+		}
+	}
+	return addr, nil
+}

plugin/ready/setup_test.go

@@ -0,0 +1,34 @@
+package ready
+
+import (
+	"testing"
+
+	"github.com/mholt/caddy"
+)
+
+func TestSetupReady(t *testing.T) {
+	tests := []struct {
+		input     string
+		shouldErr bool
+	}{
+		{`ready`, false},
+		{`ready localhost:1234`, false},
+		{`ready localhost:1234 b`, true},
+		{`ready bla`, true},
+		{`ready bla bla`, true},
+	}
+
+	for i, test := range tests {
+		_, err := parse(caddy.NewTestController("dns", test.input))
+
+		if test.shouldErr && err == nil {
+			t.Errorf("Test %d: Expected error but found none for input %s", i, test.input)
+		}
+
+		if err != nil {
+			if !test.shouldErr {
+				t.Errorf("Test %d: Expected no error but found one for input %s. Error was: %v", i, test.input, err)
+			}
+		}
+	}
+}