Allow loading the configuration from etcd.

Lading data from a configuration source now lives in its own
package. The value of the --config flag may be in the form of
"kind:path", where kind is one of "file" or "etcd". If no kind is
provided, a local file is assumed.

Author: ryszard

go/cmd/doorman/doorman_server.go

@@ -17,13 +17,10 @@ package main
 import (
 	"flag"
 	"fmt"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
-	"os/signal"
 	"strings"
-	"syscall"
 	"time"
 
 	"golang.org/x/net/context"
@@ -34,6 +31,7 @@ import (
 	rpc "google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 
+	"github.com/youtube/doorman/go/configuration"
 	"github.com/youtube/doorman/go/connection"
 	"github.com/youtube/doorman/go/server/doorman"
 	"github.com/youtube/doorman/go/server/election"
@@ -55,7 +53,7 @@ var (
 	serverRole = flag.String("server_role", "root", "Role of this server in the server tree")
 	parent     = flag.String("parent", "", "Address of the parent server which this server connects to")
 	hostname   = flag.String("hostname", "", "Use this as the hostname (if empty, use whatever the kernel reports")
-	config     = flag.String("config", "", "file to load the config from (text protobufs)")
+	config     = flag.String("config", "", "source to load the config from (text protobufs)")
 
 	rpcDialTimeout = flag.Duration("doorman_rpc_dial_timeout", 5*time.Second, "timeout to use for connecting to the doorman server")
 
@@ -142,10 +140,12 @@ func main() {
 	if *config == "" {
 		log.Exit("--config cannot be empty")
 	}
-
-	var masterElection election.Election
+	var (
+		etcdEndpointsSlice = strings.Split(*etcdEndpoints, ",")
+		masterElection     election.Election
+	)
 	if *masterElectionLock != "" {
-		etcdEndpointsSlice := strings.Split(*etcdEndpoints, ",")
+
 		if len(etcdEndpointsSlice) == 1 && etcdEndpointsSlice[0] == "" {
 			log.Exit("-etcd_endpoints cannot be empty if -master_election_lock is provided")
 		}
@@ -180,42 +180,39 @@ func main() {
 		log.Exit("-config cannot be empty")
 	}
 
-	data, err := ioutil.ReadFile(*config)
-	if err != nil {
-		log.Exitf("cannot read config file: %v", err)
-	}
-
-	cfg := new(pb.ResourceRepository)
-	if err := proto.UnmarshalText(string(data), cfg); err != nil {
-		log.Exitf("cannot load config: %v", err)
-	}
-
-	if err := dm.LoadConfig(context.Background(), cfg, map[string]*time.Time{}); err != nil {
-		log.Exitf("dm.LoadConfig: %v", err)
+	var cfg configuration.Source
+	kind, path := configuration.ParseSource(*config)
+	switch {
+	case kind == "file":
+		cfg = configuration.LocalFile(path)
+	case kind == "etcd":
+		if len(etcdEndpointsSlice) == 1 && etcdEndpointsSlice[0] == "" {
+			log.Exit("-etcd_endpoints cannot be empty if a config source etcd is provided")
+		}
+		cfg = configuration.Etcd(path, etcdEndpointsSlice)
+	default:
+		panic("unreachable")
 	}
 
-	c := make(chan os.Signal, 1)
-	signal.Notify(c, syscall.SIGHUP)
-
+	// Try to load the background. If there's a problem with loading
+	// the server for the first time, the server will keep running,
+	// but will not serve traffic.
 	go func() {
-		for range c {
-			log.Infof("Received SIGHUP, attempting to reload configuration from %v.", *config)
-			data, err := ioutil.ReadFile(*config)
+		for {
+			data, err := cfg(context.Background())
 			if err != nil {
-				log.Errorf("cannot read config file: %v", err)
+				log.Errorf("cannot load config data: %v", err)
 				continue
 			}
-
 			cfg := new(pb.ResourceRepository)
 			if err := proto.UnmarshalText(string(data), cfg); err != nil {
-				log.Errorf("cannot load config: %v", err)
+				log.Errorf("cannot unmarshal config data: %q", data)
 				continue
 			}
 
 			if err := dm.LoadConfig(context.Background(), cfg, map[string]*time.Time{}); err != nil {
-				log.Errorf("dm.LoadConfig: %v", err)
+				log.Errorf("cannot load config: %v", err)
 			}
-			log.Infof("Reloaded config from %v", *config)
 		}
 	}()
 

go/configuration/configuration.go

@@ -0,0 +1,122 @@
+// package configuration defines different sources of text-based
+// configuration.
+package configuration
+
+import (
+	"io/ioutil"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/coreos/etcd/client"
+	log "github.com/golang/glog"
+	"github.com/youtube/doorman/go/timeutil"
+	"golang.org/x/net/context"
+)
+
+// Source is a source for configuration. Calling it will block until a
+// new version of the config is available.
+type Source func(context.Context) (data []byte, err error)
+
+type pair struct {
+	data []byte
+	err  error
+}
+
+// LocalFiles is a configuration stored in a file in the local
+// filesystem. The file will be reloaded if the process receives a
+// SIGHUP.
+func LocalFile(path string) Source {
+	updates := make(chan pair, 1)
+
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, syscall.SIGHUP)
+	c <- syscall.SIGHUP
+	go func() {
+		for range c {
+			log.Infof("config: loading configuration from %v.", path)
+			data, err := ioutil.ReadFile(path)
+			updates <- pair{data: data, err: err}
+		}
+	}()
+	return func(ctx context.Context) (data []byte, err error) {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case p := <-updates:
+			return p.data, p.err
+		}
+	}
+}
+
+// Etcd is a configuration stored in etcd. It will be reloaded as soon
+// as it changes.
+func Etcd(path string, endpoints []string) Source {
+
+	updates := make(chan pair, 1)
+	req := make(chan context.Context)
+
+	go func() {
+		var c client.Client
+		for i := 0; true; i++ {
+			var err error
+			c, err = client.New(client.Config{Endpoints: endpoints})
+			if err != nil {
+				log.Errorf("configuration: cannot connect to etcd: %v", err)
+				updates <- pair{err: err}
+				time.Sleep(timeutil.Backoff(1*time.Second, 60*time.Second, i))
+				continue
+			}
+			break
+		}
+		log.V(2).Infof("configuration: connected to etcd")
+		kapi := client.NewKeysAPI(c)
+
+		r, err := kapi.Get(<-req, path, nil)
+		if err != nil {
+			updates <- pair{err: err}
+		} else {
+			updates <- pair{data: []byte(r.Node.Value)}
+		}
+
+		w := kapi.Watcher(path, nil)
+
+		for i := 0; true; i++ {
+			ctx := <-req
+			r, err := w.Next(ctx)
+			if err != nil {
+				updates <- pair{err: err}
+				time.Sleep(timeutil.Backoff(1*time.Second, 60*time.Second, i))
+				continue
+			}
+			updates <- pair{data: []byte(r.Node.Value)}
+		}
+
+	}()
+
+	return func(ctx context.Context) (data []byte, err error) {
+		req <- ctx
+		p := <-updates
+		return p.data, p.err
+	}
+
+}
+
+// ParseSource parses text and returns the kind of configuration
+// source and the desired path.
+func ParseSource(text string) (kind string, path string) {
+	parts := strings.SplitN(text, ":", 2)
+	if len(parts) == 1 {
+		return "file", text
+	}
+	switch parts[0] {
+	case "etcd":
+		return "etcd", parts[1]
+	case "file":
+		return "file", parts[1]
+	}
+
+	return "file", text
+}

go/configuration/configuration_test.go

@@ -0,0 +1,38 @@
+package configuration
+
+import "testing"
+
+func TestParseSource(t *testing.T) {
+	for _, c := range []struct {
+		text, path, kind string
+	}{
+		{
+			text: "config.prototext",
+			kind: "file",
+			path: "config.prototext",
+		},
+		{
+			text: "file:config.prototext",
+			kind: "file",
+			path: "config.prototext",
+		},
+		{
+			text: "colons:in:name:good:idea:config.prototext",
+			kind: "file",
+			path: "colons:in:name:good:idea:config.prototext",
+		},
+		{
+			text: "etcd:/config.prototext",
+			kind: "etcd",
+			path: "/config.prototext",
+		},
+	} {
+		kind, path := ParseSource(c.text)
+		if kind != c.kind {
+			t.Errorf("parseSource(%q): kind = %q; want %q", c.text, kind, c.kind)
+		}
+		if path != c.path {
+			t.Errorf("parseSource(%q): kind = %q; want %q", c.text, path, c.path)
+		}
+	}
+}