assert result[lv_path]['type'] == 'lvm'
assert result[lv_path]['human_readable_size'] == '100.00 MB'
+ def test_nvme_reads_vendor_model_rev_under_controller(
+ self, patched_get_block_devs_sysfs, fake_filesystem
+ ):
+ nvme_path = '/dev/nvme0n1'
+ patched_get_block_devs_sysfs.return_value = [
+ [nvme_path, nvme_path, 'disk', nvme_path]
+ ]
+ fake_filesystem.create_dir('/sys/block/nvme0n1/slaves')
+ fake_filesystem.create_dir('/sys/block/nvme0n1/queue')
+ fake_filesystem.create_file(
+ '/sys/block/nvme0n1/device/nvme0/device/vendor',
+ contents='Samsung',
+ )
+ fake_filesystem.create_file(
+ '/sys/block/nvme0n1/device/nvme0/device/model',
+ contents='SSD 990 PRO',
+ )
+ fake_filesystem.create_file(
+ '/sys/block/nvme0n1/device/nvme0/device/rev',
+ contents='1B2Q',
+ )
+ with patch('ceph_volume.util.disk.UdevData') as MockUdevData:
+ mock_instance = MagicMock()
+ mock_instance.is_lvm = False
+ MockUdevData.return_value = mock_instance
+ result = disk.get_devices()
+ assert result[nvme_path]['vendor'] == 'Samsung'
+ assert result[nvme_path]['model'] == 'SSD 990 PRO'
+ assert result[nvme_path]['rev'] == '1B2Q'
+
class TestSizeCalculations(object):
class TestNvmePreformat:
@patch('ceph_volume.util.nvme.process.call')
- def test_non_nvme_device_skips_preformat(self, m_call):
+ def test_non_nvme_device_skips_preformat(self, m_call, fake_filesystem):
assert nvme.preformat('/dev/sda') is False
m_call.assert_not_called()
- @patch('ceph_volume.util.nvme.disk.is_partition', return_value=False)
@patch('ceph_volume.util.nvme.disk.is_device', return_value=True)
@patch('ceph_volume.util.nvme.process.call', return_value=([], [], 0))
- def test_preformat_invokes_nvme_cli(self, m_call, *_m_disk):
+ def test_preformat_invokes_nvme_cli(self, m_call, m_is_device, fake_filesystem):
+ fake_filesystem.create_dir('/sys/block/nvme0n1/device/nvme0')
assert nvme.preformat('/dev/nvme0n1') is True
m_call.assert_called_once_with(
['nvme', 'format', '/dev/nvme0n1', '--force'],
verbose_on_failure=True
)
- @patch('ceph_volume.util.nvme.disk.is_partition', return_value=False)
@patch('ceph_volume.util.nvme.disk.is_device', return_value=True)
@patch('ceph_volume.util.nvme.process.call', return_value=([], [], 1))
- def test_preformat_handles_non_zero_rc(self, m_call, *_m_disk):
+ def test_preformat_handles_non_zero_rc(self, m_call, m_is_device, fake_filesystem):
+ fake_filesystem.create_dir('/sys/block/nvme0n1/device/nvme0')
assert nvme.preformat('/dev/nvme0n1') is False
assert m_call.called
- @patch('ceph_volume.util.nvme.disk.is_partition', return_value=False)
@patch('ceph_volume.util.nvme.disk.is_device', return_value=True)
@patch('ceph_volume.util.nvme.process.call', side_effect=FileNotFoundError('missing nvme'))
- def test_preformat_handles_missing_cli(self, m_call, *_m_disk):
+ def test_preformat_handles_missing_cli(self, m_call, m_is_device, fake_filesystem):
+ fake_filesystem.create_dir('/sys/block/nvme0n1/device/nvme0')
assert nvme.preformat('/dev/nvme0n1') is False
assert m_call.called
- @patch('ceph_volume.util.nvme.disk.is_partition', return_value=True)
- def test_partition_is_not_formatted(self, *_):
+ def test_partition_is_not_formatted(self, fake_filesystem):
+ fake_filesystem.create_file('/sys/block/nvme0n1p1/partition', contents='1')
assert nvme.preformat('/dev/nvme0n1p1') is False
-
--- /dev/null
+from ceph_volume.util import nvme_sysfs
+
+
+class TestBlockDeviceFactSysfsRelPath:
+ def test_scsi_uses_standard_path_when_present(self, fake_filesystem):
+ fake_filesystem.create_file('/sys/block/sda/device/vendor', contents='ATA')
+ assert nvme_sysfs.block_device_fact_sysfs_rel_path(
+ '/sys/block/sda', 'device/vendor'
+ ) == 'device/vendor'
+
+ def test_scsi_falls_back_when_no_vendor_file(self, fake_filesystem):
+ fake_filesystem.create_dir('/sys/block/sda/device')
+ assert nvme_sysfs.block_device_fact_sysfs_rel_path(
+ '/sys/block/sda', 'device/vendor'
+ ) == 'device/vendor'
+
+ def test_nvme_remapped_via_sysfs(self, fake_filesystem):
+ fake_filesystem.create_file(
+ '/sys/block/nvme0n1/device/nvme0/device/vendor',
+ contents='Samsung',
+ )
+ assert nvme_sysfs.block_device_fact_sysfs_rel_path(
+ '/sys/block/nvme0n1', 'device/vendor'
+ ) == 'device/nvme0/device/vendor'
+
+ def test_nvme_partition_remapped_via_sysfs(self, fake_filesystem):
+ fake_filesystem.create_file(
+ '/sys/block/nvme1n2p3/device/nvme1/device/model',
+ contents='foo',
+ )
+ assert nvme_sysfs.block_device_fact_sysfs_rel_path(
+ '/sys/block/nvme1n2p3', 'device/model'
+ ) == 'device/nvme1/device/model'
+
+ def test_nvme_nonstandard_controller_name(self, fake_filesystem):
+ """Kernel may use names like nvme0c0 for the controller sysfs link."""
+ fake_filesystem.create_file(
+ '/sys/block/nvme0c0n1/device/nvme0c0/device/vendor',
+ contents='vend',
+ )
+ assert nvme_sysfs.block_device_fact_sysfs_rel_path(
+ '/sys/block/nvme0c0n1', 'device/vendor'
+ ) == 'device/nvme0c0/device/vendor'
+
+ def test_ambiguous_multiple_matches_returns_default_rel(self, fake_filesystem):
+ fake_filesystem.create_file(
+ '/sys/block/weird/device/nvme0/device/vendor', contents='a'
+ )
+ fake_filesystem.create_file(
+ '/sys/block/weird/device/nvme1/device/vendor', contents='b'
+ )
+ assert nvme_sysfs.block_device_fact_sysfs_rel_path(
+ '/sys/block/weird', 'device/vendor'
+ ) == 'device/vendor'
+
+ def test_non_controller_fact_unchanged(self):
+ assert nvme_sysfs.block_device_fact_sysfs_rel_path(
+ '/sys/block/nvme0n1', 'removable'
+ ) == 'removable'
+
+
+class TestIsWholeNvmeNamespaceName:
+ def test_whole_namespace(self, fake_filesystem):
+ fake_filesystem.create_dir('/sys/block/nvme0n1/device/nvme0')
+ assert nvme_sysfs.is_whole_nvme_namespace_name('nvme0n1') is True
+
+ def test_partition_false(self, fake_filesystem):
+ fake_filesystem.create_file('/sys/block/nvme0n1p1/partition', contents='1')
+ fake_filesystem.create_dir('/sys/block/nvme0n1p1/device/nvme0')
+ assert nvme_sysfs.is_whole_nvme_namespace_name('nvme0n1p1') is False
+
+ def test_scsi_false(self, fake_filesystem):
+ fake_filesystem.create_dir('/sys/block/sda/device')
+ assert nvme_sysfs.is_whole_nvme_namespace_name('sda') is False
+
+ def test_no_nvme_controller_link_false(self, fake_filesystem):
+ fake_filesystem.create_dir('/sys/block/nvme0n1/device')
+ assert nvme_sysfs.is_whole_nvme_namespace_name('nvme0n1') is False
+
+ def test_missing_sysdir_false(self):
+ assert nvme_sysfs.is_whole_nvme_namespace_name(
+ 'nvme0n1', _sys_block_path='/this/path/does/not/exist'
+ ) is False
import time
import json
from ceph_volume import process, allow_loop_devices
+from ceph_volume.util import nvme_sysfs
from ceph_volume.util.system import get_file_contents
from typing import Dict, List, Any, Union, Optional
('nr_requests', 'queue/nr_requests'),
]
for key, file_ in facts:
- metadata[key] = get_file_contents(os.path.join(sysdir, file_))
+ rel = nvme_sysfs.block_device_fact_sysfs_rel_path(sysdir, file_)
+ metadata[key] = get_file_contents(os.path.join(sysdir, rel))
device_slaves = []
if block[2] != 'part':
import os
from ceph_volume import process, terminal
-from ceph_volume.util import disk
+from ceph_volume.util import disk, nvme_sysfs
logger = logging.getLogger(__name__)
We only format whole NVMe devices (e.g. /dev/nvme0n1). Partitions like
/dev/nvme0n1p1 and non-block-device paths are intentionally skipped.
"""
- if not resolved_device.startswith('/dev/nvme'):
+ if not nvme_sysfs.is_whole_nvme_namespace_name(os.path.basename(resolved_device)):
return False
if not disk.is_device(resolved_device):
# disk.is_device() already excludes partitions
--- /dev/null
+import glob
+import os
+
+_SYS_BLOCK = '/sys/block'
+
+
+def block_device_fact_sysfs_rel_path(sysdir: str, rel: str) -> str:
+ """
+ Resolve the sysfs path (relative to `sysdir`, typically `/sys/block/<devname>/`) for
+ block device metadata files such as vendor/model/rev.
+
+ If the usual path `<sysdir>/<rel>` exists, `rel` is returned.
+ Otherwise, for nvme style layouts, looks for `<sysdir>/device/nvme*/device/<basename>`
+ and returns the matching path relative to `sysdir` when exactly one match exists.
+ """
+ if rel not in ('device/vendor', 'device/model', 'device/rev'):
+ return rel
+ sysdir = os.path.normpath(sysdir)
+ default_path = os.path.join(sysdir, rel)
+ if os.path.exists(default_path):
+ return rel
+ basename = os.path.basename(rel)
+ pattern = os.path.join(sysdir, 'device', 'nvme*', 'device', basename)
+ matches = sorted(glob.glob(pattern))
+ if len(matches) == 1:
+ return os.path.relpath(matches[0], sysdir)
+ return rel
+
+
+def is_whole_nvme_namespace_name(basename: str, _sys_block_path: str = _SYS_BLOCK) -> bool:
+ """
+ True if `basename` is a whole nvme namespace (example: nvme0n1), not a partition.
+
+ Uses sysfs only: partitions expose a `partition` file; whole NVMe
+ namespaces have a controller entry under `device/nvme*`.
+ """
+ sysdir = os.path.join(_sys_block_path, basename)
+ if not os.path.isdir(sysdir):
+ return False
+ if os.path.exists(os.path.join(sysdir, 'partition')):
+ return False
+ return bool(glob.glob(os.path.join(sysdir, 'device', 'nvme*')))
projects / ceph.git / commitdiff