Embedded etcd

This monster patch embeds the etcd server. It took a good deal of
iterative work to tweak small details, and survived a rewrite from the
initial etcd v2 API implementation to the beta version of v3.

It has a notable race, and is missing some features, but it is ready for
git master and external developer consumption.

DOCUMENTATION.md

@@ -201,7 +201,7 @@ sudo mkdir -p /etc/systemd/system/mgmt.service.d/
 
 cat > /etc/systemd/system/mgmt.service.d/env.conf <<EOF
 # Environment variables:
-MGMT_SEED_ENDPOINT=http://127.0.0.1:2379
+MGMT_SEEDS=http://127.0.0.1:2379
 MGMT_CONVERGED_TIMEOUT=-1
 MGMT_MAX_RUNTIME=0
 

README.md

@@ -17,7 +17,6 @@ If you have a well phrased question that might benefit others, consider asking i
 ## Quick start:
 * Either get the golang dependencies on your own, or run `make deps` if you're comfortable with how we install them.
 * Run `make build` to get a freshly built `mgmt` binary.
-* Run `cd $(mktemp --tmpdir -d tmp.XXX) && etcd` to get etcd running. The `mgmt` software will do this automatically for you in the future.
 * Run `time ./mgmt run --file examples/graph0.yaml --converged-timeout=1` to try out a very simple example!
 * To run continuously in the default mode of operation, omit the `--converged-timeout` option.
 * Have fun hacking on our future technology!

config.go

