From ed05bcb2032242561b767927fd26296162a12bb0 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Fri, 30 Sep 2016 16:12:04 -0700 Subject: [PATCH 1/3] remove a bad osd from crushmap --- pkg/cephmgr/osd.go | 130 ++++++++++++++++++++++++----------- pkg/cephmgr/osdagent.go | 110 ++++++++++++++++++++++------- pkg/cephmgr/osdagent_test.go | 14 ++++ 3 files changed, 186 insertions(+), 68 deletions(-) diff --git a/pkg/cephmgr/osd.go b/pkg/cephmgr/osd.go index 63876b37d..f434ea17b 100644 --- a/pkg/cephmgr/osd.go +++ b/pkg/cephmgr/osd.go @@ -125,24 +125,20 @@ func createOSDBootstrapKeyring(conn client.Connection, clusterName string) error } // get-or-create-key for client.bootstrap-osd - cmd := "auth get-or-create-key" - command, err := json.Marshal(map[string]interface{}{ - "prefix": cmd, - "format": "json", + cmd := map[string]interface{}{ + "prefix": "auth get-or-create-key", "entity": "client.bootstrap-osd", "caps": []string{"mon", "allow profile bootstrap-osd"}, - }) - if err != nil { - return fmt.Errorf("command %s marshall failed: %+v", cmd, err) } - buf, _, err := conn.MonCommand(command) + buf, err := executeMonCommand(conn, cmd, "create osd bootstrap key") if err != nil { - return fmt.Errorf("mon_command %s failed: %+v", cmd, err) + return fmt.Errorf("failed to create osd bootstrap key: %+v", err) } + var resp map[string]interface{} err = json.Unmarshal(buf, &resp) if err != nil { - return fmt.Errorf("failed to unmarshall %s response: %+v", cmd, err) + return fmt.Errorf("failed to unmarshal %s response: %+v", cmd, err) } bootstrapOSDKey := resp["key"].(string) log.Printf("succeeded %s command, bootstrapOSDKey: %s", cmd, bootstrapOSDKey) @@ -405,20 +401,16 @@ func initializeOSD(factory client.ConnectionFactory, context *clusterd.Context, // creates the OSD identity in the cluster via a mon_command func createOSD(bootstrapConn client.Connection, osdUUID uuid.UUID) (int, error) { - cmd := "osd create" - command, err := json.Marshal(map[string]interface{}{ - "prefix": cmd, - "format": "json", + cmd := map[string]interface{}{ + "prefix": "osd create", "entity": "client.bootstrap-osd", "uuid": osdUUID.String(), - }) - if err != nil { - return 0, fmt.Errorf("command %s marshall failed: %+v", cmd, err) } - buf, _, err := bootstrapConn.MonCommand(command) + buf, err := executeMonCommand(bootstrapConn, cmd, fmt.Sprintf("create osd %s", osdUUID)) if err != nil { - return 0, fmt.Errorf("mon_command %s failed: %+v", cmd, err) + return 0, fmt.Errorf("failed to create osd %s: %+v", osdUUID, err) } + var resp map[string]interface{} err = json.Unmarshal(buf, &resp) if err != nil { @@ -430,18 +422,14 @@ func createOSD(bootstrapConn client.Connection, osdUUID uuid.UUID) (int, error) // gets the current mon map for the cluster func getMonMap(bootstrapConn client.Connection) ([]byte, error) { - cmd := "mon getmap" - command, err := json.Marshal(map[string]interface{}{ - "prefix": cmd, - "format": "json", + cmd := map[string]interface{}{ + "prefix": "mon getmap", "entity": "client.bootstrap-osd", - }) - if err != nil { - return nil, fmt.Errorf("command %s marshall failed: %+v", cmd, err) } - buf, _, err := bootstrapConn.MonCommand(command) + + buf, err := executeMonCommand(bootstrapConn, cmd, "get mon map") if err != nil { - return nil, fmt.Errorf("mon_command %s failed: %+v", cmd, err) + return nil, fmt.Errorf("failed to get mon map: %+v", cmd, err) } return buf, nil } @@ -532,27 +520,17 @@ func addOSDToCrushMap(osdConn client.Connection, context *clusterd.Context, osdI return err } - cmd := "osd crush create-or-move" - command, err := json.Marshal(map[string]interface{}{ - "prefix": cmd, - "format": "json", + cmd := map[string]interface{}{ + "prefix": "osd crush create-or-move", "id": osdID, "weight": weight, "args": locArgs, - }) - if err != nil { - return fmt.Errorf("command %s marshall failed: %+v", cmd, err) } - - log.Printf("%s command: '%s'", cmd, string(command)) - - _, info, err := osdConn.MonCommand(command) + _, err = executeMonCommand(osdConn, cmd, fmt.Sprintf("adding %s to crush map", osdEntity)) if err != nil { - return fmt.Errorf("mon_command %s failed: %+v", cmd, err) + return fmt.Errorf("failed adding %s to crush map: %+v", osdEntity, err) } - log.Printf("succeeded adding %s to crush map. info: %s", osdEntity, info) - if err := inventory.SetLocation(context.EtcdClient, context.NodeID, strings.Join(locArgs, ",")); err != nil { return fmt.Errorf("failed to save CRUSH location for OSD %s: %+v", osdEntity, err) } @@ -560,6 +538,74 @@ func addOSDToCrushMap(osdConn client.Connection, context *clusterd.Context, osdI return nil } +func executeMonCommand(connection client.Connection, cmd map[string]interface{}, message string) ([]byte, error) { + // ensure the json attribute is included in the request + cmd["format"] = "json" + + prefix, ok := cmd["prefix"] + if !ok { + return nil, fmt.Errorf("missing prefix for the mon_command") + } + + command, err := json.Marshal(cmd) + if err != nil { + return nil, fmt.Errorf("marshalling command %s failed: %+v", prefix, err) + } + + log.Printf("mon_command: '%s'", string(command)) + + response, info, err := connection.MonCommand(command) + if err != nil { + return nil, fmt.Errorf("mon_command %s failed: %+v", cmd, err) + } + + log.Printf("succeeded %s. info: %s", message, info) + return response, nil +} + +func markOSDOut(connection client.Connection, id int) error { + command := map[string]interface{}{ + "prefix": "osd out", + "ids": []int{id}, + } + _, err := executeMonCommand(connection, command, fmt.Sprintf("mark osd %d out", id)) + return err +} + +func purgeOSD(connection client.Connection, name string, id int) error { + // ceph osd crush remove + command := map[string]interface{}{ + "prefix": "osd crush remove", + "name": fmt.Sprintf("osd.%d", id), + } + _, err := executeMonCommand(connection, command, fmt.Sprintf("remove osd %s from crush map", name)) + if err != nil { + return fmt.Errorf("failed to remove osd %s from crush map. %v", name, err) + } + + // ceph auth del osd.$osd_num + command = map[string]interface{}{ + "prefix": "auth del", + "entity": fmt.Sprintf("osd.%d", id), + } + _, err = executeMonCommand(connection, command, fmt.Sprintf("delete auth for osd %s", name)) + if err != nil { + return fmt.Errorf("failed to delete auth for osd %s. %v", name, err) + } + + // ceph osd rm $osd_num + command = map[string]interface{}{ + "prefix": "osd rm", + "ids": []int{id}, + } + _, err = executeMonCommand(connection, command, fmt.Sprintf("rm osds %v", name)) + if err != nil { + return fmt.Errorf("failed to rm osd %s. %v", name, err) + } + + return nil +} + func formatLocation(location string) ([]string, error) { var pairs []string if location == "" { diff --git a/pkg/cephmgr/osdagent.go b/pkg/cephmgr/osdagent.go index 645122524..9f5c42f4f 100644 --- a/pkg/cephmgr/osdagent.go +++ b/pkg/cephmgr/osdagent.go @@ -29,16 +29,19 @@ const ( ) type osdAgent struct { - cluster *ClusterInfo - forceFormat bool - location string - factory client.ConnectionFactory - osdCmd map[string]*exec.Cmd - devices string + cluster *ClusterInfo + forceFormat bool + location string + factory client.ConnectionFactory + osdCmd map[string]*exec.Cmd + devices string + getIDFromName func(context *clusterd.Context, name string) (int, error) } func newOSDAgent(factory client.ConnectionFactory, devices string, forceFormat bool, location string) *osdAgent { - return &osdAgent{factory: factory, devices: devices, forceFormat: forceFormat, location: location} + a := &osdAgent{factory: factory, devices: devices, forceFormat: forceFormat, location: location} + a.getIDFromName = a.getOSDID + return a } func (a *osdAgent) Name() string { @@ -133,33 +136,88 @@ func (a *osdAgent) stopUndesiredDevices(context *clusterd.Context, connection cl continue } - cmd, ok := a.osdCmd[device] - if ok { - // stop the osd process - err = context.ProcMan.Stop(cmd) - if err != nil { - log.Printf("failed to stop osd for device %s. %v", device, err) - lastErr = err - continue - } - - delete(a.osdCmd, device) - } - - appliedKey := path.Join(getAppliedKey(context.NodeID), devicesKey, device) - _, err = context.EtcdClient.Delete(ctx.Background(), appliedKey, &etcd.DeleteOptions{Recursive: true, Dir: true}) + err := a.removeOSD(context, connection, device) if err != nil { - log.Printf("failed to remove device %s from desired state. %v", device, err) lastErr = err - continue } - - log.Printf("Stopped and removed osd device %s", device) } return lastErr } +func (a *osdAgent) getOSDID(context *clusterd.Context, name string) (int, error) { + mountPoint, err := inventory.GetDeviceMountPoint(name, context.Executor) + if err != nil { + return -1, fmt.Errorf("unable to get mount point to remove device %s: %+v", name, err) + } + + if mountPoint == "" { + return -1, fmt.Errorf("mount point not found for osd %s", name) + } + + // find the OSD data path (under the mount point/data dir) + osdDataPath, err := findOSDDataPath(mountPoint, a.cluster.Name) + if err != nil { + return -1, err + } + + var osdID int + if _, err := os.Stat(filepath.Join(osdDataPath, "whoami")); os.IsNotExist(err) { + return -1, fmt.Errorf("cannot identify osd %s to remove", name) + } + + // osd_data_dir/whoami already exists, meaning the OSD is already set up. + // look up some basic information about it so we can run it. + osdID, _, err = getOSDInfo(osdDataPath) + if err != nil { + return -1, fmt.Errorf("failed to get OSD information from %s: %+v", osdDataPath, err) + } + + return osdID, nil +} + +func (a *osdAgent) removeOSD(context *clusterd.Context, connection client.Connection, name string) error { + osdID, err := a.getIDFromName(context, name) + if err != nil { + return fmt.Errorf("failed to get osd %s id. %v", name, err) + } + + // mark the OSD as out of the cluster so its data starts to migrate + err = markOSDOut(connection, osdID) + if err != nil { + return fmt.Errorf("failed to mark out osd %s (%d). %v", name, osdID, err) + } + + // stop the osd process if running + cmd, ok := a.osdCmd[name] + if ok { + err := context.ProcMan.Stop(cmd) + if err != nil { + log.Printf("failed to stop osd for device %s. %v", name, err) + return err + } + + delete(a.osdCmd, name) + } + + err = purgeOSD(connection, name, osdID) + if err != nil { + return fmt.Errorf("faild to remove osd %s from crush map. %v", name, err) + } + + // remove the osd from the applied key + appliedKey := path.Join(getAppliedKey(context.NodeID), devicesKey, name) + _, err = context.EtcdClient.Delete(ctx.Background(), appliedKey, &etcd.DeleteOptions{Recursive: true, Dir: true}) + if err != nil { + log.Printf("failed to remove device %s from desired state. %v", name, err) + return err + } + + log.Printf("Stopped and removed osd device %s", name) + + return nil +} + func (a *osdAgent) DestroyLocalService(context *clusterd.Context) error { // stop the OSD processes for device, cmd := range a.osdCmd { diff --git a/pkg/cephmgr/osdagent_test.go b/pkg/cephmgr/osdagent_test.go index e2e54c7d9..372897277 100644 --- a/pkg/cephmgr/osdagent_test.go +++ b/pkg/cephmgr/osdagent_test.go @@ -207,6 +207,7 @@ func TestRemoveDevice(t *testing.T) { procTrap := func(action string, c *exec.Cmd) error { return nil } + context := &clusterd.Context{EtcdClient: etcdClient, NodeID: "a", Executor: executor, ProcMan: &proc.ProcManager{Trap: procTrap}} desired := util.CreateSet([]string{"sda"}) @@ -216,6 +217,19 @@ func TestRemoveDevice(t *testing.T) { etcdClient.SetValue(path.Join(root, "sdb/serial"), "456") etcdClient.SetValue(path.Join(root, "sdc/serial"), "789") + agent.getIDFromName = func(context *clusterd.Context, name string) (int, error) { + if name == "sda" { + return 1, nil + } + if name == "sdb" { + return 2, nil + } + if name == "sdc" { + return 3, nil + } + return -1, fmt.Errorf("unknown name %s", name) + } + err := agent.stopUndesiredDevices(context, conn, desired) assert.Nil(t, err) applied := etcdClient.GetChildDirs(root) From d820bf5ff6f3f0d1654a9483966771db219d52a8 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Mon, 3 Oct 2016 14:21:10 -0700 Subject: [PATCH 2/3] rest api for querying the crush map --- pkg/api/handlers.go | 23 +++++++++++++++++++++++ pkg/api/routes.go | 6 ++++++ pkg/cephmgr/mon.go | 11 ++--------- pkg/cephmgr/osd.go | 11 +++++++++++ 4 files changed, 42 insertions(+), 9 deletions(-) diff --git a/pkg/api/handlers.go b/pkg/api/handlers.go index 9e2a0a5cf..d559b3233 100644 --- a/pkg/api/handlers.go +++ b/pkg/api/handlers.go @@ -172,6 +172,29 @@ func loadDeviceFromBody(w http.ResponseWriter, r *http.Request) (*cephmgr.Device return &device, nil } +// Gets the current crush map for the cluster. +// GET +// /crushmap +func (h *Handler) GetCrushMap(w http.ResponseWriter, r *http.Request) { + // connect to ceph + conn, ok := h.connectToCeph(w) + if !ok { + return + } + defer conn.Shutdown() + + // get the monitor status + crushmap, err := cephmgr.GetCrushMap(conn) + if err != nil { + log.Printf("failed to get crush map, err: %+v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json; charset=UTF-8") + w.Write([]byte(crushmap)) +} + // Gets the monitors that have been created in this cluster. // GET // /mon diff --git a/pkg/api/routes.go b/pkg/api/routes.go index e796abe38..f39287783 100644 --- a/pkg/api/routes.go +++ b/pkg/api/routes.go @@ -26,6 +26,12 @@ func (h *Handler) GetRoutes() []Route { "/mon", h.GetMonitors, }, + { + "GetCrushMap", + "GET", + "/crushmap", + h.GetCrushMap, + }, { "AddDevice", "POST", diff --git a/pkg/cephmgr/mon.go b/pkg/cephmgr/mon.go index bb6fadd18..2b39bed09 100644 --- a/pkg/cephmgr/mon.go +++ b/pkg/cephmgr/mon.go @@ -78,15 +78,8 @@ type MonMapEntry struct { // calls mon_status mon_command func GetMonStatus(adminConn client.Connection) (MonStatusResponse, error) { - monCommand := "mon_status" - command, err := json.Marshal(map[string]string{"prefix": monCommand, "format": "json"}) - if err != nil { - return MonStatusResponse{}, fmt.Errorf("command %s marshall failed: %+v", monCommand, err) - } - buf, _, err := adminConn.MonCommand(command) - if err != nil { - return MonStatusResponse{}, fmt.Errorf("mon_command failed: %+v", err) - } + cmd := map[string]interface{}{"prefix": "mon_status"} + buf, err := executeMonCommand(adminConn, cmd, fmt.Sprintf("retrieving mon_status")) var resp MonStatusResponse err = json.Unmarshal(buf, &resp) diff --git a/pkg/cephmgr/osd.go b/pkg/cephmgr/osd.go index f434ea17b..b37491842 100644 --- a/pkg/cephmgr/osd.go +++ b/pkg/cephmgr/osd.go @@ -606,6 +606,17 @@ func purgeOSD(connection client.Connection, name string, id int) error { return nil } +// calls osd getcrushmap +func GetCrushMap(adminConn client.Connection) (string, error) { + cmd := map[string]interface{}{"prefix": "osd crush dump"} + buf, err := executeMonCommand(adminConn, cmd, fmt.Sprintf("retrieving crush map")) + if err != nil { + return "", fmt.Errorf("failed to get crush map. %v", err) + } + + return string(buf), nil +} + func formatLocation(location string) ([]string, error) { var pairs []string if location == "" { From afea5de34d2a5fd7f67773e38e74b96def9315f0 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Mon, 3 Oct 2016 16:17:48 -0700 Subject: [PATCH 3/3] increase code coverage of ceph orchestration --- pkg/api/handlers.go | 2 +- pkg/cephmgr/cephleader_test.go | 34 ++++++++++++- pkg/cephmgr/conn_test.go | 20 ++++++++ pkg/cephmgr/monleader_test.go | 48 +++++++++++++++++++ pkg/cephmgr/osd.go | 2 +- pkg/cephmgr/osd_test.go | 33 +++++++++++++ pkg/cephmgr/osdagent.go | 87 +++++++++++++++------------------- pkg/cephmgr/osdagent_test.go | 21 ++++---- 8 files changed, 181 insertions(+), 66 deletions(-) create mode 100644 pkg/cephmgr/conn_test.go diff --git a/pkg/api/handlers.go b/pkg/api/handlers.go index d559b3233..4fe93ebf6 100644 --- a/pkg/api/handlers.go +++ b/pkg/api/handlers.go @@ -183,7 +183,7 @@ func (h *Handler) GetCrushMap(w http.ResponseWriter, r *http.Request) { } defer conn.Shutdown() - // get the monitor status + // get the crush map crushmap, err := cephmgr.GetCrushMap(conn) if err != nil { log.Printf("failed to get crush map, err: %+v", err) diff --git a/pkg/cephmgr/cephleader_test.go b/pkg/cephmgr/cephleader_test.go index ee984b332..8e9b50a00 100644 --- a/pkg/cephmgr/cephleader_test.go +++ b/pkg/cephmgr/cephleader_test.go @@ -2,10 +2,13 @@ package cephmgr import ( "log" + "path" "strings" "testing" "time" + etcd "github.com/coreos/etcd/client" + "github.com/quantum/castle/pkg/cephmgr/client" testceph "github.com/quantum/castle/pkg/cephmgr/client/test" "github.com/quantum/castle/pkg/clusterd" @@ -166,11 +169,40 @@ func TestMoveUnhealthyMonitor(t *testing.T) { } func TestExtractDesiredDeviceNode(t *testing.T) { + // valid path with node id node, err := extractNodeIDFromDesiredDevice("/castle/services/ceph/osd/desired/abc/device/sdb") assert.Nil(t, err) assert.Equal(t, "abc", node) - node, err = extractNodeIDFromDesiredDevice("/castle/services/ceph/osd/desired") + // node id not found + key := "/castle/services/ceph/osd/desired" + node, err = extractNodeIDFromDesiredDevice(key) assert.NotNil(t, err) assert.Equal(t, "", node) + + // ensure the handle device changed event can run without crashing + response := &etcd.Response{Action: "create", Node: &etcd.Node{Key: key}} + handleDeviceChanged(response, nil) + +} + +func TestRefreshKeys(t *testing.T) { + factory := &testceph.MockConnectionFactory{Fsid: "fsid", SecretKey: "key"} + leader := newCephLeader(factory) + keys := leader.RefreshKeys() + assert.Equal(t, 1, len(keys)) + + expected := path.Join(cephKey, osdAgentName, desiredKey) + assert.Equal(t, expected, keys[0].Path) +} + +func TestNewCephService(t *testing.T) { + factory := &testceph.MockConnectionFactory{Fsid: "fsid", SecretKey: "key"} + + service := NewCephService(factory, "a,b,c", true, "root=default") + assert.NotNil(t, service) + assert.Equal(t, "/castle/services/ceph/osd/desired", service.Leader.RefreshKeys()[0].Path) + assert.Equal(t, 2, len(service.Agents)) + assert.Equal(t, "monitor", service.Agents[0].Name()) + assert.Equal(t, "osd", service.Agents[1].Name()) } diff --git a/pkg/cephmgr/conn_test.go b/pkg/cephmgr/conn_test.go new file mode 100644 index 000000000..fffab6fbd --- /dev/null +++ b/pkg/cephmgr/conn_test.go @@ -0,0 +1,20 @@ +package cephmgr + +import ( + "testing" + + testceph "github.com/quantum/castle/pkg/cephmgr/client/test" + "github.com/quantum/castle/pkg/util" + "github.com/stretchr/testify/assert" +) + +func TestBasicConn(t *testing.T) { + etcdClient := util.NewMockEtcdClient() + createTestClusterInfo(etcdClient, []string{"a"}) + factory := NewConnectionFactory() + fact := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"} + + conn, err := factory.ConnectAsAdmin(fact, etcdClient) + assert.Nil(t, err) + assert.NotNil(t, conn) +} diff --git a/pkg/cephmgr/monleader_test.go b/pkg/cephmgr/monleader_test.go index e0058c61d..904cabf72 100644 --- a/pkg/cephmgr/monleader_test.go +++ b/pkg/cephmgr/monleader_test.go @@ -2,6 +2,7 @@ package cephmgr import ( "fmt" + "path" "testing" "time" @@ -55,7 +56,39 @@ func TestMonSelection(t *testing.T) { assert.Equal(t, 0, len(bad)) assert.Equal(t, 3, len(chosen)) assert.Equal(t, "mon0", chosen["a"].Name) +} +func TestMonitorCount(t *testing.T) { + assert.Equal(t, 0, calculateMonitorCount(0)) + assert.Equal(t, 1, calculateMonitorCount(1)) + assert.Equal(t, 1, calculateMonitorCount(2)) + assert.Equal(t, 3, calculateMonitorCount(3)) + assert.Equal(t, 3, calculateMonitorCount(4)) + assert.Equal(t, 3, calculateMonitorCount(20)) + assert.Equal(t, 5, calculateMonitorCount(21)) + assert.Equal(t, 5, calculateMonitorCount(100)) + assert.Equal(t, 7, calculateMonitorCount(101)) + assert.Equal(t, 7, calculateMonitorCount(1001)) +} + +func TestMonOnUnhealthyNode(t *testing.T) { + etcdClient := util.NewMockEtcdClient() + + // mock a monitor + createTestClusterInfo(etcdClient, []string{"a"}) + + // the monitor is on the bad node + badNode := &clusterd.UnhealthyNode{NodeID: "a"} + context := &clusterd.Context{EtcdClient: etcdClient} + response, err := monsOnUnhealthyNode(context, []*clusterd.UnhealthyNode{badNode}) + assert.True(t, response) + assert.Nil(t, err) + + // the monitor is not on another node + badNode.NodeID = "b" + response, err = monsOnUnhealthyNode(context, []*clusterd.UnhealthyNode{badNode}) + assert.False(t, response) + assert.Nil(t, err) } func TestMaxMonID(t *testing.T) { @@ -145,3 +178,18 @@ func TestUnhealthyMon(t *testing.T) { assert.Equal(t, 1, len(bad)) assert.Equal(t, cmonName, bad["c"].Name) } + +func createTestClusterInfo(etcdClient *util.MockEtcdClient, mons []string) { + key := "/castle/services/ceph" + etcdClient.SetValue(path.Join(key, "fsid"), "12345") + etcdClient.SetValue(path.Join(key, "name"), "default") + etcdClient.SetValue(path.Join(key, "_secrets/monitor"), "foo") + etcdClient.SetValue(path.Join(key, "_secrets/admin"), "bar") + + base := "/castle/services/ceph/monitor/desired" + for i, mon := range mons { + etcdClient.SetValue(path.Join(base, mon, "id"), fmt.Sprintf("mon%d", i)) + etcdClient.SetValue(path.Join(base, mon, "ipaddress"), fmt.Sprintf("1.2.3.%d", i)) + etcdClient.SetValue(path.Join(base, mon, "port"), "4321") + } +} diff --git a/pkg/cephmgr/osd.go b/pkg/cephmgr/osd.go index b37491842..8e46c3357 100644 --- a/pkg/cephmgr/osd.go +++ b/pkg/cephmgr/osd.go @@ -556,7 +556,7 @@ func executeMonCommand(connection client.Connection, cmd map[string]interface{}, response, info, err := connection.MonCommand(command) if err != nil { - return nil, fmt.Errorf("mon_command %s failed: %+v", cmd, err) + return nil, fmt.Errorf("mon_command %+v failed: %+v", cmd, err) } log.Printf("succeeded %s. info: %s", message, info) diff --git a/pkg/cephmgr/osd_test.go b/pkg/cephmgr/osd_test.go index a5dbe8f3f..fa04806e9 100644 --- a/pkg/cephmgr/osd_test.go +++ b/pkg/cephmgr/osd_test.go @@ -9,12 +9,36 @@ import ( "strings" "testing" + "github.com/google/uuid" testceph "github.com/quantum/castle/pkg/cephmgr/client/test" "github.com/quantum/castle/pkg/clusterd" "github.com/quantum/castle/pkg/util" "github.com/stretchr/testify/assert" ) +func TestGetOSDInfo(t *testing.T) { + // error when no info is found on disk + id, guid, err := getOSDInfo("/tmp") + assert.Equal(t, -1, id) + assert.NotNil(t, err) + assert.Equal(t, uuid.UUID{}, guid) + + // write the info to disk + whoFile := "/tmp/whoami" + ioutil.WriteFile(whoFile, []byte("23"), 0644) + defer os.Remove(whoFile) + fsidFile := "/tmp/fsid" + testUUID, _ := uuid.NewUUID() + ioutil.WriteFile(fsidFile, []byte(testUUID.String()), 0644) + defer os.Remove(fsidFile) + + // check the successful osd info + id, guid, err = getOSDInfo("/tmp") + assert.Nil(t, err) + assert.Equal(t, 23, id) + assert.Equal(t, testUUID, guid) +} + func TestDesiredDeviceState(t *testing.T) { etcdClient := util.NewMockEtcdClient() @@ -94,6 +118,15 @@ func TestCrushMap(t *testing.T) { assert.Equal(t, location, etcdClient.GetValue("/castle/nodes/config/node1/location")) } +func TestGetCrushMap(t *testing.T) { + factory := &testceph.MockConnectionFactory{Fsid: "fsid", SecretKey: "key"} + conn, _ := factory.NewConnWithClusterAndUser("cluster", "user") + response, err := GetCrushMap(conn) + + assert.Nil(t, err) + assert.Equal(t, "", response) +} + func TestCrushLocation(t *testing.T) { loc := "dc=datacenter1" hostName, err := os.Hostname() diff --git a/pkg/cephmgr/osdagent.go b/pkg/cephmgr/osdagent.go index 9f5c42f4f..37f793d7d 100644 --- a/pkg/cephmgr/osdagent.go +++ b/pkg/cephmgr/osdagent.go @@ -29,18 +29,16 @@ const ( ) type osdAgent struct { - cluster *ClusterInfo - forceFormat bool - location string - factory client.ConnectionFactory - osdCmd map[string]*exec.Cmd - devices string - getIDFromName func(context *clusterd.Context, name string) (int, error) + cluster *ClusterInfo + forceFormat bool + location string + factory client.ConnectionFactory + osdCmd map[string]*exec.Cmd + devices string } func newOSDAgent(factory client.ConnectionFactory, devices string, forceFormat bool, location string) *osdAgent { a := &osdAgent{factory: factory, devices: devices, forceFormat: forceFormat, location: location} - a.getIDFromName = a.getOSDID return a } @@ -111,12 +109,13 @@ func (a *osdAgent) ConfigureLocalService(context *clusterd.Context) error { func (a *osdAgent) startDesiredDevices(context *clusterd.Context, connection client.Connection, devices *util.Set) error { // Start an OSD for each of the specified devices - if err := a.createOSDs(connection, context, devices); err != nil { + deviceMap, err := a.createOSDs(connection, context, devices) + if err != nil { return fmt.Errorf("failed to create OSDs: %+v", err) } // successful, ensure all applied devices are saved to the cluster config store - if err := a.saveAppliedOSDs(context, devices); err != nil { + if err := a.saveAppliedOSDs(context, deviceMap); err != nil { return fmt.Errorf("failed to save applied OSDs: %+v", err) } @@ -145,39 +144,23 @@ func (a *osdAgent) stopUndesiredDevices(context *clusterd.Context, connection cl return lastErr } -func (a *osdAgent) getOSDID(context *clusterd.Context, name string) (int, error) { - mountPoint, err := inventory.GetDeviceMountPoint(name, context.Executor) +func getIDFromName(context *clusterd.Context, name string) (int, error) { + key := path.Join(getAppliedKey(context.NodeID), devicesKey, name, "id") + val, err := context.EtcdClient.Get(ctx.Background(), key, nil) if err != nil { - return -1, fmt.Errorf("unable to get mount point to remove device %s: %+v", name, err) + return -1, fmt.Errorf("failed to get device id from name %s. %v", name, err) } - if mountPoint == "" { - return -1, fmt.Errorf("mount point not found for osd %s", name) - } - - // find the OSD data path (under the mount point/data dir) - osdDataPath, err := findOSDDataPath(mountPoint, a.cluster.Name) + id, err := strconv.ParseInt(val.Node.Value, 10, 32) if err != nil { - return -1, err + return -1, fmt.Errorf("failed to get parse id from name %s. %v", name, err) } - var osdID int - if _, err := os.Stat(filepath.Join(osdDataPath, "whoami")); os.IsNotExist(err) { - return -1, fmt.Errorf("cannot identify osd %s to remove", name) - } - - // osd_data_dir/whoami already exists, meaning the OSD is already set up. - // look up some basic information about it so we can run it. - osdID, _, err = getOSDInfo(osdDataPath) - if err != nil { - return -1, fmt.Errorf("failed to get OSD information from %s: %+v", osdDataPath, err) - } - - return osdID, nil + return int(id), nil } func (a *osdAgent) removeOSD(context *clusterd.Context, connection client.Connection, name string) error { - osdID, err := a.getIDFromName(context, name) + osdID, err := getIDFromName(context, name) if err != nil { return fmt.Errorf("failed to get osd %s id. %v", name, err) } @@ -235,20 +218,21 @@ func getAppliedKey(nodeID string) string { } // create and initalize OSDs for all the devices specified in the given config -func (a *osdAgent) createOSDs(adminConn client.Connection, context *clusterd.Context, devices *util.Set) error { +func (a *osdAgent) createOSDs(adminConn client.Connection, context *clusterd.Context, devices *util.Set) (map[string]int, error) { // generate and write the OSD bootstrap keyring if err := createOSDBootstrapKeyring(adminConn, a.cluster.Name); err != nil { - return err + return nil, err } // connect to the cluster using the bootstrap-osd creds, this connection will be used for config operations bootstrapConn, err := connectToCluster(a.factory, a.cluster, getBootstrapOSDDir(), "bootstrap-osd", getBootstrapOSDKeyringPath(a.cluster.Name)) if err != nil { - return err + return nil, err } defer bootstrapConn.Shutdown() // initialize all the desired OSD volumes + deviceMap := map[string]int{} for device := range devices.Iter() { var osdID int var osdUUID uuid.UUID @@ -257,44 +241,46 @@ func (a *osdAgent) createOSDs(adminConn client.Connection, context *clusterd.Con if device == localDeviceName { osdDataRoot, osdID, osdUUID, err = a.createLocalOSD(bootstrapConn) if err != nil { - return fmt.Errorf("failed to create local OSD: %+v", err) + return nil, fmt.Errorf("failed to create local OSD: %+v", err) } } else { osdDataRoot, osdID, osdUUID, err = a.createMountedOSD(device, bootstrapConn, context) if err != nil { - return fmt.Errorf("failed to create mounted OSD for device %s: %+v", device, err) + return nil, fmt.Errorf("failed to create mounted OSD for device %s: %+v", device, err) } } // find the OSD data path (under the mount point/data dir) osdDataPath, err := findOSDDataPath(osdDataRoot, a.cluster.Name) if err != nil { - return err + return nil, err } if isOSDDataNotExist(osdDataPath) { // osd_data_dir/whoami does not exist yet, create/initialize the OSD osdDataPath, err = initializeOSD(a.factory, context, osdDataRoot, osdID, osdUUID, device, bootstrapConn, a.cluster, a.location) if err != nil { - return fmt.Errorf("failed to initialize OSD at %s: %+v", osdDataRoot, err) + return nil, fmt.Errorf("failed to initialize OSD at %s: %+v", osdDataRoot, err) } } else { // osd_data_dir/whoami already exists, meaning the OSD is already set up. // look up some basic information about it so we can run it. osdID, osdUUID, err = getOSDInfo(osdDataPath) if err != nil { - return fmt.Errorf("failed to get OSD information from %s: %+v", osdDataPath, err) + return nil, fmt.Errorf("failed to get OSD information from %s: %+v", osdDataPath, err) } } // run the OSD in a child process now that it is fully initialized and ready to go err = a.runOSD(context, a.cluster.Name, osdID, osdUUID, osdDataPath, device) if err != nil { - return fmt.Errorf("failed to run osd %d: %+v", osdID, err) + return nil, fmt.Errorf("failed to run osd %d: %+v", osdID, err) } + + deviceMap[device] = osdID } - return nil + return deviceMap, nil } func (a *osdAgent) createMountedOSD(device string, bootstrapConn client.Connection, context *clusterd.Context) (string, int, uuid.UUID, error) { @@ -422,22 +408,23 @@ func GetAppliedOSDs(nodeID string, etcdClient etcd.KeysAPI) (map[string]string, return devices, nil } -func (a *osdAgent) saveAppliedOSDs(context *clusterd.Context, devices *util.Set) error { +func (a *osdAgent) saveAppliedOSDs(context *clusterd.Context, devices map[string]int) error { appliedKey := path.Join(getAppliedKey(context.NodeID), devicesKey) var settings = make(map[string]string) - for d := range devices.Iter() { + for name, id := range devices { var serial string var err error - if d == localDeviceName { + if name == localDeviceName { serial = localDeviceName } else { - serial, err = inventory.GetSerialFromDevice(d, context.NodeID, context.EtcdClient) + serial, err = inventory.GetSerialFromDevice(name, context.NodeID, context.EtcdClient) if err != nil { - return fmt.Errorf("failed to get serial for device %s: %+v", d, err) + return fmt.Errorf("failed to get serial for device %s: %+v", name, err) } } - settings[path.Join(d, "serial")] = serial + settings[path.Join(name, "serial")] = serial + settings[path.Join(name, "id")] = strconv.FormatInt(int64(id), 10) } if err := util.StoreEtcdProperties(context.EtcdClient, appliedKey, settings); err != nil { diff --git a/pkg/cephmgr/osdagent_test.go b/pkg/cephmgr/osdagent_test.go index 372897277..2c69dfff7 100644 --- a/pkg/cephmgr/osdagent_test.go +++ b/pkg/cephmgr/osdagent_test.go @@ -217,20 +217,15 @@ func TestRemoveDevice(t *testing.T) { etcdClient.SetValue(path.Join(root, "sdb/serial"), "456") etcdClient.SetValue(path.Join(root, "sdc/serial"), "789") - agent.getIDFromName = func(context *clusterd.Context, name string) (int, error) { - if name == "sda" { - return 1, nil - } - if name == "sdb" { - return 2, nil - } - if name == "sdc" { - return 3, nil - } - return -1, fmt.Errorf("unknown name %s", name) - } - + // the request will fail without the device id set err := agent.stopUndesiredDevices(context, conn, desired) + assert.NotNil(t, err) + + etcdClient.SetValue(path.Join(root, "sda/id"), "1") + etcdClient.SetValue(path.Join(root, "sdb/id"), "2") + etcdClient.SetValue(path.Join(root, "sdc/id"), "3") + + err = agent.stopUndesiredDevices(context, conn, desired) assert.Nil(t, err) applied := etcdClient.GetChildDirs(root) assert.True(t, applied.Equals(desired))