simplify etcd packages

pkg/clusterd/join.go

@@ -6,7 +6,7 @@ import (
 	"strings"
 
 	"github.com/quantum/castle/pkg/clusterd/inventory"
-	"github.com/quantum/castle/pkg/etcdmgr/manager"
+	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
 	"github.com/quantum/castle/pkg/util"
 	"github.com/quantum/castle/pkg/util/exec"
 	"github.com/quantum/castle/pkg/util/proc"
@@ -25,7 +25,7 @@ func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, con
 
 		// use the discovery URL to query the etcdmgr what the etcd client endpoints should be
 		var err error
-		etcdClients, err = manager.GetEtcdClients(configDir, discoveryURL, privateIPv4, nodeID)
+		etcdClients, err = bootstrap.GetEtcdClients(configDir, discoveryURL, privateIPv4, nodeID)
 		if err != nil {
 			return nil, err
 		}

pkg/etcdmgr/agent.go

@@ -8,7 +8,6 @@ import (
 	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"
 )
@@ -79,7 +78,7 @@ func (e *etcdMgrAgent) CreateLocalService(context *clusterd.Context, desiredKey
 	if !etcdmgrApplied {
 		log.Println("adding the current node to the etcd cluster...")
 		targetEndpoint := getPeerEndpointFromIP(ipAddr)
-		err = manager.AddMember(e.context, targetEndpoint)
+		err = AddMember(e.context, targetEndpoint)
 		if err != nil {
 			return fmt.Errorf("error in adding a member to the cluster")
 		}
@@ -120,5 +119,5 @@ func getEtcdMgrAppliedKey(nodeID string) string {
 }
 
 func getPeerEndpointFromIP(ip string) string {
-	return "http://" + ip + ":" + bootstrap.DefaultPeerPort
+	return fmt.Sprintf("http://%s:%d", ip, bootstrap.DefaultPeerPort)
 }

pkg/etcdmgr/bootstrap/client.go

@@ -0,0 +1,54 @@
+package bootstrap
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"os"
+)
+
+// GetEtcdClients bootstraps an embedded etcd instance and returns a list of
+// current etcd cluster's client URLs. (entrypoint, when it's used as a library)
+func GetEtcdClients(configDir, token, ipAddr, nodeID string) ([]string, error) {
+	// GODEBUG setting forcing use of Go's resolver
+	os.Setenv("GODEBUG", "netdns=go+1")
+
+	full, err, currentNodes := isQuorumFull(token)
+	//currentNodes, err := GetCurrentNodesFromDiscovery(token)
+	if err != nil {
+		return []string{}, errors.New("error querying discovery service")
+	}
+	log.Println("current etcd cluster nodes: ", currentNodes)
+
+	localURL := fmt.Sprintf("http://%s:%d", ipAddr, DefaultClientPort)
+
+	// Is it a restart scenario?
+	restart := false
+	log.Println("current localURL: ", localURL)
+	for _, node := range currentNodes {
+		if node == localURL {
+			log.Println("restart scenario detected.")
+			restart = true
+		}
+	}
+
+	if full && !restart {
+		log.Println("quorum is already formed, returning current cluster members: ", currentNodes)
+		return currentNodes, nil
+	}
+
+	log.Println("quorum is not complete, creating a new embedded etcd member...")
+	conf, err := generateConfig(configDir, ipAddr, nodeID)
+	if err != nil {
+		return []string{}, err
+	}
+	log.Println("conf:", conf)
+
+	factory := EmbeddedEtcdFactory{}
+	ee, err := factory.NewEmbeddedEtcd(token, conf, true)
+	if err != nil {
+		return nil, err
+	}
+
+	return ee.Server.Cluster().ClientURLs(), nil
+}

pkg/etcdmgr/bootstrap/config.go

@@ -1,6 +1,7 @@
 package bootstrap
 
 import (
+	"fmt"
 	"log"
 	"net/url"
 	"path"
@@ -11,9 +12,9 @@ import (
 
 const (
 	// DefaultClientPort is the default port for listening to client traffic
-	DefaultClientPort = "53379"
+	DefaultClientPort = 53379
 	// DefaultPeerPort is the default port for listening to peer traffic
-	DefaultPeerPort = "53380"
+	DefaultPeerPort = 53380
 	// DefaultClientTimeout is the default timeout for etcd client
 	DefaultClientTimeout = 5 * time.Second
 )
@@ -44,8 +45,8 @@ func CopyConfig(conf *Config) Config {
 	return newConf
 }
 
-// GenerateConfig automatically generates a config for embedded etcd
-func GenerateConfig(configDir, ipAddr, nodeID string) (*Config, error) {
+// generateConfig automatically generates a config for embedded etcd
+func generateConfig(configDir, ipAddr, nodeID string) (*Config, error) {
 	conf := &Config{}
 	conf.InstanceName = nodeID
 	conf.ListenPeerURLs = append(conf.ListenPeerURLs,
@@ -64,7 +65,7 @@ func GenerateConfig(configDir, ipAddr, nodeID string) (*Config, error) {
 
 // GenerateConfigFromExistingCluster automatically generates a config for joining an existing cluster
 func GenerateConfigFromExistingCluster(Context EtcdMgrContext, configDir, ipAddr, nodeID string) (*Config, error) {
-	conf, err := GenerateConfig(configDir, ipAddr, nodeID)
+	conf, err := generateConfig(configDir, ipAddr, nodeID)
 	if err != nil {
 		return nil, err
 	}
@@ -81,10 +82,10 @@ func GenerateConfigFromExistingCluster(Context EtcdMgrContext, configDir, ipAddr
 	return conf, nil
 }
 
-func getURLFromSchemeIPPort(scheme, ip, port string) url.URL {
+func getURLFromSchemeIPPort(scheme, ip string, port int) url.URL {
 	u := url.URL{
 		Scheme: scheme,
-		Host:   ip + ":" + port,
+		Host:   fmt.Sprintf("%s:%d", ip, port),
 	}
 	return u
 }

pkg/etcdmgr/bootstrap/discovery.go

@@ -15,21 +15,8 @@ import (
 	"golang.org/x/net/context/ctxhttp"
 )
 
-// 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(token string) (bool, error, []string) {
-	res, err := QueryDiscoveryService(token + "/_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{}
 	}
@@ -47,8 +34,8 @@ func IsQuorumFull(token string) (bool, error, []string) {
 
 }
 
-// QueryDiscoveryService reads a key from a discovery url.
-func QueryDiscoveryService(token string) (*store.Event, error) {
+// queryDiscoveryService reads a key from a discovery url.
+func queryDiscoveryService(token string) (*store.Event, error) {
 	ctx, _ := context.WithTimeout(context.Background(), DefaultClientTimeout)
 	resp, err := ctxhttp.Get(ctx, http.DefaultClient, token)
 	if err != nil {
@@ -70,25 +57,28 @@ func QueryDiscoveryService(token string) (*store.Event, error) {
 }
 
 func GetCurrentNodesFromDiscovery(token string) ([]string, error) {
-	res, err := QueryDiscoveryService(token)
+	res, err := queryDiscoveryService(token)
 	if err != nil {
 		return nil, err
 	}
+
 	nodes := make([]string, 0, len(res.Node.Nodes))
 	for _, nn := range res.Node.Nodes {
 		if nn.Value == nil {
 			log.Printf("Skipping %q because no value exists", nn.Key)
 		}
-		clientPort, _ := strconv.Atoi(DefaultClientPort)
-		n, err := newDiscoveryNode(*nn.Value, clientPort)
+
+		n, err := newDiscoveryNode(*nn.Value, DefaultClientPort)
 		if err != nil {
 			log.Printf("invalid peer url %q in discovery service: %v", *nn.Value, err)
 			continue
 		}
+
 		for _, node := range n {
 			nodes = append(nodes, node)
 		}
 	}
+
 	return nodes, nil
 }
 

pkg/etcdmgr/leader.go

@@ -12,7 +12,6 @@ import (
 	"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"
 )
@@ -185,7 +184,7 @@ func (e *etcdMgrLeader) shrinkEtcdQuorum(context *clusterd.Context, candidates [
 			return fmt.Errorf("error in removing desired key for node: %+v err: %+v\n", candidate, err)
 		}
 
-		err = manager.RemoveMember(e.context, targetEndpoint)
+		err = RemoveMember(e.context, targetEndpoint)
 		if err != nil {
 			return fmt.Errorf("error in removing a member from the cluster. %+v\n", err)
 		}

pkg/etcdmgr/manager/etcdmgr.go

@@ -1,149 +0,0 @@
-package manager
-
-import (
-	"errors"
-	"fmt"
-	"log"
-	"os"
-
-	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
-	ctx "golang.org/x/net/context"
-)
-
-// GetEtcdClients bootstraps an embedded etcd instance and returns a list of
-// current etcd cluster's client URLs. (entrypoint, when it's used as a library)
-func GetEtcdClients(configDir, 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 querying discovery service")
-	}
-	log.Println("current etcd cluster nodes: ", currentNodes)
-
-	localURL := "http://" + ipAddr + ":" + bootstrap.DefaultClientPort
-	// Is it a restart scenario?
-	restart := false
-	log.Println("current localURL: ", localURL)
-	for _, node := range currentNodes {
-		if node == localURL {
-			log.Println("restart scenario detected.")
-			restart = true
-		}
-	}
-
-	if full && !restart {
-		log.Println("quorum is already formed, returning current cluster members: ", currentNodes)
-		return currentNodes, nil
-	}
-
-	log.Println("quorum is not complete, creating a new embedded etcd member...")
-	conf, err := bootstrap.GenerateConfig(configDir, ipAddr, nodeID)
-	if err != nil {
-		return []string{}, err
-	}
-	log.Println("conf:", conf)
-
-	factory := bootstrap.EmbeddedEtcdFactory{}
-	ee, err := factory.NewEmbeddedEtcd(token, conf, true)
-	if err != nil {
-		return nil, err
-	}
-
-	return ee.Server.Cluster().ClientURLs(), nil
-}
-
-// 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.Config) ([]string, error) {
-	// GODEBUG setting forcing use of Go's resolver
-	os.Setenv("GODEBUG", "netdns=go+1")
-	full, err, currentNodes := bootstrap.IsQuorumFull(token)
-	if err != nil {
-		return []string{}, errors.New("error in querying discovery service")
-	}
-
-	//TODO refactor + we assumed we only have one advertiseClient
-	localURL := "http://" + config.AdvertiseClientURLs[0].Host
-	// Is it a restart scenario?
-	restart := false
-	log.Println("current localURL: ", localURL)
-	for _, node := range currentNodes {
-		if node == localURL {
-			log.Println("restart scenario detected.")
-			restart = true
-		}
-	}
-	if full && !restart {
-		log.Println("quorum is full, returning current quorum nodes...")
-		log.Println("current nodes: ", currentNodes)
-		return currentNodes, nil
-	}
-
-	log.Println("creating a new embedded etcd")
-	factory := bootstrap.EmbeddedEtcdFactory{}
-	ee, err := factory.NewEmbeddedEtcd(token, &config, true)
-	if err != nil {
-		return nil, err
-	}
-	return ee.Server.Cluster().ClientURLs(), nil
-}
-
-// 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 {
-		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
-		}
-	}
-	if nodeID == "" {
-		return errors.New("node not found")
-	}
-	err = mapi.Remove(ctx.Background(), nodeID)
-	if err != nil {
-		return err
-	}
-	log.Println("New member removed: ", node)
-	return nil
-}
-
-func sliceContains(slice []string, str string) bool {
-	for _, s := range slice {
-		if s == str {
-			return true
-		}
-	}
-	return false
-}

pkg/etcdmgr/member.go

@@ -0,0 +1,67 @@
+package etcdmgr
+
+import (
+	"errors"
+	"fmt"
+	"log"
+
+	"github.com/quantum/castle/pkg/etcdmgr/bootstrap"
+	ctx "golang.org/x/net/context"
+)
+
+// 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 {
+		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
+		}
+	}
+	if nodeID == "" {
+		return errors.New("node not found")
+	}
+	err = mapi.Remove(ctx.Background(), nodeID)
+	if err != nil {
+		return err
+	}
+	log.Println("New member removed: ", node)
+	return nil
+}
+
+func sliceContains(slice []string, str string) bool {
+	for _, s := range slice {
+		if s == str {
+			return true
+		}
+	}
+	return false
+}

pkg/etcdmgr/policy/policy.go

@@ -33,6 +33,7 @@ func ChooseEtcdCandidatesToAddOrRemove(delta int, currentEtcdNodes, currentNodes
 			sort.Strings(candidates)
 			return candidates[:numOfNodeToRemove], nil
 		}
+
 		//need more candidates from etcd nodes
 		numOfNodeToRemove = numOfNodeToRemove - len(candidates)
 		moreCandidates := util.SetDifference(currentEtcdNodes, candidates).ToSlice()
@@ -44,6 +45,7 @@ func ChooseEtcdCandidatesToAddOrRemove(delta int, currentEtcdNodes, currentNodes
 		}
 		candidates = append(candidates, moreCandidates...)
 	}
+
 	sort.Strings(candidates)
 	return candidates, nil
 }