Merge pull request #39 from travisn/remove-osd

Stop and remove an OSD from the crush map

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

pkg/api/routes.go

@@ -26,6 +26,12 @@ func (h *Handler) GetRoutes() []Route {
 			"/mon",
 			h.GetMonitors,
 		},
+		{
+			"GetCrushMap",
+			"GET",
+			"/crushmap",
+			h.GetCrushMap,
+		},
 		{
 			"AddDevice",
 			"POST",

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())
 }

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)
+}

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)

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")
+	}
+}

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 <name>
+	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 == "" {

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()

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 {

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))