diff --git a/pkg/daemon/ceph/osd/daemon.go b/pkg/daemon/ceph/osd/daemon.go index 7363a0d28..32c3724f6 100644 --- a/pkg/daemon/ceph/osd/daemon.go +++ b/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 diff --git a/pkg/daemon/ceph/osd/daemon_test.go b/pkg/daemon/ceph/osd/daemon_test.go index deefffae3..ed7580bc4 100644 --- a/pkg/daemon/ceph/osd/daemon_test.go +++ b/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) +} diff --git a/pkg/daemon/ceph/osd/device.go b/pkg/daemon/ceph/osd/device.go index d91e7dd8c..1b7ccee54 100644 --- a/pkg/daemon/ceph/osd/device.go +++ b/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) diff --git a/pkg/operator/cluster/ceph/osd/config/scheme.go b/pkg/operator/cluster/ceph/osd/config/scheme.go index bbe648251..26b83a90b 100644 --- a/pkg/operator/cluster/ceph/osd/config/scheme.go +++ b/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() diff --git a/pkg/operator/cluster/ceph/osd/osd.go b/pkg/operator/cluster/ceph/osd/osd.go index 948a10d0f..b750dec9d 100644 --- a/pkg/operator/cluster/ceph/osd/osd.go +++ b/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 -} diff --git a/pkg/operator/cluster/ceph/osd/pod.go b/pkg/operator/cluster/ceph/osd/pod.go index 68bac9ed9..d8a958892 100644 --- a/pkg/operator/cluster/ceph/osd/pod.go +++ b/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 } diff --git a/pkg/operator/cluster/ceph/osd/pod_test.go b/pkg/operator/cluster/ceph/osd/pod_test.go index b4f4178d6..eed1bdd5d 100644 --- a/pkg/operator/cluster/ceph/osd/pod_test.go +++ b/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) {