Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add the WAL and DB devices for the disk-add command #222

Merged
merged 6 commits into from
Oct 6, 2023
Merged

Add the WAL and DB devices for the disk-add command #222

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d00323d
Add the WAL and DB devices for the disk-add command
lmlg Sep 21, 2023
2cf7ddf
Add the parameter names
lmlg Sep 21, 2023
c9ae181
Fix: failure domain unit tests
sabaini Sep 22, 2023
3696435
Factor out common functionality
lmlg Sep 25, 2023
3cfab98
Fill out missing members and check error
lmlg Sep 25, 2023
61cab69
Cleanup code and improve error messages.
lmlg Sep 27, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion microceph/api/disks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,8 @@ func cmdDisksGet(s *state.State, r *http.Request) response.Response {

func cmdDisksPost(s *state.State, r *http.Request) response.Response {
var req types.DisksPost
var wal *types.DiskParameter
var db *types.DiskParameter
lmlg marked this conversation as resolved.
Show resolved Hide resolved

logger.Debugf("cmdDisksPost: %v", req)
err := json.NewDecoder(r.Body).Decode(&req)
Expand All @@ -56,7 +58,16 @@ func cmdDisksPost(s *state.State, r *http.Request) response.Response {

mu.Lock()
defer mu.Unlock()
err = ceph.AddOSD(s, req.Path, req.Wipe, req.Encrypt)

data := types.DiskParameter{req.Path, req.Wipe, req.Encrypt}
if req.WALDev != nil {
wal = &types.DiskParameter{*req.WALDev, req.WALWipe, req.WALEncrypt}
}
if req.DBDev != nil {
db = &types.DiskParameter{*req.DBDev, req.DBWipe, req.DBEncrypt}
}

err = ceph.AddOSD(s, data, wal, db)
if err != nil {
return response.SmartError(err)
}
Expand Down
18 changes: 15 additions & 3 deletions microceph/api/types/disks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,15 @@ package types

// DisksPost hold a path and a flag for enabling device wiping
type DisksPost struct {
Path string `json:"path" yaml:"path"`
Wipe bool `json:"wipe" yaml:"wipe"`
Encrypt bool `json:"encrypt" yaml:"encrypt"`
Path string `json:"path" yaml:"path"`
Wipe bool `json:"wipe" yaml:"wipe"`
Encrypt bool `json:"encrypt" yaml:"encrypt"`
WALDev *string `json:"waldev" yaml:"waldev"`
WALWipe bool `json:"walwipe" yaml:"walwipe"`
WALEncrypt bool `json:"walencrypt" yaml:"walencrypt"`
DBDev *string `json:"dbdev" yaml:"dbdev"`
DBWipe bool `json:"dbwipe" yaml:"dbwipe"`
DBEncrypt bool `json:"dbencrypt" yaml:"dbencrypt"`
}

// DisksDelete holds an OSD number and a flag for forcing the removal
Expand All @@ -25,3 +31,9 @@ type Disk struct {
Path string `json:"path" yaml:"path"`
Location string `json:"location" yaml:"location"`
}

type DiskParameter struct {
Path string
Encrypt bool
Wipe bool
}
134 changes: 76 additions & 58 deletions microceph/ceph/osd.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,12 +80,34 @@ func nextOSD(s *state.State) (int64, error) {
}
}

func prepareDisk(disk *types.DiskParameter, suffix string, osdPath string, osdID int64) error {
if disk.Wipe {
err := timeoutWipe(disk.Path)
if err != nil {
return fmt.Errorf("Failed to wipe device %s: %w", disk.Path, err)
}
}
if disk.Encrypt {
err := checkEncryptSupport()
if err != nil {
return fmt.Errorf("Encryption unsupported on this machine: %w", err)
}
path, err := setupEncryptedOSD(disk.Path, osdPath, osdID, suffix)
if err != nil {
return fmt.Errorf("Failed to encrypt device %s: %w", disk.Path, err)
}
disk.Path = path
}
return os.Symlink(disk.Path, filepath.Join(osdPath, "block"+suffix))
}

// setupEncryptedOSD sets up an encrypted OSD on the given disk.
//
// Takes a path to the disk device as well as the osd data path and the osd id.
// Takes a path to the disk device as well as the OSD data path, the OSD id and
// a suffix (to differentiate invocations between data, WAL and DB devices).
// Returns the path to the encrypted device and an error if any.
func setupEncryptedOSD(devicePath string, osdDataPath string, osdID int64) (string, error) {
if err := os.Symlink(devicePath, filepath.Join(osdDataPath, "unencrypted")); err != nil {
func setupEncryptedOSD(devicePath string, osdDataPath string, osdID int64, suffix string) (string, error) {
if err := os.Symlink(devicePath, filepath.Join(osdDataPath, "unencrypted"+suffix)); err != nil {
return "", fmt.Errorf("Failed to add unencrypted block symlink: %w", err)
}

Expand All @@ -96,7 +118,7 @@ func setupEncryptedOSD(devicePath string, osdDataPath string, osdID int64) (stri
}

// Store key in ceph key value store
if err = storeKey(key, osdID); err != nil {
if err = storeKey(key, osdID, suffix); err != nil {
return "", fmt.Errorf("Key store error: %w", err)
}

Expand All @@ -106,7 +128,7 @@ func setupEncryptedOSD(devicePath string, osdDataPath string, osdID int64) (stri
}

// Open the encrypted device
encryptedDevicePath, err := openEncryptedDevice(devicePath, osdID, key)
encryptedDevicePath, err := openEncryptedDevice(devicePath, osdID, key, suffix)
if err != nil {
return "", fmt.Errorf("Failed to open: %w", err)
}
Expand Down Expand Up @@ -151,25 +173,25 @@ func encryptDevice(path string, key []byte) error {
}

// Store the key in the ceph key value store, under a name that derives from the osd id.
func storeKey(key []byte, osdID int64) error {
func storeKey(key []byte, osdID int64, suffix string) error {
// Run the ceph config-key set command
_, err := processExec.RunCommand("ceph", "config-key", "set", fmt.Sprintf("microceph:osd.%d/key", osdID), string(key))
_, err := processExec.RunCommand("ceph", "config-key", "set", fmt.Sprintf("microceph:osd%s.%d/key", suffix, osdID), string(key))
if err != nil {
return fmt.Errorf("Failed to store key: %w", err)
}
return nil
}

// Open the encrypted device and return its path.
func openEncryptedDevice(path string, osdID int64, key []byte) (string, error) {
func openEncryptedDevice(path string, osdID int64, key []byte, suffix string) (string, error) {
// Run the cryptsetup open command, expect key on stdin
cmd := exec.Command(
"cryptsetup",
"--keyfile-size", "128",
"--key-file", "-",
"luksOpen",
path,
fmt.Sprintf("luksosd-%d", osdID),
fmt.Sprintf("luksosd%s-%d", suffix, osdID),
)
stdin, err := cmd.StdinPipe()
if err != nil {
Expand All @@ -187,7 +209,7 @@ NOTE: OSD Encryption requires a snapd >= 2.59.1
Verify your version of snapd by running "snap version"
`, path, err, out)
}
return fmt.Sprintf("/dev/mapper/luksosd-%d", osdID), nil
return fmt.Sprintf("/dev/mapper/luksosd%s-%d", suffix, osdID), nil
}

// checkEncryptSupport checks if the kernel supports encryption.
Expand Down Expand Up @@ -265,24 +287,18 @@ func updateFailureDomain(s *state.State) error {
}

// AddOSD adds an OSD to the cluster, given a device path and a flag for wiping
lmlg marked this conversation as resolved.
Show resolved Hide resolved
func AddOSD(s *state.State, path string, wipe bool, encrypt bool) error {
logger.Debugf("Adding OSD %s", path)
func AddOSD(s *state.State, data types.DiskParameter, wal *types.DiskParameter, db *types.DiskParameter) error {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, is there a reason to pass data as by value and wal / db by ref?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

wal and db may be nil, whereas data cannot.

logger.Debugf("Adding OSD %s", data.Path)

revert := revert.New()
defer revert.Fail()

// Validate the path.
if !shared.IsBlockdevPath(path) {
return fmt.Errorf("Invalid disk path: %s", path)
}
// Check if we need to support encryption
if encrypt {
if err := checkEncryptSupport(); err != nil {
return fmt.Errorf("Encryption unsupported on this machine: %w", err)
}
if !shared.IsBlockdevPath(data.Path) {
return fmt.Errorf("Invalid disk path: %s", data.Path)
}

_, _, major, minor, _, _, err := shared.GetFileStat(path)
_, _, major, minor, _, _, err := shared.GetFileStat(data.Path)
lmlg marked this conversation as resolved.
Show resolved Hide resolved
if err != nil {
return fmt.Errorf("Invalid disk path: %w", err)
}
Expand All @@ -302,11 +318,11 @@ func AddOSD(s *state.State, path string, wipe bool, encrypt bool) error {

// check if candidate exists
if shared.PathExists(candidate) && !shared.IsDir(candidate) {
path = candidate
data.Path = candidate
} else {
candidate = fmt.Sprintf("/dev/disk/by-path/%s", disk.DevicePath)
if shared.PathExists(candidate) && !shared.IsDir(candidate) {
path = candidate
data.Path = candidate
}
}

Expand All @@ -319,11 +335,11 @@ func AddOSD(s *state.State, path string, wipe bool, encrypt bool) error {
if part.Device == dev {
candidate := fmt.Sprintf("/dev/disk/by-id/%s-part%d", disk.DeviceID, part.Partition)
if shared.PathExists(candidate) {
path = candidate
data.Path = candidate
} else {
candidate = fmt.Sprintf("/dev/disk/by-path/%s-part%d", disk.DevicePath, part.Partition)
if shared.PathExists(candidate) {
path = candidate
data.Path = candidate
}
}

Expand All @@ -337,24 +353,16 @@ func AddOSD(s *state.State, path string, wipe bool, encrypt bool) error {
// Fallthrough. We didn't find a /dev/disk path for this device, use the original path.
}

// Wipe the block device if requested.
if wipe {
err = timeoutWipe(path)
if err != nil {
return fmt.Errorf("Failed to wipe the device: %w", err)
}
}

// Get a OSD number.
nr, err := nextOSD(s)
if err != nil {
return fmt.Errorf("Failed to find next OSD number: %w", err)
}
logger.Debugf("nextOSD number is %d for disk %s", nr, path)
logger.Debugf("nextOSD number is %d for disk %s", nr, data.Path)

// Record the disk.
err = s.Database.Transaction(s.Context, func(ctx context.Context, tx *sql.Tx) error {
_, err := database.CreateDisk(ctx, tx, database.Disk{Member: s.Name(), Path: path, OSD: int(nr)})
_, err := database.CreateDisk(ctx, tx, database.Disk{Member: s.Name(), Path: data.Path, OSD: int(nr)})
if err != nil {
return fmt.Errorf("Failed to record disk: %w", err)
}
Expand All @@ -370,42 +378,36 @@ func AddOSD(s *state.State, path string, wipe bool, encrypt bool) error {
dataPath := filepath.Join(os.Getenv("SNAP_COMMON"), "data")
osdDataPath := filepath.Join(dataPath, "osd", fmt.Sprintf("ceph-%d", nr))

// Keep the old path in case it changes after encrypting.
oldPath := data.Path

// Create directory.
err = os.MkdirAll(osdDataPath, 0700)
if err != nil {
return fmt.Errorf("Failed to bootstrap monitor: %w", err)
lmlg marked this conversation as resolved.
Show resolved Hide resolved
}

// Wipe and/or encrypt the disk if needed.
err = prepareDisk(&data, "", osdDataPath, nr)
if err != nil {
return fmt.Errorf("Failed to prepare data device: %w", err)
}

// if we fail later, make sure we free up the record
revert.Add(func() {
os.RemoveAll(osdDataPath)
s.Database.Transaction(s.Context, func(ctx context.Context, tx *sql.Tx) error {
database.DeleteDisk(ctx, tx, s.Name(), path)
database.DeleteDisk(ctx, tx, s.Name(), oldPath)
return nil
})
})

// Create directory.
err = os.MkdirAll(osdDataPath, 0700)
if err != nil {
return fmt.Errorf("Failed to bootstrap monitor: %w", err)
}

// Generate keyring.
err = genAuth(filepath.Join(osdDataPath, "keyring"), fmt.Sprintf("osd.%d", nr), []string{"mgr", "allow profile osd"}, []string{"mon", "allow profile osd"}, []string{"osd", "allow *"})
if err != nil {
return fmt.Errorf("Failed to generate OSD keyring: %w", err)
}

var blockPath string
if encrypt {
blockPath, err = setupEncryptedOSD(path, osdDataPath, nr)
if err != nil {
return err
}
} else {
blockPath = path
}

// Setup device symlink.
if err = os.Symlink(blockPath, filepath.Join(osdDataPath, "block")); err != nil {
return fmt.Errorf("Failed to add block symlink: %w", err)
}

// Generate OSD uuid.
fsid := uuid.NewRandom().String()

Expand All @@ -416,7 +418,23 @@ func AddOSD(s *state.State, path string, wipe bool, encrypt bool) error {
}

// Bootstrap OSD.
_, err = processExec.RunCommand("ceph-osd", "--mkfs", "--no-mon-config", "-i", fmt.Sprintf("%d", nr))
args := []string{"--mkfs", "--no-mon-config", "-i", fmt.Sprintf("%d", nr)}
if wal != nil {
err = prepareDisk(wal, ".wal", osdDataPath, nr)
if err != nil {
return fmt.Errorf("Failed to set up WAL device: %w", err)
}
args = append(args, []string{"--bluestore-block-wal-path", wal.Path}...)
}
if db != nil {
err = prepareDisk(db, ".db", osdDataPath, nr)
if err != nil {
return fmt.Errorf("Failed to set up DB device: %w", err)
}
args = append(args, []string{"--bluestore-block-db-path", db.Path}...)
}

_, err = processExec.RunCommand("ceph-osd", args...)
if err != nil {
return fmt.Errorf("Failed to bootstrap OSD: %w", err)
}
Expand Down
28 changes: 12 additions & 16 deletions microceph/ceph/osd_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,20 +113,18 @@ func (s *osdSuite) SetupTest() {
// TestSwitchHostFailureDomain tests the switchFailureDomain function
func (s *osdSuite) TestSwitchHostFailureDomain() {
r := mocks.NewRunner(s.T())
// list and create two crush rules
addCrushRuleLsExpectations(r)
addCrushRuleCreateExpectations(r)
addCrushRuleCreateExpectations(r)
// dump crush rules

// dump crush rules to resolve names
addCrushRuleDumpExpectations(r)
// set default crush rule
addSetDefaultRuleExpectations(r)
// list and dump crush rule
// list to check if crush rule exists
addCrushRuleLsExpectations(r)
// dump crush rules to resolve names
addCrushRuleDumpExpectations(r)
// list crush rule json
// list pools
addCrushRuleLsJsonExpectations(r)
// set osd pool
// set pool crush rule
addOsdPoolSetExpectations(r)

processExec = r
Expand All @@ -149,20 +147,18 @@ func (s *osdSuite) TestUpdateFailureDomain() {
}

r := mocks.NewRunner(s.T())
// list and create two crush rules
addCrushRuleLsExpectations(r)
addCrushRuleCreateExpectations(r)
addCrushRuleCreateExpectations(r)
// dump crush rules

// dump crush rules to resolve names
addCrushRuleDumpExpectations(r)
// set default crush rule
addSetDefaultRuleExpectations(r)
// list and dump crush rule
// list to check if crush rule exists
addCrushRuleLsExpectations(r)
// dump crush rules to resolve names
addCrushRuleDumpExpectations(r)
// list crush rule json
// list pools
addCrushRuleLsJsonExpectations(r)
// set osd pool
// set pool crush rule
addOsdPoolSetExpectations(r)

processExec = r
Expand Down
Loading
Loading