From 97b6a1d70db274ec9bf2ac4e579f10a57c2eaa80 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Wed, 19 Oct 2016 23:30:20 -0700 Subject: [PATCH] simplify etcd packages --- pkg/clusterd/join.go | 4 +- pkg/etcdmgr/{etcdmgragent.go => agent.go} | 5 +- .../{etcdmgragent_test.go => agent_test.go} | 0 pkg/etcdmgr/bootstrap/client.go | 54 +++++++ pkg/etcdmgr/bootstrap/config.go | 15 +- .../bootstrap/{etcdcontext.go => context.go} | 0 pkg/etcdmgr/bootstrap/discovery.go | 30 ++-- pkg/etcdmgr/{etcdmgrleader.go => leader.go} | 3 +- .../{etcdmgrleader_test.go => leader_test.go} | 0 pkg/etcdmgr/manager/etcdmgr.go | 149 ------------------ pkg/etcdmgr/member.go | 67 ++++++++ pkg/etcdmgr/policy/policy.go | 2 + 12 files changed, 146 insertions(+), 183 deletions(-) rename pkg/etcdmgr/{etcdmgragent.go => agent.go} (95%) rename pkg/etcdmgr/{etcdmgragent_test.go => agent_test.go} (100%) create mode 100644 pkg/etcdmgr/bootstrap/client.go rename pkg/etcdmgr/bootstrap/{etcdcontext.go => context.go} (100%) rename pkg/etcdmgr/{etcdmgrleader.go => leader.go} (98%) rename pkg/etcdmgr/{etcdmgrleader_test.go => leader_test.go} (100%) delete mode 100644 pkg/etcdmgr/manager/etcdmgr.go create mode 100644 pkg/etcdmgr/member.go diff --git a/pkg/clusterd/join.go b/pkg/clusterd/join.go index a4d16ba0e..d8eab32da 100644 --- a/pkg/clusterd/join.go +++ b/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 } diff --git a/pkg/etcdmgr/etcdmgragent.go b/pkg/etcdmgr/agent.go similarity index 95% rename from pkg/etcdmgr/etcdmgragent.go rename to pkg/etcdmgr/agent.go index 01815505a..88f9f6621 100644 --- a/pkg/etcdmgr/etcdmgragent.go +++ b/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) } diff --git a/pkg/etcdmgr/etcdmgragent_test.go b/pkg/etcdmgr/agent_test.go similarity index 100% rename from pkg/etcdmgr/etcdmgragent_test.go rename to pkg/etcdmgr/agent_test.go diff --git a/pkg/etcdmgr/bootstrap/client.go b/pkg/etcdmgr/bootstrap/client.go new file mode 100644 index 000000000..a4afb69dd --- /dev/null +++ b/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 +} diff --git a/pkg/etcdmgr/bootstrap/config.go b/pkg/etcdmgr/bootstrap/config.go index 9b64a6646..f866bd13c 100644 --- a/pkg/etcdmgr/bootstrap/config.go +++ b/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 } diff --git a/pkg/etcdmgr/bootstrap/etcdcontext.go b/pkg/etcdmgr/bootstrap/context.go similarity index 100% rename from pkg/etcdmgr/bootstrap/etcdcontext.go rename to pkg/etcdmgr/bootstrap/context.go diff --git a/pkg/etcdmgr/bootstrap/discovery.go b/pkg/etcdmgr/bootstrap/discovery.go index 6b47fc0cd..0850a557e 100644 --- a/pkg/etcdmgr/bootstrap/discovery.go +++ b/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 } diff --git a/pkg/etcdmgr/etcdmgrleader.go b/pkg/etcdmgr/leader.go similarity index 98% rename from pkg/etcdmgr/etcdmgrleader.go rename to pkg/etcdmgr/leader.go index 3ee258b23..a991189ef 100644 --- a/pkg/etcdmgr/etcdmgrleader.go +++ b/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) } diff --git a/pkg/etcdmgr/etcdmgrleader_test.go b/pkg/etcdmgr/leader_test.go similarity index 100% rename from pkg/etcdmgr/etcdmgrleader_test.go rename to pkg/etcdmgr/leader_test.go diff --git a/pkg/etcdmgr/manager/etcdmgr.go b/pkg/etcdmgr/manager/etcdmgr.go deleted file mode 100644 index b3d0e760a..000000000 --- a/pkg/etcdmgr/manager/etcdmgr.go +++ /dev/null @@ -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 -} diff --git a/pkg/etcdmgr/member.go b/pkg/etcdmgr/member.go new file mode 100644 index 000000000..d2123b0c0 --- /dev/null +++ b/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 +} diff --git a/pkg/etcdmgr/policy/policy.go b/pkg/etcdmgr/policy/policy.go index d2cc7c7b0..cbd70ed7d 100644 --- a/pkg/etcdmgr/policy/policy.go +++ b/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 }