When marking an OSD device to be removed, ensure the correct data partition name is used.

If watching the osd orchestration status map fails, ensure the loop is broken out of so a new watcher can be created.
Log the osd device/dir mapping structures in a more friendly way

Signed-off-by: Jared Watts <jbw976@gmail.com>

pkg/daemon/ceph/osd/daemon.go

@@ -297,8 +297,16 @@ func getRemovedDevices(agent *OsdAgent) (*config.PerfScheme, *DeviceOsdMapping,
 	}
 
 	for _, entry := range scheme.Entries {
+		// determine which partition the data lives on for this entry
+		dataDetails, ok := entry.Partitions[entry.GetDataPartitionType()]
+		if !ok || dataDetails == nil {
+			return nil, nil, fmt.Errorf("failed to find data partition for entry %+v", entry)
+		}
+
+		// add the current scheme entry to the removed devices scheme and its device to the removed
+		// devices mapping
 		removedDevicesScheme.Entries = append(removedDevicesScheme.Entries, entry)
-		removedDevicesMapping.Entries[entry.Partitions[config.BlockPartitionType].Device] = &DeviceOsdIDEntry{Data: entry.ID}
+		removedDevicesMapping.Entries[dataDetails.Device] = &DeviceOsdIDEntry{Data: entry.ID}
 	}
 
 	return removedDevicesScheme, removedDevicesMapping, nil

pkg/daemon/ceph/osd/daemon_test.go

@@ -223,3 +223,37 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil
 	assert.Equal(t, -1, mapping.Entries["rdb"].Data)
 	assert.Equal(t, -1, mapping.Entries["nvme01"].Data)
 }
+
+func TestGetRemovedDevices(t *testing.T) {
+	testGetRemovedDevicesHelper(t, &rookalpha.StoreConfig{StoreType: config.Bluestore})
+	testGetRemovedDevicesHelper(t, &rookalpha.StoreConfig{StoreType: config.Filestore})
+}
+
+func testGetRemovedDevicesHelper(t *testing.T, storeConfig *rookalpha.StoreConfig) {
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
+	os.MkdirAll(configDir, 0755)
+	nodeName := "node3391"
+	agent, _, _ := createTestAgent(t, "none", configDir, nodeName, storeConfig)
+
+	// mock the pre-existence of osd 1 on device sdx
+	_, _, _ = mockPartitionSchemeEntry(t, 1, "sdx", &agent.storeConfig, agent.kv, nodeName)
+
+	// get the removed devices for this configuration (note we said to use devices "none" above),
+	// it should be osd 1 on device sdx
+	scheme, mapping, err := getRemovedDevices(agent)
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(mapping.Entries))
+	assert.Equal(t, 1, len(scheme.Entries))
+
+	// assert the scheme has an entry for osd 1 and its data partition is on sdx
+	schemeEntry := scheme.Entries[0]
+	assert.Equal(t, 1, schemeEntry.ID)
+	assert.Equal(t, "sdx", schemeEntry.Partitions[schemeEntry.GetDataPartitionType()].Device)
+
+	// assert the removed device mapping has an entry for device sdx and it points to osd 1
+	mappingEntry, ok := mapping.Entries["sdx"]
+	assert.True(t, ok)
+	assert.NotNil(t, mappingEntry)
+	assert.Equal(t, 1, mappingEntry.Data)
+}

pkg/daemon/ceph/osd/device.go

@@ -77,6 +77,11 @@ type DeviceOsdIDEntry struct {
 	Metadata []int // OSD IDs (multiple) that have metadata stored here
 }
 
