From f1d56803bbf134a9534769e9bb7e97a55357beba Mon Sep 17 00:00:00 2001 From: Mehdi Azarmi Date: Fri, 23 Sep 2016 10:13:55 -0700 Subject: [PATCH] Dynamic etcd cluster resizing added. --- cmd/castled/castled.go | 6 +- pkg/cephmgr/cephleader_test.go | 38 +---- pkg/cephmgr/monagent.go | 1 - pkg/cephmgr/osdagent.go | 1 - pkg/clusterd/agent.go | 4 +- pkg/clusterd/clusterMember.go | 26 ---- pkg/clusterd/clusterMember_test.go | 13 -- pkg/clusterd/join.go | 13 +- pkg/clusterd/status.go | 14 +- pkg/clusterd/test/nonevent.go | 39 +++++ pkg/etcdmgr/bootstrap/config.go | 72 ++++++++- pkg/etcdmgr/bootstrap/discovery.go | 29 ++-- pkg/etcdmgr/bootstrap/etcd.go | 85 ++++++---- pkg/etcdmgr/bootstrap/etcdcontext.go | 92 +++++++++++ pkg/etcdmgr/etcdmgr.go | 25 +++ pkg/etcdmgr/etcdmgragent.go | 124 +++++++++++++++ pkg/etcdmgr/etcdmgragent_test.go | 45 ++++++ pkg/etcdmgr/etcdmgrleader.go | 224 +++++++++++++++++++++++++++ pkg/etcdmgr/etcdmgrleader_test.go | 95 ++++++++++++ pkg/etcdmgr/manager/etcdmgr.go | 126 +++++++-------- pkg/etcdmgr/policy/policy.go | 62 ++++++++ pkg/etcdmgr/test/mocketcdmgr.go | 118 ++++++++++++++ pkg/util/id.go | 31 ++++ pkg/util/id_test.go | 21 +++ 24 files changed, 1101 insertions(+), 203 deletions(-) create mode 100644 pkg/clusterd/test/nonevent.go create mode 100644 pkg/etcdmgr/bootstrap/etcdcontext.go create mode 100644 pkg/etcdmgr/etcdmgr.go create mode 100644 pkg/etcdmgr/etcdmgragent.go create mode 100644 pkg/etcdmgr/etcdmgragent_test.go create mode 100644 pkg/etcdmgr/etcdmgrleader.go create mode 100644 pkg/etcdmgr/etcdmgrleader_test.go create mode 100644 pkg/etcdmgr/policy/policy.go create mode 100644 pkg/etcdmgr/test/mocketcdmgr.go create mode 100644 pkg/util/id.go create mode 100644 pkg/util/id_test.go diff --git a/cmd/castled/castled.go b/cmd/castled/castled.go index ac7070f86..3032c5d29 100644 --- a/cmd/castled/castled.go +++ b/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() diff --git a/pkg/cephmgr/cephleader_test.go b/pkg/cephmgr/cephleader_test.go index 8e9b50a00..138b10b33 100644 --- a/pkg/cephmgr/cephleader_test.go +++ b/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) diff --git a/pkg/cephmgr/monagent.go b/pkg/cephmgr/monagent.go index ba35e106f..edc081aad 100644 --- a/pkg/cephmgr/monagent.go +++ b/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) diff --git a/pkg/cephmgr/osdagent.go b/pkg/cephmgr/osdagent.go index fc9256938..87879317a 100644 --- a/pkg/cephmgr/osdagent.go +++ b/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) diff --git a/pkg/clusterd/agent.go b/pkg/clusterd/agent.go index b30292816..94239fa42 100644 --- a/pkg/clusterd/agent.go +++ b/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) diff --git a/pkg/clusterd/clusterMember.go b/pkg/clusterd/clusterMember.go index 99d030c10..47213a532 100644 --- a/pkg/clusterd/clusterMember.go +++ b/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 diff --git a/pkg/clusterd/clusterMember_test.go b/pkg/clusterd/clusterMember_test.go index 09006c6ab..e0968c3d8 100644 --- a/pkg/clusterd/clusterMember_test.go +++ b/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() diff --git a/pkg/clusterd/join.go b/pkg/clusterd/join.go index d5662cec4..4726e69d1 100644 --- a/pkg/clusterd/join.go +++ b/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 { diff --git a/pkg/clusterd/status.go b/pkg/clusterd/status.go index 2f295fde1..63b4d2e79 100644 --- a/pkg/clusterd/status.go +++ b/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) diff --git a/pkg/clusterd/test/nonevent.go b/pkg/clusterd/test/nonevent.go new file mode 100644 index 000000000..61cbc9465 --- /dev/null +++ b/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 +} diff --git a/pkg/etcdmgr/bootstrap/config.go b/pkg/etcdmgr/bootstrap/config.go index 96999fa67..65457ade3 100644 --- a/pkg/etcdmgr/bootstrap/config.go +++ b/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 +} diff --git a/pkg/etcdmgr/bootstrap/discovery.go b/pkg/etcdmgr/bootstrap/discovery.go index ff889a879..6b47fc0cd 100644 --- a/pkg/etcdmgr/bootstrap/discovery.go +++ b/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 diff --git a/pkg/etcdmgr/bootstrap/etcd.go b/pkg/etcdmgr/bootstrap/etcd.go index ef8f6b46f..1c9d7b679 100644 --- a/pkg/etcdmgr/bootstrap/etcd.go +++ b/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, + } +} diff --git a/pkg/etcdmgr/bootstrap/etcdcontext.go b/pkg/etcdmgr/bootstrap/etcdcontext.go new file mode 100644 index 000000000..61c4288ae --- /dev/null +++ b/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 +} diff --git a/pkg/etcdmgr/etcdmgr.go b/pkg/etcdmgr/etcdmgr.go new file mode 100644 index 000000000..1beadb538 --- /dev/null +++ b/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{}}, + }, + } +} diff --git a/pkg/etcdmgr/etcdmgragent.go b/pkg/etcdmgr/etcdmgragent.go new file mode 100644 index 000000000..0c2360d33 --- /dev/null +++ b/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 +} diff --git a/pkg/etcdmgr/etcdmgragent_test.go b/pkg/etcdmgr/etcdmgragent_test.go new file mode 100644 index 000000000..0f3b09225 --- /dev/null +++ b/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) +} diff --git a/pkg/etcdmgr/etcdmgrleader.go b/pkg/etcdmgr/etcdmgrleader.go new file mode 100644 index 000000000..3ee258b23 --- /dev/null +++ b/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 +} diff --git a/pkg/etcdmgr/etcdmgrleader_test.go b/pkg/etcdmgr/etcdmgrleader_test.go new file mode 100644 index 000000000..c4ea09169 --- /dev/null +++ b/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) +} diff --git a/pkg/etcdmgr/manager/etcdmgr.go b/pkg/etcdmgr/manager/etcdmgr.go index 5d9b36645..577b73158 100644 --- a/pkg/etcdmgr/manager/etcdmgr.go +++ b/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 } diff --git a/pkg/etcdmgr/policy/policy.go b/pkg/etcdmgr/policy/policy.go new file mode 100644 index 000000000..d2cc7c7b0 --- /dev/null +++ b/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 +} diff --git a/pkg/etcdmgr/test/mocketcdmgr.go b/pkg/etcdmgr/test/mocketcdmgr.go new file mode 100644 index 000000000..341cf0b41 --- /dev/null +++ b/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 +} diff --git a/pkg/util/id.go b/pkg/util/id.go new file mode 100644 index 000000000..f635245bc --- /dev/null +++ b/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] +} diff --git a/pkg/util/id_test.go b/pkg/util/id_test.go new file mode 100644 index 000000000..17360744d --- /dev/null +++ b/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) +}