Merge pull request #41 from mazarmi/feature-etcdmgr-add-remove

Dynamic etcd cluster resizing added.

cmd/castled/castled.go

@@ -17,6 +17,7 @@ import (
 	"github.com/quantum/castle/pkg/cephmgr"
 	"github.com/quantum/castle/pkg/cephmgr/cephd"
 	"github.com/quantum/castle/pkg/clusterd"
+	"github.com/quantum/castle/pkg/etcdmgr"
 	"github.com/quantum/castle/pkg/util/flags"
 	"github.com/quantum/castle/pkg/util/proc"
 )
@@ -87,7 +88,10 @@ func joinCluster(cmd *cobra.Command, args []string) error {
 		cfg.discoveryURL = discURL
 	}
 
-	services := []*clusterd.ClusterService{cephmgr.NewCephService(cephd.New(), cfg.devices, cfg.forceFormat, cfg.location)}
+	services := []*clusterd.ClusterService{
+		cephmgr.NewCephService(cephd.New(), cfg.devices, cfg.forceFormat, cfg.location),
+		etcdmgr.NewEtcdMgrService(cfg.discoveryURL),
+	}
 	procMan := &proc.ProcManager{}
 	defer func() {
 		procMan.Shutdown()

pkg/cephmgr/cephleader_test.go

@@ -1,7 +1,6 @@
 package cephmgr
 
 import (
-	"log"
 	"path"
 	"strings"
 	"testing"
@@ -13,6 +12,7 @@ import (
 	testceph "github.com/quantum/castle/pkg/cephmgr/client/test"
 	"github.com/quantum/castle/pkg/clusterd"
 	"github.com/quantum/castle/pkg/clusterd/inventory"
+	clusterdtest "github.com/quantum/castle/pkg/clusterd/test"
 	"github.com/quantum/castle/pkg/util"
 	"github.com/stretchr/testify/assert"
 )
@@ -52,7 +52,7 @@ func TestCephLeaders(t *testing.T) {
 	leader.Events() <- refresh
 
 	// wait for the event queue to be empty
-	waitForEvents(leader)
+	clusterdtest.WaitForEvents(leader)
 
 	assert.True(t, etcdClient.GetChildDirs("/castle/services/ceph/osd/desired").Equals(util.CreateSet([]string{"a"})))
 	assert.Equal(t, "mon0", etcdClient.GetValue("/castle/services/ceph/monitor/desired/a/id"))
@@ -65,45 +65,13 @@ func TestCephLeaders(t *testing.T) {
 	leader.Events() <- addNode
 
 	// wait for the event queue to be empty
-	waitForEvents(leader)
+	clusterdtest.WaitForEvents(leader)
 
 	assert.True(t, etcdClient.GetChildDirs("/castle/services/ceph/osd/desired").Equals(util.CreateSet([]string{"a", "b"})))
 	assert.Equal(t, "myfsid", etcdClient.GetValue("/castle/services/ceph/fsid"))
 	assert.Equal(t, "mykey", etcdClient.GetValue("/castle/services/ceph/_secrets/admin"))
 }
 
-func waitForEvents(leader *cephLeader) {
-	// add a placeholder event to the queue. When it is dequeued we know the rest of the events have completed.
-	e := newNonEvent()
-	leader.Events() <- e
-
-	// wait for the Name() method to be called on the nonevent, which means it was dequeued
-	log.Printf("waiting for event queue to empty")
-	<-e.signaled
-	log.Printf("event queue is empty")
-}
-
-// Empty event for testing
-type nonEvent struct {
-	signaled   chan bool
-	nameCalled bool
-}
-
-func newNonEvent() *nonEvent {
-	return &nonEvent{signaled: make(chan bool)}
-}
-
-func (e *nonEvent) Name() string {
-	if !e.nameCalled {
-		e.nameCalled = true
-		e.signaled <- true
-	}
-	return "nonevent"
-}
-func (e *nonEvent) Context() *clusterd.Context {
-	return nil
-}
-
 func TestMoveUnhealthyMonitor(t *testing.T) {
 	factory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	leader := newCephLeader(factory)

pkg/cephmgr/monagent.go

@@ -33,7 +33,6 @@ func (a *monAgent) Initialize(context *clusterd.Context) error {
 }
 
 func (a *monAgent) ConfigureLocalService(context *clusterd.Context) error {
-
 	// check if the monitor is in the desired state for this node
 	key := path.Join(cephKey, monitorAgentName, desiredKey, context.NodeID)
 	monDesired, err := util.EtcdDirExists(context.EtcdClient, key)

pkg/cephmgr/osdagent.go

@@ -68,7 +68,6 @@ func (a *osdAgent) Initialize(context *clusterd.Context) error {
 }
 
 func (a *osdAgent) ConfigureLocalService(context *clusterd.Context) error {
-
 	// check if the osd is in the desired state for this node
 	key := path.Join(cephKey, osdAgentName, desiredKey, context.NodeID, "ready")
 	osdDesired, err := context.EtcdClient.Get(ctx.Background(), key, nil)

pkg/clusterd/agent.go

@@ -10,9 +10,8 @@ import (
 	ctx "golang.org/x/net/context"
 )
 
-// Listen for notifications from the orchestrator that components should be installed on the aSSgent
+// Listen for notifications from the orchestrator that components should be installed on the agent
 func watchForAgentServiceConfig(context *Context) {
-
 	restoreDesiredState(context)
 
 	// Get the initial status of the configuration
@@ -77,7 +76,6 @@ func restoreDesiredState(context *Context) {
 
 func ConfigureServiceIfTriggered(context *Context, agent ServiceAgent) bool {
 	state, _, err := GetNodeConfigStatus(context.EtcdClient, agent.Name(), context.NodeID)
-
 	// Start the install action if it was just triggered or if it had already been running and now has restarted
 	if err == nil && (state == NodeConfigStatusTriggered || state == NodeConfigStatusRunning) {
 		RunConfigureAgent(context, agent)

pkg/clusterd/clusterMember.go

@@ -2,10 +2,8 @@ package clusterd
 
 import (
 	"errors"
-	"io/ioutil"
 	"log"
 	"path"
-	"strings"
 	"sync/atomic"
 	"time"
 
@@ -24,7 +22,6 @@ const (
 	hardwareDiscoveryIntervalSeconds = 120 * 60
 	watchErrorRetrySeconds           = 2
 	machineTtlMinutes                = 5
-	maxMachineIDLength               = 12
 )
 
 var (
@@ -41,29 +38,6 @@ type Leader interface {
 	GetLeaseName() string
 }
 
-func GetMachineID() (string, error) {
-	buf, err := ioutil.ReadFile("/etc/machine-id")
-	if err != nil {
-		return "", err
-	}
-
-	return trimMachineID(string(buf)), nil
-}
-
-func trimMachineID(id string) string {
-	// Trim the machine ID to a length that is statistically unlikely to collide with another node in the cluster
-	// while allowing us to use an ID that is both unique and succinct.
-	// Using the birthday collision algorithm, if we have a length of 12 hex characters, that gives us
-	// 16^12 possibilities. If we have a cluster with 1,000 nodes, we have a likelihood with node IDs
-	// colliding in less than 1 in a billion clusters.
-	id = strings.TrimSpace(id)
-	if len(id) <= maxMachineIDLength {
-		return id
-	}
-
-	return id[0:maxMachineIDLength]
-}
-
 func IsLeader(l Lease, nodeID string) bool {
 	if l == nil {
 		return false

pkg/clusterd/clusterMember_test.go

@@ -113,19 +113,6 @@ func (l *MockLeader) GetLeaseName() string {
 //
 // ************************************************************************************************
 // ************************************************************************************************
-func TestTrimMachineID(t *testing.T) {
-	testTrimMachineID(t, " 123 		", "123")
-	testTrimMachineID(t, " 1234567890", "1234567890")
-	testTrimMachineID(t, " 123456789012", "123456789012")
-	testTrimMachineID(t, " 1234567890123", "123456789012")
-	testTrimMachineID(t, "1234567890123", "123456789012")
-	testTrimMachineID(t, "123456789012345678", "123456789012")
-}
-
-func testTrimMachineID(t *testing.T, input, expected string) {
-	result := trimMachineID(input)
-	assert.Equal(t, expected, result)
-}
 
 func TestElectLeaderAcquireNil(t *testing.T) {
 	_, context, mockLeaseManager, leader := createDefaultDependencies()

pkg/clusterd/join.go

@@ -11,8 +11,12 @@ import (
 	"github.com/quantum/castle/pkg/util/proc"
 )
 
-// Initialize the cluster services to enable joining the cluster and listening for orchestration.
+// StartJoinCluster initializes the cluster services to enable joining the cluster and listening for orchestration.
 func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, discoveryURL, etcdMembers, privateIPv4 string) (*Context, error) {
+	nodeID, err := util.GetMachineID()
+	if err != nil {
+		return nil, err
+	}
 
 	etcdClients := []string{}
 	if etcdMembers != "" {
@@ -22,7 +26,7 @@ func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, dis
 
 		// use the discovery URL to query the etcdmgr what the etcd client endpoints should be
 		var err error
-		etcdClients, err = manager.GetEtcdClients(discoveryURL, privateIPv4)
+		etcdClients, err = manager.GetEtcdClients(discoveryURL, privateIPv4, nodeID)
 		if err != nil {
 			return nil, err
 		}
@@ -33,11 +37,6 @@ func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, dis
 		return nil, err
 	}
 
-	nodeID, err := GetMachineID()
-	if err != nil {
-		return nil, err
-	}
-
 	// set the basic node config in etcd
 	key := inventory.GetNodeConfigKey(nodeID)
 	if err := util.CreateEtcdDir(etcdClient, key); err != nil {

pkg/clusterd/status.go

@@ -89,7 +89,7 @@ func WaitForNodeConfigCompletion(etcdClient etcd.KeysAPI, taskKey string, nodes
 
 	var waitgroup sync.WaitGroup
 	waitgroup.Add(len(nodes))
-	var nodesSuccessful int32 = 0
+	var nodesSuccessful int32
 
 	// Start a go routine for each node that is expecting status updates for the configuration task
 	for _, node := range nodes {
@@ -114,10 +114,10 @@ func WaitForNodeConfigCompletion(etcdClient etcd.KeysAPI, taskKey string, nodes
 		}(node)
 	}
 
-	log.Printf("Waiting for %d nodes to complete task %s", len(nodes), taskKey)
+	log.Printf("Waiting for %d nodes to complete task: %s", len(nodes), taskKey)
 	waitgroup.Wait()
 
-	log.Printf("%d/%d nodes successful for task %s", nodesSuccessful, len(nodes), taskKey)
+	log.Printf("%d/%d nodes successful for task: %s", nodesSuccessful, len(nodes), taskKey)
 	if int(nodesSuccessful) < len(nodes) {
 		return int(nodesSuccessful), errors.New("not all nodes succeeded configuration")
 	}
@@ -134,16 +134,16 @@ func GetNodeProgressKey(nodeID string) string {
 func GetNodeStatusKey(service, nodeID string) string {
 	if service == "" {
 		return path.Join(GetNodeProgressKey(nodeID), StatusValue)
-	} else {
-		return path.Join(fmt.Sprintf(NodeStatusServiceKey, nodeID, service), StatusValue)
 	}
+	return path.Join(fmt.Sprintf(NodeStatusServiceKey, nodeID, service), StatusValue)
 }
 
-// Set the node configuration status.
+// SetNodeConfigStatus sets the node configuration status.
 // If a taskKey is specified, set the status for a specific task.
 // If the taskKey is the empty string, set the status for the node.
 func SetNodeConfigStatus(etcdClient etcd.KeysAPI, nodeID, taskKey string, nodeStatus NodeConfigStatus) error {
 	key := GetNodeStatusKey(taskKey, nodeID)
+	fmt.Println("key: ", key)
 	_, err := etcdClient.Set(ctx.Background(), key, nodeStatus.String(), nil)
 	return err
 }
@@ -171,7 +171,7 @@ func GetNodeConfigStatus(etcdClient etcd.KeysAPI, taskKey, nodeID string) (NodeC
 	return retVal, value.Index, nil
 }
 
-// Watch for changes to the node config status etcd key
+// WatchNodeConfigStatus watches for changes to the node config status etcd key
 func WatchNodeConfigStatus(etcdClient etcd.KeysAPI, taskKey, nodeID string, timeout int, index *uint64) (NodeConfigStatus, error) {
 	key := GetNodeStatusKey(taskKey, nodeID)
 	value, timedOut, err := util.WatchEtcdKey(etcdClient, key, index, timeout)

pkg/clusterd/test/nonevent.go

@@ -0,0 +1,39 @@
+package test
+
+import (
+	"log"
+
+	"github.com/quantum/castle/pkg/clusterd"
+)
+
+func WaitForEvents(leader clusterd.ServiceLeader) {
+	// add a placeholder event to the queue. When it is dequeued we know the rest of the events have completed.
+	e := newNonEvent()
+	leader.Events() <- e
+
+	// wait for the Name() method to be called on the nonevent, which means it was dequeued
+	log.Printf("waiting for event queue to empty")
+	<-e.signaled
+	log.Printf("event queue is empty")
+}
+
+// Empty event for testing
+type nonEvent struct {
+	signaled   chan bool
+	nameCalled bool
+}
+
+func newNonEvent() *nonEvent {
+	return &nonEvent{signaled: make(chan bool)}
+}
+
+func (e *nonEvent) Name() string {
+	if !e.nameCalled {
+		e.nameCalled = true
+		e.signaled <- true
+	}
+	return "nonevent"
+}
+func (e *nonEvent) Context() *clusterd.Context {
+	return nil
+}

pkg/etcdmgr/bootstrap/config.go

@@ -1,27 +1,89 @@
 package bootstrap
 
-import "net/url"
+import (
+	"log"
+	"net/url"
+	"time"
 
-// EtcdConfig holds the configuration of elastic etcd.
-type EtcdConfig struct {
+	"github.com/coreos/etcd/pkg/types"
+)
+
+const (
+	// DefaultClientPort is the default port for listening to client traffic
+	DefaultClientPort = "53379"
+	// DefaultPeerPort is the default port for listening to peer traffic
+	DefaultPeerPort = "53380"
+	// DefaultClientTimeout is the default timeout for etcd client
+	DefaultClientTimeout = 5 * time.Second
+)
+
+// Config holds the configuration of elastic etcd.
+type Config struct {
 	InstanceName        string
 	ListenPeerURLs      []url.URL
 	ListenClientURLs    []url.URL
 	AdvertisePeerURLs   []url.URL
 	AdvertiseClientURLs []url.URL
 	DataDir             string
+	URLsMap             types.URLsMap
 }
 
 // CopyConfig returns a copy of given etcd config
-func CopyConfig(conf *EtcdConfig) EtcdConfig {
-	newConf := EtcdConfig{
+func CopyConfig(conf *Config) Config {
+	newConf := Config{
 		InstanceName:        conf.InstanceName,
 		ListenPeerURLs:      conf.ListenPeerURLs,
 		ListenClientURLs:    conf.ListenClientURLs,
 		AdvertisePeerURLs:   conf.AdvertisePeerURLs,
 		AdvertiseClientURLs: conf.AdvertiseClientURLs,
 		DataDir:             conf.DataDir,
+		URLsMap:             conf.URLsMap,
 	}
 
 	return newConf
 }
+
+// GenerateConfig automatically generates a config for embedded etcd
+func GenerateConfig(ipAddr, nodeID string) (*Config, error) {
+	conf := &Config{}
+	conf.InstanceName = nodeID
+	conf.ListenPeerURLs = append(conf.ListenPeerURLs,
+		getURLFromSchemeIPPort("http", ipAddr, DefaultPeerPort))
+	conf.ListenClientURLs = append(conf.ListenClientURLs,
+		getURLFromSchemeIPPort("http", ipAddr, DefaultClientPort))
+	conf.AdvertisePeerURLs = append(conf.AdvertisePeerURLs,
+		getURLFromSchemeIPPort("http", ipAddr, DefaultPeerPort))
+	conf.AdvertiseClientURLs = append(conf.AdvertiseClientURLs,
+		getURLFromSchemeIPPort("http", ipAddr, DefaultClientPort))
+	conf.DataDir = "/tmp/etcd-data"
+	conf.URLsMap = types.URLsMap{}
+
+	return conf, nil
+}
+
+// GenerateConfigFromExistingCluster automatically generates a config for joining an existing cluster
+func GenerateConfigFromExistingCluster(Context EtcdMgrContext, ipAddr, nodeID string) (*Config, error) {
+	conf, err := GenerateConfig(ipAddr, nodeID)
+	if err != nil {
+		return nil, err
+	}
+
+	// get current urlmap of the etcd cluster
+	_, conf.URLsMap, err = Context.Members()
+	if err != nil {
+		return nil, err
+	}
+
+	// add the entry for the new member
+	conf.URLsMap[conf.InstanceName] = conf.AdvertisePeerURLs
+	log.Println("conf.URLsMap: ", conf.URLsMap)
+	return conf, nil
+}
+
+func getURLFromSchemeIPPort(scheme, ip, port string) url.URL {
+	u := url.URL{
+		Scheme: scheme,
+		Host:   ip + ":" + port,
+	}
+	return u
+}

pkg/etcdmgr/bootstrap/discovery.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/coreos/etcd/client"
 	"github.com/coreos/etcd/store"
@@ -16,12 +15,21 @@ import (
 	"golang.org/x/net/context/ctxhttp"
 )
 
-const (
-	Timeout = 30 * time.Second
-)
+// QuorumFormed tries to connect to the etcd cluster. If it fails it interprets it as an incomplete quorum.
+func QuorumFormed(currentNodes []string) bool {
+	_, err := client.New(client.Config{
+		Endpoints:               currentNodes,
+		Transport:               client.DefaultTransport,
+		HeaderTimeoutPerRequest: DefaultClientTimeout,
+	})
+	if err != nil {
+		return false
+	}
+	return true
+}
 
-func IsQuorumFull(discovery string) (bool, error, []string) {
-	res, err := QueryDiscoveryService(discovery + "/_config/size")
+func IsQuorumFull(token string) (bool, error, []string) {
+	res, err := QueryDiscoveryService(token + "/_config/size")
 	if err != nil {
 		return false, fmt.Errorf("cannot get discovery url cluster size: %v", err), []string{}
 	}
@@ -30,7 +38,7 @@ func IsQuorumFull(discovery string) (bool, error, []string) {
 	clusterSize := int(size)
 	log.Println("cluster max size is: ", clusterSize)
 
-	currentNodes, _ := GetCurrentNodes(discovery)
+	currentNodes, _ := GetCurrentNodesFromDiscovery(token)
 	log.Println("currentNodes: ", currentNodes)
 	if len(currentNodes) < clusterSize {
 		return false, nil, []string{}
@@ -41,7 +49,7 @@ func IsQuorumFull(discovery string) (bool, error, []string) {
 
 // QueryDiscoveryService reads a key from a discovery url.
 func QueryDiscoveryService(token string) (*store.Event, error) {
-	ctx, _ := context.WithTimeout(context.Background(), Timeout)
+	ctx, _ := context.WithTimeout(context.Background(), DefaultClientTimeout)
 	resp, err := ctxhttp.Get(ctx, http.DefaultClient, token)
 	if err != nil {
 		return nil, err
@@ -61,7 +69,7 @@ func QueryDiscoveryService(token string) (*store.Event, error) {
 	return &res, nil
 }
 
-func GetCurrentNodes(token string) ([]string, error) {
+func GetCurrentNodesFromDiscovery(token string) ([]string, error) {
 	res, err := QueryDiscoveryService(token)
 	if err != nil {
 		return nil, err
@@ -71,7 +79,8 @@ func GetCurrentNodes(token string) ([]string, error) {
 		if nn.Value == nil {
 			log.Printf("Skipping %q because no value exists", nn.Key)
 		}
-		n, err := newDiscoveryNode(*nn.Value, 2379)
+		clientPort, _ := strconv.Atoi(DefaultClientPort)
+		n, err := newDiscoveryNode(*nn.Value, clientPort)
 		if err != nil {
 			log.Printf("invalid peer url %q in discovery service: %v", *nn.Value, err)
 			continue

pkg/etcdmgr/bootstrap/etcd.go

@@ -1,6 +1,7 @@
 package bootstrap
 
 import (
+	"fmt"
 	"log"
 	"net"
 	"net/http"
@@ -12,70 +13,60 @@ import (
 	"github.com/coreos/etcd/pkg/types"
 )
 
+type EtcdFactory interface {
+	NewEmbeddedEtcd(token string, conf *Config, newCluster bool) (*EmbeddedEtcd, error)
+}
+
+type EmbeddedEtcdFactory struct {
+}
+
 // EmbeddedEtcd represents an instance of etcd server.
 type EmbeddedEtcd struct {
 	Server          *etcdserver.EtcdServer
 	peerListeners   []net.Listener
 	clientListeners []net.Listener
-	config          EtcdConfig
+	config          *Config
 }
 
 // NewEmbeddedEtcd creates a new inproc instance of etcd.
-func NewEmbeddedEtcd(token string, conf EtcdConfig, newCluster bool) (*EmbeddedEtcd, error) {
+func (e *EmbeddedEtcdFactory) NewEmbeddedEtcd(token string, conf *Config, newCluster bool) (*EmbeddedEtcd, error) {
 	var err error
-	ee := &EmbeddedEtcd{config: conf}
-	err = ee.initializeListeners()
+	instance := &EmbeddedEtcd{config: conf}
+	err = instance.InitializeListeners()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error initializing listeners. %+v", err)
 	}
 
 	serverConfig := getServerConfig(token, conf, newCluster)
 
-	ee.Server, err = etcdserver.NewServer(serverConfig)
+	instance.Server, err = etcdserver.NewServer(serverConfig)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error in creating etcd server. %+v", err)
 	}
 
-	ee.Server.Start()
-	for _, pl := range ee.peerListeners {
+	instance.Server.Start()
+	for _, pl := range instance.peerListeners {
 		go func(l net.Listener) {
-			http.Serve(l, v2http.NewPeerHandler(ee.Server))
+			http.Serve(l, v2http.NewPeerHandler(instance.Server))
 		}(pl)
 	}
 
 	clientHandler := http.Handler(&cors.CORSHandler{
-		Handler: v2http.NewClientHandler(ee.Server, ee.Server.Cfg.ReqTimeout()),
+		Handler: v2http.NewClientHandler(instance.Server, instance.Server.Cfg.ReqTimeout()),
 		Info:    &cors.CORSInfo{},
 	})
-	for _, cl := range ee.clientListeners {
+	for _, cl := range instance.clientListeners {
 		go func(l net.Listener) {
 			http.Serve(l, clientHandler)
 		}(cl)
 	}
 
 	// wait until server is stable and ready
-	<-ee.Server.ReadyNotify()
-	return ee, nil
+	<-instance.Server.ReadyNotify()
+	return instance, nil
 }
 
-func getServerConfig(token string, conf EtcdConfig, newCluster bool) *etcdserver.ServerConfig {
-	return &etcdserver.ServerConfig{
-		Name:                conf.InstanceName,
-		DiscoveryURL:        token,
-		InitialClusterToken: token,
-		ClientURLs:          conf.AdvertiseClientURLs,
-		PeerURLs:            conf.AdvertisePeerURLs,
-		DataDir:             conf.DataDir,
-		NewCluster:          newCluster,
-		TickMs:              100,
-		ElectionTicks:       10,
-		InitialPeerURLsMap: types.URLsMap{
-			conf.InstanceName: conf.AdvertisePeerURLs,
-		},
-	}
-}
-
-func (ee *EmbeddedEtcd) initializeListeners() error {
+func (ee *EmbeddedEtcd) InitializeListeners() error {
 	//initializing client listeners
 	log.Println("client urls to set listeners for: ", ee.config.ListenClientURLs)
 	for _, url := range ee.config.ListenClientURLs {
@@ -100,7 +91,7 @@ func (ee *EmbeddedEtcd) initializeListeners() error {
 }
 
 // Destroy wipes out the current instance of etcd and makes the necessary cleanup.
-func (ee *EmbeddedEtcd) Destroy(conf EtcdConfig) error {
+func (ee *EmbeddedEtcd) Destroy(conf *Config) error {
 	var err error
 
 	for _, pl := range ee.peerListeners {
@@ -127,3 +118,31 @@ func (ee *EmbeddedEtcd) Destroy(conf EtcdConfig) error {
 
 	return err
 }
+
+func getServerConfig(token string, conf *Config, newCluster bool) *etcdserver.ServerConfig {
+	// if this config if for a member that is joining an exsiting cluster, the token should be empty
+	if !newCluster {
+		token = ""
+	}
+
+	var initialPeerURLsMap types.URLsMap
+	if len(conf.URLsMap) == 0 {
+		initialPeerURLsMap = types.URLsMap{
+			conf.InstanceName: conf.AdvertisePeerURLs,
+		}
+	} else {
+		initialPeerURLsMap = conf.URLsMap
+	}
+	return &etcdserver.ServerConfig{
+		Name:                conf.InstanceName,
+		DiscoveryURL:        token,
+		InitialClusterToken: token,
+		ClientURLs:          conf.AdvertiseClientURLs,
+		PeerURLs:            conf.AdvertisePeerURLs,
+		DataDir:             conf.DataDir,
+		NewCluster:          newCluster,
+		TickMs:              100,
+		ElectionTicks:       10,
+		InitialPeerURLsMap:  initialPeerURLsMap,
+	}
+}

pkg/etcdmgr/bootstrap/etcdcontext.go

@@ -0,0 +1,92 @@
+package bootstrap
+
+import (
+	"log"
+
+	"github.com/coreos/etcd/client"
+	"github.com/coreos/etcd/pkg/types"
+	ctx "golang.org/x/net/context"
+)
+
+type EtcdMgrContext interface {
+	Client(initialNodes []string) (client.Client, error)
+	MembersAPI() (client.MembersAPI, error)
+	KeysAPI() (client.KeysAPI, error)
+	Members() ([]string, types.URLsMap, error)
+}
+
+type Context struct {
+	ClusterToken string
+}
+
+func (e *Context) Client(initialNodes []string) (client.Client, error) {
+	c, err := client.New(client.Config{
+		Endpoints:               initialNodes,
+		Transport:               client.DefaultTransport,
+		HeaderTimeoutPerRequest: DefaultClientTimeout,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
+// MembersAPI returns an instance of MembersAPI for the current etcd cluster
+func (e *Context) MembersAPI() (client.MembersAPI, error) {
+	members, _, err := e.Members()
+	if err != nil {
+		return nil, err
+	}
+	c, err := e.Client(members)
+	if err != nil {
+		return nil, err
+	}
+	mAPI := client.NewMembersAPI(c)
+	return mAPI, nil
+}
+
+func (e *Context) KeysAPI() (client.KeysAPI, error) {
+	members, _, err := e.Members()
+	if err != nil {
+		return nil, err
+	}
+	c, err := e.Client(members)
+	if err != nil {
+		return nil, err
+	}
+	kAPI := client.NewKeysAPI(c)
+	return kAPI, nil
+}
+
+func (e *Context) Members() ([]string, types.URLsMap, error) {
+	urlsMap := types.URLsMap{}
+	var nodes []string
+	initialNodes, err := GetCurrentNodesFromDiscovery(e.ClusterToken)
+	if err != nil {
+		log.Println("error in GetCurrentNodesFromDiscovery")
+		return nodes, urlsMap, err
+	}
+
+	c, err := e.Client(initialNodes)
+	if err != nil {
+		return nodes, urlsMap, err
+	}
+
+	mAPI := client.NewMembersAPI(c)
+	members, err := mAPI.List(ctx.Background())
+	if err != nil {
+		return nodes, urlsMap, err
+	}
+
+	for _, member := range members {
+		urls, err := types.NewURLs(member.PeerURLs)
+		if err != nil {
+			return nodes, urlsMap, err
+		}
+		urlsMap[member.ID] = urls
+		// ClientURLs of a member is a url which is used by this member to listen to the clients' requests. This url could be used
+		// to create etcd client objects. In some use cases, multiple urls might be used, but we don't use that pattern.
+		nodes = append(nodes, member.ClientURLs...)
+	}
+	return nodes, urlsMap, nil
+}

pkg/etcdmgr/etcdmgr.go

@@ -0,0 +1,25 @@
+package etcdmgr
+
+import (
+	"log"
+
+	"github.com/quantum/castle/pkg/clusterd"
+	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
+)
+
+const (
+	etcdMgrName = "etcdmgr"
+)
+
+// NewEtcdMgrService creates a new etcdmgr service
+func NewEtcdMgrService(token string) *clusterd.ClusterService {
+	log.Println("creating instances of etcdMgrLeader and etcdMgrAgent")
+
+	return &clusterd.ClusterService{
+		Name:   etcdMgrName,
+		Leader: &etcdMgrLeader{context: &bootstrap.Context{ClusterToken: token}},
+		Agents: []clusterd.ServiceAgent{
+			&etcdMgrAgent{context: &bootstrap.Context{ClusterToken: token}, etcdFactory: &bootstrap.EmbeddedEtcdFactory{}},
+		},
+	}
+}

pkg/etcdmgr/etcdmgragent.go

@@ -0,0 +1,124 @@
+package etcdmgr
+
+import (
+	"fmt"
+	"log"
+	"path"
+
+	etcd "github.com/coreos/etcd/client"
+	"github.com/quantum/castle/pkg/clusterd"
+	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
+	"github.com/quantum/castle/pkg/etcdmgr/manager"
+	"github.com/quantum/castle/pkg/util"
+	ctx "golang.org/x/net/context"
+)
+
+const (
+	etcdMgrAgentName = "etcd"
+)
+
+type etcdMgrAgent struct {
+	embeddedEtcd *bootstrap.EmbeddedEtcd
+	conf         *bootstrap.Config
+	context      bootstrap.EtcdMgrContext
+	etcdFactory  bootstrap.EtcdFactory
+}
+
+func (e *etcdMgrAgent) Name() string {
+	return etcdMgrAgentName
+}
+
+func (e *etcdMgrAgent) Initialize(context *clusterd.Context) error {
+	return nil
+}
+
+func (e *etcdMgrAgent) ConfigureLocalService(context *clusterd.Context) error {
+	log.Printf("inside ConfigureLocalService")
+	// check if the etcdmgr is in the desired state for this node
+	desiredKey := path.Join(etcdmgrKey, etcdDesiredKey, context.NodeID)
+	etcdmgrDesired, err := util.EtcdDirExists(context.EtcdClient, desiredKey)
+	if err != nil {
+		return fmt.Errorf("error in checking existence of desired key: %+v", err)
+	}
+	appliedKey := getEtcdMgrAppliedKey(context.NodeID)
+	etcdmgrApplied, err := util.EtcdDirExists(context.EtcdClient, appliedKey)
+	if err != nil {
+		return fmt.Errorf("error in checking existence of applied key: %+v", err)
+	}
+
+	// Add or remote embedded etcd instance as needed
+	if etcdmgrDesired {
+		e.CreateLocalService(context, desiredKey, etcdmgrApplied)
+
+	} else if !etcdmgrDesired && etcdmgrApplied {
+		err := e.DestroyLocalService(context)
+		if err != nil {
+			return fmt.Errorf("error in removing node: %+v", err)
+		}
+
+	}
+
+	return nil
+}
+
+// get ip address for the target agent (will be used to bootstrap embedded etcd)
+func (e *etcdMgrAgent) CreateLocalService(context *clusterd.Context, desiredKey string, etcdmgrApplied bool) error {
+	ipAddrKey := path.Join(desiredKey, "ipaddress")
+	resp, err := context.EtcdClient.Get(ctx.Background(), ipAddrKey, nil)
+	if err != nil {
+		return fmt.Errorf("error in getting the ip address key: %+v. err: %+v", ipAddrKey, err)
+	}
+	ipAddr := resp.Node.Value
+	log.Println("ipAddress: ", ipAddr)
+	e.conf, err = bootstrap.GenerateConfigFromExistingCluster(e.context, ipAddr, context.NodeID)
+	log.Println("config: ", e.conf)
+	if err != nil {
+		return err
+	}
+
+	if !etcdmgrApplied {
+		log.Println("adding the current node to the etcd cluster...")
+		targetEndpoint := getPeerEndpointFromIP(ipAddr)
+		err = manager.AddMember(e.context, targetEndpoint)
+		if err != nil {
+			return fmt.Errorf("error in adding a member to the cluster")
+		}
+
+		ipKey := path.Join(etcdmgrKey, etcdAppliedKey, context.NodeID, "ipaddress")
+		log.Println("ipKey for new instance: ", ipKey)
+		_, err = context.EtcdClient.Set(ctx.Background(), ipKey, ipAddr, nil)
+		if err != nil {
+			return fmt.Errorf("error in setting applied key for ip key. %+v", err)
+		}
+	}
+	e.embeddedEtcd, err = e.etcdFactory.NewEmbeddedEtcd("", e.conf, false)
+	if err != nil {
+		return fmt.Errorf("error in creating a new instance of embedded etcd. err: %+v: ", err)
+	}
+
+	return nil
+}
+
+func (e *etcdMgrAgent) DestroyLocalService(context *clusterd.Context) error {
+	fmt.Println("destroying the local embedded etcd instance")
+	err := e.embeddedEtcd.Destroy(e.conf)
+	e.embeddedEtcd = nil
+	if err != nil {
+		return fmt.Errorf("error in destroying local embedded etcd. err: %+v", err)
+	}
+	// successful, remove the current node from desired state
+	appliedKey := getEtcdMgrAppliedKey(context.NodeID)
+	_, err = context.EtcdClient.Delete(ctx.Background(), appliedKey, &etcd.DeleteOptions{Dir: true, Recursive: true})
+	if err != nil {
+		return fmt.Errorf("error in removing etcdmgr applied key. err: %+v", err)
+	}
+	return nil
+}
+
+func getEtcdMgrAppliedKey(nodeID string) string {
+	return path.Join(etcdmgrKey, etcdAppliedKey, nodeID)
+}
+
+func getPeerEndpointFromIP(ip string) string {
+	return "http://" + ip + ":" + bootstrap.DefaultPeerPort
+}

pkg/etcdmgr/etcdmgragent_test.go

@@ -0,0 +1,45 @@
+package etcdmgr
+
+import (
+	"path"
+	"testing"
+
+	"github.com/quantum/castle/pkg/clusterd"
+	"github.com/quantum/castle/pkg/etcdmgr/test"
+	"github.com/quantum/castle/pkg/util"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEtcdMgrAgent(t *testing.T) {
+	mockContext := test.MockContext{}
+	// adding 1.2.3.4 as the first/existing cluster member
+	mockContext.AddMembers([]string{"http://1.2.3.4:53379"})
+	mockEmbeddedEtcdFactory := test.MockEmbeddedEtcdFactory{}
+
+	// agent2 is the agent on node 2 which is going to create a new embedded etcd
+	agent2 := &etcdMgrAgent{context: &mockContext, etcdFactory: &mockEmbeddedEtcdFactory}
+	etcdClient2 := util.NewMockEtcdClient()
+	context2 := &clusterd.Context{
+		EtcdClient: etcdClient2,
+		NodeID:     "node2",
+	}
+	err := agent2.Initialize(context2)
+	assert.Equal(t, "etcd", agent2.Name())
+	assert.Nil(t, agent2.embeddedEtcd)
+
+	err = agent2.ConfigureLocalService(context2)
+	assert.Nil(t, err)
+
+	//set the agent in the desired state
+	desiredKey := path.Join(etcdmgrKey, etcdDesiredKey, context2.NodeID)
+	etcdClient2.SetValue(path.Join(desiredKey, "ipaddress"), "2.3.4.5")
+
+	err = agent2.ConfigureLocalService(context2)
+	assert.Nil(t, err)
+	assert.NotNil(t, agent2.embeddedEtcd)
+	//remove the desired status
+	etcdClient2.DeleteDir(desiredKey)
+	err = agent2.ConfigureLocalService(context2)
+	assert.Nil(t, err)
+	assert.Nil(t, agent2.embeddedEtcd)
+}

pkg/etcdmgr/etcdmgrleader.go

@@ -0,0 +1,224 @@
+package etcdmgr
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"net/url"
+	"path"
+
+	"github.com/coreos/etcd/client"
+	"github.com/quantum/castle/pkg/clusterd"
+	"github.com/quantum/castle/pkg/clusterd/inventory"
+	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
+	"github.com/quantum/castle/pkg/etcdmgr/manager"
+	"github.com/quantum/castle/pkg/etcdmgr/policy"
+	ctx "golang.org/x/net/context"
+)
+
+const (
+	etcdmgrKey     = "/castle/services/etcd"
+	etcdDesiredKey = "desired"
+	etcdAppliedKey = "applied"
+)
+
+type etcdMgrLeader struct {
+	context bootstrap.EtcdMgrContext
+	events  chan clusterd.LeaderEvent
+}
+
+func (e *etcdMgrLeader) RefreshKeys() []*clusterd.RefreshKey {
+	return []*clusterd.RefreshKey{}
+}
+
+func (e *etcdMgrLeader) StartWatchEvents() {
+	log.Println("etcdMgrLeader.StartWatchEvents")
+	if e.events != nil {
+		close(e.events)
+	}
+	e.events = make(chan clusterd.LeaderEvent, 10)
+	go e.handleOrchestratorEvents()
+}
+
+func (e *etcdMgrLeader) Events() chan clusterd.LeaderEvent {
+	return e.events
+}
+
+// Close closes the event queue for the etcdmanager
+func (e *etcdMgrLeader) Close() error {
+	close(e.events)
+	e.events = nil
+	return nil
+}
+
+func (e *etcdMgrLeader) handleOrchestratorEvents() {
+	// Listen for events from the orchestrator indicating Refresh, AddNode, RemoveNode, and StaleNode events
+	for evt := range e.events {
+		log.Printf("etcdmgr leader received event %s", evt.Name())
+
+		if refreshEvt, ok := evt.(*clusterd.RefreshEvent); ok {
+			// Perform a full refresh of the cluster to make sure the cluster is in the desired state
+			err := e.ConfigureEtcdServices(refreshEvt.Context(), []*clusterd.UnhealthyNode{})
+			if err != nil {
+				log.Println("the ConfigureEtcdServices wasn't successful for RefreshEvent: ", err)
+			}
+		} else if nodeAddEvt, ok := evt.(*clusterd.AddNodeEvent); ok {
+			err := e.ConfigureEtcdServices(nodeAddEvt.Context(), []*clusterd.UnhealthyNode{})
+			if err != nil {
+				log.Println("the ConfigureEtcdServices wasn't successful for AddNodeEvent: ", err)
+			}
+		} else if unHealthyNodeEvt, ok := evt.(*clusterd.UnhealthyNodeEvent); ok {
+			// if the removed node has been an etcd member, then we need to add a new node to the cluster if needed
+			// and then remove it.
+			err := e.ConfigureEtcdServices(unHealthyNodeEvt.Context(), unHealthyNodeEvt.Nodes())
+			if err != nil {
+				log.Println("the ConfigureEtcdServices wasn't successful for UnhealthyNodeEvent: ", err)
+			}
+		}
+
+		log.Printf("etcdmgr leader completed event %s", evt.Name())
+	}
+}
+
+// ConfigureEtcdServices
+func (e *etcdMgrLeader) ConfigureEtcdServices(context *clusterd.Context, unhealthyNodes []*clusterd.UnhealthyNode) error {
+	log.Printf("entered etcdMgr.ConfigureEtcdservices")
+
+	currentEtcdMembers, _, err := e.context.Members()
+	if err != nil {
+		return err
+	}
+	// currentEtcdMembers are full URLs (scheme, ip, port). We want to convert them to a list of node IDs.
+	currentEtcdMemberIDs, err := getNodeIDs(currentEtcdMembers, context.Inventory.Nodes)
+	if err != nil {
+		log.Println("error in converting etcd member urls to node ids.")
+		return err
+	}
+	log.Printf("current etcd cluster members: %v | Nodes: %v | IDs: %v | unhealthy nodes: %v",
+		currentEtcdMembers, context.Inventory.Nodes, currentEtcdMemberIDs, unhealthyNodes)
+	currentEtcdQuorumSize := len(currentEtcdMembers)
+	currentClusterSize := len(context.Inventory.Nodes) - len(unhealthyNodes)
+	log.Printf("currentEtcdQuorumSize: %v | currentClusterSize: %v ", currentEtcdQuorumSize, currentClusterSize)
+	desiredEtcdQuorumSize := policy.CalculateDesiredEtcdCount(currentClusterSize)
+	var candidates []string
+	delta := desiredEtcdQuorumSize - currentEtcdQuorumSize
+	var clusterNodes []string
+	for node := range context.Inventory.Nodes {
+		clusterNodes = append(clusterNodes, node)
+	}
+	var unhealthyIDs []string
+	for _, node := range unhealthyNodes {
+		unhealthyIDs = append(unhealthyIDs, node.NodeID)
+	}
+	log.Println("unhealthy nodeIDs: ", unhealthyIDs)
+	if delta != 0 {
+		log.Printf("desiredEtcdQuorumSize: %v, delta: %v, clusterNodes: %v", desiredEtcdQuorumSize, delta, clusterNodes)
+		log.Println("currentEtcdMemberIDs: ", currentEtcdMemberIDs)
+		candidates, err = policy.ChooseEtcdCandidatesToAddOrRemove(delta, currentEtcdMemberIDs, clusterNodes, unhealthyIDs)
+		if err != nil {
+			return err
+		}
+		log.Println("candidates: ", candidates)
+	}
+
+	if delta > 0 {
+		log.Println("target nodes to run new instances of embedded etcds on: ", candidates)
+		err := e.growEtcdQuorum(context, candidates)
+		if err != nil {
+			return fmt.Errorf("error in growing etcd quorum: %+v", err)
+		}
+
+	} else if delta < 0 {
+		log.Println("target nodes to remove the instances of embedded etcds from: ", candidates)
+		err := e.shrinkEtcdQuorum(context, candidates)
+		if err != nil {
+			return fmt.Errorf("error in shrinking etcd quorum: %+v", err)
+		}
+	}
+	return nil
+}
+
+func (e *etcdMgrLeader) growEtcdQuorum(context *clusterd.Context, candidates []string) error {
+	// We need to do the operations one by one to prevent quorum corruption
+	for _, candidate := range candidates {
+		// add target node to the current etcd cluster
+		var targetIP string
+		if node, ok := context.Inventory.Nodes[candidate]; ok {
+			targetIP = node.IPAddress
+		} else {
+			return errors.New("candidate ip not found in the inventory")
+		}
+
+		// set ip address for the target agent (will be used to bootstrap embedded etcd)
+		ipKey := path.Join(etcdmgrKey, etcdDesiredKey, candidate, "ipaddress")
+		log.Println("address key for new instance: ", ipKey)
+		_, err := context.EtcdClient.Set(ctx.Background(), ipKey, targetIP, nil)
+		if err != nil {
+			return err
+		}
+
+		log.Printf("triggering the agent on %v to create an instance of embedded etcd\n", targetIP)
+		err = clusterd.TriggerAgentsAndWaitForCompletion(context.EtcdClient, []string{candidate}, etcdMgrAgentName, len(candidates))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (e *etcdMgrLeader) shrinkEtcdQuorum(context *clusterd.Context, candidates []string) error {
+	// We need to do the operations one by one to prevent quorum corruption
+	for _, candidate := range candidates {
+		// remove target node from the current etcd cluster
+		var targetEndpoint string
+		if node, ok := context.Inventory.Nodes[candidate]; ok {
+			targetEndpoint = getPeerEndpointFromIP(node.IPAddress)
+		} else {
+			return errors.New("candidate endpoint not found in the inventory")
+		}
+
+		key := path.Join(etcdmgrKey, etcdDesiredKey, candidate, "ipaddress")
+		log.Println("address key for new instance: ", key)
+		_, err := context.EtcdClient.Delete(ctx.Background(), key, &client.DeleteOptions{Dir: true, Recursive: true})
+		if err != nil {
+			return fmt.Errorf("error in removing desired key for node: %+v err: %+v\n", candidate, err)
+		}
+
+		err = manager.RemoveMember(e.context, targetEndpoint)
+		if err != nil {
+			return fmt.Errorf("error in removing a member from the cluster. %+v\n", err)
+		}
+
+		// Note: For remove, we first try to remove it from cluster and then delete the corresponding instance. (different from add case)
+		// handling a case when a node is not down but demoted.
+		log.Println("node was successfully removed from the cluster. Now triggering the agent to cleanup the etcd instance...")
+		// we set the timout to 10sec since the node might already be failed
+		err = clusterd.TriggerAgentsAndWait(context.EtcdClient, []string{candidate}, etcdMgrAgentName, len(candidates), 10)
+		if err != nil {
+			return fmt.Errorf("error in cleaning up the target etcdmgr agent (this might have happened due to failure of the node): %+v", err)
+		}
+	}
+	return nil
+}
+
+// TODO: can we make it more efficient?
+func getNodeIDs(nodeURLs []string, Nodes map[string]*inventory.NodeConfig) ([]string, error) {
+	log.Println("nodeURLs: ", nodeURLs)
+	nodeIDs := []string{}
+	for _, u := range nodeURLs {
+		uu, err := url.Parse(u)
+		if err != nil {
+			return nil, err
+		}
+		ip, _, _ := net.SplitHostPort(uu.Host)
+		for nodeID, config := range Nodes {
+			log.Println("nodeID: ", nodeID)
+			if config.IPAddress == ip {
+				log.Printf("matched, ip: %v | nodeID: %v\n", ip, nodeID)
+				nodeIDs = append(nodeIDs, nodeID)
+			}
+		}
+	}
+	return nodeIDs, nil
+}

pkg/etcdmgr/etcdmgrleader_test.go

@@ -0,0 +1,95 @@
+package etcdmgr
+
+import (
+	"testing"
+
+	"github.com/quantum/castle/pkg/clusterd"
+	"github.com/quantum/castle/pkg/clusterd/inventory"
+	clusterdtest "github.com/quantum/castle/pkg/clusterd/test"
+	"github.com/quantum/castle/pkg/etcdmgr/test"
+	"github.com/quantum/castle/pkg/util"
+	"github.com/stretchr/testify/assert"
+)
+
+// TestEtcdMgrLeaders tests Etcd Cluster's grow scenario
+func TestEtcdMgrLeaderGrow(t *testing.T) {
+	mockContext := test.MockContext{}
+	nodes := make(map[string]*inventory.NodeConfig)
+	inv := &inventory.Config{Nodes: nodes}
+
+	// adding 1.2.3.4 as the first/existing cluster member
+	nodes["a"] = &inventory.NodeConfig{IPAddress: "1.2.3.4"}
+	mockContext.AddMembers([]string{"http://1.2.3.4:53379"})
+	etcdmgrService := etcdMgrLeader{context: &mockContext}
+	etcdmgrService.StartWatchEvents()
+	defer etcdmgrService.Close()
+
+	nodes["b"] = &inventory.NodeConfig{IPAddress: "2.3.4.5"}
+	nodes["c"] = &inventory.NodeConfig{IPAddress: "3.4.5.6"}
+
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{
+		EtcdClient: etcdClient,
+		Inventory:  inv,
+	}
+
+	// mock the agent responses that the deployments were successful to create an instance of embeddedEtcd
+	etcdClient.WatcherResponses["/castle/_notify/b/etcd/status"] = "succeeded"
+	etcdClient.WatcherResponses["/castle/_notify/c/etcd/status"] = "succeeded"
+
+	addNodeEvt := clusterd.NewAddNodeEvent(context, "b")
+	etcdmgrService.Events() <- addNodeEvt
+
+	// wait for the event queue to be empty
+	clusterdtest.WaitForEvents(&etcdmgrService)
+
+	// there might be a&c or b&c in the desired states
+	assert.True(t, etcdClient.GetChildDirs("/castle/services/etcd/desired").Count() == 2)
+	assert.Equal(t, etcdClient.GetValue("/castle/services/etcd/desired/b/ipaddress"), "2.3.4.5")
+	assert.Equal(t, etcdClient.GetValue("/castle/services/etcd/desired/c/ipaddress"), "3.4.5.6")
+}
+
+func TestEtcdMgrLeaderShrink(t *testing.T) {
+	mockContext := test.MockContext{}
+	nodes := make(map[string]*inventory.NodeConfig)
+	inv := &inventory.Config{Nodes: nodes}
+
+	// adding 1.2.3.4 as the first/existing cluster member
+	nodes["a"] = &inventory.NodeConfig{IPAddress: "1.2.3.4"}
+	mockContext.AddMembers([]string{"http://1.2.3.4:53379"})
+	nodes["b"] = &inventory.NodeConfig{IPAddress: "2.3.4.5"}
+	mockContext.AddMembers([]string{"http://2.3.4.5:53379"})
+	nodes["c"] = &inventory.NodeConfig{IPAddress: "3.4.5.6"}
+	mockContext.AddMembers([]string{"http://3.4.5.6:53379"})
+	etcdmgrService := etcdMgrLeader{context: &mockContext}
+	etcdmgrService.StartWatchEvents()
+	defer etcdmgrService.Close()
+
+	nodes["a"] = &inventory.NodeConfig{IPAddress: "1.2.3.4"}
+	nodes["b"] = &inventory.NodeConfig{IPAddress: "2.3.4.5"}
+	nodes["c"] = &inventory.NodeConfig{IPAddress: "3.4.5.6"}
+
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{
+		EtcdClient: etcdClient,
+		Inventory:  inv,
+	}
+
+	// mock the agent responses that the deployments were successful to create an instance of embeddedEtcd
+	etcdClient.WatcherResponses["/castle/_notify/a/etcd/status"] = "succeeded"
+	etcdClient.WatcherResponses["/castle/_notify/c/etcd/status"] = "succeeded"
+	etcdClient.WatcherResponses["/castle/services/etcd/desired/a/ipaddress"] = "1.2.3.4"
+	etcdClient.WatcherResponses["/castle/services/etcd/desired/b/ipaddress"] = "2.3.4.5"
+	etcdClient.WatcherResponses["/castle/services/etcd/desired/c/ipaddress"] = "2.3.4.5"
+	etcdClient.Dump()
+
+	unhealthyNode := clusterd.UnhealthyNode{AgeSeconds: 10, NodeID: "c"}
+	unhealthyNodeEvt := clusterd.NewUnhealthyNodeEvent(context, []*clusterd.UnhealthyNode{&unhealthyNode})
+	etcdmgrService.Events() <- unhealthyNodeEvt
+
+	// wait for the event queue to be empty
+	clusterdtest.WaitForEvents(&etcdmgrService)
+	etcdClient.Dump()
+	// two possibilities: there might be a&c or b&c in the desired states
+	assert.True(t, etcdClient.GetChildDirs("/castle/services/etcd/desired").Count() == 0)
+}

pkg/etcdmgr/manager/etcdmgr.go

@@ -1,40 +1,29 @@
-// Package manager is the etcdmgr's entrypoint when it is used as a library.
 package manager
 
 import (
 	"errors"
-	"io/ioutil"
+	"fmt"
 	"log"
-	"net/url"
 	"os"
-	"strings"
-	"time"
 
-	"github.com/jmcvetta/randutil"
 	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
-)
-
-const (
-	// DefaultClientPort is the default port for listening to client traffic
-	DefaultClientPort = "2379"
-	// DefaultPeerPort is the default port for listening to peer traffic
-	DefaultPeerPort = "2380"
-	// DefaultClientTimeout is the default timeout for etcd client
-	DefaultClientTimeout = 30 * time.Second
+	ctx "golang.org/x/net/context"
 )
 
 // GetEtcdClients bootstraps an embedded etcd instance and returns a list of
-// current etcd cluster's client urls. (used in library scenarios)
-func GetEtcdClients(token, ipAddr string) ([]string, error) {
+// current etcd cluster's client URLs. (entrypoint, when it's used as a library)
+func GetEtcdClients(token, ipAddr, nodeID string) ([]string, error) {
 	// GODEBUG setting forcing use of Go's resolver
 	os.Setenv("GODEBUG", "netdns=go+1")
 
-	full, err, currentNodes := bootstrap.IsQuorumFull(token)
+	currentNodes, err := bootstrap.GetCurrentNodesFromDiscovery(token)
 	if err != nil {
-		return []string{}, errors.New("error in querying discovery service")
+		return nil, err
 	}
+	log.Println("current etcd cluster nodes: ", currentNodes)
 
-	localURL := "http://" + ipAddr + ":" + DefaultClientPort
+	quorum := bootstrap.QuorumFormed(currentNodes)
+	localURL := "http://" + ipAddr + ":" + bootstrap.DefaultClientPort
 	// Is it a restart scenario?
 	restart := false
 	log.Println("current localURL: ", localURL)
@@ -44,19 +33,19 @@ func GetEtcdClients(token, ipAddr string) ([]string, error) {
 			restart = true
 		}
 	}
-	if full && !restart {
-		log.Println("quorum is full, returning current quorum nodes...")
-		log.Println("current nodes: ", currentNodes)
+	if quorum && !restart {
+		log.Println("quorum is already formed, returning current cluster members: ", currentNodes)
 		return currentNodes, nil
 	}
-	log.Println("creating a new embedded etcd...")
-	conf, err := GenerateEtcdConfig(ipAddr)
+	log.Println("quorum is not complete, creating a new embedded etcd member...")
+	conf, err := bootstrap.GenerateConfig(ipAddr, nodeID)
 	if err != nil {
 		return []string{}, err
 	}
 	log.Println("conf:", conf)
 
-	ee, err := bootstrap.NewEmbeddedEtcd(token, conf, true)
+	factory := bootstrap.EmbeddedEtcdFactory{}
+	ee, err := factory.NewEmbeddedEtcd(token, conf, true)
 	if err != nil {
 		return nil, err
 	}
@@ -66,7 +55,7 @@ func GetEtcdClients(token, ipAddr string) ([]string, error) {
 
 // GetEtcdClientsWithConfig bootstraps an embedded etcd instance based on a given config and returns
 // a list of current etcd cluster's client urls. (used in CLI scenarios).
-func GetEtcdClientsWithConfig(token string, config bootstrap.EtcdConfig) ([]string, error) {
+func GetEtcdClientsWithConfig(token string, config bootstrap.Config) ([]string, error) {
 	// GODEBUG setting forcing use of Go's resolver
 	os.Setenv("GODEBUG", "netdns=go+1")
 	full, err, currentNodes := bootstrap.IsQuorumFull(token)
@@ -92,52 +81,67 @@ func GetEtcdClientsWithConfig(token string, config bootstrap.EtcdConfig) ([]stri
 	}
 
 	log.Println("creating a new embedded etcd")
-	ee, err := bootstrap.NewEmbeddedEtcd(token, config, true)
+	factory := bootstrap.EmbeddedEtcdFactory{}
+	ee, err := factory.NewEmbeddedEtcd(token, &config, true)
 	if err != nil {
 		return nil, err
 	}
 	return ee.Server.Cluster().ClientURLs(), nil
 }
 
-func getMachineID() (string, error) {
-	buf, err := ioutil.ReadFile("/etc/machine-id")
+// AddMember adds a node to the etcd cluster as a member
+func AddMember(context bootstrap.EtcdMgrContext, node string) error {
+	log.Println("Adding a node to the etcd cluster as a member: ", node)
+	mapi, err := context.MembersAPI()
 	if err != nil {
-		randomID, err := randutil.AlphaString(10)
-		if err != nil {
-			return "", err
+		return fmt.Errorf("error in Context.MembersAPI(). %+v", err)
+	}
+	member, err := mapi.Add(ctx.Background(), node)
+	if err != nil {
+		return err
+	}
+	log.Println("New member added: ", *member)
+	return nil
+}
+
+// RemoveMember removes a member from the etcd cluster
+func RemoveMember(context bootstrap.EtcdMgrContext, node string) error {
+	log.Println("removing a member from the etcd cluster: ", node)
+	mapi, err := context.MembersAPI()
+	if err != nil {
+		return err
+	}
+	// Get the list of members.
+	members, err := mapi.List(ctx.Background())
+	if err != nil {
+		return err
+	}
+	log.Println("members: ", members)
+	var nodeID string
+	for _, m := range members {
+		// PeerURLs is taken from the etcd's terminology. It means what endpoints a member use to communicates by its peers.
+		// In our implementation, we always use 1 endpoint for each member. In some cases, someone could bind to localhost
+		// too which is not useful for us and we don't do that.
+		if sliceContains(m.PeerURLs, node) {
+			nodeID = m.ID
 		}
-		log.Println("err reading machine-id. generated random id: ", randomID)
-		return randomID, nil
 	}
-
-	return strings.TrimSpace(string(buf)), nil
-}
-
-// GenerateEtcdConfig automatically generates a config for library scenarios
-func GenerateEtcdConfig(ipAddr string) (bootstrap.EtcdConfig, error) {
-	conf := bootstrap.EtcdConfig{}
-	machineID, err := getMachineID()
+	if nodeID == "" {
+		return errors.New("node not found")
+	}
+	err = mapi.Remove(ctx.Background(), nodeID)
 	if err != nil {
-		return bootstrap.EtcdConfig{}, err
+		return err
 	}
-	conf.InstanceName = machineID
-	conf.ListenPeerURLs = append(conf.ListenPeerURLs,
-		getURLFromSchemeIPPort("http", ipAddr, DefaultPeerPort))
-	conf.ListenClientURLs = append(conf.ListenClientURLs,
-		getURLFromSchemeIPPort("http", ipAddr, DefaultClientPort))
-	conf.AdvertisePeerURLs = append(conf.AdvertisePeerURLs,
-		getURLFromSchemeIPPort("http", ipAddr, DefaultPeerPort))
-	conf.AdvertiseClientURLs = append(conf.AdvertiseClientURLs,
-		getURLFromSchemeIPPort("http", ipAddr, DefaultClientPort))
-	conf.DataDir = "/tmp/etcd-data"
-
-	return conf, nil
+	log.Println("New member removed: ", node)
+	return nil
 }
 
-func getURLFromSchemeIPPort(scheme, ip, port string) url.URL {
-	u := url.URL{
-		Scheme: scheme,
-		Host:   ip + ":" + port,
+func sliceContains(slice []string, str string) bool {
+	for _, s := range slice {
+		if s == str {
+			return true
+		}
 	}
-	return u
+	return false
 }

pkg/etcdmgr/policy/policy.go

@@ -0,0 +1,62 @@
+package policy
+
+import (
+	"errors"
+	"log"
+	"sort"
+
+	"github.com/quantum/castle/pkg/util"
+)
+
+// ChooseEtcdCandidatesToAddOrRemove decides which node to become member..
+func ChooseEtcdCandidatesToAddOrRemove(delta int, currentEtcdNodes, currentNodes, unhealthyNodes []string) ([]string, error) {
+	var candidates []string
+	if delta > 0 {
+		// We assume unhealthynode event happens separate from add node event. So, in this case unhealthyNodes
+		// should be empty
+		if len(unhealthyNodes) > 0 {
+			return []string{}, errors.New("request for adding nodes to the etcd cluster, while there are unhealthy nodes")
+		}
+		candidates = util.SetDifference(currentNodes, currentEtcdNodes).ToSlice()
+		if len(candidates) > delta {
+			return candidates[:delta], nil
+		}
+	} else {
+		// Choosing etcd nodes to remove
+		currentEtcdNodeSet := util.CreateSet(currentEtcdNodes)
+		currentEtcdNodeSet.Subtract(util.SetDifference(currentEtcdNodes, unhealthyNodes))
+		unhealthyEtcdNodes := currentEtcdNodeSet.ToSlice()
+		log.Println("unhealthyEtcdNodes: ", unhealthyEtcdNodes)
+		candidates = unhealthyEtcdNodes
+		numOfNodeToRemove := -delta
+		if len(candidates) >= numOfNodeToRemove {
+			sort.Strings(candidates)
+			return candidates[:numOfNodeToRemove], nil
+		}
+		//need more candidates from etcd nodes
+		numOfNodeToRemove = numOfNodeToRemove - len(candidates)
+		moreCandidates := util.SetDifference(currentEtcdNodes, candidates).ToSlice()
+		sort.Strings(moreCandidates)
+		if len(moreCandidates) >= numOfNodeToRemove {
+			candidates = append(candidates, moreCandidates[:numOfNodeToRemove]...)
+			sort.Strings(candidates)
+			return candidates, nil
+		}
+		candidates = append(candidates, moreCandidates...)
+	}
+	sort.Strings(candidates)
+	return candidates, nil
+}
+
+// CalculateDesiredEtcdCount the number of etcd instances that should be deployed
+// TODO: make the logic more intelligent.
+func CalculateDesiredEtcdCount(nodeCount int) int {
+	if nodeCount > 100 {
+		return 7
+	} else if nodeCount > 20 {
+		return 5
+	} else if nodeCount > 2 {
+		return 3
+	}
+	return 1
+}

pkg/etcdmgr/test/mocketcdmgr.go

@@ -0,0 +1,118 @@
+package test
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"errors"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/coreos/etcd/client"
+	"github.com/coreos/etcd/pkg/types"
+	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
+	ctx "golang.org/x/net/context"
+)
+
+type MockContext struct {
+	members    []string
+	urlsMap    types.URLsMap
+	membersAPI client.MembersAPI
+}
+
+func (m *MockContext) Client(initialNodes []string) (client.Client, error) {
+	c, err := client.New(client.Config{
+		Endpoints:               initialNodes,
+		Transport:               client.DefaultTransport,
+		HeaderTimeoutPerRequest: 0 * time.Second,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
+// MembersAPI returns an instance of MembersAPI for the current etcd cluster
+func (m *MockContext) MembersAPI() (client.MembersAPI, error) {
+	if m.membersAPI == nil {
+		m.membersAPI = &MockMembersAPI{}
+	}
+
+	return m.membersAPI, nil
+}
+
+func (m *MockContext) KeysAPI() (client.KeysAPI, error) {
+	members, _, _ := m.Members()
+	c, err := m.Client(members)
+	if err != nil {
+		return nil, err
+	}
+	kAPI := client.NewKeysAPI(c)
+	return kAPI, nil
+}
+
+func (m *MockContext) Members() ([]string, types.URLsMap, error) {
+	return m.members, m.urlsMap, nil
+}
+
+// AddMembers is added for easier unit testing. In real implementation, we always get the latest
+// membership from the cluster itself.
+func (m *MockContext) AddMembers(members []string) {
+	m.members = append(m.members, members...)
+	mAPI, _ := m.MembersAPI()
+	if m.urlsMap == nil {
+		m.urlsMap = types.URLsMap{}
+	}
+	for _, member := range members {
+		uu, _ := url.Parse(member)
+		memberURL := "http://" + strings.Split(uu.Host, ":")[0] + ":53380"
+		urls, err := types.NewURLs([]string{memberURL})
+		if err != nil {
+			return // we don't care about the error case as this method is only used for the unit testing.
+		}
+		mAPI.Add(ctx.Background(), memberURL)
+		u, _ := url.Parse(member)
+		m.urlsMap[strings.Split(u.Host, ".")[0]] = urls
+	}
+}
+
+type MockEmbeddedEtcdFactory struct {
+}
+
+func (m *MockEmbeddedEtcdFactory) NewEmbeddedEtcd(token string, conf *bootstrap.Config, newCluster bool) (*bootstrap.EmbeddedEtcd, error) {
+	return &bootstrap.EmbeddedEtcd{}, nil
+}
+
+type MockMembersAPI struct {
+	members []client.Member
+}
+
+func (m *MockMembersAPI) List(ctx ctx.Context) ([]client.Member, error) {
+	return m.members, nil
+}
+
+func (m *MockMembersAPI) Add(ctx ctx.Context, peerURL string) (*client.Member, error) {
+	hasher := md5.New()
+	hasher.Write([]byte(peerURL))
+	member := client.Member{PeerURLs: []string{peerURL}, ID: hex.EncodeToString(hasher.Sum(nil))}
+	m.members = append(m.members, member)
+	return &member, nil
+}
+
+func (m *MockMembersAPI) Remove(ctx ctx.Context, mID string) error {
+	for i, member := range m.members {
+		if member.ID == mID {
+			m.members = append(m.members[:i], m.members[i+1:]...)
+			return nil
+		}
+	}
+	return errors.New("member not found")
+}
+
+func (m *MockMembersAPI) Update(ctx ctx.Context, mID string, peerURLs []string) error {
+	return nil
+}
+
+func (m *MockMembersAPI) Leader(ctx ctx.Context) (*client.Member, error) {
+	return &client.Member{}, nil
+}

pkg/util/id.go

@@ -0,0 +1,31 @@
+package util
+
+import (
+	"io/ioutil"
+	"strings"
+)
+
+const maxMachineIDLength = 12
+
+func GetMachineID() (string, error) {
+	buf, err := ioutil.ReadFile("/etc/machine-id")
+	if err != nil {
+		return "", err
+	}
+
+	return trimMachineID(string(buf)), nil
+}
+
+func trimMachineID(id string) string {
+	// Trim the machine ID to a length that is statistically unlikely to collide with another node in the cluster
+	// while allowing us to use an ID that is both unique and succinct.
+	// Using the birthday collision algorithm, if we have a length of 12 hex characters, that gives us
+	// 16^12 possibilities. If we have a cluster with 1,000 nodes, we have a likelihood with node IDs
+	// colliding in less than 1 in a billion clusters.
+	id = strings.TrimSpace(id)
+	if len(id) <= maxMachineIDLength {
+		return id
+	}
+
+	return id[0:maxMachineIDLength]
+}

pkg/util/id_test.go

@@ -0,0 +1,21 @@
+package util
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTrimMachineID(t *testing.T) {
+	testTrimMachineID(t, " 123 		", "123")
+	testTrimMachineID(t, " 1234567890", "1234567890")
+	testTrimMachineID(t, " 123456789012", "123456789012")
+	testTrimMachineID(t, " 1234567890123", "123456789012")
+	testTrimMachineID(t, "1234567890123", "123456789012")
+	testTrimMachineID(t, "123456789012345678", "123456789012")
+}
+
+func testTrimMachineID(t *testing.T, input, expected string) {
+	result := trimMachineID(input)
+	assert.Equal(t, expected, result)
+}