remove a bad osd from crushmap

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 <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
+}
+
 func formatLocation(location string) ([]string, error) {
 	var pairs []string
 	if location == "" {

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 {

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)