+func (m *DeviceOsdMapping) String() string {
+	b, _ := json.Marshal(m)
+	return string(b)
+}
+
 // format the given device for usage by an OSD
 func formatDevice(context *clusterd.Context, config *osdConfig, forceFormat bool, storeConfig rookalpha.StoreConfig) error {
 	dataDetails, err := getDataPartitionDetails(config)

pkg/operator/cluster/ceph/osd/config/scheme.go

@@ -159,6 +159,11 @@ func RemoveFromScheme(e *PerfSchemeEntry, kv *k8sutil.ConfigMapKVStore, storeNam
 	return nil
 }
 
+func (s *PerfScheme) String() string {
+	b, _ := json.Marshal(s)
+	return string(b)
+}
+
 func (s *PerfScheme) UpdateSchemeEntry(e *PerfSchemeEntry) error {
 	return s.doSchemeEntryAction(e, func(scheme *PerfScheme, index int, entry *PerfSchemeEntry) {
 		// the action to perform if the entry is found is to update the entry
@@ -363,6 +368,11 @@ func (m *MetadataDeviceInfo) GetPartitionArgs() []string {
 	return args
 }
 
+func (e *PerfSchemeEntry) String() string {
+	b, _ := json.Marshal(e)
+	return string(b)
+}
+
 func (e *PerfSchemeEntry) GetPartitionArgs() []string {
 	// first determine if all the partitions are collocated
 	collocated := e.IsCollocated()

pkg/operator/cluster/ceph/osd/osd.go

@@ -223,7 +223,7 @@ func (c *Cluster) Start() error {
 
 		// trigger orchestration on the removed node by telling it not to use any storage at all.  note that the directories are still passed in
 		// so that the pod will be able to mount them and migrate data from them.
-		rs := c.makeReplicaSet(n.Name, nil, rookalpha.Selection{DeviceFilter: "none", Directories: n.Directories}, v1.ResourceRequirements{}, rookalpha.Config{})
+		rs := c.makeReplicaSet(n.Name, nil, rookalpha.Selection{DeviceFilter: "none", Directories: n.Directories}, v1.ResourceRequirements{}, n.Config)
 		rs, err := c.context.Clientset.Extensions().ReplicaSets(c.Namespace).Update(rs)
 		if err != nil {
 			message := fmt.Sprintf("failed to update osd replica set for removed node %s. %+v", n.Name, err)
@@ -422,14 +422,15 @@ func (c *Cluster) waitForCompletion(node string) error {
 		}
 		defer w.Stop()
 
+	ResultLoop:
 		for {
 			select {
 			case e, ok := <-w.ResultChan():
 				if !ok {
-					logger.Warning("orchestration status config map result channel closed, restarting watch.")
-					<-time.After(100 * time.Millisecond)
+					logger.Warning("orchestration status config map result channel closed, will restart watch.")
 					w.Stop()
-					break
+					<-time.After(100 * time.Millisecond)
+					break ResultLoop
 				}
 
 				if e.Type == watch.Modified {
@@ -520,6 +521,7 @@ func (c *Cluster) discoverStorageNodes() ([]rookalpha.Node, error) {
 			Selection: rookalpha.Selection{
 				Directories: getDirectoriesFromContainer(osdContainer),
 			},
+			Config: getConfigFromContainer(osdContainer),
 		}
 
 		discoveredNodes[i] = node
@@ -647,24 +649,3 @@ func (c *Cluster) getOSDsForNode(node rookalpha.Node) ([]int, error) {
 
 	return osdsForNode, nil
 }
-
-func getDirectoriesFromContainer(osdContainer v1.Container) []rookalpha.Directory {
-	var dirsArg string
-	for _, envVar := range osdContainer.Env {
-		if envVar.Name == dataDirsEnvVarName {
-			dirsArg = envVar.Value
-		}
-	}
-
-	var dirsList []string
-	if dirsArg != "" {
-		dirsList = strings.Split(dirsArg, ",")
-	}
-
-	dirs := make([]rookalpha.Directory, len(dirsList))
-	for dirNum, dir := range dirsList {
-		dirs[dirNum] = rookalpha.Directory{Path: dir}
-	}
-
-	return dirs
-}

pkg/operator/cluster/ceph/osd/pod.go

@@ -32,7 +32,12 @@ import (
 )
 
 const (
-	dataDirsEnvVarName = "ROOK_DATA_DIRECTORIES"
+	dataDirsEnvVarName        = "ROOK_DATA_DIRECTORIES"
+	osdStoreEnvVarName        = "ROOK_OSD_STORE"
+	osdDatabaseSizeEnvVarName = "ROOK_OSD_DATABASE_SIZE"
+	osdWalSizeEnvVarName      = "ROOK_OSD_WAL_SIZE"
+	osdJournalSizeEnvVarName  = "ROOK_OSD_JOURNAL_SIZE"
+	locationEnvVarName        = "ROOK_LOCATION"
 )
 
 func (c *Cluster) makeDaemonSet(selection rookalpha.Selection, config rookalpha.Config) *extensions.DaemonSet {
@@ -267,21 +272,67 @@ func dataDirectoriesEnvVar(dataDirectories string) v1.EnvVar {
 }
 
 func osdStoreEnvVar(osdStore string) v1.EnvVar {
-	return v1.EnvVar{Name: "ROOK_OSD_STORE", Value: osdStore}
+	return v1.EnvVar{Name: osdStoreEnvVarName, Value: osdStore}
 }
 
 func osdDatabaseSizeEnvVar(databaseSize int) v1.EnvVar {
-	return v1.EnvVar{Name: "ROOK_OSD_DATABASE_SIZE", Value: strconv.Itoa(databaseSize)}
+	return v1.EnvVar{Name: osdDatabaseSizeEnvVarName, Value: strconv.Itoa(databaseSize)}
 }
 
 func osdWalSizeEnvVar(walSize int) v1.EnvVar {
-	return v1.EnvVar{Name: "ROOK_OSD_WAL_SIZE", Value: strconv.Itoa(walSize)}
+	return v1.EnvVar{Name: osdWalSizeEnvVarName, Value: strconv.Itoa(walSize)}
 }
 
 func osdJournalSizeEnvVar(journalSize int) v1.EnvVar {
-	return v1.EnvVar{Name: "ROOK_OSD_JOURNAL_SIZE", Value: strconv.Itoa(journalSize)}
+	return v1.EnvVar{Name: osdJournalSizeEnvVarName, Value: strconv.Itoa(journalSize)}
 }
 
 func locationEnvVar(location string) v1.EnvVar {
-	return v1.EnvVar{Name: "ROOK_LOCATION", Value: location}
+	return v1.EnvVar{Name: locationEnvVarName, Value: location}
+}
+
+func getDirectoriesFromContainer(osdContainer v1.Container) []rookalpha.Directory {
+	var dirsArg string
+	for _, envVar := range osdContainer.Env {
+		if envVar.Name == dataDirsEnvVarName {
+			dirsArg = envVar.Value
+		}
+	}
+
+	var dirsList []string
+	if dirsArg != "" {
+		dirsList = strings.Split(dirsArg, ",")
+	}
+
+	dirs := make([]rookalpha.Directory, len(dirsList))
+	for dirNum, dir := range dirsList {
+		dirs[dirNum] = rookalpha.Directory{Path: dir}
+	}
+
+	return dirs
+}
+
+func getConfigFromContainer(osdContainer v1.Container) rookalpha.Config {
+	cfg := rookalpha.Config{}
+
+	for _, envVar := range osdContainer.Env {
+		switch envVar.Name {
+		case osdStoreEnvVarName:
+			cfg.StoreConfig.StoreType = envVar.Value
+		case osdDatabaseSizeEnvVarName:
+			size, _ := strconv.Atoi(envVar.Value)
+			cfg.StoreConfig.DatabaseSizeMB = size
+		case osdWalSizeEnvVarName:
+			size, _ := strconv.Atoi(envVar.Value)
+			cfg.StoreConfig.WalSizeMB = size
+		case osdJournalSizeEnvVarName:
+			size, _ := strconv.Atoi(envVar.Value)
+			cfg.StoreConfig.JournalSizeMB = size
+		case locationEnvVarName:
+			cfg.Location = envVar.Value
+		}
+
+	}
+
+	return cfg
 }

pkg/operator/cluster/ceph/osd/pod_test.go

@@ -216,6 +216,10 @@ func TestStorageSpecConfig(t *testing.T) {
 
 	assert.Equal(t, "100", container.Resources.Limits.Cpu().String())
 	assert.Equal(t, "1337", container.Resources.Requests.Memory().String())
+
+	// verify that osd config can be discovered from the container and matches the original config from the spec
+	cfg := getConfigFromContainer(container)
+	assert.Equal(t, storageSpec.Nodes[0].Config, cfg)
 }
 
 func TestHostNetwork(t *testing.T) {