Skip to content

Commit

Permalink
osd: support create osd with metadata partition
Browse files Browse the repository at this point in the history 
Signed-off-by: Liang Zheng <zhengliang0901@gmail.com>
  • Loading branch information
@microyahoo
microyahoo committed Dec 4, 2023
1 parent 3097455 commit 95a99d6
Showing 1 changed file with 86 additions and 39 deletions.
125 changes: 86 additions & 39 deletions pkg/daemon/ceph/osd/volume.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,10 @@ const (
dbDeviceFlag = "--db-devices"
cephVolumeCmd = "ceph-volume"
cephVolumeMinDBSize = 1024 // 1GB

blockDBFlag = "--block.db"
blockDBSizeFlag = "--block.db-size"
dataFlag = "--data"
)

// These are not constants because they are used by the tests
Expand Down Expand Up @@ -606,6 +610,8 @@ func (a *OsdAgent) initializeDevicesLVMMode(context *clusterd.Context, devices *
osdsPerDeviceCount := sanitizeOSDsPerDevice(a.storeConfig.OSDsPerDevice)
batchArgs := baseArgs

var hasMetadataPartition bool

metadataDevices := make(map[string]map[string]string)
for name, device := range devices.Entries {
if device.Data == -1 {
Expand Down Expand Up @@ -649,6 +655,9 @@ func (a *OsdAgent) initializeDevicesLVMMode(context *clusterd.Context, devices *
if metadataDevice.Type != sys.LVMType {
md = metadataDevice.Name
}
if metadataDevice.Type == sys.PartType {
hasMetadataPartition = true // ceph-volume lvm batch only supports disk and lvm
}

logger.Infof("using %s as metadataDevice for device %s and let ceph-volume lvm batch decide how to create volumes", md, deviceArg)
if _, ok := metadataDevices[md]; ok {
Expand Down Expand Up @@ -719,73 +728,111 @@ func (a *OsdAgent) initializeDevicesLVMMode(context *clusterd.Context, devices *
}
}

var prepareArgs []string
if hasMetadataPartition {
// ceph-volume lvm prepare --data {vg/lv} --block.wal {partition} --block.db {/path/to/device}
baseArgs := []string{"-oL", cephVolumeCmd, "--log-path", logPath, "lvm", "prepare", storeFlag}
if a.storeConfig.EncryptedDevice {
baseArgs = append(baseArgs, encryptedFlag)
}
prepareArgs = baseArgs
}

for md, conf := range metadataDevices {

mdArgs := batchArgs
if _, ok := conf["osdsperdevice"]; ok {
mdArgs = append(mdArgs, []string{
osdsPerDeviceFlag,
conf["osdsperdevice"],
}...)
if hasMetadataPartition {
mdArgs = prepareArgs
} else {
if _, ok := conf["osdsperdevice"]; ok {
mdArgs = append(mdArgs, []string{
osdsPerDeviceFlag,
conf["osdsperdevice"],
}...)
}
}
if _, ok := conf["deviceclass"]; ok {
mdArgs = append(mdArgs, []string{
crushDeviceClassFlag,
conf["deviceclass"],
}...)
}
if _, ok := conf["databasesizemb"]; ok {
if hasMetadataPartition {
devices := strings.Split(conf["devices"], " ")
if len(devices) > 1 {
logger.Warningf("device partition %s can only be used by one device", md)
}
mdArgs = append(mdArgs, []string{
databaseSizeFlag,
conf["databasesizemb"],
dataFlag,
devices[0],
}...)
if _, ok := conf["databasesizemb"]; ok {
mdArgs = append(mdArgs, []string{
blockDBSizeFlag,
conf["databasesizemb"],
}...)
}
} else {
if _, ok := conf["databasesizemb"]; ok {
mdArgs = append(mdArgs, []string{
databaseSizeFlag,
conf["databasesizemb"],
}...)
}
mdArgs = append(mdArgs, strings.Split(conf["devices"], " ")...)
}
mdArgs = append(mdArgs, strings.Split(conf["devices"], " ")...)

// Do not change device names if udev persistent names are passed
mdPath := md
if !strings.HasPrefix(mdPath, "/dev") {
mdPath = path.Join("/dev", md)
}

mdArgs = append(mdArgs, []string{
dbDeviceFlag,
mdPath,
}...)
if hasMetadataPartition {
mdArgs = append(mdArgs, []string{
blockDBFlag,
mdPath,
}...)
} else {
mdArgs = append(mdArgs, []string{
dbDeviceFlag,
mdPath,
}...)

// Reporting
reportArgs := append(mdArgs, []string{
"--report",
}...)
// Reporting
reportArgs := append(mdArgs, []string{
"--report",
}...)

if err := context.Executor.ExecuteCommand(baseCommand, reportArgs...); err != nil {
return errors.Wrap(err, "failed ceph-volume report") // fail return here as validation provided by ceph-volume
}
if err := context.Executor.ExecuteCommand(baseCommand, reportArgs...); err != nil {
return errors.Wrap(err, "failed ceph-volume report") // fail return here as validation provided by ceph-volume
}

reportArgs = append(reportArgs, []string{
"--format",
"json",
}...)
reportArgs = append(reportArgs, []string{
"--format",
"json",
}...)

cvOut, err := context.Executor.ExecuteCommandWithOutput(baseCommand, reportArgs...)
if err != nil {
return errors.Wrapf(err, "failed ceph-volume json report: %s", cvOut) // fail return here as validation provided by ceph-volume
}
cvOut, err := context.Executor.ExecuteCommandWithOutput(baseCommand, reportArgs...)
if err != nil {
return errors.Wrapf(err, "failed ceph-volume json report: %s", cvOut) // fail return here as validation provided by ceph-volume
}

logger.Debugf("ceph-volume reports: %+v", cvOut)
logger.Debugf("ceph-volume reports: %+v", cvOut)

var cvReports []cephVolReportV2
if err = json.Unmarshal([]byte(cvOut), &cvReports); err != nil {
return errors.Wrap(err, "failed to unmarshal ceph-volume report json")
}
var cvReports []cephVolReportV2
if err = json.Unmarshal([]byte(cvOut), &cvReports); err != nil {
return errors.Wrap(err, "failed to unmarshal ceph-volume report json")
}

if len(strings.Split(conf["devices"], " ")) != len(cvReports) {
return errors.Errorf("failed to create enough required devices, required: %s, actual: %v", cvOut, cvReports)
}
if len(strings.Split(conf["devices"], " ")) != len(cvReports) {
return errors.Errorf("failed to create enough required devices, required: %s, actual: %v", cvOut, cvReports)
}

for _, report := range cvReports {
if report.BlockDB != mdPath && !strings.HasSuffix(mdPath, report.BlockDB) {
return errors.Errorf("wrong db device for %s, required: %s, actual: %s", report.Data, mdPath, report.BlockDB)
for _, report := range cvReports {
if report.BlockDB != mdPath && !strings.HasSuffix(mdPath, report.BlockDB) {
return errors.Errorf("wrong db device for %s, required: %s, actual: %s", report.Data, mdPath, report.BlockDB)
}
}
}

Expand Down

0 comments on commit 95a99d6

Please sign in to comment.