import logging
import os
import uuid
-import re
from itertools import repeat
from math import floor
from ceph_volume import process, util, conf
except ValueError:
res_lv = None
return res_lv
-
-def get_lv_path_from_mapper(mapper):
- """
- This functions translates a given mapper device under the format:
- /dev/mapper/LV to the format /dev/VG/LV.
- eg:
- from:
- /dev/mapper/ceph--c1a97e46--234c--46aa--a549--3ca1d1f356a9-osd--block--32e8e896--172e--4a38--a06a--3702598510ec
- to:
- /dev/ceph-c1a97e46-234c-46aa-a549-3ca1d1f356a9/osd-block-32e8e896-172e-4a38-a06a-3702598510ec
- """
- results = re.split(r'^\/dev\/mapper\/(.+\w)-(\w.+)', mapper)
- results = list(filter(None, results))
-
- if len(results) != 2:
- return None
-
- return f"/dev/{results[0].replace('--', '-')}/{results[1].replace('--', '-')}"
-
-def get_mapper_from_lv_path(lv_path):
- """
- This functions translates a given lv path under the format:
- /dev/VG/LV to the format /dev/mapper/LV.
- eg:
- from:
- /dev/ceph-c1a97e46-234c-46aa-a549-3ca1d1f356a9/osd-block-32e8e896-172e-4a38-a06a-3702598510ec
- to:
- /dev/mapper/ceph--c1a97e46--234c--46aa--a549--3ca1d1f356a9-osd--block--32e8e896--172e--4a38--a06a--3702598510ec
- """
- results = re.split(r'^\/dev\/(.+\w)-(\w.+)', lv_path)
- results = list(filter(None, results))
-
- if len(results) != 2:
- return None
-
- return f"/dev/mapper/{results[0].replace('-', '--')}/{results[1].replace('-', '--')}"
assert isinstance(lv_, api.Volume)
assert lv_.name == 'lv1'
-
-
-class TestHelpers:
- def test_get_lv_path_from_mapper(self):
- mapper = '/dev/mapper/ceph--c1a97e46--234c--46aa--a549--3ca1d1f356a9-osd--block--32e8e896--172e--4a38--a06a--3702598510ec'
- lv_path = api.get_lv_path_from_mapper(mapper)
- assert lv_path == '/dev/ceph-c1a97e46-234c-46aa-a549-3ca1d1f356a9/osd-block-32e8e896-172e-4a38-a06a-3702598510ec'
-
- def test_get_mapper_from_lv_path(self):
- lv_path = '/dev/ceph-c1a97e46-234c-46aa-a549-3ca1d1f356a9/osd-block-32e8e896-172e-4a38-a06a-3702598510ec'
- mapper = api.get_mapper_from_lv_path(lv_path)
- assert mapper == '/dev/mapper/ceph--c1a97e46--234c--46aa--a549--3ca1d1f356a9/osd--block--32e8e896--172e--4a38--a06a/3702598510ec'
import pytest
+import stat
from ceph_volume.util import disk
from mock.mock import patch, Mock, MagicMock, mock_open
from pyfakefs.fake_filesystem_unittest import TestCase
assert b.active_mappers()['dm-1']
assert b.active_mappers()['dm-1']['type'] == 'LVM'
assert b.active_mappers()['dm-1']['uuid'] == 'abcdef'
+
+
+class TestUdevData(TestCase):
+ def setUp(self) -> None:
+ udev_data_lv_device: str = """
+S:disk/by-id/dm-uuid-LVM-1f1RaxWlzQ61Sbc7oCIHRMdh0M8zRTSnU03ekuStqWuiA6eEDmwoGg3cWfFtE2li
+S:mapper/vg1-lv1
+S:disk/by-id/dm-name-vg1-lv1
+S:vg1/lv1
+I:837060642207
+E:DM_UDEV_DISABLE_OTHER_RULES_FLAG=
+E:DM_UDEV_DISABLE_LIBRARY_FALLBACK_FLAG=1
+E:DM_UDEV_PRIMARY_SOURCE_FLAG=1
+E:DM_UDEV_RULES_VSN=2
+E:DM_NAME=fake_vg1-fake-lv1
+E:DM_UUID=LVM-1f1RaxWlzQ61Sbc7oCIHRMdh0M8zRTSnU03ekuStqWuiA6eEDmwoGg3cWfFtE2li
+E:DM_SUSPENDED=0
+E:DM_VG_NAME=fake_vg1
+E:DM_LV_NAME=fake-lv1
+E:DM_LV_LAYER=
+E:NVME_HOST_IFACE=none
+E:SYSTEMD_READY=1
+G:systemd
+Q:systemd
+V:1"""
+ udev_data_bare_device: str = """
+S:disk/by-path/pci-0000:00:02.0
+S:disk/by-path/virtio-pci-0000:00:02.0
+S:disk/by-diskseq/1
+I:3037919
+E:ID_PATH=pci-0000:00:02.0
+E:ID_PATH_TAG=pci-0000_00_02_0
+E:ID_PART_TABLE_UUID=baefa409
+E:ID_PART_TABLE_TYPE=dos
+E:NVME_HOST_IFACE=none
+G:systemd
+Q:systemd
+V:1"""
+ self.fake_device: str = '/dev/cephtest'
+ self.setUpPyfakefs()
+ self.fs.create_file(self.fake_device, st_mode=(stat.S_IFBLK | 0o600))
+ self.fs.create_file('/run/udev/data/b999:0', create_missing_dirs=True, contents=udev_data_bare_device)
+ self.fs.create_file('/run/udev/data/b998:1', create_missing_dirs=True, contents=udev_data_lv_device)
+
+ def test_device_not_found(self) -> None:
+ self.fs.remove(self.fake_device)
+ with pytest.raises(RuntimeError):
+ disk.UdevData(self.fake_device)
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=0))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=999))
+ def test_no_data(self) -> None:
+ self.fs.remove('/run/udev/data/b999:0')
+ with pytest.raises(RuntimeError):
+ disk.UdevData(self.fake_device)
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=0))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=999))
+ def test_is_dm_false(self) -> None:
+ assert not disk.UdevData(self.fake_device).is_dm
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=1))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=998))
+ def test_is_dm_true(self) -> None:
+ assert disk.UdevData(self.fake_device).is_dm
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=1))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=998))
+ def test_is_lvm_true(self) -> None:
+ assert disk.UdevData(self.fake_device).is_dm
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=0))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=999))
+ def test_is_lvm_false(self) -> None:
+ assert not disk.UdevData(self.fake_device).is_dm
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=1))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=998))
+ def test_slashed_path_with_lvm(self) -> None:
+ assert disk.UdevData(self.fake_device).slashed_path == '/dev/fake_vg1/fake-lv1'
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=1))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=998))
+ def test_dashed_path_with_lvm(self) -> None:
+ assert disk.UdevData(self.fake_device).dashed_path == '/dev/mapper/fake_vg1-fake-lv1'
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=0))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=999))
+ def test_slashed_path_with_bare_device(self) -> None:
+ assert disk.UdevData(self.fake_device).slashed_path == '/dev/cephtest'
+
+ @patch('ceph_volume.util.disk.os.stat', MagicMock())
+ @patch('ceph_volume.util.disk.os.minor', Mock(return_value=0))
+ @patch('ceph_volume.util.disk.os.major', Mock(return_value=999))
+ def test_dashed_path_with_bare_device(self) -> None:
+ assert disk.UdevData(self.fake_device).dashed_path == '/dev/cephtest'
\ No newline at end of file
for block in block_devs:
metadata: Dict[str, Any] = {}
if block[2] == 'lvm':
- block[1] = lvm.get_lv_path_from_mapper(block[1])
+ block[1] = UdevData(block[1]).slashed_path
devname = os.path.basename(block[0])
diskname = block[1]
if block[2] not in block_types:
if mapper_type == 'LVM':
result[holder]['uuid'] = content_split[1]
return result
+
+class UdevData:
+ """
+ Class representing udev data for a specific device.
+ This class extracts and stores relevant information about the device from udev files.
+
+ Attributes:
+ -----------
+ path : str
+ The initial device path (e.g., /dev/sda).
+ realpath : str
+ The resolved real path of the device.
+ stats : os.stat_result
+ The result of the os.stat() call to retrieve device metadata.
+ major : int
+ The device's major number.
+ minor : int
+ The device's minor number.
+ udev_data_path : str
+ The path to the udev metadata for the device (e.g., /run/udev/data/b<major>:<minor>).
+ symlinks : List[str]
+ A list of symbolic links pointing to the device.
+ id : str
+ A unique identifier for the device.
+ environment : Dict[str, str]
+ A dictionary containing environment variables extracted from the udev data.
+ group : str
+ The group associated with the device.
+ queue : str
+ The queue associated with the device.
+ version : str
+ The version of the device or its metadata.
+ """
+ def __init__(self, path: str) -> None:
+ """Initialize an instance of the UdevData class and load udev information.
+
+ Args:
+ path (str): The path to the device to be analyzed (e.g., /dev/sda).
+
+ Raises:
+ RuntimeError: Raised if no udev data file is found for the specified device.
+ """
+ if not os.path.exists(path):
+ raise RuntimeError(f'{path} not found.')
+ self.path: str = path
+ self.realpath: str = os.path.realpath(self.path)
+ self.stats: os.stat_result = os.stat(self.realpath)
+ self.major: int = os.major(self.stats.st_rdev)
+ self.minor: int = os.minor(self.stats.st_rdev)
+ self.udev_data_path: str = f'/run/udev/data/b{self.major}:{self.minor}'
+ self.symlinks: List[str] = []
+ self.id: str = ''
+ self.environment: Dict[str, str] = {}
+ self.group: str = ''
+ self.queue: str = ''
+ self.version: str = ''
+
+ if not os.path.exists(self.udev_data_path):
+ raise RuntimeError(f'No udev data could be retrieved for {self.path}')
+
+ with open(self.udev_data_path, 'r') as f:
+ content: str = f.read().strip()
+ self.raw_data: List[str] = content.split('\n')
+
+ for line in self.raw_data:
+ data_type, data = line.split(':', 1)
+ if data_type == 'S':
+ self.symlinks.append(data)
+ if data_type == 'I':
+ self.id = data
+ if data_type == 'E':
+ key, value = data.split('=')
+ self.environment[key] = value
+ if data_type == 'G':
+ self.group = data
+ if data_type == 'Q':
+ self.queue = data
+ if data_type == 'V':
+ self.version = data
+
+ @property
+ def is_dm(self) -> bool:
+ """Check if the device is a device mapper (DM).
+
+ Returns:
+ bool: True if the device is a device mapper, otherwise False.
+ """
+ return 'DM_UUID' in self.environment.keys()
+
+ @property
+ def is_lvm(self) -> bool:
+ """Check if the device is a Logical Volume Manager (LVM) volume.
+
+ Returns:
+ bool: True if the device is an LVM volume, otherwise False.
+ """
+ return self.environment.get('DM_UUID', '').startswith('LVM')
+
+ @property
+ def slashed_path(self) -> str:
+ """Get the LVM path structured with slashes.
+
+ Returns:
+ str: A path using slashes if the device is an LVM volume (e.g., /dev/vgname/lvname),
+ otherwise the original path.
+ """
+ result: str = self.path
+ if self.is_lvm:
+ vg: str = self.environment.get('DM_VG_NAME')
+ lv: str = self.environment.get('DM_LV_NAME')
+ result = f'/dev/{vg}/{lv}'
+ return result
+
+ @property
+ def dashed_path(self) -> str:
+ """Get the LVM path structured with dashes.
+
+ Returns:
+ str: A path using dashes if the device is an LVM volume (e.g., /dev/mapper/vgname-lvname),
+ otherwise the original path.
+ """
+ result: str = self.path
+ if self.is_lvm:
+ name: str = self.environment.get('DM_NAME')
+ result = f'/dev/mapper/{name}'
+ return result
projects / ceph.git / commitdiff