@@ -85,7 +85,7 @@ func ParseConfigFromFile(filename string) *GraphConfig {
 
 // NewGraphFromConfig returns a new graph from existing input, such as from the
 // existing graph, and a GraphConfig struct.
-func (g *Graph) NewGraphFromConfig(config *GraphConfig, etcdO *EtcdWObject, hostname string, noop bool) (*Graph, error) {
+func (g *Graph) NewGraphFromConfig(config *GraphConfig, embdEtcd *EmbdEtcd, hostname string, noop bool) (*Graph, error) {
 
 	var graph *Graph // new graph to return
 	if g == nil {    // FIXME: how can we check for an empty graph?
@@ -102,7 +102,7 @@ func (g *Graph) NewGraphFromConfig(config *GraphConfig, etcdO *EtcdWObject, host
 	graph.SetName(config.Graph) // set graph name
 
 	var keep []*Vertex  // list of vertex which are the same in new graph
-
+	var resources []Res // list of resources to export
 	// use reflection to avoid duplicating code... better options welcome!
 	value := reflect.Indirect(reflect.ValueOf(config.Resources))
 	vtype := value.Type()
@@ -118,91 +118,106 @@ func (g *Graph) NewGraphFromConfig(config *GraphConfig, etcdO *EtcdWObject, host
 		}
 		for j := 0; j < slice.Len(); j++ { // loop through resources of same kind
 			x := slice.Index(j).Interface()
-			obj, ok := x.(Res) // convert to Res type
+			res, ok := x.(Res) // convert to Res type
 			if !ok {
 				return nil, fmt.Errorf("Error: Config: Can't convert: %v of type: %T to Res.", x, x)
 			}
 			if noop {
-				obj.Meta().Noop = noop
+				res.Meta().Noop = noop
 			}
 			if _, exists := lookup[kind]; !exists {
 				lookup[kind] = make(map[string]*Vertex)
 			}
 			// XXX: should we export based on a @@ prefix, or a metaparam
 			// like exported => true || exported => (host pattern)||(other pattern?)
-			if !strings.HasPrefix(obj.GetName(), "@@") { // exported resource
+			if !strings.HasPrefix(res.GetName(), "@@") { // not exported resource
 				// XXX: we don't have a way of knowing if any of the
 				// metaparams are undefined, and as a result to set the
 				// defaults that we want! I hate the go yaml parser!!!
-				v := graph.GetVertexMatch(obj)
+				v := graph.GetVertexMatch(res)
 				if v == nil { // no match found
-					obj.Init()
-					v = NewVertex(obj)
+					res.Init()
+					v = NewVertex(res)
 					graph.AddVertex(v) // call standalone in case not part of an edge
 				}
-				lookup[kind][obj.GetName()] = v // used for constructing edges
+				lookup[kind][res.GetName()] = v // used for constructing edges
 				keep = append(keep, v)          // append
 
 			} else if !noop { // do not export any resources if noop
-				// XXX: do this in a different function...
-				// add to etcd storage...
-				obj.SetName(obj.GetName()[2:]) //slice off @@
-
-				data, err := ResToB64(obj)
-				if err != nil {
-					return nil, fmt.Errorf("Config: Could not encode %v resource: %v, error: %v", kind, obj.GetName(), err)
+				// store for addition to etcd storage...
+				res.SetName(res.GetName()[2:]) //slice off @@
+				res.setKind(kind)              // cheap init
+				resources = append(resources, res)
+			}
+		}
+	}
+	// store in etcd
+	if err := EtcdSetResources(embdEtcd, hostname, resources); err != nil {
+		return nil, fmt.Errorf("Config: Could not export resources: %v", err)
 	}
 
-				if !etcdO.EtcdPut(hostname, obj.GetName(), kind, data) {
-					return nil, fmt.Errorf("Config: Could not export %v resource: %v", kind, obj.GetName())
-				}
-			}
-		}
-	}
-
-	// lookup from etcd graph
-	// do all the graph look ups in one single step, so that if the etcd
-	// database changes, we don't have a partial state of affairs...
-	nodes, ok := etcdO.EtcdGet()
-	if ok {
+	// lookup from etcd
+	var hostnameFilter []string // empty to get from everyone
+	kindFilter := []string{}
+	for _, t := range config.Collector {
+		// XXX: should we just drop these everywhere and have the kind strings be all lowercase?
+		kind := FirstToUpper(t.Kind)
+		kindFilter = append(kindFilter, kind)
+	}
+	// do all the graph look ups in one single step, so that if the etcd
+	// database changes, we don't have a partial state of affairs...
+	if len(kindFilter) > 0 { // if kindFilter is empty, don't need to do lookups!
+		var err error
+		resources, err = EtcdGetResources(embdEtcd, hostnameFilter, kindFilter)
+		if err != nil {
+			return nil, fmt.Errorf("Config: Could not collect resources: %v", err)
+		}
+	}
+	for _, res := range resources {
+		matched := false
+		// see if we find a collect pattern that matches
 		for _, t := range config.Collector {
 			// XXX: should we just drop these everywhere and have the kind strings be all lowercase?
 			kind := FirstToUpper(t.Kind)
-
 			// use t.Kind and optionally t.Pattern to collect from etcd storage
 			log.Printf("Collect: %v; Pattern: %v", kind, t.Pattern)
-			for _, str := range etcdO.EtcdGetProcess(nodes, kind) {
-				obj, err := B64ToRes(str)
-				if err != nil {
-					log.Printf("B64ToRes failed to decode: %v", err)
-					log.Printf("Collect: %v: not collected!", kind)
+
+			// XXX: expand to more complex pattern matching here...
+			if res.Kind() != kind {
 				continue
 			}
 
+			if matched {
+				// we've already matched this resource, should we match again?
+				log.Printf("Config: Warning: Matching %v[%v] again!", kind, res.GetName())
+			}
+			matched = true
+
 			// collect resources but add the noop metaparam
 			if noop {
-					obj.Meta().Noop = noop
+				res.Meta().Noop = noop
 			}
 
 			if t.Pattern != "" { // XXX: simplistic for now
-					obj.CollectPattern(t.Pattern) // obj.Dirname = t.Pattern
+				res.CollectPattern(t.Pattern) // res.Dirname = t.Pattern
 			}
 
-				log.Printf("Collect: %v[%v]: collected!", kind, obj.GetName())
+			log.Printf("Collect: %v[%v]: collected!", kind, res.GetName())
 
 			// XXX: similar to other resource add code:
 			if _, exists := lookup[kind]; !exists {
 				lookup[kind] = make(map[string]*Vertex)
 			}
-				v := graph.GetVertexMatch(obj)
+			v := graph.GetVertexMatch(res)
 			if v == nil { // no match found
-					obj.Init() // initialize go channels or things won't work!!!
-					v = NewVertex(obj)
+				res.Init() // initialize go channels or things won't work!!!
+				v = NewVertex(res)
 				graph.AddVertex(v) // call standalone in case not part of an edge
 			}
-				lookup[kind][obj.GetName()] = v // used for constructing edges
+			lookup[kind][res.GetName()] = v // used for constructing edges
 			keep = append(keep, v)          // append
-			}
+
+			//break // let's see if another resource even matches
 		}
 	}
 

examples/etcd1d.yaml

@@ -0,0 +1,18 @@
+---
+graph: mygraph
+resources:
+  file:
+  - name: file1d
+    path: "/tmp/mgmtD/f1d"
+    content: |
+      i am f1
+    state: exists
+  - name: "@@file2d"
+    path: "/tmp/mgmtD/f2d"
+    content: |
+      i am f2, exported from host D
+    state: exists
+collect:
+- kind: file
+  pattern: "/tmp/mgmtD/"
+edges: []

main.go

@@ -19,6 +19,7 @@ package main
 
 import (
 	"github.com/codegangsta/cli"
+	etcdtypes "github.com/coreos/etcd/pkg/types"
 	"github.com/coreos/pkg/capnslog"
 	"log"
 	"os"
@@ -35,7 +36,9 @@ var (
 )
 
 const (
-	DEBUG = false
+	DEBUG   = false // add additional log messages
+	TRACE   = false // add execution flow log messages
+	VERBOSE = false // add extra log message output
 )
 
 // signal handler
@@ -59,16 +62,51 @@ func waitForSignal(exit chan bool) {
 
 func run(c *cli.Context) error {
 	var start = time.Now().UnixNano()
-	var wg sync.WaitGroup
-	exit := make(chan bool) // exit signal
 	log.Printf("This is: %v, version: %v", program, version)
 	log.Printf("Main: Start: %v", start)
-	var G, fullGraph *Graph
 
-	if c.IsSet("file") && c.IsSet("puppet") {
-		log.Println("the --file and --puppet parameters cannot be used together")
+	hostname := c.String("hostname")
+	if hostname == "" {
+		hostname, _ = os.Hostname()
+	}
+	noop := c.Bool("noop")
+
+	seeds, err := etcdtypes.NewURLs(
+		FlattenListWithSplit(c.StringSlice("seeds"), []string{",", ";", " "}),
+	)
+	if err != nil && len(c.StringSlice("seeds")) > 0 {
+		log.Printf("Main: Error: seeds didn't parse correctly!")
 		return cli.NewExitError("", 1)
 	}
+	clientURLs, err := etcdtypes.NewURLs(
+		FlattenListWithSplit(c.StringSlice("client-urls"), []string{",", ";", " "}),
+	)
+	if err != nil && len(c.StringSlice("client-urls")) > 0 {
+		log.Printf("Main: Error: clientURLs didn't parse correctly!")
+		return cli.NewExitError("", 1)
+	}
+	serverURLs, err := etcdtypes.NewURLs(
+		FlattenListWithSplit(c.StringSlice("server-urls"), []string{",", ";", " "}),
+	)
+	if err != nil && len(c.StringSlice("server-urls")) > 0 {
+		log.Printf("Main: Error: serverURLs didn't parse correctly!")
+		return cli.NewExitError("", 1)
+	}
+
+	idealClusterSize := uint16(c.Int("ideal-cluster-size"))
+	if idealClusterSize < 1 {
+		log.Printf("Main: Error: idealClusterSize should be at least one!")
+		return cli.NewExitError("", 1)
+	}
+
+	if c.IsSet("file") && c.IsSet("puppet") {
+		log.Println("Main: Error: the --file and --puppet parameters cannot be used together!")
+		return cli.NewExitError("", 1)
+	}
+
+	var wg sync.WaitGroup
+	exit := make(chan bool) // exit signal
+	var G, fullGraph *Graph
 
 	// exit after `max-runtime` seconds for no reason at all...
 	if i := c.Int("max-runtime"); i > 0 {
@@ -88,28 +126,26 @@ func run(c *cli.Context) error {
 	)
 	go converger.Loop(true) // main loop for converger, true to start paused
 
-	// initial etcd peer endpoint
-	seed := c.String("seed")
-	if seed == "" {
-		// XXX: start up etcd server, others will join me!
-		seed = "http://127.0.0.1:2379" // thus we use the local server!
+	// embedded etcd
+	if len(seeds) == 0 {
+		log.Printf("Main: Seeds: No seeds specified!")
+	} else {
+		log.Printf("Main: Seeds(%v): %v", len(seeds), seeds)
 	}
-	// then, connect to `seed` as a client
-
-	// FIXME: validate seed, or wait for it to fail in etcd init?
-
-	// etcd
-	etcdO := &EtcdWObject{
-		seed:      seed,
-		converger: converger,
+	EmbdEtcd := NewEmbdEtcd(
+		hostname,
+		seeds,
+		clientURLs,
+		serverURLs,
+		c.Bool("no-server"),
+		idealClusterSize,
+		converger,
+	)
+	if err := EmbdEtcd.Startup(); err != nil { // startup (returns when etcd main loop is running)
+		log.Printf("Main: Etcd: Startup failed: %v", err)
+		exit <- true
 	}
 
-	hostname := c.String("hostname")
-	if hostname == "" {
-		hostname, _ = os.Hostname() // etcd watch key // XXX: this is not the correct key name this is the set key name... WOOPS
-	}
-	noop := c.Bool("noop")
-
 	exitchan := make(chan Event) // exit event
 	go func() {
 		startchan := make(chan struct{}) // start signal
@@ -124,26 +160,23 @@ func run(c *cli.Context) error {
 			puppetchan = time.Tick(time.Duration(interval) * time.Second)
 		}
 		log.Println("Etcd: Starting...")
-		etcdchan := etcdO.EtcdWatch()
+		etcdchan := EtcdWatch(EmbdEtcd)
 		first := true // first loop or not
 		for {
 			log.Println("Main: Waiting...")
 			select {
-			case _ = <-startchan: // kick the loop once at start
+			case <-startchan: // kick the loop once at start
 				// pass
-			case msg := <-etcdchan:
-				switch msg {
-				// some types of messages we ignore...
-				case etcdFoo, etcdBar:
+
+			case b := <-etcdchan:
+				if !b { // ignore the message
 					continue
-				// while others passthrough and cause a compile!
-				case etcdStart, etcdEvent:
-					// pass
-				default:
-					log.Fatal("Etcd: Unhandled message: ", msg)
 				}
-			case _ = <-puppetchan:
+				// everything else passes through to cause a compile!
+
+			case <-puppetchan:
 				// nothing, just go on
+
 			case msg := <-configchan:
 				if c.Bool("no-watch") || !msg {
 					continue // not ready to read config
@@ -174,7 +207,7 @@ func run(c *cli.Context) error {
 
 			// build graph from yaml file on events (eg: from etcd)
 			// we need the vertices to be paused to work on them
-			if newFullgraph, err := fullGraph.NewGraphFromConfig(config, etcdO, hostname, noop); err == nil { // keep references to all original elements
+			if newFullgraph, err := fullGraph.NewGraphFromConfig(config, EmbdEtcd, hostname, noop); err == nil { // keep references to all original elements
 				fullGraph = newFullgraph
 			} else {
 				log.Printf("Config: Error making new graph from config: %v", err)
@@ -215,11 +248,19 @@ func run(c *cli.Context) error {
 
 	waitForSignal(exit) // pass in exit channel to watch
 
+	log.Println("Destroy...")
+
 	G.Exit() // tell all the children to exit
 
 	// tell inner main loop to exit
 	resp := NewResp()
-	exitchan <- Event{eventExit, resp, "", false}
+	go func() { exitchan <- Event{eventExit, resp, "", false} }()
+
+	// cleanup etcd main loop last so it can process everything first
+	if err := EmbdEtcd.Destroy(); err != nil { // shutdown and cleanup etcd
+		log.Printf("Etcd exited poorly with: %v", err)
+	}
+
 	resp.ACKWait() // let inner main loop finish cleanly just in case
 
 	if DEBUG {
@@ -243,7 +284,11 @@ func main() {
 
 	// un-hijack from capnslog...
 	log.SetOutput(os.Stderr)
+	if VERBOSE {
 		capnslog.SetFormatter(capnslog.NewLogFormatter(os.Stderr, "(etcd) ", flags))
+	} else {
+		capnslog.SetFormatter(capnslog.NewNilFormatter())
+	}
 
 	// test for sanity
 	if program == "" || version == "" {
@@ -294,11 +339,35 @@ func main() {
 					Usage: "hostname to use",
 				},
 				// if empty, it will startup a new server
-				cli.StringFlag{
-					Name:   "seed, s",
-					Value:  "",
-					Usage:  "default etc peer endpoint",
-					EnvVar: "MGMT_SEED_ENDPOINT",
+				cli.StringSliceFlag{
+					Name:   "seeds, s",
+					Value:  &cli.StringSlice{}, // empty slice
+					Usage:  "default etc client endpoint",
+					EnvVar: "MGMT_SEEDS",
+				},
+				// port 2379 and 4001 are common
+				cli.StringSliceFlag{
+					Name:   "client-urls",
+					Value:  &cli.StringSlice{},
+					Usage:  "list of URLs to listen on for client traffic",
+					EnvVar: "MGMT_CLIENT_URLS",
+				},
+				// port 2380 and 7001 are common
+				cli.StringSliceFlag{
+					Name:   "server-urls, peer-urls",
+					Value:  &cli.StringSlice{},
+					Usage:  "list of URLs to listen on for server (peer) traffic",
+					EnvVar: "MGMT_SERVER_URLS",
+				},
+				cli.BoolFlag{
+					Name:  "no-server",
+					Usage: "do not let other servers peer with me",
+				},
+				cli.IntFlag{
+					Name:   "ideal-cluster-size",
+					Value:  defaultIdealClusterSize,
+					Usage:  "ideal number of server peers in cluster, only read by initial server",
+					EnvVar: "MGMT_IDEAL_CLUSTER_SIZE",
 				},
 				cli.IntFlag{
 					Name:   "converged-timeout, t",

misc.go

@@ -40,6 +40,15 @@ func StrInList(needle string, haystack []string) bool {
 	return false
 }
 
+func Uint64KeyFromStrInMap(needle string, haystack map[uint64]string) (uint64, bool) {
+	for k, v := range haystack {
+		if v == needle {
+			return k, true
+		}
+	}
+	return 0, false
+}
+
 // remove any duplicate values in the list
 // possibly sub-optimal, O(n^2)? implementation
 func StrRemoveDuplicatesInList(list []string) []string {
@@ -87,7 +96,7 @@ func ReverseStringList(in []string) []string {
 
 // return the sorted list of string keys in a map with string keys
 // NOTE: i thought it would be nice for this to use: map[string]interface{} but
-// it turns out that's not allowed. I know we don't have generics, but common!
+// it turns out that's not allowed. I know we don't have generics, but come on!
 func StrMapKeys(m map[string]string) []string {
 	result := []string{}
 	for k, _ := range m {
@@ -97,6 +106,15 @@ func StrMapKeys(m map[string]string) []string {
 	return result
 }
 
+func StrMapKeysUint64(m map[string]uint64) []string {
+	result := []string{}
+	for k, _ := range m {
+		result = append(result, k)
+	}
+	sort.Strings(result) // deterministic order
+	return result
+}
+
 // return the sorted list of bool values in a map with string values
 func BoolMapValues(m map[string]bool) []bool {
 	result := []bool{}
@@ -117,6 +135,16 @@ func StrMapValues(m map[string]string) []string {
 	return result
 }
 
+// return the sorted list of string values in a map with string values
+func StrMapValuesUint64(m map[uint64]string) []string {
+	result := []string{}
+	for _, v := range m {
+		result = append(result, v)
+	}
+	sort.Strings(result) // deterministic order
+	return result
+}
+
 // return true if everyone is true
 func BoolMapTrue(l []bool) bool {
 	for _, b := range l {

resources.go

@@ -68,6 +68,7 @@ type MetaParams struct {
 type Base interface {
 	GetName() string // can't be named "Name()" because of struct field
 	SetName(string)
+	setKind(string)
 	Kind() string
 	Meta() *MetaParams
 	AssociateData(Converger)
@@ -162,7 +163,12 @@ func (obj *BaseRes) SetName(name string) {
 	obj.Name = name
 }
 
-// return the kind of resource this is
+// setKind sets the kind. This is used internally for exported resources.
+func (obj *BaseRes) setKind(kind string) {
+	obj.kind = kind
+}
+
+// Kind returns the kind of resource this is
 func (obj *BaseRes) Kind() string {
 	return obj.kind
 }