plugin/metrcs: fix datarace on listeners (#2835)

This fixes a data race on the listener(s) that get started in the
metrics plugins.

It also restore pkg/uniq to its former glory and removes and state being
carried in there; this means for metrics that registry.go was to
replicate that behavior *with* locking (as pkg/uniq doesn't do, or need
that).

Also renamed uniqAddr to just u, to make it slightly shorter.

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

plugin/metrics/metrics.go

@@ -15,22 +15,24 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
-// Metrics holds the prometheus configuration. The metrics' path is fixed to be /metrics
+// Metrics holds the prometheus configuration. The metrics' path is fixed to be /metrics .
 type Metrics struct {
-	Next    plugin.Handler
-	Addr    string
-	Reg     *prometheus.Registry
+	Next plugin.Handler
+	Addr string
+	Reg  *prometheus.Registry
+
 	ln      net.Listener
 	lnSetup bool
-	mux     *http.ServeMux
-	srv     *http.Server
+
+	mux *http.ServeMux
+	srv *http.Server
 
 	zoneNames []string
 	zoneMap   map[string]struct{}
 	zoneMu    sync.RWMutex
 }
 
-// New returns a new instance of Metrics with the given address
+// New returns a new instance of Metrics with the given address.
 func New(addr string) *Metrics {
 	met := &Metrics{
 		Addr:    addr,
@@ -101,14 +103,19 @@ func (m *Metrics) OnStartup() error {
 
 	m.ln = ln
 	m.lnSetup = true
-	ListenAddr = m.ln.Addr().String() // For tests
 
 	m.mux = http.NewServeMux()
 	m.mux.Handle("/metrics", promhttp.HandlerFor(m.Reg, promhttp.HandlerOpts{}))
-	m.srv = &http.Server{Handler: m.mux}
+
+	// creating some helper variables to avoid data races on m.srv and m.ln
+	server := &http.Server{Handler: m.mux}
+	m.srv = server
+
 	go func() {
-		m.srv.Serve(m.ln)
+		server.Serve(ln)
 	}()
+
+	ListenAddr = ln.Addr().String() // For tests.
 	return nil
 }
 
@@ -117,7 +124,7 @@ func (m *Metrics) OnRestart() error {
 	if !m.lnSetup {
 		return nil
 	}
-	uniqAddr.Unset(m.Addr)
+	u.Unset(m.Addr)
 	return m.stopServer()
 }
 

plugin/metrics/registry.go

@@ -0,0 +1,28 @@
+package metrics
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type reg struct {
+	sync.RWMutex
+	r map[string]*prometheus.Registry
+}
+
+func newReg() *reg { return &reg{r: make(map[string]*prometheus.Registry)} }
+
+// update sets the registry if not already there and returns the input. Or it returns
+// a previous set value.
+func (r *reg) getOrSet(addr string, pr *prometheus.Registry) *prometheus.Registry {
+	r.Lock()
+	defer r.Unlock()
+
+	if v, ok := r.r[addr]; ok {
+		return v
+	}
+
+	r.r[addr] = pr
+	return pr
+}

plugin/metrics/setup.go

@@ -16,7 +16,8 @@ import (
 
 var (
 	log      = clog.NewWithPlugin("prometheus")
-	uniqAddr = uniq.New()
+	u        = uniq.New()
+	registry = newReg()
 )
 
 func init() {
@@ -31,12 +32,13 @@ func setup(c *caddy.Controller) error {
 	if err != nil {
 		return plugin.Error("prometheus", err)
 	}
+	m.Reg = registry.getOrSet(m.Addr, m.Reg)
 
-	c.OnStartup(func() error { m.Reg = uniqAddr.Set(m.Addr, m.OnStartup, m).(*Metrics).Reg; return nil })
-	c.OnRestartFailed(func() error { m.Reg = uniqAddr.Set(m.Addr, m.OnStartup, m).(*Metrics).Reg; return nil })
+	c.OnStartup(func() error { m.Reg = registry.getOrSet(m.Addr, m.Reg); u.Set(m.Addr, m.OnStartup); return nil })
+	c.OnRestartFailed(func() error { m.Reg = registry.getOrSet(m.Addr, m.Reg); u.Set(m.Addr, m.OnStartup); return nil })
 
-	c.OnStartup(func() error { return uniqAddr.ForEach() })
-	c.OnRestartFailed(func() error { return uniqAddr.ForEach() })
+	c.OnStartup(func() error { return u.ForEach() })
+	c.OnRestartFailed(func() error { return u.ForEach() })
 
 	c.OnStartup(func() error {
 		conf := dnsserver.GetConfig(c)
@@ -75,7 +77,7 @@ func setup(c *caddy.Controller) error {
 }
 
 func parse(c *caddy.Controller) (*Metrics, error) {
-	var met = New(defaultAddr)
+	met := New(defaultAddr)
 
 	i := 0
 	for c.Next() {

plugin/pkg/uniq/uniq.go

@@ -10,27 +10,22 @@ type U struct {
 type item struct {
 	state int          // either todo or done
 	f     func() error // function to be executed.
-	obj   interface{}  // any object to return when needed
 }
 
 // New returns a new initialized U.
 func New() U { return U{u: make(map[string]item)} }
 
-// Set sets function f in U under key. If the key already exists
-// it is not overwritten.
-func (u U) Set(key string, f func() error, o interface{}) interface{} {
-	if item, ok := u.u[key]; ok {
-		return item.obj
+// Set sets function f in U under key. If the key already exists it is not overwritten.
+func (u U) Set(key string, f func() error) {
+	if _, ok := u.u[key]; ok {
+		return
 	}
-	u.u[key] = item{todo, f, o}
-	return o
+	u.u[key] = item{todo, f}
 }
 
 // Unset removes the key.
 func (u U) Unset(key string) {
-	if _, ok := u.u[key]; ok {
-		delete(u.u, key)
-	}
+	delete(u.u, key)
 }
 
 // ForEach iterates for u executes f for each element that is 'todo' and sets it to 'done'.

plugin/pkg/uniq/uniq_test.go

@@ -4,7 +4,7 @@ import "testing"
 
 func TestForEach(t *testing.T) {
 	u, i := New(), 0
-	u.Set("test", func() error { i++; return nil }, nil)
+	u.Set("test", func() error { i++; return nil })
 
 	u.ForEach()
 	if i != 1 {

plugin/ready/setup.go

@@ -24,7 +24,7 @@ func setup(c *caddy.Controller) error {
 
 	rd := &ready{Addr: addr}
 
-	uniqAddr.Set(addr, rd.onStartup, rd)
+	uniqAddr.Set(addr, rd.onStartup)
 
 	c.OncePerServerBlock(func() error {
 		c.OnStartup(func() error {