diff --git a/glide.lock b/glide.lock index 515bfef1c..1f87e9bf5 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 03c94526023e69d4ddf4f795690b2d5e77c3de32415c17d668efe9e08b676d51 -updated: 2016-09-08T20:14:35.505788174-07:00 +hash: 81a5b4931ce8ec3eb293e6fe995ed8fc33a46da835d5012324cea5345d58083d +updated: 2016-09-09T10:18:59.830943693-07:00 imports: - name: bitbucket.org/ww/goautoneg version: 75cd24fc2f2c2a2088577d12123ddee5f54e0675 @@ -12,54 +12,54 @@ imports: - name: github.com/coreos/etcd version: a6d22b96c3813b546494cdc72160c7cb407c3297 subpackages: - - client - - error - - pkg/pathutil - - pkg/types - - etcdserver - - etcdserver/api/v2http - - pkg/cors - alarm - auth + - auth/authpb + - client - compactor - discovery + - error + - etcdserver - etcdserver/api + - etcdserver/api/v2http - etcdserver/api/v2http/httptypes + - etcdserver/auth - etcdserver/etcdserverpb - etcdserver/membership - etcdserver/stats - lease - lease/leasehttp + - lease/leasepb - mvcc - mvcc/backend - mvcc/mvccpb + - pkg/adt - pkg/contention + - pkg/cors + - pkg/crc - pkg/fileutil - pkg/httputil - pkg/idutil + - pkg/ioutil + - pkg/logutil - pkg/netutil + - pkg/pathutil - pkg/pbutil - pkg/runtime - pkg/schedule + - pkg/tlsutil - pkg/transport + - pkg/types - pkg/wait - raft - raft/raftpb - rafthttp - snap + - snap/snappb - store - version - wal - wal/walpb - - etcdserver/auth - - pkg/logutil - - auth/authpb - - lease/leasepb - - pkg/adt - - pkg/tlsutil - - pkg/ioutil - - snap/snappb - - pkg/crc - name: github.com/coreos/go-semver version: 568e959cd89871e61434c1143528d9162da89ef2 subpackages: @@ -67,9 +67,7 @@ imports: - name: github.com/coreos/go-systemd version: bfdc81d0d7e0fb19447b08571f63b774495251ce subpackages: - - daemon - journal - - util - name: github.com/coreos/pkg version: fa29b1d70f0beaddd4c7021607cc3c3be8ce94b8 subpackages: @@ -121,17 +119,17 @@ imports: - name: github.com/prometheus/procfs version: 454a56f35412459b5e684fd5ec0f9211b94f002a - name: github.com/quantum/clusterd - version: a315649618080b8a52479111c1e0ad229d9a36da + version: 186d5668fd1006c30f57a23aaf0fab1be2a5fe12 subpackages: - pkg/inventory - pkg/orchestrator - - pkg/util - pkg/proc + - pkg/util - name: github.com/quantum/etcdmgr version: f375501d312c13a3f0d2f3e1b8ae844524e43420 subpackages: - - manager - bootstrap + - manager - name: github.com/spf13/cobra version: dbb7c2d02e284aa24423e3e6e105969dffe6e4d6 subpackages: @@ -172,4 +170,8 @@ imports: - naming - peer - transport -testImports: [] +testImports: +- name: github.com/stretchr/testify + version: 9cc77fa25329013ce07362c7742952ff887361f2 + subpackages: + - assert diff --git a/glide.yaml b/glide.yaml index ad76f2f7b..10494f991 100644 --- a/glide.yaml +++ b/glide.yaml @@ -8,7 +8,7 @@ import: - client - error - package: github.com/quantum/clusterd - version: a315649618080b8a52479111c1e0ad229d9a36da + version: 186d5668fd1006c30f57a23aaf0fab1be2a5fe12 subpackages: - pkg/inventory - pkg/orchestrator @@ -18,4 +18,3 @@ import: - package: github.com/google/uuid - package: github.com/go-ini/ini - package: github.com/quantum/etcdmgr - version: f375501d312c13a3f0d2f3e1b8ae844524e43420 diff --git a/pkg/castled/cephleader.go b/pkg/castled/cephleader.go index da6328ea4..5cd74b844 100644 --- a/pkg/castled/cephleader.go +++ b/pkg/castled/cephleader.go @@ -15,8 +15,9 @@ import ( // Interface implemented by a service that has been elected leader type cephLeader struct { - cluster *clusterInfo - events chan orchestrator.LeaderEvent + cluster *clusterInfo + events chan orchestrator.LeaderEvent + mockCeph bool } func (c *cephLeader) StartWatchEvents() { @@ -52,6 +53,7 @@ func (c *cephLeader) handleOrchestratorEvents() { } else if _, ok := e.(*orchestrator.StaleNodeEvent); ok { // TODO: Move a monitor to another node and/or declare OSDs dead } + log.Printf("ceph leader completed event %s", e.Name()) } } @@ -66,7 +68,7 @@ func (c *cephLeader) configureCephServices(context *orchestrator.ClusterContext) } // Select the monitors, instruct them to start, and wait for quorum - err = createMonitors(context, c.cluster) + err = createMonitors(context, c.cluster, c.mockCeph) if err != nil { return err } diff --git a/pkg/castled/cephleader_test.go b/pkg/castled/cephleader_test.go new file mode 100644 index 000000000..3b52955ea --- /dev/null +++ b/pkg/castled/cephleader_test.go @@ -0,0 +1,94 @@ +package castled + +import ( + "log" + "testing" + + "github.com/quantum/clusterd/pkg/inventory" + "github.com/quantum/clusterd/pkg/orchestrator" + "github.com/quantum/clusterd/pkg/util" + "github.com/stretchr/testify/assert" +) + +// ************************************************************************************************ +// ************************************************************************************************ +// +// unit test functions +// +// ********************************"**************************************************************** +// ************************************************************************************************ +func TestCephLeaders(t *testing.T) { + leader := &cephLeader{mockCeph: true} + leader.StartWatchEvents() + defer leader.Close() + + nodes := make(map[string]*inventory.NodeConfig) + inv := &inventory.Config{Nodes: nodes} + nodes["a"] = &inventory.NodeConfig{IPAddress: "1.2.3.4"} + + etcdClient := util.NewMockEtcdClient() + context := &orchestrator.ClusterContext{ + EtcdClient: etcdClient, + Inventory: inv, + } + + // mock the agent responses that the deployments were successful to start mons and osds + etcdClient.WatcherResponses["/castle/_notify/a/monitor/status"] = "succeeded" + etcdClient.WatcherResponses["/castle/_notify/b/monitor/status"] = "succeeded" + etcdClient.WatcherResponses["/castle/_notify/a/osd/status"] = "succeeded" + etcdClient.WatcherResponses["/castle/_notify/b/osd/status"] = "succeeded" + + // trigger a refresh event + refresh := orchestrator.NewRefreshEvent(context) + leader.Events() <- refresh + + // wait for the event queue to be empty + 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")) + assert.Equal(t, "1.2.3.4", etcdClient.GetValue("/castle/services/ceph/monitor/desired/a/ipaddress")) + assert.Equal(t, "6790", etcdClient.GetValue("/castle/services/ceph/monitor/desired/a/port")) + + // trigger an add node event + nodes["b"] = &inventory.NodeConfig{IPAddress: "2.3.4.5"} + addNode := orchestrator.NewAddNodeEvent(context, []string{"b"}) + leader.Events() <- addNode + + // wait for the event queue to be empty + waitForEvents(leader) + + assert.True(t, etcdClient.GetChildDirs("/castle/services/ceph/osd/desired").Equals(util.CreateSet([]string{"a", "b"}))) +} + +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() *orchestrator.ClusterContext { + return nil +} diff --git a/pkg/castled/monleader.go b/pkg/castled/monleader.go index 888f7c49e..040357a0a 100644 --- a/pkg/castled/monleader.go +++ b/pkg/castled/monleader.go @@ -21,7 +21,8 @@ const ( // Create the ceph monitors // Must be idempotent -func createMonitors(context *orchestrator.ClusterContext, cluster *clusterInfo) error { +func createMonitors(context *orchestrator.ClusterContext, cluster *clusterInfo, skipQuorum bool) error { + log.Printf("Creating monitors with %d nodes available", len(context.Inventory.Nodes)) // Choose the nodes where the monitors will run var err error @@ -41,10 +42,12 @@ func createMonitors(context *orchestrator.ClusterContext, cluster *clusterInfo) return err } - // Wait for quorum - err = waitForQuorum(context, cluster) - if err != nil { - return err + if !skipQuorum { + // Wait for quorum + err = waitForQuorum(context, cluster) + if err != nil { + return err + } } return nil