diff --git a/pkg/api/handlers.go b/pkg/api/handlers.go index 9e2a0a5cf..4fe93ebf6 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 crush map + 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/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/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/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 63876b37d..8e46c3357 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,85 @@ 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 %+v 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 +} + +// 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 == "" { 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 645122524..37f793d7d 100644 --- a/pkg/cephmgr/osdagent.go +++ b/pkg/cephmgr/osdagent.go @@ -38,7 +38,8 @@ type osdAgent struct { } 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} + return a } func (a *osdAgent) Name() string { @@ -108,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) } @@ -133,33 +135,72 @@ 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 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("failed to get device id from name %s. %v", name, err) + } + + id, err := strconv.ParseInt(val.Node.Value, 10, 32) + if err != nil { + return -1, fmt.Errorf("failed to get parse id from name %s. %v", name, err) + } + + return int(id), nil +} + +func (a *osdAgent) removeOSD(context *clusterd.Context, connection client.Connection, name string) error { + osdID, err := 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 { @@ -177,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 @@ -199,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) { @@ -364,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 e2e54c7d9..2c69dfff7 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,7 +217,15 @@ func TestRemoveDevice(t *testing.T) { etcdClient.SetValue(path.Join(root, "sdb/serial"), "456") etcdClient.SetValue(path.Join(root, "sdc/serial"), "789") + // 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))