From f06dd0009e4f04ea9484cf0ca99c091fac857faa Mon Sep 17 00:00:00 2001 From: Jared Watts Date: Tue, 13 Feb 2018 22:08:50 -0800 Subject: [PATCH] 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 --- pkg/daemon/ceph/osd/daemon.go | 10 ++- pkg/daemon/ceph/osd/daemon_test.go | 34 ++++++++++ pkg/daemon/ceph/osd/device.go | 5 ++ .../cluster/ceph/osd/config/scheme.go | 10 +++ pkg/operator/cluster/ceph/osd/osd.go | 31 ++------- pkg/operator/cluster/ceph/osd/pod.go | 63 +++++++++++++++++-- pkg/operator/cluster/ceph/osd/pod_test.go | 4 ++ 7 files changed, 125 insertions(+), 32 deletions(-) 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) {