diff options
Diffstat (limited to '')
33 files changed, 5353 insertions, 0 deletions
diff --git a/src/python-common/.gitignore b/src/python-common/.gitignore new file mode 100644 index 000000000..c2de8195a --- /dev/null +++ b/src/python-common/.gitignore @@ -0,0 +1,3 @@ +ceph.egg-info +build +setup.cfg diff --git a/src/python-common/CMakeLists.txt b/src/python-common/CMakeLists.txt new file mode 100644 index 000000000..e89bbe2fe --- /dev/null +++ b/src/python-common/CMakeLists.txt @@ -0,0 +1,7 @@ +include(Distutils) +distutils_install_module(ceph) + +if(WITH_TESTS) + include(AddCephTest) + add_tox_test(python-common TOX_ENVS py3 lint) +endif() diff --git a/src/python-common/README.rst b/src/python-common/README.rst new file mode 100644 index 000000000..3900ec4f9 --- /dev/null +++ b/src/python-common/README.rst @@ -0,0 +1,22 @@ +ceph-python-common +================== + +This library is meant to be used to keep common data structures and +functions usable throughout the Ceph project. + +Like for example: + +- All different Cython bindings. +- MGR modules. +- ``ceph`` command line interface and other Ceph tools. +- Also external tools. + +Usage +===== + +From within the Ceph git, just import it: + +.. code:: python + + from ceph.deployment_utils import DriveGroupSpec + from ceph.exceptions import OSError diff --git a/src/python-common/ceph/__init__.py b/src/python-common/ceph/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/src/python-common/ceph/__init__.py diff --git a/src/python-common/ceph/deployment/__init__.py b/src/python-common/ceph/deployment/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/src/python-common/ceph/deployment/__init__.py diff --git a/src/python-common/ceph/deployment/drive_group.py b/src/python-common/ceph/deployment/drive_group.py new file mode 100644 index 000000000..dc36a60d9 --- /dev/null +++ b/src/python-common/ceph/deployment/drive_group.py @@ -0,0 +1,348 @@ +import enum +import yaml + +from ceph.deployment.inventory import Device +from ceph.deployment.service_spec import ServiceSpec, PlacementSpec +from ceph.deployment.hostspec import SpecValidationError + +try: + from typing import Optional, List, Dict, Any, Union +except ImportError: + pass + + +class OSDMethod(str, enum.Enum): + raw = 'raw' + lvm = 'lvm' + + +class DeviceSelection(object): + """ + Used within :class:`ceph.deployment.drive_group.DriveGroupSpec` to specify the devices + used by the Drive Group. + + Any attributes (even none) can be included in the device + specification structure. + """ + + _supported_filters = [ + "paths", "size", "vendor", "model", "rotational", "limit", "all" + ] + + def __init__(self, + paths=None, # type: Optional[List[str]] + model=None, # type: Optional[str] + size=None, # type: Optional[str] + rotational=None, # type: Optional[bool] + limit=None, # type: Optional[int] + vendor=None, # type: Optional[str] + all=False, # type: bool + ): + """ + ephemeral drive group device specification + """ + #: List of Device objects for devices paths. + self.paths = [] if paths is None else [Device(path) for path in paths] # type: List[Device] + + #: A wildcard string. e.g: "SDD*" or "SanDisk SD8SN8U5" + self.model = model + + #: Match on the VENDOR property of the drive + self.vendor = vendor + + #: Size specification of format LOW:HIGH. + #: Can also take the the form :HIGH, LOW: + #: or an exact value (as ceph-volume inventory reports) + self.size: Optional[str] = size + + #: is the drive rotating or not + self.rotational = rotational + + #: Limit the number of devices added to this Drive Group. Devices + #: are used from top to bottom in the output of ``ceph-volume inventory`` + self.limit = limit + + #: Matches all devices. Can only be used for data devices + self.all = all + + def validate(self, name: str) -> None: + props = [self.model, self.vendor, self.size, self.rotational] # type: List[Any] + if self.paths and any(p is not None for p in props): + raise DriveGroupValidationError( + name, + 'device selection: `paths` and other parameters are mutually exclusive') + is_empty = not any(p is not None and p != [] for p in [self.paths] + props) + if not self.all and is_empty: + raise DriveGroupValidationError(name, 'device selection cannot be empty') + + if self.all and not is_empty: + raise DriveGroupValidationError( + name, + 'device selection: `all` and other parameters are mutually exclusive. {}'.format( + repr(self))) + + @classmethod + def from_json(cls, device_spec): + # type: (dict) -> Optional[DeviceSelection] + if not device_spec: + return None + for applied_filter in list(device_spec.keys()): + if applied_filter not in cls._supported_filters: + raise KeyError(applied_filter) + + return cls(**device_spec) + + def to_json(self): + # type: () -> Dict[str, Any] + ret: Dict[str, Any] = {} + if self.paths: + ret['paths'] = [p.path for p in self.paths] + if self.model: + ret['model'] = self.model + if self.vendor: + ret['vendor'] = self.vendor + if self.size: + ret['size'] = self.size + if self.rotational is not None: + ret['rotational'] = self.rotational + if self.limit: + ret['limit'] = self.limit + if self.all: + ret['all'] = self.all + + return ret + + def __repr__(self) -> str: + keys = [ + key for key in self._supported_filters + ['limit'] if getattr(self, key) is not None + ] + if 'paths' in keys and self.paths == []: + keys.remove('paths') + return "DeviceSelection({})".format( + ', '.join('{}={}'.format(key, repr(getattr(self, key))) for key in keys) + ) + + def __eq__(self, other: Any) -> bool: + return repr(self) == repr(other) + + +class DriveGroupValidationError(SpecValidationError): + """ + Defining an exception here is a bit problematic, cause you cannot properly catch it, + if it was raised in a different mgr module. + """ + + def __init__(self, name: Optional[str], msg: str): + name = name or "<unnamed>" + super(DriveGroupValidationError, self).__init__( + f'Failed to validate OSD spec "{name}": {msg}') + + +class DriveGroupSpec(ServiceSpec): + """ + Describe a drive group in the same form that ceph-volume + understands. + """ + + _supported_features = [ + "encrypted", "block_wal_size", "osds_per_device", + "db_slots", "wal_slots", "block_db_size", "placement", "service_id", "service_type", + "data_devices", "db_devices", "wal_devices", "journal_devices", + "data_directories", "osds_per_device", "objectstore", "osd_id_claims", + "journal_size", "unmanaged", "filter_logic", "preview_only", "extra_container_args", + "data_allocate_fraction", "method", "crush_device_class", + ] + + def __init__(self, + placement=None, # type: Optional[PlacementSpec] + service_id=None, # type: Optional[str] + data_devices=None, # type: Optional[DeviceSelection] + db_devices=None, # type: Optional[DeviceSelection] + wal_devices=None, # type: Optional[DeviceSelection] + journal_devices=None, # type: Optional[DeviceSelection] + data_directories=None, # type: Optional[List[str]] + osds_per_device=None, # type: Optional[int] + objectstore='bluestore', # type: str + encrypted=False, # type: bool + db_slots=None, # type: Optional[int] + wal_slots=None, # type: Optional[int] + osd_id_claims=None, # type: Optional[Dict[str, List[str]]] + block_db_size=None, # type: Union[int, str, None] + block_wal_size=None, # type: Union[int, str, None] + journal_size=None, # type: Union[int, str, None] + service_type=None, # type: Optional[str] + unmanaged=False, # type: bool + filter_logic='AND', # type: str + preview_only=False, # type: bool + extra_container_args=None, # type: Optional[List[str]] + data_allocate_fraction=None, # type: Optional[float] + method=None, # type: Optional[OSDMethod] + crush_device_class=None, # type: Optional[str] + ): + assert service_type is None or service_type == 'osd' + super(DriveGroupSpec, self).__init__('osd', service_id=service_id, + placement=placement, + unmanaged=unmanaged, + preview_only=preview_only, + extra_container_args=extra_container_args) + + #: A :class:`ceph.deployment.drive_group.DeviceSelection` + self.data_devices = data_devices + + #: A :class:`ceph.deployment.drive_group.DeviceSelection` + self.db_devices = db_devices + + #: A :class:`ceph.deployment.drive_group.DeviceSelection` + self.wal_devices = wal_devices + + #: A :class:`ceph.deployment.drive_group.DeviceSelection` + self.journal_devices = journal_devices + + #: Set (or override) the "bluestore_block_wal_size" value, in bytes + self.block_wal_size: Union[int, str, None] = block_wal_size + + #: Set (or override) the "bluestore_block_db_size" value, in bytes + self.block_db_size: Union[int, str, None] = block_db_size + + #: set journal_size in bytes + self.journal_size: Union[int, str, None] = journal_size + + #: Number of osd daemons per "DATA" device. + #: To fully utilize nvme devices multiple osds are required. + self.osds_per_device = osds_per_device + + #: A list of strings, containing paths which should back OSDs + self.data_directories = data_directories + + #: ``filestore`` or ``bluestore`` + self.objectstore = objectstore + + #: ``true`` or ``false`` + self.encrypted = encrypted + + #: How many OSDs per DB device + self.db_slots = db_slots + + #: How many OSDs per WAL device + self.wal_slots = wal_slots + + #: Optional: mapping of host -> List of osd_ids that should be replaced + #: See :ref:`orchestrator-osd-replace` + self.osd_id_claims = osd_id_claims or dict() + + #: The logic gate we use to match disks with filters. + #: defaults to 'AND' + self.filter_logic = filter_logic.upper() + + #: If this should be treated as a 'preview' spec + self.preview_only = preview_only + + #: Allocate a fraction of the data device (0,1.0] + self.data_allocate_fraction = data_allocate_fraction + + self.method = method + + #: Crush device class to assign to OSDs + self.crush_device_class = crush_device_class + + @classmethod + def _from_json_impl(cls, json_drive_group): + # type: (dict) -> DriveGroupSpec + """ + Initialize 'Drive group' structure + + :param json_drive_group: A valid json string with a Drive Group + specification + """ + args: Dict[str, Any] = json_drive_group.copy() + # legacy json (pre Octopus) + if 'host_pattern' in args and 'placement' not in args: + args['placement'] = {'host_pattern': args['host_pattern']} + del args['host_pattern'] + + s_id = args.get('service_id', '<unnamed>') + + # spec: was not mandatory in octopus + if 'spec' in args: + args['spec'].update(cls._drive_group_spec_from_json(s_id, args['spec'])) + else: + args.update(cls._drive_group_spec_from_json(s_id, args)) + + return super(DriveGroupSpec, cls)._from_json_impl(args) + + @classmethod + def _drive_group_spec_from_json(cls, name: str, json_drive_group: dict) -> dict: + for applied_filter in list(json_drive_group.keys()): + if applied_filter not in cls._supported_features: + raise DriveGroupValidationError( + name, + "Feature `{}` is not supported".format(applied_filter)) + + try: + def to_selection(key: str, vals: dict) -> Optional[DeviceSelection]: + try: + return DeviceSelection.from_json(vals) + except KeyError as e: + raise DriveGroupValidationError( + f'{name}.{key}', + f"Filtering for `{e.args[0]}` is not supported") + + args = {k: (to_selection(k, v) if k.endswith('_devices') else v) for k, v in + json_drive_group.items()} + if not args: + raise DriveGroupValidationError(name, "Didn't find drive selections") + return args + except (KeyError, TypeError) as e: + raise DriveGroupValidationError(name, str(e)) + + def validate(self): + # type: () -> None + super(DriveGroupSpec, self).validate() + + if self.placement.is_empty(): + raise DriveGroupValidationError(self.service_id, '`placement` required') + + if self.data_devices is None: + raise DriveGroupValidationError(self.service_id, "`data_devices` element is required.") + + specs_names = "data_devices db_devices wal_devices journal_devices".split() + specs = dict(zip(specs_names, [getattr(self, k) for k in specs_names])) + for k, s in [ks for ks in specs.items() if ks[1] is not None]: + assert s is not None + s.validate(f'{self.service_id}.{k}') + for s in filter(None, [self.db_devices, self.wal_devices, self.journal_devices]): + if s.all: + raise DriveGroupValidationError( + self.service_id, + "`all` is only allowed for data_devices") + + if self.objectstore not in ('bluestore'): + raise DriveGroupValidationError(self.service_id, + f"{self.objectstore} is not supported. Must be " + f"one of ('bluestore')") + + if self.block_wal_size is not None and type(self.block_wal_size) not in [int, str]: + raise DriveGroupValidationError( + self.service_id, + 'block_wal_size must be of type int or string') + if self.block_db_size is not None and type(self.block_db_size) not in [int, str]: + raise DriveGroupValidationError( + self.service_id, + 'block_db_size must be of type int or string') + if self.journal_size is not None and type(self.journal_size) not in [int, str]: + raise DriveGroupValidationError( + self.service_id, + 'journal_size must be of type int or string') + + if self.filter_logic not in ['AND', 'OR']: + raise DriveGroupValidationError( + self.service_id, + 'filter_logic must be either <AND> or <OR>') + + if self.method not in [None, 'lvm', 'raw']: + raise DriveGroupValidationError(self.service_id, 'method must be one of None, lvm, raw') + if self.method == 'raw' and self.objectstore == 'filestore': + raise DriveGroupValidationError(self.service_id, 'method raw only supports bluestore') + + +yaml.add_representer(DriveGroupSpec, DriveGroupSpec.yaml_representer) diff --git a/src/python-common/ceph/deployment/drive_selection/__init__.py b/src/python-common/ceph/deployment/drive_selection/__init__.py new file mode 100644 index 000000000..994e2f2da --- /dev/null +++ b/src/python-common/ceph/deployment/drive_selection/__init__.py @@ -0,0 +1,2 @@ +from .selector import DriveSelection # NOQA +from .matchers import Matcher, SubstringMatcher, EqualityMatcher, AllMatcher, SizeMatcher # NOQA diff --git a/src/python-common/ceph/deployment/drive_selection/example.yaml b/src/python-common/ceph/deployment/drive_selection/example.yaml new file mode 100644 index 000000000..2851e7dbb --- /dev/null +++ b/src/python-common/ceph/deployment/drive_selection/example.yaml @@ -0,0 +1,21 @@ +# default: +# target: 'data*' +# data_devices: +# size: 20G +# db_devices: +# size: 10G +# rotational: 1 +# allflash: +# target: 'fast_nodes*' +# data_devices: +# size: 100G +# db_devices: +# size: 50G +# rotational: 0 + +# This is the default configuration and +# will create an OSD on all available drives +default: + target: 'fnmatch_target' + data_devices: + all: true diff --git a/src/python-common/ceph/deployment/drive_selection/filter.py b/src/python-common/ceph/deployment/drive_selection/filter.py new file mode 100644 index 000000000..36cacfaa5 --- /dev/null +++ b/src/python-common/ceph/deployment/drive_selection/filter.py @@ -0,0 +1,34 @@ +# -*- coding: utf-8 -*- + +import logging + +from ceph.deployment.drive_group import DeviceSelection + +try: + from typing import Generator +except ImportError: + pass + +from .matchers import Matcher, SubstringMatcher, AllMatcher, SizeMatcher, EqualityMatcher + +logger = logging.getLogger(__name__) + + +class FilterGenerator(object): + def __init__(self, device_filter): + # type: (DeviceSelection) -> None + self.device_filter = device_filter + + def __iter__(self): + # type: () -> Generator[Matcher, None, None] + if self.device_filter.size: + yield SizeMatcher('size', self.device_filter.size) + if self.device_filter.model: + yield SubstringMatcher('model', self.device_filter.model) + if self.device_filter.vendor: + yield SubstringMatcher('vendor', self.device_filter.vendor) + if self.device_filter.rotational is not None: + val = '1' if self.device_filter.rotational else '0' + yield EqualityMatcher('rotational', val) + if self.device_filter.all: + yield AllMatcher('all', str(self.device_filter.all)) diff --git a/src/python-common/ceph/deployment/drive_selection/matchers.py b/src/python-common/ceph/deployment/drive_selection/matchers.py new file mode 100644 index 000000000..f423c2f43 --- /dev/null +++ b/src/python-common/ceph/deployment/drive_selection/matchers.py @@ -0,0 +1,412 @@ +# -*- coding: utf-8 -*- + +from typing import Tuple, Optional, Any, Union, Iterator + +from ceph.deployment.inventory import Device + +import re +import logging + +logger = logging.getLogger(__name__) + + +class _MatchInvalid(Exception): + pass + + +# pylint: disable=too-few-public-methods +class Matcher(object): + """ The base class to all Matchers + + It holds utility methods such as _get_disk_key + and handles the initialization. + + """ + + def __init__(self, key, value): + # type: (str, Any) -> None + """ Initialization of Base class + + :param str key: Attribute like 'model, size or vendor' + :param str value: Value of attribute like 'X123, 5G or samsung' + """ + self.key = key + self.value = value + self.fallback_key = '' # type: Optional[str] + + def _get_disk_key(self, device): + # type: (Device) -> Any + """ Helper method to safely extract values form the disk dict + + There is a 'key' and a _optional_ 'fallback' key that can be used. + The reason for this is that the output of ceph-volume is not always + consistent (due to a bug currently, but you never know). + There is also a safety measure for a disk_key not existing on + virtual environments. ceph-volume apparently sources its information + from udev which seems to not populate certain fields on VMs. + + :raises: A generic Exception when no disk_key could be found. + :return: A disk value + :rtype: str + """ + # using the . notation, but some keys are nested, and hidden behind + # a different hierarchy, which makes it harder to access programatically + # hence, make it a dict. + disk = device.to_json() + + def findkeys(node: Union[list, dict], key_val: str) -> Iterator[str]: + """ Find keys in non-flat dict recursively """ + if isinstance(node, list): + for i in node: + for key in findkeys(i, key_val): + yield key + elif isinstance(node, dict): + if key_val in node: + yield node[key_val] + for j in node.values(): + for key in findkeys(j, key_val): + yield key + + disk_value = list(findkeys(disk, self.key)) + if not disk_value and self.fallback_key: + disk_value = list(findkeys(disk, self.fallback_key)) + + if disk_value: + return disk_value[0] + else: + raise _MatchInvalid("No value found for {} or {}".format( + self.key, self.fallback_key)) + + def compare(self, disk): + # type: (Device) -> bool + """ Implements a valid comparison method for a SubMatcher + This will get overwritten by the individual classes + + :param dict disk: A disk representation + """ + raise NotImplementedError + + +# pylint: disable=too-few-public-methods +class SubstringMatcher(Matcher): + """ Substring matcher subclass + """ + + def __init__(self, key, value, fallback_key=None): + # type: (str, str, Optional[str]) -> None + Matcher.__init__(self, key, value) + self.fallback_key = fallback_key + + def compare(self, disk): + # type: (Device) -> bool + """ Overwritten method to match substrings + + This matcher does substring matching + :param dict disk: A disk representation (see base for examples) + :return: True/False if the match succeeded + :rtype: bool + """ + if not disk: + return False + disk_value = self._get_disk_key(disk) + if str(self.value) in disk_value: + return True + return False + + +# pylint: disable=too-few-public-methods +class AllMatcher(Matcher): + """ All matcher subclass + """ + + def __init__(self, key, value, fallback_key=None): + # type: (str, Any, Optional[str]) -> None + + Matcher.__init__(self, key, value) + self.fallback_key = fallback_key + + def compare(self, disk): + # type: (Device) -> bool + + """ Overwritten method to match all + + A rather dumb matcher that just accepts all disks + (regardless of the value) + :param dict disk: A disk representation (see base for examples) + :return: always True + :rtype: bool + """ + if not disk: + return False + return True + + +# pylint: disable=too-few-public-methods +class EqualityMatcher(Matcher): + """ Equality matcher subclass + """ + + def __init__(self, key, value): + # type: (str, Any) -> None + + Matcher.__init__(self, key, value) + + def compare(self, disk): + # type: (Device) -> bool + + """ Overwritten method to match equality + + This matcher does value comparison + :param dict disk: A disk representation + :return: True/False if the match succeeded + :rtype: bool + """ + if not disk: + return False + disk_value = self._get_disk_key(disk) + ret = disk_value == self.value + if not ret: + logger.debug('{} != {}'.format(disk_value, self.value)) + return ret + + +class SizeMatcher(Matcher): + """ Size matcher subclass + """ + + SUFFIXES = ( + ["KB", "MB", "GB", "TB"], + ["K", "M", "G", "T"], + [1e+3, 1e+6, 1e+9, 1e+12] + ) + + supported_suffixes = SUFFIXES[0] + SUFFIXES[1] + + # pylint: disable=too-many-instance-attributes + def __init__(self, key, value): + # type: (str, str) -> None + + # The 'key' value is overwritten here because + # the user_defined attribute does not neccessarily + # correspond to the desired attribute + # requested from the inventory output + Matcher.__init__(self, key, value) + self.key = "human_readable_size" + self.fallback_key = "size" + self._high = None # type: Optional[str] + self._high_suffix = None # type: Optional[str] + self._low = None # type: Optional[str] + self._low_suffix = None # type: Optional[str] + self._exact = None # type: Optional[str] + self._exact_suffix = None # type: Optional[str] + self._parse_filter() + + @property + def low(self): + # type: () -> Tuple[Optional[str], Optional[str]] + """ Getter for 'low' matchers + """ + return self._low, self._low_suffix + + @low.setter + def low(self, low): + # type: (Tuple[str, str]) -> None + """ Setter for 'low' matchers + """ + self._low, self._low_suffix = low + + @property + def high(self): + # type: () -> Tuple[Optional[str], Optional[str]] + """ Getter for 'high' matchers + """ + return self._high, self._high_suffix + + @high.setter + def high(self, high): + # type: (Tuple[str, str]) -> None + """ Setter for 'high' matchers + """ + self._high, self._high_suffix = high + + @property + def exact(self): + # type: () -> Tuple[Optional[str], Optional[str]] + """ Getter for 'exact' matchers + """ + return self._exact, self._exact_suffix + + @exact.setter + def exact(self, exact): + # type: (Tuple[str, str]) -> None + """ Setter for 'exact' matchers + """ + self._exact, self._exact_suffix = exact + + @classmethod + def _normalize_suffix(cls, suffix): + # type: (str) -> str + """ Normalize any supported suffix + + Since the Drive Groups are user facing, we simply + can't make sure that all users type in the requested + form. That's why we have to internally agree on one format. + It also checks if any of the supported suffixes was used + and raises an Exception otherwise. + + :param str suffix: A suffix ('G') or ('M') + :return: A normalized output + :rtype: str + """ + suffix = suffix.upper() + if suffix not in cls.supported_suffixes: + raise _MatchInvalid("Unit '{}' not supported".format(suffix)) + return dict(zip( + cls.SUFFIXES[1], + cls.SUFFIXES[0], + )).get(suffix, suffix) + + @classmethod + def _parse_suffix(cls, obj): + # type: (str) -> str + """ Wrapper method to find and normalize a prefix + + :param str obj: A size filtering string ('10G') + :return: A normalized unit ('GB') + :rtype: str + """ + return cls._normalize_suffix(re.findall(r"[a-zA-Z]+", obj)[0]) + + @classmethod + def _get_k_v(cls, data): + # type: (str) -> Tuple[str, str] + """ Helper method to extract data from a string + + It uses regex to extract all digits and calls _parse_suffix + which also uses a regex to extract all letters and normalizes + the resulting suffix. + + :param str data: A size filtering string ('10G') + :return: A Tuple with normalized output (10, 'GB') + :rtype: tuple + """ + return re.findall(r"\d+\.?\d*", data)[0], cls._parse_suffix(data) + + def _parse_filter(self) -> None: + """ Identifies which type of 'size' filter is applied + + There are four different filtering modes: + + 1) 10G:50G (high-low) + At least 10G but at max 50G of size + + 2) :60G + At max 60G of size + + 3) 50G: + At least 50G of size + + 4) 20G + Exactly 20G in size + + This method uses regex to identify and extract this information + and raises if none could be found. + """ + low_high = re.match(r"\d+[A-Z]{1,2}:\d+[A-Z]{1,2}", self.value) + if low_high: + low, high = low_high.group().split(":") + self.low = self._get_k_v(low) + self.high = self._get_k_v(high) + + low = re.match(r"\d+[A-Z]{1,2}:$", self.value) + if low: + self.low = self._get_k_v(low.group()) + + high = re.match(r"^:\d+[A-Z]{1,2}", self.value) + if high: + self.high = self._get_k_v(high.group()) + + exact = re.match(r"^\d+\.?\d*[A-Z]{1,2}$", self.value) + if exact: + self.exact = self._get_k_v(exact.group()) + + if not self.low and not self.high and not self.exact: + raise _MatchInvalid("Couldn't parse {}".format(self.value)) + + @staticmethod + # pylint: disable=inconsistent-return-statements + def to_byte(tpl): + # type: (Tuple[Optional[str], Optional[str]]) -> float + + """ Convert any supported unit to bytes + + :param tuple tpl: A tuple with ('10', 'GB') + :return: The converted byte value + :rtype: float + """ + val_str, suffix = tpl + value = float(val_str) if val_str is not None else 0.0 + return dict(zip( + SizeMatcher.SUFFIXES[0], + SizeMatcher.SUFFIXES[2], + )).get(str(suffix), 0.00) * value + + @staticmethod + def str_to_byte(input): + # type: (str) -> float + return SizeMatcher.to_byte(SizeMatcher._get_k_v(input)) + + # pylint: disable=inconsistent-return-statements, too-many-return-statements + def compare(self, disk): + # type: (Device) -> bool + """ Convert MB/GB/TB down to bytes and compare + + 1) Extracts information from the to-be-inspected disk. + 2) Depending on the mode, apply checks and return + + # This doesn't seem very solid and _may_ + be re-factored + + + """ + if not disk: + return False + disk_value = self._get_disk_key(disk) + # This doesn't neccessarily have to be a float. + # The current output from ceph-volume gives a float.. + # This may change in the future.. + # todo: harden this paragraph + if not disk_value: + logger.warning("Could not retrieve value for disk") + return False + + disk_size = re.findall(r"\d+\.\d+", disk_value)[0] + disk_suffix = self._parse_suffix(disk_value) + disk_size_in_byte = self.to_byte((disk_size, disk_suffix)) + + if all(self.high) and all(self.low): + if disk_size_in_byte <= self.to_byte( + self.high) and disk_size_in_byte >= self.to_byte(self.low): + return True + # is a else: return False neccessary here? + # (and in all other branches) + logger.debug("Disk didn't match for 'high/low' filter") + + elif all(self.low) and not all(self.high): + if disk_size_in_byte >= self.to_byte(self.low): + return True + logger.debug("Disk didn't match for 'low' filter") + + elif all(self.high) and not all(self.low): + if disk_size_in_byte <= self.to_byte(self.high): + return True + logger.debug("Disk didn't match for 'high' filter") + + elif all(self.exact): + if disk_size_in_byte == self.to_byte(self.exact): + return True + logger.debug("Disk didn't match for 'exact' filter") + else: + logger.debug("Neither high, low, nor exact was given") + raise _MatchInvalid("No filters applied") + return False diff --git a/src/python-common/ceph/deployment/drive_selection/selector.py b/src/python-common/ceph/deployment/drive_selection/selector.py new file mode 100644 index 000000000..07b0549f3 --- /dev/null +++ b/src/python-common/ceph/deployment/drive_selection/selector.py @@ -0,0 +1,179 @@ +import logging + +from typing import List, Optional, Dict, Callable + +from ..inventory import Device +from ..drive_group import DriveGroupSpec, DeviceSelection, DriveGroupValidationError + +from .filter import FilterGenerator +from .matchers import _MatchInvalid + +logger = logging.getLogger(__name__) + + +def to_dg_exception(f: Callable) -> Callable[['DriveSelection', str, + Optional['DeviceSelection']], + List['Device']]: + def wrapper(self: 'DriveSelection', name: str, ds: Optional['DeviceSelection']) -> List[Device]: + try: + return f(self, ds) + except _MatchInvalid as e: + raise DriveGroupValidationError(f'{self.spec.service_id}.{name}', e.args[0]) + return wrapper + + +class DriveSelection(object): + def __init__(self, + spec, # type: DriveGroupSpec + disks, # type: List[Device] + existing_daemons=None, # type: Optional[int] + ): + self.disks = disks.copy() + self.spec = spec + self.existing_daemons = existing_daemons or 0 + + self._data = self.assign_devices('data_devices', self.spec.data_devices) + self._wal = self.assign_devices('wal_devices', self.spec.wal_devices) + self._db = self.assign_devices('db_devices', self.spec.db_devices) + self._journal = self.assign_devices('journal_devices', self.spec.journal_devices) + + def data_devices(self): + # type: () -> List[Device] + return self._data + + def wal_devices(self): + # type: () -> List[Device] + return self._wal + + def db_devices(self): + # type: () -> List[Device] + return self._db + + def journal_devices(self): + # type: () -> List[Device] + return self._journal + + def _limit_reached(self, device_filter, len_devices, + disk_path): + # type: (DeviceSelection, int, str) -> bool + """ Check for the <limit> property and apply logic + + If a limit is set in 'device_attrs' we have to stop adding + disks at some point. + + If limit is set (>0) and len(devices) >= limit + + :param int len_devices: Length of the already populated device set/list + :param str disk_path: The disk identifier (for logging purposes) + :return: True/False if the device should be added to the list of devices + :rtype: bool + """ + limit = device_filter.limit or 0 + + if limit > 0 and (len_devices + self.existing_daemons >= limit): + logger.info("Refuse to add {} due to limit policy of <{}>".format( + disk_path, limit)) + return True + return False + + @staticmethod + def _has_mandatory_idents(disk): + # type: (Device) -> bool + """ Check for mandatory identification fields + """ + if disk.path: + logger.debug("Found matching disk: {}".format(disk.path)) + return True + else: + raise Exception( + "Disk {} doesn't have a 'path' identifier".format(disk)) + + @to_dg_exception + def assign_devices(self, device_filter): + # type: (Optional[DeviceSelection]) -> List[Device] + """ Assign drives based on used filters + + Do not add disks when: + + 1) Filter didn't match + 2) Disk doesn't have a mandatory identification item (path) + 3) The set :limit was reached + + After the disk was added we make sure not to re-assign this disk + for another defined type[wal/db/journal devices] + + return a sorted(by path) list of devices + """ + + if not device_filter: + logger.debug('device_filter is None') + return [] + + if not self.spec.data_devices: + logger.debug('data_devices is None') + return [] + + if device_filter.paths: + logger.debug('device filter is using explicit paths') + return device_filter.paths + + devices = list() # type: List[Device] + for disk in self.disks: + logger.debug("Processing disk {}".format(disk.path)) + + if not disk.available and not disk.ceph_device: + logger.debug( + ("Ignoring disk {}. " + "Disk is unavailable due to {}".format(disk.path, disk.rejected_reasons)) + ) + continue + + if not self._has_mandatory_idents(disk): + logger.debug( + "Ignoring disk {}. Missing mandatory idents".format( + disk.path)) + continue + + # break on this condition. + if self._limit_reached(device_filter, len(devices), disk.path): + logger.debug("Ignoring disk {}. Limit reached".format( + disk.path)) + break + + if disk in devices: + continue + + if self.spec.filter_logic == 'AND': + if not all(m.compare(disk) for m in FilterGenerator(device_filter)): + logger.debug( + "Ignoring disk {}. Not all filter did match the disk".format( + disk.path)) + continue + + if self.spec.filter_logic == 'OR': + if not any(m.compare(disk) for m in FilterGenerator(device_filter)): + logger.debug( + "Ignoring disk {}. No filter matched the disk".format( + disk.path)) + continue + + logger.debug('Adding disk {}'.format(disk.path)) + devices.append(disk) + + # This disk is already taken and must not be re-assigned. + for taken_device in devices: + if taken_device in self.disks: + self.disks.remove(taken_device) + + return sorted([x for x in devices], key=lambda dev: dev.path) + + def __repr__(self) -> str: + selection: Dict[str, List[str]] = { + 'data devices': [d.path for d in self._data], + 'wal_devices': [d.path for d in self._wal], + 'db devices': [d.path for d in self._db], + 'journal devices': [d.path for d in self._journal] + } + return "DeviceSelection({})".format( + ', '.join('{}={}'.format(key, selection[key]) for key in selection.keys()) + ) diff --git a/src/python-common/ceph/deployment/hostspec.py b/src/python-common/ceph/deployment/hostspec.py new file mode 100644 index 000000000..1bf686f97 --- /dev/null +++ b/src/python-common/ceph/deployment/hostspec.py @@ -0,0 +1,135 @@ +from collections import OrderedDict +import errno +import re +from typing import Optional, List, Any, Dict + + +def assert_valid_host(name: str) -> None: + p = re.compile('^[a-zA-Z0-9-]+$') + try: + assert len(name) <= 250, 'name is too long (max 250 chars)' + for part in name.split('.'): + assert len(part) > 0, '.-delimited name component must not be empty' + assert len(part) <= 63, '.-delimited name component must not be more than 63 chars' + assert p.match(part), 'name component must include only a-z, 0-9, and -' + except AssertionError as e: + raise SpecValidationError(str(e)) + + +class SpecValidationError(Exception): + """ + Defining an exception here is a bit problematic, cause you cannot properly catch it, + if it was raised in a different mgr module. + """ + def __init__(self, + msg: str, + errno: int = -errno.EINVAL): + super(SpecValidationError, self).__init__(msg) + self.errno = errno + + +class HostSpec(object): + """ + Information about hosts. Like e.g. ``kubectl get nodes`` + """ + def __init__(self, + hostname: str, + addr: Optional[str] = None, + labels: Optional[List[str]] = None, + status: Optional[str] = None, + location: Optional[Dict[str, str]] = None, + ): + self.service_type = 'host' + + #: the bare hostname on the host. Not the FQDN. + self.hostname = hostname # type: str + + #: DNS name or IP address to reach it + self.addr = addr or hostname # type: str + + #: label(s), if any + self.labels = labels or [] # type: List[str] + + #: human readable status + self.status = status or '' # type: str + + self.location = location + + def validate(self) -> None: + assert_valid_host(self.hostname) + + def to_json(self) -> Dict[str, Any]: + r: Dict[str, Any] = { + 'hostname': self.hostname, + 'addr': self.addr, + 'labels': list(OrderedDict.fromkeys((self.labels))), + 'status': self.status, + } + if self.location: + r['location'] = self.location + return r + + @classmethod + def from_json(cls, host_spec: dict) -> 'HostSpec': + host_spec = cls.normalize_json(host_spec) + _cls = cls( + host_spec['hostname'], + host_spec['addr'] if 'addr' in host_spec else None, + list(OrderedDict.fromkeys( + host_spec['labels'])) if 'labels' in host_spec else None, + host_spec['status'] if 'status' in host_spec else None, + host_spec.get('location'), + ) + return _cls + + @staticmethod + def normalize_json(host_spec: dict) -> dict: + labels = host_spec.get('labels') + if labels is not None: + if isinstance(labels, str): + host_spec['labels'] = [labels] + elif ( + not isinstance(labels, list) + or any(not isinstance(v, str) for v in labels) + ): + raise SpecValidationError( + f'Labels ({labels}) must be a string or list of strings' + ) + + loc = host_spec.get('location') + if loc is not None: + if ( + not isinstance(loc, dict) + or any(not isinstance(k, str) for k in loc.keys()) + or any(not isinstance(v, str) for v in loc.values()) + ): + raise SpecValidationError( + f'Location ({loc}) must be a dictionary of strings to strings' + ) + + return host_spec + + def __repr__(self) -> str: + args = [self.hostname] # type: List[Any] + if self.addr is not None: + args.append(self.addr) + if self.labels: + args.append(self.labels) + if self.status: + args.append(self.status) + if self.location: + args.append(self.location) + + return "HostSpec({})".format(', '.join(map(repr, args))) + + def __str__(self) -> str: + if self.hostname != self.addr: + return f'{self.hostname} ({self.addr})' + return self.hostname + + def __eq__(self, other: Any) -> bool: + # Let's omit `status` for the moment, as it is still the very same host. + return self.hostname == other.hostname and \ + self.addr == other.addr and \ + sorted(self.labels) == sorted(other.labels) and \ + self.location == other.location diff --git a/src/python-common/ceph/deployment/inventory.py b/src/python-common/ceph/deployment/inventory.py new file mode 100644 index 000000000..570c1dbb3 --- /dev/null +++ b/src/python-common/ceph/deployment/inventory.py @@ -0,0 +1,132 @@ +try: + from typing import List, Optional, Dict, Any, Union +except ImportError: + pass # for type checking + +from ceph.utils import datetime_now, datetime_to_str, str_to_datetime +import datetime +import json + + +class Devices(object): + """ + A container for Device instances with reporting + """ + + def __init__(self, devices): + # type: (List[Device]) -> None + self.devices = devices # type: List[Device] + + def __eq__(self, other: Any) -> bool: + if not isinstance(other, Devices): + return NotImplemented + if len(self.devices) != len(other.devices): + return False + for d1, d2 in zip(other.devices, self.devices): + if d1 != d2: + return False + return True + + def to_json(self): + # type: () -> List[dict] + return [d.to_json() for d in self.devices] + + @classmethod + def from_json(cls, input): + # type: (List[Dict[str, Any]]) -> Devices + return cls([Device.from_json(i) for i in input]) + + def copy(self): + # type: () -> Devices + return Devices(devices=list(self.devices)) + + +class Device(object): + report_fields = [ + 'ceph_device', + 'rejected_reasons', + 'available', + 'path', + 'sys_api', + 'created', + 'lvs', + 'human_readable_type', + 'device_id', + 'lsm_data', + ] + + def __init__(self, + path, # type: str + sys_api=None, # type: Optional[Dict[str, Any]] + available=None, # type: Optional[bool] + rejected_reasons=None, # type: Optional[List[str]] + lvs=None, # type: Optional[List[str]] + device_id=None, # type: Optional[str] + lsm_data=None, # type: Optional[Dict[str, Dict[str, str]]] + created=None, # type: Optional[datetime.datetime] + ceph_device=None # type: Optional[bool] + ): + self.path = path + self.sys_api = sys_api if sys_api is not None else {} # type: Dict[str, Any] + self.available = available + self.rejected_reasons = rejected_reasons if rejected_reasons is not None else [] + self.lvs = lvs + self.device_id = device_id + self.lsm_data = lsm_data if lsm_data is not None else {} # type: Dict[str, Dict[str, str]] + self.created = created if created is not None else datetime_now() + self.ceph_device = ceph_device + + def __eq__(self, other): + # type: (Any) -> bool + if not isinstance(other, Device): + return NotImplemented + diff = [k for k in self.report_fields if k != 'created' and (getattr(self, k) + != getattr(other, k))] + return not diff + + def to_json(self): + # type: () -> dict + return { + k: (getattr(self, k) if k != 'created' + or not isinstance(getattr(self, k), datetime.datetime) + else datetime_to_str(getattr(self, k))) + for k in self.report_fields + } + + @classmethod + def from_json(cls, input): + # type: (Dict[str, Any]) -> Device + if not isinstance(input, dict): + raise ValueError('Device: Expected dict. Got `{}...`'.format(json.dumps(input)[:10])) + ret = cls( + **{ + key: (input.get(key, None) if key != 'created' + or not input.get(key, None) + else str_to_datetime(input.get(key, None))) + for key in Device.report_fields + if key != 'human_readable_type' + } + ) + if ret.rejected_reasons: + ret.rejected_reasons = sorted(ret.rejected_reasons) + return ret + + @property + def human_readable_type(self): + # type: () -> str + if self.sys_api is None or 'rotational' not in self.sys_api: + return "unknown" + return 'hdd' if self.sys_api["rotational"] == "1" else 'ssd' + + def __repr__(self) -> str: + device_desc: Dict[str, Union[str, List[str]]] = { + 'path': self.path if self.path is not None else 'unknown', + 'lvs': self.lvs if self.lvs else 'None', + 'available': str(self.available), + 'ceph_device': str(self.ceph_device) + } + if not self.available and self.rejected_reasons: + device_desc['rejection reasons'] = self.rejected_reasons + return "Device({})".format( + ', '.join('{}={}'.format(key, device_desc[key]) for key in device_desc.keys()) + ) diff --git a/src/python-common/ceph/deployment/service_spec.py b/src/python-common/ceph/deployment/service_spec.py new file mode 100644 index 000000000..17130ea9a --- /dev/null +++ b/src/python-common/ceph/deployment/service_spec.py @@ -0,0 +1,1384 @@ +import fnmatch +import re +import enum +from collections import OrderedDict +from contextlib import contextmanager +from functools import wraps +from ipaddress import ip_network, ip_address +from typing import Optional, Dict, Any, List, Union, Callable, Iterable, Type, TypeVar, cast, \ + NamedTuple, Mapping, Iterator + +import yaml + +from ceph.deployment.hostspec import HostSpec, SpecValidationError, assert_valid_host +from ceph.deployment.utils import unwrap_ipv6, valid_addr +from ceph.utils import is_hex + +ServiceSpecT = TypeVar('ServiceSpecT', bound='ServiceSpec') +FuncT = TypeVar('FuncT', bound=Callable) + + +def handle_type_error(method: FuncT) -> FuncT: + @wraps(method) + def inner(cls: Any, *args: Any, **kwargs: Any) -> Any: + try: + return method(cls, *args, **kwargs) + except (TypeError, AttributeError) as e: + error_msg = '{}: {}'.format(cls.__name__, e) + raise SpecValidationError(error_msg) + return cast(FuncT, inner) + + +class HostPlacementSpec(NamedTuple): + hostname: str + network: str + name: str + + def __str__(self) -> str: + res = '' + res += self.hostname + if self.network: + res += ':' + self.network + if self.name: + res += '=' + self.name + return res + + @classmethod + @handle_type_error + def from_json(cls, data: Union[dict, str]) -> 'HostPlacementSpec': + if isinstance(data, str): + return cls.parse(data) + return cls(**data) + + def to_json(self) -> str: + return str(self) + + @classmethod + def parse(cls, host, require_network=True): + # type: (str, bool) -> HostPlacementSpec + """ + Split host into host, network, and (optional) daemon name parts. The network + part can be an IP, CIDR, or ceph addrvec like '[v2:1.2.3.4:3300,v1:1.2.3.4:6789]'. + e.g., + "myhost" + "myhost=name" + "myhost:1.2.3.4" + "myhost:1.2.3.4=name" + "myhost:1.2.3.0/24" + "myhost:1.2.3.0/24=name" + "myhost:[v2:1.2.3.4:3000]=name" + "myhost:[v2:1.2.3.4:3000,v1:1.2.3.4:6789]=name" + """ + # Matches from start to : or = or until end of string + host_re = r'^(.*?)(:|=|$)' + # Matches from : to = or until end of string + ip_re = r':(.*?)(=|$)' + # Matches from = to end of string + name_re = r'=(.*?)$' + + # assign defaults + host_spec = cls('', '', '') + + match_host = re.search(host_re, host) + if match_host: + host_spec = host_spec._replace(hostname=match_host.group(1)) + + name_match = re.search(name_re, host) + if name_match: + host_spec = host_spec._replace(name=name_match.group(1)) + + ip_match = re.search(ip_re, host) + if ip_match: + host_spec = host_spec._replace(network=ip_match.group(1)) + + if not require_network: + return host_spec + + networks = list() # type: List[str] + network = host_spec.network + # in case we have [v2:1.2.3.4:3000,v1:1.2.3.4:6478] + if ',' in network: + networks = [x for x in network.split(',')] + else: + if network != '': + networks.append(network) + + for network in networks: + # only if we have versioned network configs + if network.startswith('v') or network.startswith('[v'): + # if this is ipv6 we can't just simply split on ':' so do + # a split once and rsplit once to leave us with just ipv6 addr + network = network.split(':', 1)[1] + network = network.rsplit(':', 1)[0] + try: + # if subnets are defined, also verify the validity + if '/' in network: + ip_network(network) + else: + ip_address(unwrap_ipv6(network)) + except ValueError as e: + # logging? + raise e + host_spec.validate() + return host_spec + + def validate(self) -> None: + assert_valid_host(self.hostname) + + +class PlacementSpec(object): + """ + For APIs that need to specify a host subset + """ + + def __init__(self, + label=None, # type: Optional[str] + hosts=None, # type: Union[List[str],List[HostPlacementSpec], None] + count=None, # type: Optional[int] + count_per_host=None, # type: Optional[int] + host_pattern=None, # type: Optional[str] + ): + # type: (...) -> None + self.label = label + self.hosts = [] # type: List[HostPlacementSpec] + + if hosts: + self.set_hosts(hosts) + + self.count = count # type: Optional[int] + self.count_per_host = count_per_host # type: Optional[int] + + #: fnmatch patterns to select hosts. Can also be a single host. + self.host_pattern = host_pattern # type: Optional[str] + + self.validate() + + def is_empty(self) -> bool: + return ( + self.label is None + and not self.hosts + and not self.host_pattern + and self.count is None + and self.count_per_host is None + ) + + def __eq__(self, other: Any) -> bool: + if isinstance(other, PlacementSpec): + return self.label == other.label \ + and self.hosts == other.hosts \ + and self.count == other.count \ + and self.host_pattern == other.host_pattern \ + and self.count_per_host == other.count_per_host + return NotImplemented + + def set_hosts(self, hosts: Union[List[str], List[HostPlacementSpec]]) -> None: + # To backpopulate the .hosts attribute when using labels or count + # in the orchestrator backend. + if all([isinstance(host, HostPlacementSpec) for host in hosts]): + self.hosts = hosts # type: ignore + else: + self.hosts = [HostPlacementSpec.parse(x, require_network=False) # type: ignore + for x in hosts if x] + + # deprecated + def filter_matching_hosts(self, _get_hosts_func: Callable) -> List[str]: + return self.filter_matching_hostspecs(_get_hosts_func(as_hostspec=True)) + + def filter_matching_hostspecs(self, hostspecs: Iterable[HostSpec]) -> List[str]: + if self.hosts: + all_hosts = [hs.hostname for hs in hostspecs] + return [h.hostname for h in self.hosts if h.hostname in all_hosts] + if self.label: + return [hs.hostname for hs in hostspecs if self.label in hs.labels] + all_hosts = [hs.hostname for hs in hostspecs] + if self.host_pattern: + return fnmatch.filter(all_hosts, self.host_pattern) + return all_hosts + + def get_target_count(self, hostspecs: Iterable[HostSpec]) -> int: + if self.count: + return self.count + return len(self.filter_matching_hostspecs(hostspecs)) * (self.count_per_host or 1) + + def pretty_str(self) -> str: + """ + >>> #doctest: +SKIP + ... ps = PlacementSpec(...) # For all placement specs: + ... PlacementSpec.from_string(ps.pretty_str()) == ps + """ + kv = [] + if self.hosts: + kv.append(';'.join([str(h) for h in self.hosts])) + if self.count: + kv.append('count:%d' % self.count) + if self.count_per_host: + kv.append('count-per-host:%d' % self.count_per_host) + if self.label: + kv.append('label:%s' % self.label) + if self.host_pattern: + kv.append(self.host_pattern) + return ';'.join(kv) + + def __repr__(self) -> str: + kv = [] + if self.count: + kv.append('count=%d' % self.count) + if self.count_per_host: + kv.append('count_per_host=%d' % self.count_per_host) + if self.label: + kv.append('label=%s' % repr(self.label)) + if self.hosts: + kv.append('hosts={!r}'.format(self.hosts)) + if self.host_pattern: + kv.append('host_pattern={!r}'.format(self.host_pattern)) + return "PlacementSpec(%s)" % ', '.join(kv) + + @classmethod + @handle_type_error + def from_json(cls, data: dict) -> 'PlacementSpec': + c = data.copy() + hosts = c.get('hosts', []) + if hosts: + c['hosts'] = [] + for host in hosts: + c['hosts'].append(HostPlacementSpec.from_json(host)) + _cls = cls(**c) + _cls.validate() + return _cls + + def to_json(self) -> dict: + r: Dict[str, Any] = {} + if self.label: + r['label'] = self.label + if self.hosts: + r['hosts'] = [host.to_json() for host in self.hosts] + if self.count: + r['count'] = self.count + if self.count_per_host: + r['count_per_host'] = self.count_per_host + if self.host_pattern: + r['host_pattern'] = self.host_pattern + return r + + def validate(self) -> None: + if self.hosts and self.label: + # TODO: a less generic Exception + raise SpecValidationError('Host and label are mutually exclusive') + if self.count is not None: + try: + intval = int(self.count) + except (ValueError, TypeError): + raise SpecValidationError("num/count must be a numeric value") + if self.count != intval: + raise SpecValidationError("num/count must be an integer value") + if self.count < 1: + raise SpecValidationError("num/count must be >= 1") + if self.count_per_host is not None: + try: + intval = int(self.count_per_host) + except (ValueError, TypeError): + raise SpecValidationError("count-per-host must be a numeric value") + if self.count_per_host != intval: + raise SpecValidationError("count-per-host must be an integer value") + if self.count_per_host < 1: + raise SpecValidationError("count-per-host must be >= 1") + if self.count_per_host is not None and not ( + self.label + or self.hosts + or self.host_pattern + ): + raise SpecValidationError( + "count-per-host must be combined with label or hosts or host_pattern" + ) + if self.count is not None and self.count_per_host is not None: + raise SpecValidationError("cannot combine count and count-per-host") + if ( + self.count_per_host is not None + and self.hosts + and any([hs.network or hs.name for hs in self.hosts]) + ): + raise SpecValidationError( + "count-per-host cannot be combined explicit placement with names or networks" + ) + if self.host_pattern: + if not isinstance(self.host_pattern, str): + raise SpecValidationError('host_pattern must be of type string') + if self.hosts: + raise SpecValidationError('cannot combine host patterns and hosts') + + for h in self.hosts: + h.validate() + + @classmethod + def from_string(cls, arg): + # type: (Optional[str]) -> PlacementSpec + """ + A single integer is parsed as a count: + + >>> PlacementSpec.from_string('3') + PlacementSpec(count=3) + + A list of names is parsed as host specifications: + + >>> PlacementSpec.from_string('host1 host2') + PlacementSpec(hosts=[HostPlacementSpec(hostname='host1', network='', name=''), HostPlacemen\ +tSpec(hostname='host2', network='', name='')]) + + You can also prefix the hosts with a count as follows: + + >>> PlacementSpec.from_string('2 host1 host2') + PlacementSpec(count=2, hosts=[HostPlacementSpec(hostname='host1', network='', name=''), Hos\ +tPlacementSpec(hostname='host2', network='', name='')]) + + You can specify labels using `label:<label>` + + >>> PlacementSpec.from_string('label:mon') + PlacementSpec(label='mon') + + Labels also support a count: + + >>> PlacementSpec.from_string('3 label:mon') + PlacementSpec(count=3, label='mon') + + fnmatch is also supported: + + >>> PlacementSpec.from_string('data[1-3]') + PlacementSpec(host_pattern='data[1-3]') + + >>> PlacementSpec.from_string(None) + PlacementSpec() + """ + if arg is None or not arg: + strings = [] + elif isinstance(arg, str): + if ' ' in arg: + strings = arg.split(' ') + elif ';' in arg: + strings = arg.split(';') + elif ',' in arg and '[' not in arg: + # FIXME: this isn't quite right. we want to avoid breaking + # a list of mons with addrvecs... so we're basically allowing + # , most of the time, except when addrvecs are used. maybe + # ok? + strings = arg.split(',') + else: + strings = [arg] + else: + raise SpecValidationError('invalid placement %s' % arg) + + count = None + count_per_host = None + if strings: + try: + count = int(strings[0]) + strings = strings[1:] + except ValueError: + pass + for s in strings: + if s.startswith('count:'): + try: + count = int(s[len('count:'):]) + strings.remove(s) + break + except ValueError: + pass + for s in strings: + if s.startswith('count-per-host:'): + try: + count_per_host = int(s[len('count-per-host:'):]) + strings.remove(s) + break + except ValueError: + pass + + advanced_hostspecs = [h for h in strings if + (':' in h or '=' in h or not any(c in '[]?*:=' for c in h)) and + 'label:' not in h] + for a_h in advanced_hostspecs: + strings.remove(a_h) + + labels = [x for x in strings if 'label:' in x] + if len(labels) > 1: + raise SpecValidationError('more than one label provided: {}'.format(labels)) + for l in labels: + strings.remove(l) + label = labels[0][6:] if labels else None + + host_patterns = strings + if len(host_patterns) > 1: + raise SpecValidationError( + 'more than one host pattern provided: {}'.format(host_patterns)) + + ps = PlacementSpec(count=count, + count_per_host=count_per_host, + hosts=advanced_hostspecs, + label=label, + host_pattern=host_patterns[0] if host_patterns else None) + return ps + + +_service_spec_from_json_validate = True + + +@contextmanager +def service_spec_allow_invalid_from_json() -> Iterator[None]: + """ + I know this is evil, but unfortunately `ceph orch ls` + may return invalid OSD specs for OSDs not associated to + and specs. If you have a better idea, please! + """ + global _service_spec_from_json_validate + _service_spec_from_json_validate = False + yield + _service_spec_from_json_validate = True + + +class ServiceSpec(object): + """ + Details of service creation. + + Request to the orchestrator for a cluster of daemons + such as MDS, RGW, iscsi gateway, MONs, MGRs, Prometheus + + This structure is supposed to be enough information to + start the services. + """ + KNOWN_SERVICE_TYPES = 'alertmanager crash grafana iscsi mds mgr mon nfs ' \ + 'node-exporter osd prometheus rbd-mirror rgw ' \ + 'container cephadm-exporter ingress cephfs-mirror snmp-gateway'.split() + REQUIRES_SERVICE_ID = 'iscsi mds nfs rgw container ingress '.split() + MANAGED_CONFIG_OPTIONS = [ + 'mds_join_fs', + ] + + @classmethod + def _cls(cls: Type[ServiceSpecT], service_type: str) -> Type[ServiceSpecT]: + from ceph.deployment.drive_group import DriveGroupSpec + + ret = { + 'rgw': RGWSpec, + 'nfs': NFSServiceSpec, + 'osd': DriveGroupSpec, + 'mds': MDSSpec, + 'iscsi': IscsiServiceSpec, + 'alertmanager': AlertManagerSpec, + 'ingress': IngressSpec, + 'container': CustomContainerSpec, + 'grafana': GrafanaSpec, + 'node-exporter': MonitoringSpec, + 'prometheus': PrometheusSpec, + 'snmp-gateway': SNMPGatewaySpec, + }.get(service_type, cls) + if ret == ServiceSpec and not service_type: + raise SpecValidationError('Spec needs a "service_type" key.') + return ret + + def __new__(cls: Type[ServiceSpecT], *args: Any, **kwargs: Any) -> ServiceSpecT: + """ + Some Python foo to make sure, we don't have an object + like `ServiceSpec('rgw')` of type `ServiceSpec`. Now we have: + + >>> type(ServiceSpec('rgw')) == type(RGWSpec('rgw')) + True + + """ + if cls != ServiceSpec: + return object.__new__(cls) + service_type = kwargs.get('service_type', args[0] if args else None) + sub_cls: Any = cls._cls(service_type) + return object.__new__(sub_cls) + + def __init__(self, + service_type: str, + service_id: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + count: Optional[int] = None, + config: Optional[Dict[str, str]] = None, + unmanaged: bool = False, + preview_only: bool = False, + networks: Optional[List[str]] = None, + extra_container_args: Optional[List[str]] = None, + ): + + #: See :ref:`orchestrator-cli-placement-spec`. + self.placement = PlacementSpec() if placement is None else placement # type: PlacementSpec + + assert service_type in ServiceSpec.KNOWN_SERVICE_TYPES, service_type + #: The type of the service. Needs to be either a Ceph + #: service (``mon``, ``crash``, ``mds``, ``mgr``, ``osd`` or + #: ``rbd-mirror``), a gateway (``nfs`` or ``rgw``), part of the + #: monitoring stack (``alertmanager``, ``grafana``, ``node-exporter`` or + #: ``prometheus``) or (``container``) for custom containers. + self.service_type = service_type + + #: The name of the service. Required for ``iscsi``, ``mds``, ``nfs``, ``osd``, ``rgw``, + #: ``container``, ``ingress`` + self.service_id = None + + if self.service_type in self.REQUIRES_SERVICE_ID or self.service_type == 'osd': + self.service_id = service_id + + #: If set to ``true``, the orchestrator will not deploy nor remove + #: any daemon associated with this service. Placement and all other properties + #: will be ignored. This is useful, if you do not want this service to be + #: managed temporarily. For cephadm, See :ref:`cephadm-spec-unmanaged` + self.unmanaged = unmanaged + self.preview_only = preview_only + + #: A list of network identities instructing the daemons to only bind + #: on the particular networks in that list. In case the cluster is distributed + #: across multiple networks, you can add multiple networks. See + #: :ref:`cephadm-monitoring-networks-ports`, + #: :ref:`cephadm-rgw-networks` and :ref:`cephadm-mgr-networks`. + self.networks: List[str] = networks or [] + + self.config: Optional[Dict[str, str]] = None + if config: + self.config = {k.replace(' ', '_'): v for k, v in config.items()} + + self.extra_container_args: Optional[List[str]] = extra_container_args + + @classmethod + @handle_type_error + def from_json(cls: Type[ServiceSpecT], json_spec: Dict) -> ServiceSpecT: + """ + Initialize 'ServiceSpec' object data from a json structure + + There are two valid styles for service specs: + + the "old" style: + + .. code:: yaml + + service_type: nfs + service_id: foo + pool: mypool + namespace: myns + + and the "new" style: + + .. code:: yaml + + service_type: nfs + service_id: foo + config: + some_option: the_value + networks: [10.10.0.0/16] + spec: + pool: mypool + namespace: myns + + In https://tracker.ceph.com/issues/45321 we decided that we'd like to + prefer the new style as it is more readable and provides a better + understanding of what fields are special for a give service type. + + Note, we'll need to stay compatible with both versions for the + the next two major releases (octoups, pacific). + + :param json_spec: A valid dict with ServiceSpec + + :meta private: + """ + if not isinstance(json_spec, dict): + raise SpecValidationError( + f'Service Spec is not an (JSON or YAML) object. got "{str(json_spec)}"') + + json_spec = cls.normalize_json(json_spec) + + c = json_spec.copy() + + # kludge to make `from_json` compatible to `Orchestrator.describe_service` + # Open question: Remove `service_id` form to_json? + if c.get('service_name', ''): + service_type_id = c['service_name'].split('.', 1) + + if not c.get('service_type', ''): + c['service_type'] = service_type_id[0] + if not c.get('service_id', '') and len(service_type_id) > 1: + c['service_id'] = service_type_id[1] + del c['service_name'] + + service_type = c.get('service_type', '') + _cls = cls._cls(service_type) + + if 'status' in c: + del c['status'] # kludge to make us compatible to `ServiceDescription.to_json()` + + return _cls._from_json_impl(c) # type: ignore + + @staticmethod + def normalize_json(json_spec: dict) -> dict: + networks = json_spec.get('networks') + if networks is None: + return json_spec + if isinstance(networks, list): + return json_spec + if not isinstance(networks, str): + raise SpecValidationError(f'Networks ({networks}) must be a string or list of strings') + json_spec['networks'] = [networks] + return json_spec + + @classmethod + def _from_json_impl(cls: Type[ServiceSpecT], json_spec: dict) -> ServiceSpecT: + args = {} # type: Dict[str, Any] + for k, v in json_spec.items(): + if k == 'placement': + v = PlacementSpec.from_json(v) + if k == 'spec': + args.update(v) + continue + args.update({k: v}) + _cls = cls(**args) + if _service_spec_from_json_validate: + _cls.validate() + return _cls + + def service_name(self) -> str: + n = self.service_type + if self.service_id: + n += '.' + self.service_id + return n + + def get_port_start(self) -> List[int]: + # If defined, we will allocate and number ports starting at this + # point. + return [] + + def get_virtual_ip(self) -> Optional[str]: + return None + + def to_json(self): + # type: () -> OrderedDict[str, Any] + ret: OrderedDict[str, Any] = OrderedDict() + ret['service_type'] = self.service_type + if self.service_id: + ret['service_id'] = self.service_id + ret['service_name'] = self.service_name() + if self.placement.to_json(): + ret['placement'] = self.placement.to_json() + if self.unmanaged: + ret['unmanaged'] = self.unmanaged + if self.networks: + ret['networks'] = self.networks + if self.extra_container_args: + ret['extra_container_args'] = self.extra_container_args + + c = {} + for key, val in sorted(self.__dict__.items(), key=lambda tpl: tpl[0]): + if key in ret: + continue + if hasattr(val, 'to_json'): + val = val.to_json() + if val: + c[key] = val + if c: + ret['spec'] = c + return ret + + def validate(self) -> None: + if not self.service_type: + raise SpecValidationError('Cannot add Service: type required') + + if self.service_type != 'osd': + if self.service_type in self.REQUIRES_SERVICE_ID and not self.service_id: + raise SpecValidationError('Cannot add Service: id required') + if self.service_type not in self.REQUIRES_SERVICE_ID and self.service_id: + raise SpecValidationError( + f'Service of type \'{self.service_type}\' should not contain a service id') + + if self.service_id: + if not re.match('^[a-zA-Z0-9_.-]+$', str(self.service_id)): + raise SpecValidationError('Service id contains invalid characters, ' + 'only [a-zA-Z0-9_.-] allowed') + + if self.placement is not None: + self.placement.validate() + if self.config: + for k, v in self.config.items(): + if k in self.MANAGED_CONFIG_OPTIONS: + raise SpecValidationError( + f'Cannot set config option {k} in spec: it is managed by cephadm' + ) + for network in self.networks or []: + try: + ip_network(network) + except ValueError as e: + raise SpecValidationError( + f'Cannot parse network {network}: {e}' + ) + + def __repr__(self) -> str: + y = yaml.dump(cast(dict, self), default_flow_style=False) + return f"{self.__class__.__name__}.from_json(yaml.safe_load('''{y}'''))" + + def __eq__(self, other: Any) -> bool: + return (self.__class__ == other.__class__ + and + self.__dict__ == other.__dict__) + + def one_line_str(self) -> str: + return '<{} for service_name={}>'.format(self.__class__.__name__, self.service_name()) + + @staticmethod + def yaml_representer(dumper: 'yaml.SafeDumper', data: 'ServiceSpec') -> Any: + return dumper.represent_dict(cast(Mapping, data.to_json().items())) + + +yaml.add_representer(ServiceSpec, ServiceSpec.yaml_representer) + + +class NFSServiceSpec(ServiceSpec): + def __init__(self, + service_type: str = 'nfs', + service_id: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + port: Optional[int] = None, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'nfs' + super(NFSServiceSpec, self).__init__( + 'nfs', service_id=service_id, + placement=placement, unmanaged=unmanaged, preview_only=preview_only, + config=config, networks=networks, extra_container_args=extra_container_args) + + self.port = port + + def get_port_start(self) -> List[int]: + if self.port: + return [self.port] + return [] + + def rados_config_name(self): + # type: () -> str + return 'conf-' + self.service_name() + + +yaml.add_representer(NFSServiceSpec, ServiceSpec.yaml_representer) + + +class RGWSpec(ServiceSpec): + """ + Settings to configure a (multisite) Ceph RGW + + .. code-block:: yaml + + service_type: rgw + service_id: myrealm.myzone + spec: + rgw_realm: myrealm + rgw_zone: myzone + ssl: true + rgw_frontend_port: 1234 + rgw_frontend_type: beast + rgw_frontend_ssl_certificate: ... + + See also: :ref:`orchestrator-cli-service-spec` + """ + + MANAGED_CONFIG_OPTIONS = ServiceSpec.MANAGED_CONFIG_OPTIONS + [ + 'rgw_zone', + 'rgw_realm', + 'rgw_frontends', + ] + + def __init__(self, + service_type: str = 'rgw', + service_id: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + rgw_realm: Optional[str] = None, + rgw_zone: Optional[str] = None, + rgw_frontend_port: Optional[int] = None, + rgw_frontend_ssl_certificate: Optional[List[str]] = None, + rgw_frontend_type: Optional[str] = None, + unmanaged: bool = False, + ssl: bool = False, + preview_only: bool = False, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + subcluster: Optional[str] = None, # legacy, only for from_json on upgrade + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'rgw', service_type + + # for backward compatibility with octopus spec files, + if not service_id and (rgw_realm and rgw_zone): + service_id = rgw_realm + '.' + rgw_zone + + super(RGWSpec, self).__init__( + 'rgw', service_id=service_id, + placement=placement, unmanaged=unmanaged, + preview_only=preview_only, config=config, networks=networks, + extra_container_args=extra_container_args) + + #: The RGW realm associated with this service. Needs to be manually created + self.rgw_realm: Optional[str] = rgw_realm + #: The RGW zone associated with this service. Needs to be manually created + self.rgw_zone: Optional[str] = rgw_zone + #: Port of the RGW daemons + self.rgw_frontend_port: Optional[int] = rgw_frontend_port + #: List of SSL certificates + self.rgw_frontend_ssl_certificate: Optional[List[str]] = rgw_frontend_ssl_certificate + #: civetweb or beast (default: beast). See :ref:`rgw_frontends` + self.rgw_frontend_type: Optional[str] = rgw_frontend_type + #: enable SSL + self.ssl = ssl + + def get_port_start(self) -> List[int]: + return [self.get_port()] + + def get_port(self) -> int: + if self.rgw_frontend_port: + return self.rgw_frontend_port + if self.ssl: + return 443 + else: + return 80 + + def validate(self) -> None: + super(RGWSpec, self).validate() + + if self.rgw_realm and not self.rgw_zone: + raise SpecValidationError( + 'Cannot add RGW: Realm specified but no zone specified') + if self.rgw_zone and not self.rgw_realm: + raise SpecValidationError( + 'Cannot add RGW: Zone specified but no realm specified') + + +yaml.add_representer(RGWSpec, ServiceSpec.yaml_representer) + + +class IscsiServiceSpec(ServiceSpec): + def __init__(self, + service_type: str = 'iscsi', + service_id: Optional[str] = None, + pool: Optional[str] = None, + trusted_ip_list: Optional[str] = None, + api_port: Optional[int] = None, + api_user: Optional[str] = None, + api_password: Optional[str] = None, + api_secure: Optional[bool] = None, + ssl_cert: Optional[str] = None, + ssl_key: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'iscsi' + super(IscsiServiceSpec, self).__init__('iscsi', service_id=service_id, + placement=placement, unmanaged=unmanaged, + preview_only=preview_only, + config=config, networks=networks, + extra_container_args=extra_container_args) + + #: RADOS pool where ceph-iscsi config data is stored. + self.pool = pool + #: list of trusted IP addresses + self.trusted_ip_list = trusted_ip_list + #: ``api_port`` as defined in the ``iscsi-gateway.cfg`` + self.api_port = api_port + #: ``api_user`` as defined in the ``iscsi-gateway.cfg`` + self.api_user = api_user + #: ``api_password`` as defined in the ``iscsi-gateway.cfg`` + self.api_password = api_password + #: ``api_secure`` as defined in the ``iscsi-gateway.cfg`` + self.api_secure = api_secure + #: SSL certificate + self.ssl_cert = ssl_cert + #: SSL private key + self.ssl_key = ssl_key + + if not self.api_secure and self.ssl_cert and self.ssl_key: + self.api_secure = True + + def validate(self) -> None: + super(IscsiServiceSpec, self).validate() + + if not self.pool: + raise SpecValidationError( + 'Cannot add ISCSI: No Pool specified') + + # Do not need to check for api_user and api_password as they + # now default to 'admin' when setting up the gateway url. Older + # iSCSI specs from before this change should be fine as they will + # have been required to have an api_user and api_password set and + # will be unaffected by the new default value. + + +yaml.add_representer(IscsiServiceSpec, ServiceSpec.yaml_representer) + + +class IngressSpec(ServiceSpec): + def __init__(self, + service_type: str = 'ingress', + service_id: Optional[str] = None, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + placement: Optional[PlacementSpec] = None, + backend_service: Optional[str] = None, + frontend_port: Optional[int] = None, + ssl_cert: Optional[str] = None, + ssl_key: Optional[str] = None, + ssl_dh_param: Optional[str] = None, + ssl_ciphers: Optional[List[str]] = None, + ssl_options: Optional[List[str]] = None, + monitor_port: Optional[int] = None, + monitor_user: Optional[str] = None, + monitor_password: Optional[str] = None, + enable_stats: Optional[bool] = None, + keepalived_password: Optional[str] = None, + virtual_ip: Optional[str] = None, + virtual_ips_list: Optional[List[str]] = None, + virtual_interface_networks: Optional[List[str]] = [], + unmanaged: bool = False, + ssl: bool = False, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'ingress' + + super(IngressSpec, self).__init__( + 'ingress', service_id=service_id, + placement=placement, config=config, + networks=networks, + extra_container_args=extra_container_args + ) + self.backend_service = backend_service + self.frontend_port = frontend_port + self.ssl_cert = ssl_cert + self.ssl_key = ssl_key + self.ssl_dh_param = ssl_dh_param + self.ssl_ciphers = ssl_ciphers + self.ssl_options = ssl_options + self.monitor_port = monitor_port + self.monitor_user = monitor_user + self.monitor_password = monitor_password + self.keepalived_password = keepalived_password + self.virtual_ip = virtual_ip + self.virtual_ips_list = virtual_ips_list + self.virtual_interface_networks = virtual_interface_networks or [] + self.unmanaged = unmanaged + self.ssl = ssl + + def get_port_start(self) -> List[int]: + return [cast(int, self.frontend_port), + cast(int, self.monitor_port)] + + def get_virtual_ip(self) -> Optional[str]: + return self.virtual_ip + + def validate(self) -> None: + super(IngressSpec, self).validate() + + if not self.backend_service: + raise SpecValidationError( + 'Cannot add ingress: No backend_service specified') + if not self.frontend_port: + raise SpecValidationError( + 'Cannot add ingress: No frontend_port specified') + if not self.monitor_port: + raise SpecValidationError( + 'Cannot add ingress: No monitor_port specified') + if not self.virtual_ip and not self.virtual_ips_list: + raise SpecValidationError( + 'Cannot add ingress: No virtual_ip provided') + if self.virtual_ip is not None and self.virtual_ips_list is not None: + raise SpecValidationError( + 'Cannot add ingress: Single and multiple virtual IPs specified') + + +yaml.add_representer(IngressSpec, ServiceSpec.yaml_representer) + + +class CustomContainerSpec(ServiceSpec): + def __init__(self, + service_type: str = 'container', + service_id: Optional[str] = None, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + image: Optional[str] = None, + entrypoint: Optional[str] = None, + uid: Optional[int] = None, + gid: Optional[int] = None, + volume_mounts: Optional[Dict[str, str]] = {}, + args: Optional[List[str]] = [], + envs: Optional[List[str]] = [], + privileged: Optional[bool] = False, + bind_mounts: Optional[List[List[str]]] = None, + ports: Optional[List[int]] = [], + dirs: Optional[List[str]] = [], + files: Optional[Dict[str, Any]] = {}, + ): + assert service_type == 'container' + assert service_id is not None + assert image is not None + + super(CustomContainerSpec, self).__init__( + service_type, service_id, + placement=placement, unmanaged=unmanaged, + preview_only=preview_only, config=config, + networks=networks) + + self.image = image + self.entrypoint = entrypoint + self.uid = uid + self.gid = gid + self.volume_mounts = volume_mounts + self.args = args + self.envs = envs + self.privileged = privileged + self.bind_mounts = bind_mounts + self.ports = ports + self.dirs = dirs + self.files = files + + def config_json(self) -> Dict[str, Any]: + """ + Helper function to get the value of the `--config-json` cephadm + command line option. It will contain all specification properties + that haven't a `None` value. Such properties will get default + values in cephadm. + :return: Returns a dictionary containing all specification + properties. + """ + config_json = {} + for prop in ['image', 'entrypoint', 'uid', 'gid', 'args', + 'envs', 'volume_mounts', 'privileged', + 'bind_mounts', 'ports', 'dirs', 'files']: + value = getattr(self, prop) + if value is not None: + config_json[prop] = value + return config_json + + +yaml.add_representer(CustomContainerSpec, ServiceSpec.yaml_representer) + + +class MonitoringSpec(ServiceSpec): + def __init__(self, + service_type: str, + service_id: Optional[str] = None, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + port: Optional[int] = None, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type in ['grafana', 'node-exporter', 'prometheus', 'alertmanager'] + + super(MonitoringSpec, self).__init__( + service_type, service_id, + placement=placement, unmanaged=unmanaged, + preview_only=preview_only, config=config, + networks=networks, extra_container_args=extra_container_args) + + self.service_type = service_type + self.port = port + + def get_port_start(self) -> List[int]: + return [self.get_port()] + + def get_port(self) -> int: + if self.port: + return self.port + else: + return {'prometheus': 9095, + 'node-exporter': 9100, + 'alertmanager': 9093, + 'grafana': 3000}[self.service_type] + + +yaml.add_representer(MonitoringSpec, ServiceSpec.yaml_representer) + + +class AlertManagerSpec(MonitoringSpec): + def __init__(self, + service_type: str = 'alertmanager', + service_id: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + user_data: Optional[Dict[str, Any]] = None, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + port: Optional[int] = None, + secure: bool = False, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'alertmanager' + super(AlertManagerSpec, self).__init__( + 'alertmanager', service_id=service_id, + placement=placement, unmanaged=unmanaged, + preview_only=preview_only, config=config, networks=networks, port=port, + extra_container_args=extra_container_args) + + # Custom configuration. + # + # Example: + # service_type: alertmanager + # service_id: xyz + # user_data: + # default_webhook_urls: + # - "https://foo" + # - "https://bar" + # + # Documentation: + # default_webhook_urls - A list of additional URL's that are + # added to the default receivers' + # <webhook_configs> configuration. + self.user_data = user_data or {} + self.secure = secure + + def get_port_start(self) -> List[int]: + return [self.get_port(), 9094] + + def validate(self) -> None: + super(AlertManagerSpec, self).validate() + + if self.port == 9094: + raise SpecValidationError( + 'Port 9094 is reserved for AlertManager cluster listen address') + + +yaml.add_representer(AlertManagerSpec, ServiceSpec.yaml_representer) + + +class GrafanaSpec(MonitoringSpec): + def __init__(self, + service_type: str = 'grafana', + service_id: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + port: Optional[int] = None, + initial_admin_password: Optional[str] = None, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'grafana' + super(GrafanaSpec, self).__init__( + 'grafana', service_id=service_id, + placement=placement, unmanaged=unmanaged, + preview_only=preview_only, config=config, networks=networks, port=port, + extra_container_args=extra_container_args) + + self.initial_admin_password = initial_admin_password + + +yaml.add_representer(GrafanaSpec, ServiceSpec.yaml_representer) + + +class PrometheusSpec(MonitoringSpec): + def __init__(self, + service_type: str = 'prometheus', + service_id: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + config: Optional[Dict[str, str]] = None, + networks: Optional[List[str]] = None, + port: Optional[int] = None, + retention_time: Optional[str] = None, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'prometheus' + super(PrometheusSpec, self).__init__( + 'prometheus', service_id=service_id, + placement=placement, unmanaged=unmanaged, + preview_only=preview_only, config=config, networks=networks, port=port, + extra_container_args=extra_container_args) + + self.retention_time = retention_time + + +yaml.add_representer(PrometheusSpec, ServiceSpec.yaml_representer) + + +class SNMPGatewaySpec(ServiceSpec): + class SNMPVersion(str, enum.Enum): + V2c = 'V2c' + V3 = 'V3' + + def to_json(self) -> str: + return self.value + + class SNMPAuthType(str, enum.Enum): + MD5 = 'MD5' + SHA = 'SHA' + + def to_json(self) -> str: + return self.value + + class SNMPPrivacyType(str, enum.Enum): + DES = 'DES' + AES = 'AES' + + def to_json(self) -> str: + return self.value + + valid_destination_types = [ + 'Name:Port', + 'IPv4:Port' + ] + + def __init__(self, + service_type: str = 'snmp-gateway', + snmp_version: Optional[SNMPVersion] = None, + snmp_destination: str = '', + credentials: Dict[str, str] = {}, + engine_id: Optional[str] = None, + auth_protocol: Optional[SNMPAuthType] = None, + privacy_protocol: Optional[SNMPPrivacyType] = None, + placement: Optional[PlacementSpec] = None, + unmanaged: bool = False, + preview_only: bool = False, + port: Optional[int] = None, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'snmp-gateway' + + super(SNMPGatewaySpec, self).__init__( + service_type, + placement=placement, + unmanaged=unmanaged, + preview_only=preview_only, + extra_container_args=extra_container_args) + + self.service_type = service_type + self.snmp_version = snmp_version + self.snmp_destination = snmp_destination + self.port = port + self.credentials = credentials + self.engine_id = engine_id + self.auth_protocol = auth_protocol + self.privacy_protocol = privacy_protocol + + @classmethod + def _from_json_impl(cls, json_spec: dict) -> 'SNMPGatewaySpec': + + cpy = json_spec.copy() + types = [ + ('snmp_version', SNMPGatewaySpec.SNMPVersion), + ('auth_protocol', SNMPGatewaySpec.SNMPAuthType), + ('privacy_protocol', SNMPGatewaySpec.SNMPPrivacyType), + ] + for d in cpy, cpy.get('spec', {}): + for key, enum_cls in types: + try: + if key in d: + d[key] = enum_cls(d[key]) + except ValueError: + raise SpecValidationError(f'{key} unsupported. Must be one of ' + f'{", ".join(enum_cls)}') + return super(SNMPGatewaySpec, cls)._from_json_impl(cpy) + + @property + def ports(self) -> List[int]: + return [self.port or 9464] + + def get_port_start(self) -> List[int]: + return self.ports + + def validate(self) -> None: + super(SNMPGatewaySpec, self).validate() + + if not self.credentials: + raise SpecValidationError( + 'Missing authentication information (credentials). ' + 'SNMP V2c and V3 require credential information' + ) + elif not self.snmp_version: + raise SpecValidationError( + 'Missing SNMP version (snmp_version)' + ) + + creds_requirement = { + 'V2c': ['snmp_community'], + 'V3': ['snmp_v3_auth_username', 'snmp_v3_auth_password'] + } + if self.privacy_protocol: + creds_requirement['V3'].append('snmp_v3_priv_password') + + missing = [parm for parm in creds_requirement[self.snmp_version] + if parm not in self.credentials] + # check that credentials are correct for the version + if missing: + raise SpecValidationError( + f'SNMP {self.snmp_version} credentials are incomplete. Missing {", ".join(missing)}' + ) + + if self.engine_id: + if 10 <= len(self.engine_id) <= 64 and \ + is_hex(self.engine_id) and \ + len(self.engine_id) % 2 == 0: + pass + else: + raise SpecValidationError( + 'engine_id must be a string containing 10-64 hex characters. ' + 'Its length must be divisible by 2' + ) + + else: + if self.snmp_version == 'V3': + raise SpecValidationError( + 'Must provide an engine_id for SNMP V3 notifications' + ) + + if not self.snmp_destination: + raise SpecValidationError( + 'SNMP destination (snmp_destination) must be provided' + ) + else: + valid, description = valid_addr(self.snmp_destination) + if not valid: + raise SpecValidationError( + f'SNMP destination (snmp_destination) is invalid: {description}' + ) + if description not in self.valid_destination_types: + raise SpecValidationError( + f'SNMP destination (snmp_destination) type ({description}) is invalid. ' + f'Must be either: {", ".join(sorted(self.valid_destination_types))}' + ) + + +yaml.add_representer(SNMPGatewaySpec, ServiceSpec.yaml_representer) + + +class MDSSpec(ServiceSpec): + def __init__(self, + service_type: str = 'mds', + service_id: Optional[str] = None, + placement: Optional[PlacementSpec] = None, + config: Optional[Dict[str, str]] = None, + unmanaged: bool = False, + preview_only: bool = False, + extra_container_args: Optional[List[str]] = None, + ): + assert service_type == 'mds' + super(MDSSpec, self).__init__('mds', service_id=service_id, + placement=placement, + config=config, + unmanaged=unmanaged, + preview_only=preview_only, + extra_container_args=extra_container_args) + + def validate(self) -> None: + super(MDSSpec, self).validate() + + if str(self.service_id)[0].isdigit(): + raise SpecValidationError('MDS service id cannot start with a numeric digit') + + +yaml.add_representer(MDSSpec, ServiceSpec.yaml_representer) diff --git a/src/python-common/ceph/deployment/translate.py b/src/python-common/ceph/deployment/translate.py new file mode 100644 index 000000000..2d373732c --- /dev/null +++ b/src/python-common/ceph/deployment/translate.py @@ -0,0 +1,119 @@ +import logging + +try: + from typing import Optional, List +except ImportError: + pass + +from ceph.deployment.drive_selection.selector import DriveSelection + +logger = logging.getLogger(__name__) + + +# TODO refactor this to a DriveSelection method +class to_ceph_volume(object): + + def __init__(self, + selection, # type: DriveSelection + osd_id_claims=None, # type: Optional[List[str]] + preview=False # type: bool + ): + + self.selection = selection + self.spec = selection.spec + self.preview = preview + self.osd_id_claims = osd_id_claims + + def run(self): + # type: () -> List[str] + """ Generate ceph-volume commands based on the DriveGroup filters """ + data_devices = [x.path for x in self.selection.data_devices()] + db_devices = [x.path for x in self.selection.db_devices()] + wal_devices = [x.path for x in self.selection.wal_devices()] + journal_devices = [x.path for x in self.selection.journal_devices()] + + if not data_devices: + return [] + + cmds: List[str] = [] + if self.spec.method == 'raw': + assert self.spec.objectstore == 'bluestore' + # ceph-volume raw prepare only support 1:1 ratio of data to db/wal devices + if data_devices and db_devices: + if len(data_devices) != len(db_devices): + raise ValueError('Number of data devices must match number of ' + 'db devices for raw mode osds') + if data_devices and wal_devices: + if len(data_devices) != len(wal_devices): + raise ValueError('Number of data devices must match number of ' + 'wal devices for raw mode osds') + # for raw prepare each data device needs its own prepare command + dev_counter = 0 + while dev_counter < len(data_devices): + cmd = "raw prepare --bluestore" + cmd += " --data {}".format(data_devices[dev_counter]) + if db_devices: + cmd += " --block.db {}".format(db_devices[dev_counter]) + if wal_devices: + cmd += " --block.wal {}".format(wal_devices[dev_counter]) + cmds.append(cmd) + dev_counter += 1 + + elif self.spec.objectstore == 'filestore': + # for lvm batch we can just do all devices in one command + cmd = "lvm batch --no-auto" + + cmd += " {}".format(" ".join(data_devices)) + + if self.spec.journal_size: + cmd += " --journal-size {}".format(self.spec.journal_size) + + if journal_devices: + cmd += " --journal-devices {}".format( + ' '.join(journal_devices)) + + cmd += " --filestore" + cmds.append(cmd) + + elif self.spec.objectstore == 'bluestore': + # for lvm batch we can just do all devices in one command + cmd = "lvm batch --no-auto {}".format(" ".join(data_devices)) + + if db_devices: + cmd += " --db-devices {}".format(" ".join(db_devices)) + + if wal_devices: + cmd += " --wal-devices {}".format(" ".join(wal_devices)) + + if self.spec.block_wal_size: + cmd += " --block-wal-size {}".format(self.spec.block_wal_size) + + if self.spec.block_db_size: + cmd += " --block-db-size {}".format(self.spec.block_db_size) + cmds.append(cmd) + + for i in range(len(cmds)): + if self.spec.encrypted: + cmds[i] += " --dmcrypt" + + if self.spec.osds_per_device: + cmds[i] += " --osds-per-device {}".format(self.spec.osds_per_device) + + if self.spec.data_allocate_fraction: + cmds[i] += " --data-allocate-fraction {}".format(self.spec.data_allocate_fraction) + + if self.osd_id_claims: + cmds[i] += " --osd-ids {}".format(" ".join(self.osd_id_claims)) + + if self.spec.method != 'raw': + cmds[i] += " --yes" + cmds[i] += " --no-systemd" + + if self.spec.crush_device_class: + cmds[i] += " --crush-device-class {}".format(self.spec.crush_device_class) + + if self.preview: + cmds[i] += " --report" + cmds[i] += " --format json" + + return cmds diff --git a/src/python-common/ceph/deployment/utils.py b/src/python-common/ceph/deployment/utils.py new file mode 100644 index 000000000..6aad15b75 --- /dev/null +++ b/src/python-common/ceph/deployment/utils.py @@ -0,0 +1,102 @@ +import ipaddress +import socket +from typing import Tuple, Optional +from urllib.parse import urlparse + + +def unwrap_ipv6(address): + # type: (str) -> str + if address.startswith('[') and address.endswith(']'): + return address[1:-1] + return address + + +def wrap_ipv6(address): + # type: (str) -> str + + # We cannot assume it's already wrapped or even an IPv6 address if + # it's already wrapped it'll not pass (like if it's a hostname) and trigger + # the ValueError + try: + if ipaddress.ip_address(address).version == 6: + return f"[{address}]" + except ValueError: + pass + + return address + + +def is_ipv6(address): + # type: (str) -> bool + address = unwrap_ipv6(address) + try: + return ipaddress.ip_address(address).version == 6 + except ValueError: + return False + + +def valid_addr(addr: str) -> Tuple[bool, str]: + """check that an address string is valid + Valid in this case means that a name is resolvable, or the + IP address string is a correctly formed IPv4 or IPv6 address, + with or without a port + + Args: + addr (str): address + + Returns: + Tuple[bool, str]: Validity of the address, either + True, address type (IPv4[:Port], IPv6[:Port], Name[:Port]) + False, <error description> + """ + + def _dns_lookup(addr: str, port: Optional[int]) -> Tuple[bool, str]: + try: + socket.getaddrinfo(addr, None) + except socket.gaierror: + # not resolvable + return False, 'DNS lookup failed' + return True, 'Name:Port' if port else 'Name' + + def _ip_lookup(addr: str, port: Optional[int]) -> Tuple[bool, str]: + unwrapped = unwrap_ipv6(addr) + try: + ip_addr = ipaddress.ip_address(unwrapped) + except ValueError: + return False, 'Invalid IP v4 or v6 address format' + return True, f'IPv{ip_addr.version}:Port' if port else f'IPv{ip_addr.version}' + + dots = addr.count('.') + colons = addr.count(':') + addr_as_url = f'http://{addr}' + + try: + res = urlparse(addr_as_url) + except ValueError as e: + if str(e) == 'Invalid IPv6 URL': + return False, 'Address has incorrect/incomplete use of enclosing brackets' + return False, f'Unknown urlparse error {str(e)} for {addr_as_url}' + + addr = res.netloc + port = None + try: + port = res.port + if port: + addr = addr[:-len(f':{port}')] + except ValueError: + if colons == 1: + return False, 'Port must be numeric' + elif ']:' in addr: + return False, 'Port must be numeric' + + if addr.startswith('[') and dots: + return False, "IPv4 address wrapped in brackets is invalid" + + # catch partial address like 10.8 which would be valid IPaddress schemes + # but are classed as invalid here since they're not usable + if dots and addr[0].isdigit() and dots != 3: + return False, 'Invalid partial IPv4 address' + + if addr[0].isalpha() and '.' in addr: + return _dns_lookup(addr, port) + return _ip_lookup(addr, port) diff --git a/src/python-common/ceph/py.typed b/src/python-common/ceph/py.typed new file mode 100644 index 000000000..444b02d77 --- /dev/null +++ b/src/python-common/ceph/py.typed @@ -0,0 +1 @@ +# Marker file for PEP 561. This package uses inline types.
\ No newline at end of file diff --git a/src/python-common/ceph/tests/__init__.py b/src/python-common/ceph/tests/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/src/python-common/ceph/tests/__init__.py diff --git a/src/python-common/ceph/tests/c-v-inventory.json b/src/python-common/ceph/tests/c-v-inventory.json new file mode 100644 index 000000000..c24345525 --- /dev/null +++ b/src/python-common/ceph/tests/c-v-inventory.json @@ -0,0 +1,155 @@ +[ + { + "available": false, + "created": "2022-02-11T10:58:23.177450Z", + "rejected_reasons": [ + "locked" + ], + "sys_api": { + "scheduler_mode": "", + "rotational": "0", + "vendor": "", + "human_readable_size": "50.00 GB", + "sectors": 0, + "sas_device_handle": "", + "partitions": {}, + "rev": "", + "sas_address": "", + "locked": 1, + "sectorsize": "512", + "removable": "0", + "path": "/dev/dm-0", + "support_discard": "", + "model": "", + "ro": "0", + "nr_requests": "128", + "size": 53687091200 + }, + "lvs": [], + "path": "/dev/dm-0" + }, + { + "available": false, + "rejected_reasons": [ + "locked" + ], + "sys_api": { + "scheduler_mode": "", + "rotational": "0", + "vendor": "", + "human_readable_size": "31.47 GB", + "sectors": 0, + "sas_device_handle": "", + "partitions": {}, + "rev": "", + "sas_address": "", + "locked": 1, + "sectorsize": "512", + "removable": "0", + "path": "/dev/dm-1", + "support_discard": "", + "model": "", + "ro": "0", + "nr_requests": "128", + "size": 33789313024 + }, + "lvs": [], + "path": "/dev/dm-1" + }, + { + "available": false, + "created": "2022-02-11T10:58:23.177450Z", + "rejected_reasons": [ + "locked" + ], + "sys_api": { + "scheduler_mode": "", + "rotational": "0", + "vendor": "", + "human_readable_size": "394.27 GB", + "sectors": 0, + "sas_device_handle": "", + "partitions": {}, + "rev": "", + "sas_address": "", + "locked": 1, + "sectorsize": "512", + "removable": "0", + "path": "/dev/dm-2", + "support_discard": "", + "model": "", + "ro": "0", + "nr_requests": "128", + "size": 423347879936 + }, + "lvs": [], + "path": "/dev/dm-2" + }, + { + "available": false, + "rejected_reasons": [ + "locked" + ], + "sys_api": { + "scheduler_mode": "cfq", + "rotational": "0", + "vendor": "ATA", + "human_readable_size": "476.94 GB", + "sectors": 0, + "sas_device_handle": "", + "partitions": { + "sda2": { + "start": "411648", + "holders": [], + "sectorsize": 512, + "sectors": "2097152", + "size": "1024.00 MB" + }, + "sda3": { + "start": "2508800", + "holders": [ + "dm-1", + "dm-2", + "dm-0" + ], + "sectorsize": 512, + "sectors": "997705728", + "size": "475.74 GB" + }, + "sda1": { + "start": "2048", + "holders": [], + "sectorsize": 512, + "sectors": "409600", + "size": "200.00 MB" + } + }, + "rev": "0000", + "sas_address": "", + "locked": 1, + "sectorsize": "512", + "removable": "0", + "path": "/dev/sda", + "support_discard": "", + "model": "SanDisk SD8SN8U5", + "ro": "0", + "nr_requests": "128", + "size": 512110190592 + }, + "lvs": [ + { + "comment": "not used by ceph", + "name": "swap" + }, + { + "comment": "not used by ceph", + "name": "home" + }, + { + "comment": "not used by ceph", + "name": "root" + } + ], + "path": "/dev/sda" + } +] diff --git a/src/python-common/ceph/tests/factories.py b/src/python-common/ceph/tests/factories.py new file mode 100644 index 000000000..6938fd084 --- /dev/null +++ b/src/python-common/ceph/tests/factories.py @@ -0,0 +1,101 @@ +from ceph.deployment.inventory import Device + + +class InventoryFactory(object): + def __init__(self): + self.taken_paths = [] + + def _make_path(self, ident='b'): + return "/dev/{}{}".format(self.prefix, ident) + + def _find_new_path(self): + cnt = 0 + if len(self.taken_paths) >= 25: + raise Exception( + "Double-character disks are not implemetend. Maximum amount" + "of disks reached.") + + while self.path in self.taken_paths: + ident = chr(ord('b') + cnt) + self.path = "/dev/{}{}".format(self.prefix, ident) + cnt += 1 + + def assemble(self): + if self.empty: + return {} + self._find_new_path() + inventory_sample = { + 'available': self.available, + 'lvs': [], + 'path': self.path, + 'rejected_reasons': self.rejected_reason, + 'sys_api': { + 'human_readable_size': self.human_readable_size, + 'locked': 1, + 'model': self.model, + 'nr_requests': '256', + 'partitions': + { # partitions are not as relevant for now, todo for later + 'sda1': { + 'sectors': '41940992', + 'sectorsize': 512, + 'size': self.human_readable_size, + 'start': '2048' + } + }, + 'path': self.path, + 'removable': '0', + 'rev': '', + 'ro': '0', + 'rotational': str(self.rotational), + 'sas_address': '', + 'sas_device_handle': '', + 'scheduler_mode': 'mq-deadline', + 'sectors': 0, + 'sectorsize': '512', + 'size': self.size, + 'support_discard': '', + 'vendor': self.vendor + } + } + + if self.available: + self.taken_paths.append(self.path) + return inventory_sample + return {} + + def _init(self, **kwargs): + self.prefix = 'sd' + self.path = kwargs.get('path', self._make_path()) + self.human_readable_size = kwargs.get('human_readable_size', + '50.00 GB') + self.vendor = kwargs.get('vendor', 'samsung') + self.model = kwargs.get('model', '42-RGB') + self.available = kwargs.get('available', True) + self.rejected_reason = kwargs.get('rejected_reason', ['']) + self.rotational = kwargs.get('rotational', '1') + if not self.available: + self.rejected_reason = ['locked'] + self.empty = kwargs.get('empty', False) + self.size = kwargs.get('size', 5368709121) + + def produce(self, pieces=1, **kwargs): + if kwargs.get('path') and pieces > 1: + raise Exception("/path/ and /pieces/ are mutually exclusive") + # Move to custom init to track _taken_paths. + # class is invoked once in each context. + # if disks with different properties are being created + # we'd have to re-init the class and loose track of the + # taken_paths + self._init(**kwargs) + return [self.assemble() for x in range(0, pieces)] + + +class DeviceFactory(object): + def __init__(self, device_setup): + self.device_setup = device_setup + self.pieces = device_setup.get('pieces', 1) + self.device_conf = device_setup.get('device_config', {}) + + def produce(self): + return [Device(**self.device_conf) for x in range(0, self.pieces)] diff --git a/src/python-common/ceph/tests/test_datetime.py b/src/python-common/ceph/tests/test_datetime.py new file mode 100644 index 000000000..d03a82930 --- /dev/null +++ b/src/python-common/ceph/tests/test_datetime.py @@ -0,0 +1,61 @@ +import datetime + +import pytest + +from ceph.utils import datetime_now, datetime_to_str, str_to_datetime + + +def test_datetime_to_str_1(): + dt = datetime.datetime.now() + assert type(datetime_to_str(dt)) is str + + +def test_datetime_to_str_2(): + # note: tz isn't specified in the string, so explicitly store this as UTC + dt = datetime.datetime.strptime( + '2019-04-24T17:06:53.039991', + '%Y-%m-%dT%H:%M:%S.%f' + ).replace(tzinfo=datetime.timezone.utc) + assert datetime_to_str(dt) == '2019-04-24T17:06:53.039991Z' + + +def test_datetime_to_str_3(): + dt = datetime.datetime.strptime('2020-11-02T04:40:12.748172-0800', + '%Y-%m-%dT%H:%M:%S.%f%z') + assert datetime_to_str(dt) == '2020-11-02T12:40:12.748172Z' + + +def test_str_to_datetime_1(): + dt = str_to_datetime('2020-03-03T09:21:43.636153304Z') + assert type(dt) is datetime.datetime + assert dt.tzinfo is not None + + +def test_str_to_datetime_2(): + dt = str_to_datetime('2020-03-03T15:52:30.136257504-0600') + assert type(dt) is datetime.datetime + assert dt.tzinfo is not None + + +def test_str_to_datetime_3(): + dt = str_to_datetime('2020-03-03T15:52:30.136257504') + assert type(dt) is datetime.datetime + assert dt.tzinfo is not None + + +def test_str_to_datetime_invalid_format_1(): + with pytest.raises(ValueError): + str_to_datetime('2020-03-03 15:52:30.136257504') + + +def test_str_to_datetime_invalid_format_2(): + with pytest.raises(ValueError): + str_to_datetime('2020-03-03') + + +def test_datetime_now_1(): + dt = str_to_datetime('2020-03-03T09:21:43.636153304Z') + dt_now = datetime_now() + assert type(dt_now) is datetime.datetime + assert dt_now.tzinfo is not None + assert dt < dt_now diff --git a/src/python-common/ceph/tests/test_disk_selector.py b/src/python-common/ceph/tests/test_disk_selector.py new file mode 100644 index 000000000..b08236130 --- /dev/null +++ b/src/python-common/ceph/tests/test_disk_selector.py @@ -0,0 +1,560 @@ +# flake8: noqa +import pytest + +from ceph.deployment.drive_selection.matchers import _MatchInvalid +from ceph.deployment.inventory import Devices, Device + +from ceph.deployment.drive_group import DriveGroupSpec, DeviceSelection, \ + DriveGroupValidationError + +from ceph.deployment import drive_selection +from ceph.deployment.service_spec import PlacementSpec +from ceph.tests.factories import InventoryFactory +from ceph.tests.utils import _mk_inventory, _mk_device + + +class TestMatcher(object): + """ Test Matcher base class + """ + + def test_get_disk_key_3(self): + """ + virtual is False + key is found + retrun value of key is expected + """ + disk_map = Device(path='/dev/vdb', sys_api={'foo': 'bar'}) + ret = drive_selection.Matcher('foo', 'bar')._get_disk_key(disk_map) + assert ret is disk_map.sys_api.get('foo') + + def test_get_disk_key_4(self): + """ + virtual is False + key is not found + expect raise Exception + """ + disk_map = Device(path='/dev/vdb') + with pytest.raises(Exception): + drive_selection.Matcher('bar', 'foo')._get_disk_key(disk_map) + pytest.fail("No disk_key found for foo or None") + + +class TestSubstringMatcher(object): + def test_compare(self): + disk_dict = Device(path='/dev/vdb', sys_api=dict(model='samsung')) + matcher = drive_selection.SubstringMatcher('model', 'samsung') + ret = matcher.compare(disk_dict) + assert ret is True + + def test_compare_false(self): + disk_dict = Device(path='/dev/vdb', sys_api=dict(model='nothing_matching')) + matcher = drive_selection.SubstringMatcher('model', 'samsung') + ret = matcher.compare(disk_dict) + assert ret is False + + +class TestEqualityMatcher(object): + def test_compare(self): + disk_dict = Device(path='/dev/vdb', sys_api=dict(rotates='1')) + matcher = drive_selection.EqualityMatcher('rotates', '1') + ret = matcher.compare(disk_dict) + assert ret is True + + def test_compare_false(self): + disk_dict = Device(path='/dev/vdb', sys_api=dict(rotates='1')) + matcher = drive_selection.EqualityMatcher('rotates', '0') + ret = matcher.compare(disk_dict) + assert ret is False + + +class TestAllMatcher(object): + def test_compare(self): + disk_dict = Device(path='/dev/vdb') + matcher = drive_selection.AllMatcher('all', 'True') + ret = matcher.compare(disk_dict) + assert ret is True + + def test_compare_value_not_true(self): + disk_dict = Device(path='/dev/vdb') + matcher = drive_selection.AllMatcher('all', 'False') + ret = matcher.compare(disk_dict) + assert ret is True + + +class TestSizeMatcher(object): + def test_parse_filter_exact(self): + """ Testing exact notation with 20G """ + matcher = drive_selection.SizeMatcher('size', '20G') + assert isinstance(matcher.exact, tuple) + assert matcher.exact[0] == '20' + assert matcher.exact[1] == 'GB' + + def test_parse_filter_exact_GB_G(self): + """ Testing exact notation with 20G """ + matcher = drive_selection.SizeMatcher('size', '20GB') + assert isinstance(matcher.exact, tuple) + assert matcher.exact[0] == '20' + assert matcher.exact[1] == 'GB' + + def test_parse_filter_high_low(self): + """ Testing high-low notation with 20G:50G """ + + matcher = drive_selection.SizeMatcher('size', '20G:50G') + assert isinstance(matcher.exact, tuple) + assert matcher.low[0] == '20' + assert matcher.high[0] == '50' + assert matcher.low[1] == 'GB' + assert matcher.high[1] == 'GB' + + def test_parse_filter_max_high(self): + """ Testing high notation with :50G """ + + matcher = drive_selection.SizeMatcher('size', ':50G') + assert isinstance(matcher.exact, tuple) + assert matcher.high[0] == '50' + assert matcher.high[1] == 'GB' + + def test_parse_filter_min_low(self): + """ Testing low notation with 20G: """ + + matcher = drive_selection.SizeMatcher('size', '50G:') + assert isinstance(matcher.exact, tuple) + assert matcher.low[0] == '50' + assert matcher.low[1] == 'GB' + + def test_to_byte_KB(self): + """ I doubt anyone ever thought we'd need to understand KB """ + + ret = drive_selection.SizeMatcher('size', '4K').to_byte(('4', 'KB')) + assert ret == 4 * 1e+3 + + def test_to_byte_GB(self): + """ Pretty nonesense test..""" + + ret = drive_selection.SizeMatcher('size', '10G').to_byte(('10', 'GB')) + assert ret == 10 * 1e+9 + + def test_to_byte_MB(self): + """ Pretty nonesense test..""" + + ret = drive_selection.SizeMatcher('size', '10M').to_byte(('10', 'MB')) + assert ret == 10 * 1e+6 + + def test_to_byte_TB(self): + """ Pretty nonesense test..""" + + ret = drive_selection.SizeMatcher('size', '10T').to_byte(('10', 'TB')) + assert ret == 10 * 1e+12 + + def test_to_byte_PB(self): + """ Expect to raise """ + + with pytest.raises(_MatchInvalid): + drive_selection.SizeMatcher('size', '10P').to_byte(('10', 'PB')) + assert 'Unit \'P\' is not supported' + + def test_compare_exact(self): + + matcher = drive_selection.SizeMatcher('size', '20GB') + disk_dict = Device(path='/dev/vdb', sys_api=dict(size='20.00 GB')) + ret = matcher.compare(disk_dict) + assert ret is True + + def test_compare_exact_decimal(self): + + matcher = drive_selection.SizeMatcher('size', '20.12GB') + disk_dict = Device(path='/dev/vdb', sys_api=dict(size='20.12 GB')) + ret = matcher.compare(disk_dict) + assert ret is True + + @pytest.mark.parametrize("test_input,expected", [ + ("1.00 GB", False), + ("20.00 GB", True), + ("50.00 GB", True), + ("100.00 GB", True), + ("101.00 GB", False), + ("1101.00 GB", False), + ]) + def test_compare_high_low(self, test_input, expected): + + matcher = drive_selection.SizeMatcher('size', '20GB:100GB') + disk_dict = Device(path='/dev/vdb', sys_api=dict(size=test_input)) + ret = matcher.compare(disk_dict) + assert ret is expected + + @pytest.mark.parametrize("test_input,expected", [ + ("1.00 GB", True), + ("20.00 GB", True), + ("50.00 GB", True), + ("100.00 GB", False), + ("101.00 GB", False), + ("1101.00 GB", False), + ]) + def test_compare_high(self, test_input, expected): + + matcher = drive_selection.SizeMatcher('size', ':50GB') + disk_dict = Device(path='/dev/vdb', sys_api=dict(size=test_input)) + ret = matcher.compare(disk_dict) + assert ret is expected + + @pytest.mark.parametrize("test_input,expected", [ + ("1.00 GB", False), + ("20.00 GB", False), + ("50.00 GB", True), + ("100.00 GB", True), + ("101.00 GB", True), + ("1101.00 GB", True), + ]) + def test_compare_low(self, test_input, expected): + + matcher = drive_selection.SizeMatcher('size', '50GB:') + disk_dict = Device(path='/dev/vdb', sys_api=dict(size=test_input)) + ret = matcher.compare(disk_dict) + assert ret is expected + + @pytest.mark.parametrize("test_input,expected", [ + ("1.00 GB", False), + ("20.00 GB", False), + ("50.00 GB", False), + ("100.00 GB", False), + ("101.00 GB", False), + ("1101.00 GB", True), + ("9.10 TB", True), + ]) + def test_compare_at_least_1TB(self, test_input, expected): + + matcher = drive_selection.SizeMatcher('size', '1TB:') + disk_dict = Device(path='/dev/sdz', sys_api=dict(size=test_input)) + ret = matcher.compare(disk_dict) + assert ret is expected + + def test_compare_raise(self): + + matcher = drive_selection.SizeMatcher('size', 'None') + disk_dict = Device(path='/dev/vdb', sys_api=dict(size='20.00 GB')) + with pytest.raises(Exception): + matcher.compare(disk_dict) + pytest.fail("Couldn't parse size") + + @pytest.mark.parametrize("test_input,expected", [ + ("10G", ('10', 'GB')), + ("20GB", ('20', 'GB')), + ("10g", ('10', 'GB')), + ("20gb", ('20', 'GB')), + ]) + def test_get_k_v(self, test_input, expected): + assert drive_selection.SizeMatcher('size', '10G')._get_k_v(test_input) == expected + + @pytest.mark.parametrize("test_input,expected", [ + ("10G", ('GB')), + ("10g", ('GB')), + ("20GB", ('GB')), + ("20gb", ('GB')), + ("20TB", ('TB')), + ("20tb", ('TB')), + ("20T", ('TB')), + ("20t", ('TB')), + ("20MB", ('MB')), + ("20mb", ('MB')), + ("20M", ('MB')), + ("20m", ('MB')), + ]) + def test_parse_suffix(self, test_input, expected): + assert drive_selection.SizeMatcher('size', '10G')._parse_suffix(test_input) == expected + + @pytest.mark.parametrize("test_input,expected", [ + ("G", 'GB'), + ("GB", 'GB'), + ("TB", 'TB'), + ("T", 'TB'), + ("MB", 'MB'), + ("M", 'MB'), + ]) + def test_normalize_suffix(self, test_input, expected): + + assert drive_selection.SizeMatcher('10G', 'size')._normalize_suffix(test_input) == expected + + def test_normalize_suffix_raises(self): + + with pytest.raises(_MatchInvalid): + drive_selection.SizeMatcher('10P', 'size')._normalize_suffix("P") + pytest.fail("Unit 'P' not supported") + + +class TestDriveGroup(object): + @pytest.fixture(scope='class') + def test_fix(self, empty=None): + def make_sample_data(empty=empty, + data_limit=0, + wal_limit=0, + db_limit=0, + osds_per_device='', + disk_format='bluestore'): + raw_sample_bluestore = { + 'service_type': 'osd', + 'service_id': 'foo', + 'placement': {'host_pattern': 'data*'}, + 'data_devices': { + 'size': '30G:50G', + 'model': '42-RGB', + 'vendor': 'samsung', + 'limit': data_limit + }, + 'wal_devices': { + 'model': 'fast', + 'limit': wal_limit + }, + 'db_devices': { + 'size': ':20G', + 'limit': db_limit + }, + 'db_slots': 5, + 'wal_slots': 5, + 'block_wal_size': '5G', + 'block_db_size': '10G', + 'objectstore': disk_format, + 'osds_per_device': osds_per_device, + 'encrypted': True, + } + raw_sample_filestore = { + 'service_type': 'osd', + 'service_id': 'foo', + 'placement': {'host_pattern': 'data*'}, + 'objectstore': 'filestore', + 'data_devices': { + 'size': '30G:50G', + 'model': 'foo', + 'vendor': '1x', + 'limit': data_limit + }, + 'journal_devices': { + 'size': ':20G' + }, + 'journal_size': '5G', + 'osds_per_device': osds_per_device, + 'encrypted': True, + } + if disk_format == 'filestore': + raw_sample = raw_sample_filestore + else: + raw_sample = raw_sample_bluestore + + if empty: + raw_sample = { + 'service_type': 'osd', + 'service_id': 'foo', + 'placement': {'host_pattern': 'data*'}, + 'data_devices': { + 'all': True + }, + } + + dgo = DriveGroupSpec.from_json(raw_sample) + return dgo + + return make_sample_data + + def test_encryption_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.encrypted is True + + def test_encryption_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.encrypted is False + + def test_db_slots_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.db_slots == 5 + + def test_db_slots_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.db_slots is None + + def test_wal_slots_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.wal_slots == 5 + + def test_wal_slots_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.wal_slots is None + + def test_block_wal_size_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.block_wal_size == '5G' + + def test_block_wal_size_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.block_wal_size is None + + def test_block_db_size_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.block_db_size == '10G' + + def test_block_db_size_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.block_db_size is None + + def test_data_devices_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.data_devices == DeviceSelection( + model='42-RGB', + size='30G:50G', + vendor='samsung', + limit=0, + ) + + def test_data_devices_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.db_devices is None + + def test_db_devices_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.db_devices == DeviceSelection( + size=':20G', + limit=0, + ) + + def test_db_devices_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.db_devices is None + + def test_wal_device_prop(self, test_fix): + test_fix = test_fix() + assert test_fix.wal_devices == DeviceSelection( + model='fast', + limit=0, + ) + + def test_wal_device_prop_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.wal_devices is None + + def test_bluestore_format_prop(self, test_fix): + test_fix = test_fix(disk_format='bluestore') + assert test_fix.objectstore == 'bluestore' + + def test_default_format_prop(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.objectstore == 'bluestore' + + def test_osds_per_device(self, test_fix): + test_fix = test_fix(osds_per_device='3') + assert test_fix.osds_per_device == '3' + + def test_osds_per_device_default(self, test_fix): + test_fix = test_fix() + assert test_fix.osds_per_device == '' + + def test_journal_size_empty(self, test_fix): + test_fix = test_fix(empty=True) + assert test_fix.journal_size is None + + @pytest.fixture + def inventory(self, available=True): + def make_sample_data(available=available, + data_devices=10, + wal_devices=0, + db_devices=2, + human_readable_size_data='50.00 GB', + human_readable_size_wal='20.00 GB', + size=5368709121, + human_readable_size_db='20.00 GB'): + factory = InventoryFactory() + inventory_sample = [] + data_disks = factory.produce( + pieces=data_devices, + available=available, + size=size, + human_readable_size=human_readable_size_data) + wal_disks = factory.produce( + pieces=wal_devices, + human_readable_size=human_readable_size_wal, + rotational='0', + model='ssd_type_model', + size=size, + available=available) + db_disks = factory.produce( + pieces=db_devices, + human_readable_size=human_readable_size_db, + rotational='0', + size=size, + model='ssd_type_model', + available=available) + inventory_sample.extend(data_disks) + inventory_sample.extend(wal_disks) + inventory_sample.extend(db_disks) + + return Devices(devices=inventory_sample) + + return make_sample_data + + +class TestDriveSelection(object): + + testdata = [ + ( + DriveGroupSpec( + placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(all=True)), + _mk_inventory(_mk_device() * 5), + ['/dev/sda', '/dev/sdb', '/dev/sdc', '/dev/sdd', '/dev/sde'], [] + ), + ( + DriveGroupSpec( + placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(all=True, limit=3), + db_devices=DeviceSelection(all=True) + ), + _mk_inventory(_mk_device() * 5), + ['/dev/sda', '/dev/sdb', '/dev/sdc'], ['/dev/sdd', '/dev/sde'] + ), + ( + DriveGroupSpec( + placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True), + db_devices=DeviceSelection(rotational=False) + ), + _mk_inventory(_mk_device(rotational=False) + _mk_device(rotational=True)), + ['/dev/sdb'], ['/dev/sda'] + ), + ( + DriveGroupSpec( + placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True), + db_devices=DeviceSelection(rotational=False) + ), + _mk_inventory(_mk_device(rotational=True)*2 + _mk_device(rotational=False)), + ['/dev/sda', '/dev/sdb'], ['/dev/sdc'] + ), + ( + DriveGroupSpec( + placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True), + db_devices=DeviceSelection(rotational=False) + ), + _mk_inventory(_mk_device(rotational=True)*2), + ['/dev/sda', '/dev/sdb'], [] + ), + ] + + @pytest.mark.parametrize("spec,inventory,expected_data,expected_db", testdata) + def test_disk_selection(self, spec, inventory, expected_data, expected_db): + sel = drive_selection.DriveSelection(spec, inventory) + assert [d.path for d in sel.data_devices()] == expected_data + assert [d.path for d in sel.db_devices()] == expected_db + + def test_disk_selection_raise(self): + spec = DriveGroupSpec( + placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(size='wrong'), + ) + inventory = _mk_inventory(_mk_device(rotational=True)*2) + m = 'Failed to validate OSD spec "foobar.data_devices": No filters applied' + with pytest.raises(DriveGroupValidationError, match=m): + drive_selection.DriveSelection(spec, inventory)
\ No newline at end of file diff --git a/src/python-common/ceph/tests/test_drive_group.py b/src/python-common/ceph/tests/test_drive_group.py new file mode 100644 index 000000000..2fc66b737 --- /dev/null +++ b/src/python-common/ceph/tests/test_drive_group.py @@ -0,0 +1,336 @@ +# flake8: noqa +import re + +import pytest +import yaml + +from ceph.deployment import drive_selection, translate +from ceph.deployment.hostspec import HostSpec, SpecValidationError +from ceph.deployment.inventory import Device +from ceph.deployment.service_spec import PlacementSpec +from ceph.tests.utils import _mk_inventory, _mk_device +from ceph.deployment.drive_group import DriveGroupSpec, DeviceSelection, \ + DriveGroupValidationError + +@pytest.mark.parametrize("test_input", +[ + ( # new style json + """service_type: osd +service_id: testing_drivegroup +placement: + host_pattern: hostname +data_devices: + paths: + - /dev/sda +""" + ), + ( + """service_type: osd +service_id: testing_drivegroup +placement: + host_pattern: hostname +data_devices: + paths: + - /dev/sda""" + ), +]) +def test_DriveGroup(test_input): + dg = DriveGroupSpec.from_json(yaml.safe_load(test_input)) + assert dg.service_id == 'testing_drivegroup' + assert all([isinstance(x, Device) for x in dg.data_devices.paths]) + assert dg.data_devices.paths[0].path == '/dev/sda' + +@pytest.mark.parametrize("match,test_input", +[ + ( + re.escape('Service Spec is not an (JSON or YAML) object. got "None"'), + '' + ), + ( + 'Failed to validate OSD spec "<unnamed>": `placement` required', + """data_devices: + all: True +""" + ), + ( + 'Failed to validate OSD spec "mydg.data_devices": device selection cannot be empty', """ +service_type: osd +service_id: mydg +placement: + host_pattern: '*' +data_devices: + limit: 1 +""" + ), + ( + 'Failed to validate OSD spec "mydg": filter_logic must be either <AND> or <OR>', """ +service_type: osd +service_id: mydg +placement: + host_pattern: '*' +data_devices: + all: True +filter_logic: XOR +""" + ), + ( + 'Failed to validate OSD spec "mydg": `data_devices` element is required.', """ +service_type: osd +service_id: mydg +placement: + host_pattern: '*' +spec: + db_devices: + model: model +""" + ), + ( + 'Failed to validate OSD spec "mydg.db_devices": Filtering for `unknown_key` is not supported', """ +service_type: osd +service_id: mydg +placement: + host_pattern: '*' +spec: + db_devices: + unknown_key: 1 +""" + ), + ( + 'Failed to validate OSD spec "mydg": Feature `unknown_key` is not supported', """ +service_type: osd +service_id: mydg +placement: + host_pattern: '*' +spec: + db_devices: + all: true + unknown_key: 1 +""" + ), +]) +def test_DriveGroup_fail(match, test_input): + with pytest.raises(SpecValidationError, match=match): + osd_spec = DriveGroupSpec.from_json(yaml.safe_load(test_input)) + osd_spec.validate() + + +def test_drivegroup_pattern(): + dg = DriveGroupSpec( + PlacementSpec(host_pattern='node[1-3]'), + service_id='foobar', + data_devices=DeviceSelection(all=True)) + assert dg.placement.filter_matching_hostspecs([HostSpec('node{}'.format(i)) for i in range(10)]) == ['node1', 'node2', 'node3'] + + +def test_drive_selection(): + devs = DeviceSelection(paths=['/dev/sda']) + spec = DriveGroupSpec( + PlacementSpec('node_name'), + service_id='foobar', + data_devices=devs) + assert all([isinstance(x, Device) for x in spec.data_devices.paths]) + assert spec.data_devices.paths[0].path == '/dev/sda' + + with pytest.raises(DriveGroupValidationError, match='exclusive'): + ds = DeviceSelection(paths=['/dev/sda'], rotational=False) + ds.validate('') + + +def test_ceph_volume_command_0(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(all=True) + ) + spec.validate() + inventory = _mk_inventory(_mk_device()*2) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --yes --no-systemd' for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_1(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True), + db_devices=DeviceSelection(rotational=False) + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True)*2 + _mk_device(rotational=False)*2) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == ('lvm batch --no-auto /dev/sda /dev/sdb ' + '--db-devices /dev/sdc /dev/sdd --yes --no-systemd') for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_2(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(size='200GB:350GB', rotational=True), + db_devices=DeviceSelection(size='200GB:350GB', rotational=False), + wal_devices=DeviceSelection(size='10G') + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True, size="300.00 GB")*2 + + _mk_device(rotational=False, size="300.00 GB")*2 + + _mk_device(size="10.0 GB", rotational=False)*2 + ) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == ('lvm batch --no-auto /dev/sda /dev/sdb ' + '--db-devices /dev/sdc /dev/sdd --wal-devices /dev/sde /dev/sdf ' + '--yes --no-systemd') for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_3(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(size='200GB:350GB', rotational=True), + db_devices=DeviceSelection(size='200GB:350GB', rotational=False), + wal_devices=DeviceSelection(size='10G'), + encrypted=True + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True, size="300.00 GB")*2 + + _mk_device(rotational=False, size="300.00 GB")*2 + + _mk_device(size="10.0 GB", rotational=False)*2 + ) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == ('lvm batch --no-auto /dev/sda /dev/sdb ' + '--db-devices /dev/sdc /dev/sdd ' + '--wal-devices /dev/sde /dev/sdf --dmcrypt ' + '--yes --no-systemd') for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_4(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(size='200GB:350GB', rotational=True), + db_devices=DeviceSelection(size='200GB:350GB', rotational=False), + wal_devices=DeviceSelection(size='10G'), + block_db_size='500M', + block_wal_size='500M', + osds_per_device=3, + encrypted=True + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True, size="300.00 GB")*2 + + _mk_device(rotational=False, size="300.00 GB")*2 + + _mk_device(size="10.0 GB", rotational=False)*2 + ) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == ('lvm batch --no-auto /dev/sda /dev/sdb ' + '--db-devices /dev/sdc /dev/sdd --wal-devices /dev/sde /dev/sdf ' + '--block-wal-size 500M --block-db-size 500M --dmcrypt ' + '--osds-per-device 3 --yes --no-systemd') for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_5(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True), + objectstore='filestore' + ) + with pytest.raises(DriveGroupValidationError): + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True)*2) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --filestore --yes --no-systemd' for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_6(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=False), + journal_devices=DeviceSelection(rotational=True), + journal_size='500M', + objectstore='filestore' + ) + with pytest.raises(DriveGroupValidationError): + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True)*2 + _mk_device(rotational=False)*2) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == ('lvm batch --no-auto /dev/sdc /dev/sdd ' + '--journal-size 500M --journal-devices /dev/sda /dev/sdb ' + '--filestore --yes --no-systemd') for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_7(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(all=True), + osd_id_claims={'host1': ['0', '1']} + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True)*2) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, ['0', '1']).run() + assert all(cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --osd-ids 0 1 --yes --no-systemd' for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_8(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True, model='INTEL SSDS'), + db_devices=DeviceSelection(model='INTEL SSDP'), + filter_logic='OR', + osd_id_claims={} + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True, size='1.82 TB', model='ST2000DM001-1ER1') + # data + _mk_device(rotational=False, size="223.0 GB", model='INTEL SSDSC2KG24') + # data + _mk_device(rotational=False, size="349.0 GB", model='INTEL SSDPED1K375GA') # wal/db + ) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --db-devices /dev/sdc --yes --no-systemd' for cmd in cmds), f'Expected {cmd} in {cmds}' + + +def test_ceph_volume_command_9(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(all=True), + data_allocate_fraction=0.8 + ) + spec.validate() + inventory = _mk_inventory(_mk_device()*2) + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --data-allocate-fraction 0.8 --yes --no-systemd' for cmd in cmds), f'Expected {cmd} in {cmds}' + +def test_raw_ceph_volume_command_0(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True), + db_devices=DeviceSelection(rotational=False), + method='raw', + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True) + # data + _mk_device(rotational=True) + # data + _mk_device(rotational=False) + # db + _mk_device(rotational=False) # db + ) + exp_cmds = ['raw prepare --bluestore --data /dev/sda --block.db /dev/sdc', 'raw prepare --bluestore --data /dev/sdb --block.db /dev/sdd'] + sel = drive_selection.DriveSelection(spec, inventory) + cmds = translate.to_ceph_volume(sel, []).run() + assert all(cmd in exp_cmds for cmd in cmds), f'Expected {exp_cmds} to match {cmds}' + +def test_raw_ceph_volume_command_1(): + spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'), + service_id='foobar', + data_devices=DeviceSelection(rotational=True), + db_devices=DeviceSelection(rotational=False), + method='raw', + ) + spec.validate() + inventory = _mk_inventory(_mk_device(rotational=True) + # data + _mk_device(rotational=True) + # data + _mk_device(rotational=False) # db + ) + sel = drive_selection.DriveSelection(spec, inventory) + with pytest.raises(ValueError): + cmds = translate.to_ceph_volume(sel, []).run() diff --git a/src/python-common/ceph/tests/test_hostspec.py b/src/python-common/ceph/tests/test_hostspec.py new file mode 100644 index 000000000..b6817579e --- /dev/null +++ b/src/python-common/ceph/tests/test_hostspec.py @@ -0,0 +1,40 @@ +# flake8: noqa +import json +import yaml + +import pytest + +from ceph.deployment.hostspec import HostSpec, SpecValidationError + + +@pytest.mark.parametrize( + "test_input,expected", + [ + ({"hostname": "foo"}, HostSpec('foo')), + ({"hostname": "foo", "labels": "l1"}, HostSpec('foo', labels=['l1'])), + ({"hostname": "foo", "labels": ["l1", "l2"]}, HostSpec('foo', labels=['l1', 'l2'])), + ({"hostname": "foo", "location": {"rack": "foo"}}, HostSpec('foo', location={'rack': 'foo'})), + ] +) +def test_parse_host_specs(test_input, expected): + hs = HostSpec.from_json(test_input) + assert hs == expected + + +@pytest.mark.parametrize( + "bad_input", + [ + ({"hostname": "foo", "labels": 124}), + ({"hostname": "foo", "labels": {"a", "b"}}), + ({"hostname": "foo", "labels": {"a", "b"}}), + ({"hostname": "foo", "labels": ["a", 2]}), + ({"hostname": "foo", "location": "rack=bar"}), + ({"hostname": "foo", "location": ["a"]}), + ({"hostname": "foo", "location": {"rack", 1}}), + ({"hostname": "foo", "location": {1: "rack"}}), + ] +) +def test_parse_host_specs(bad_input): + with pytest.raises(SpecValidationError): + hs = HostSpec.from_json(bad_input) + diff --git a/src/python-common/ceph/tests/test_inventory.py b/src/python-common/ceph/tests/test_inventory.py new file mode 100644 index 000000000..69c1c306c --- /dev/null +++ b/src/python-common/ceph/tests/test_inventory.py @@ -0,0 +1,21 @@ +import json +import os +import pytest + +from ceph.deployment.inventory import Devices + + +@pytest.mark.parametrize("filename", + [ + os.path.dirname(__file__) + '/c-v-inventory.json', + os.path.dirname(__file__) + '/../../../pybind/mgr/test_orchestrator/du' + 'mmy_data.json', + ]) +def test_from_json(filename): + with open(filename) as f: + data = json.load(f) + if 'inventory' in data: + data = data['inventory'] + ds = Devices.from_json(data) + assert len(ds.devices) == len(data) + assert Devices.from_json(ds.to_json()) == ds diff --git a/src/python-common/ceph/tests/test_service_spec.py b/src/python-common/ceph/tests/test_service_spec.py new file mode 100644 index 000000000..d3fb43296 --- /dev/null +++ b/src/python-common/ceph/tests/test_service_spec.py @@ -0,0 +1,866 @@ +# flake8: noqa +import json +import re + +import yaml + +import pytest + +from ceph.deployment.service_spec import HostPlacementSpec, PlacementSpec, \ + ServiceSpec, RGWSpec, NFSServiceSpec, IscsiServiceSpec, AlertManagerSpec, \ + CustomContainerSpec +from ceph.deployment.drive_group import DriveGroupSpec +from ceph.deployment.hostspec import SpecValidationError + + +@pytest.mark.parametrize("test_input,expected, require_network", + [("myhost", ('myhost', '', ''), False), + ("myhost=sname", ('myhost', '', 'sname'), False), + ("myhost:10.1.1.10", ('myhost', '10.1.1.10', ''), True), + ("myhost:10.1.1.10=sname", ('myhost', '10.1.1.10', 'sname'), True), + ("myhost:10.1.1.0/32", ('myhost', '10.1.1.0/32', ''), True), + ("myhost:10.1.1.0/32=sname", ('myhost', '10.1.1.0/32', 'sname'), True), + ("myhost:[v1:10.1.1.10:6789]", ('myhost', '[v1:10.1.1.10:6789]', ''), True), + ("myhost:[v1:10.1.1.10:6789]=sname", ('myhost', '[v1:10.1.1.10:6789]', 'sname'), True), + ("myhost:[v1:10.1.1.10:6789,v2:10.1.1.11:3000]", ('myhost', '[v1:10.1.1.10:6789,v2:10.1.1.11:3000]', ''), True), + ("myhost:[v1:10.1.1.10:6789,v2:10.1.1.11:3000]=sname", ('myhost', '[v1:10.1.1.10:6789,v2:10.1.1.11:3000]', 'sname'), True), + ]) +def test_parse_host_placement_specs(test_input, expected, require_network): + ret = HostPlacementSpec.parse(test_input, require_network=require_network) + assert ret == expected + assert str(ret) == test_input + + ps = PlacementSpec.from_string(test_input) + assert ps.pretty_str() == test_input + assert ps == PlacementSpec.from_string(ps.pretty_str()) + + # Testing the old verbose way of generating json. Don't remove: + assert ret == HostPlacementSpec.from_json({ + 'hostname': ret.hostname, + 'network': ret.network, + 'name': ret.name + }) + + assert ret == HostPlacementSpec.from_json(ret.to_json()) + + + + +@pytest.mark.parametrize( + "test_input,expected", + [ + ('', "PlacementSpec()"), + ("count:2", "PlacementSpec(count=2)"), + ("3", "PlacementSpec(count=3)"), + ("host1 host2", "PlacementSpec(hosts=[HostPlacementSpec(hostname='host1', network='', name=''), HostPlacementSpec(hostname='host2', network='', name='')])"), + ("host1;host2", "PlacementSpec(hosts=[HostPlacementSpec(hostname='host1', network='', name=''), HostPlacementSpec(hostname='host2', network='', name='')])"), + ("host1,host2", "PlacementSpec(hosts=[HostPlacementSpec(hostname='host1', network='', name=''), HostPlacementSpec(hostname='host2', network='', name='')])"), + ("host1 host2=b", "PlacementSpec(hosts=[HostPlacementSpec(hostname='host1', network='', name=''), HostPlacementSpec(hostname='host2', network='', name='b')])"), + ("host1=a host2=b", "PlacementSpec(hosts=[HostPlacementSpec(hostname='host1', network='', name='a'), HostPlacementSpec(hostname='host2', network='', name='b')])"), + ("host1:1.2.3.4=a host2:1.2.3.5=b", "PlacementSpec(hosts=[HostPlacementSpec(hostname='host1', network='1.2.3.4', name='a'), HostPlacementSpec(hostname='host2', network='1.2.3.5', name='b')])"), + ("myhost:[v1:10.1.1.10:6789]", "PlacementSpec(hosts=[HostPlacementSpec(hostname='myhost', network='[v1:10.1.1.10:6789]', name='')])"), + ('2 host1 host2', "PlacementSpec(count=2, hosts=[HostPlacementSpec(hostname='host1', network='', name=''), HostPlacementSpec(hostname='host2', network='', name='')])"), + ('label:foo', "PlacementSpec(label='foo')"), + ('3 label:foo', "PlacementSpec(count=3, label='foo')"), + ('*', "PlacementSpec(host_pattern='*')"), + ('3 data[1-3]', "PlacementSpec(count=3, host_pattern='data[1-3]')"), + ('3 data?', "PlacementSpec(count=3, host_pattern='data?')"), + ('3 data*', "PlacementSpec(count=3, host_pattern='data*')"), + ("count-per-host:4 label:foo", "PlacementSpec(count_per_host=4, label='foo')"), + ]) +def test_parse_placement_specs(test_input, expected): + ret = PlacementSpec.from_string(test_input) + assert str(ret) == expected + assert PlacementSpec.from_string(ret.pretty_str()) == ret, f'"{ret.pretty_str()}" != "{test_input}"' + +@pytest.mark.parametrize( + "test_input", + [ + ("host=a host*"), + ("host=a label:wrong"), + ("host? host*"), + ('host=a count-per-host:0'), + ('host=a count-per-host:-10'), + ('count:2 count-per-host:1'), + ('host1=a host2=b count-per-host:2'), + ('host1:10/8 count-per-host:2'), + ('count-per-host:2'), + ] +) +def test_parse_placement_specs_raises(test_input): + with pytest.raises(SpecValidationError): + PlacementSpec.from_string(test_input) + +@pytest.mark.parametrize("test_input", + # wrong subnet + [("myhost:1.1.1.1/24"), + # wrong ip format + ("myhost:1"), + ]) +def test_parse_host_placement_specs_raises_wrong_format(test_input): + with pytest.raises(ValueError): + HostPlacementSpec.parse(test_input) + + +@pytest.mark.parametrize( + "p,hosts,size", + [ + ( + PlacementSpec(count=3), + ['host1', 'host2', 'host3', 'host4', 'host5'], + 3 + ), + ( + PlacementSpec(host_pattern='*'), + ['host1', 'host2', 'host3', 'host4', 'host5'], + 5 + ), + ( + PlacementSpec(count_per_host=2, host_pattern='*'), + ['host1', 'host2', 'host3', 'host4', 'host5'], + 10 + ), + ( + PlacementSpec(host_pattern='foo*'), + ['foo1', 'foo2', 'bar1', 'bar2'], + 2 + ), + ( + PlacementSpec(count_per_host=2, host_pattern='foo*'), + ['foo1', 'foo2', 'bar1', 'bar2'], + 4 + ), + ]) +def test_placement_target_size(p, hosts, size): + assert p.get_target_count( + [HostPlacementSpec(n, '', '') for n in hosts] + ) == size + + +def _get_dict_spec(s_type, s_id): + dict_spec = { + "service_id": s_id, + "service_type": s_type, + "placement": + dict(hosts=["host1:1.1.1.1"]) + } + if s_type == 'nfs': + pass + elif s_type == 'iscsi': + dict_spec['pool'] = 'pool' + dict_spec['api_user'] = 'api_user' + dict_spec['api_password'] = 'api_password' + elif s_type == 'osd': + dict_spec['spec'] = { + 'data_devices': { + 'all': True + } + } + elif s_type == 'rgw': + dict_spec['rgw_realm'] = 'realm' + dict_spec['rgw_zone'] = 'zone' + + return dict_spec + + +@pytest.mark.parametrize( + "s_type,o_spec,s_id", + [ + ("mgr", ServiceSpec, 'test'), + ("mon", ServiceSpec, 'test'), + ("mds", ServiceSpec, 'test'), + ("rgw", RGWSpec, 'realm.zone'), + ("nfs", NFSServiceSpec, 'test'), + ("iscsi", IscsiServiceSpec, 'test'), + ("osd", DriveGroupSpec, 'test'), + ]) +def test_servicespec_map_test(s_type, o_spec, s_id): + spec = ServiceSpec.from_json(_get_dict_spec(s_type, s_id)) + assert isinstance(spec, o_spec) + assert isinstance(spec.placement, PlacementSpec) + assert isinstance(spec.placement.hosts[0], HostPlacementSpec) + assert spec.placement.hosts[0].hostname == 'host1' + assert spec.placement.hosts[0].network == '1.1.1.1' + assert spec.placement.hosts[0].name == '' + assert spec.validate() is None + ServiceSpec.from_json(spec.to_json()) + +def test_osd_unmanaged(): + osd_spec = {"placement": {"host_pattern": "*"}, + "service_id": "all-available-devices", + "service_name": "osd.all-available-devices", + "service_type": "osd", + "spec": {"data_devices": {"all": True}, "filter_logic": "AND", "objectstore": "bluestore"}, + "unmanaged": True} + + dg_spec = ServiceSpec.from_json(osd_spec) + assert dg_spec.unmanaged == True + + +@pytest.mark.parametrize("y", +"""service_type: crash +service_name: crash +placement: + host_pattern: '*' +--- +service_type: crash +service_name: crash +placement: + host_pattern: '*' +unmanaged: true +--- +service_type: rgw +service_id: default-rgw-realm.eu-central-1.1 +service_name: rgw.default-rgw-realm.eu-central-1.1 +placement: + hosts: + - ceph-001 +networks: +- 10.0.0.0/8 +- 192.168.0.0/16 +spec: + rgw_frontend_type: civetweb + rgw_realm: default-rgw-realm + rgw_zone: eu-central-1 +--- +service_type: osd +service_id: osd_spec_default +service_name: osd.osd_spec_default +placement: + host_pattern: '*' +spec: + data_devices: + model: MC-55-44-XZ + db_devices: + model: SSD-123-foo + filter_logic: AND + objectstore: bluestore + wal_devices: + model: NVME-QQQQ-987 +--- +service_type: alertmanager +service_name: alertmanager +spec: + port: 1234 + user_data: + default_webhook_urls: + - foo +--- +service_type: grafana +service_name: grafana +spec: + port: 1234 +--- +service_type: grafana +service_name: grafana +spec: + initial_admin_password: secure + port: 1234 +--- +service_type: ingress +service_id: rgw.foo +service_name: ingress.rgw.foo +placement: + hosts: + - host1 + - host2 + - host3 +spec: + backend_service: rgw.foo + frontend_port: 8080 + monitor_port: 8081 + virtual_ip: 192.168.20.1/24 +--- +service_type: nfs +service_id: mynfs +service_name: nfs.mynfs +spec: + port: 1234 +--- +service_type: iscsi +service_id: iscsi +service_name: iscsi.iscsi +networks: +- ::0/8 +spec: + api_user: api_user + pool: pool + trusted_ip_list: + - ::1 + - ::2 +--- +service_type: container +service_id: hello-world +service_name: container.hello-world +spec: + args: + - --foo + bind_mounts: + - - type=bind + - source=lib/modules + - destination=/lib/modules + - ro=true + dirs: + - foo + - bar + entrypoint: /usr/bin/bash + envs: + - FOO=0815 + files: + bar.conf: + - foo + - bar + foo.conf: 'foo + + bar' + gid: 2000 + image: docker.io/library/hello-world:latest + ports: + - 8080 + - 8443 + uid: 1000 + volume_mounts: + foo: /foo +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public + snmp_destination: 192.168.1.42:162 + snmp_version: V2c +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + auth_protocol: MD5 + credentials: + snmp_v3_auth_password: mypassword + snmp_v3_auth_username: myuser + engine_id: 8000C53F00000000 + port: 9464 + snmp_destination: 192.168.1.42:162 + snmp_version: V3 +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_password: mypassword + snmp_v3_auth_username: myuser + snmp_v3_priv_password: mysecret + engine_id: 8000C53F00000000 + privacy_protocol: AES + snmp_destination: 192.168.1.42:162 + snmp_version: V3 +""".split('---\n')) +def test_yaml(y): + data = yaml.safe_load(y) + object = ServiceSpec.from_json(data) + + assert yaml.dump(object) == y + assert yaml.dump(ServiceSpec.from_json(object.to_json())) == y + + +def test_alertmanager_spec_1(): + spec = AlertManagerSpec() + assert spec.service_type == 'alertmanager' + assert isinstance(spec.user_data, dict) + assert len(spec.user_data.keys()) == 0 + assert spec.get_port_start() == [9093, 9094] + + +def test_alertmanager_spec_2(): + spec = AlertManagerSpec(user_data={'default_webhook_urls': ['foo']}) + assert isinstance(spec.user_data, dict) + assert 'default_webhook_urls' in spec.user_data.keys() + + + +def test_repr(): + val = """ServiceSpec.from_json(yaml.safe_load('''service_type: crash +service_name: crash +placement: + count: 42 +'''))""" + obj = eval(val) + assert obj.service_type == 'crash' + assert val == repr(obj) + +@pytest.mark.parametrize("spec1, spec2, eq", + [ + ( + ServiceSpec( + service_type='mon' + ), + ServiceSpec( + service_type='mon' + ), + True + ), + ( + ServiceSpec( + service_type='mon' + ), + ServiceSpec( + service_type='mon', + service_id='foo' + ), + True + ), + # Add service_type='mgr' + ( + ServiceSpec( + service_type='osd' + ), + ServiceSpec( + service_type='osd', + ), + True + ), + ( + ServiceSpec( + service_type='osd' + ), + DriveGroupSpec(), + True + ), + ( + ServiceSpec( + service_type='osd' + ), + ServiceSpec( + service_type='osd', + service_id='foo', + ), + False + ), + ( + ServiceSpec( + service_type='rgw', + service_id='foo', + ), + RGWSpec(service_id='foo'), + True + ), + ]) +def test_spec_hash_eq(spec1: ServiceSpec, + spec2: ServiceSpec, + eq: bool): + + assert (spec1 == spec2) is eq + +@pytest.mark.parametrize( + "s_type,s_id,s_name", + [ + ('mgr', 's_id', 'mgr'), + ('mon', 's_id', 'mon'), + ('mds', 's_id', 'mds.s_id'), + ('rgw', 's_id', 'rgw.s_id'), + ('nfs', 's_id', 'nfs.s_id'), + ('iscsi', 's_id', 'iscsi.s_id'), + ('osd', 's_id', 'osd.s_id'), + ]) +def test_service_name(s_type, s_id, s_name): + spec = ServiceSpec.from_json(_get_dict_spec(s_type, s_id)) + spec.validate() + assert spec.service_name() == s_name + +@pytest.mark.parametrize( + 's_type,s_id', + [ + ('mds', 's:id'), # MDS service_id cannot contain an invalid char ':' + ('mds', '1abc'), # MDS service_id cannot start with a numeric digit + ('mds', ''), # MDS service_id cannot be empty + ('rgw', '*s_id'), + ('nfs', 's/id'), + ('iscsi', 's@id'), + ('osd', 's;id'), + ]) + +def test_service_id_raises_invalid_char(s_type, s_id): + with pytest.raises(SpecValidationError): + spec = ServiceSpec.from_json(_get_dict_spec(s_type, s_id)) + spec.validate() + +def test_custom_container_spec(): + spec = CustomContainerSpec(service_id='hello-world', + image='docker.io/library/hello-world:latest', + entrypoint='/usr/bin/bash', + uid=1000, + gid=2000, + volume_mounts={'foo': '/foo'}, + args=['--foo'], + envs=['FOO=0815'], + bind_mounts=[ + [ + 'type=bind', + 'source=lib/modules', + 'destination=/lib/modules', + 'ro=true' + ] + ], + ports=[8080, 8443], + dirs=['foo', 'bar'], + files={ + 'foo.conf': 'foo\nbar', + 'bar.conf': ['foo', 'bar'] + }) + assert spec.service_type == 'container' + assert spec.entrypoint == '/usr/bin/bash' + assert spec.uid == 1000 + assert spec.gid == 2000 + assert spec.volume_mounts == {'foo': '/foo'} + assert spec.args == ['--foo'] + assert spec.envs == ['FOO=0815'] + assert spec.bind_mounts == [ + [ + 'type=bind', + 'source=lib/modules', + 'destination=/lib/modules', + 'ro=true' + ] + ] + assert spec.ports == [8080, 8443] + assert spec.dirs == ['foo', 'bar'] + assert spec.files == { + 'foo.conf': 'foo\nbar', + 'bar.conf': ['foo', 'bar'] + } + + +def test_custom_container_spec_config_json(): + spec = CustomContainerSpec(service_id='foo', image='foo', dirs=None) + config_json = spec.config_json() + for key in ['entrypoint', 'uid', 'gid', 'bind_mounts', 'dirs']: + assert key not in config_json + + +def test_ingress_spec(): + yaml_str = """service_type: ingress +service_id: rgw.foo +placement: + hosts: + - host1 + - host2 + - host3 +spec: + virtual_ip: 192.168.20.1/24 + backend_service: rgw.foo + frontend_port: 8080 + monitor_port: 8081 +""" + yaml_file = yaml.safe_load(yaml_str) + spec = ServiceSpec.from_json(yaml_file) + assert spec.service_type == "ingress" + assert spec.service_id == "rgw.foo" + assert spec.virtual_ip == "192.168.20.1/24" + assert spec.frontend_port == 8080 + assert spec.monitor_port == 8081 + + +@pytest.mark.parametrize("y, error_match", [ + (""" +service_type: rgw +service_id: foo +placement: + count_per_host: "twelve" +""", "count-per-host must be a numeric value",), + (""" +service_type: rgw +service_id: foo +placement: + count_per_host: "2" +""", "count-per-host must be an integer value",), + (""" +service_type: rgw +service_id: foo +placement: + count_per_host: 7.36 +""", "count-per-host must be an integer value",), + (""" +service_type: rgw +service_id: foo +placement: + count: "fifteen" +""", "num/count must be a numeric value",), + (""" +service_type: rgw +service_id: foo +placement: + count: "4" +""", "num/count must be an integer value",), + (""" +service_type: rgw +service_id: foo +placement: + count: 7.36 +""", "num/count must be an integer value",), + (""" +service_type: rgw +service_id: foo +placement: + count: 0 +""", "num/count must be >= 1",), + (""" +service_type: rgw +service_id: foo +placement: + count_per_host: 0 +""", "count-per-host must be >= 1",), + (""" +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_password: mypassword + snmp_v3_auth_username: myuser + snmp_v3_priv_password: mysecret + port: 9464 + engine_id: 8000c53f0000000000 + privacy_protocol: WEIRD + snmp_destination: 192.168.122.1:162 + auth_protocol: BIZARRE + snmp_version: V3 +""", "auth_protocol unsupported. Must be one of MD5, SHA"), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public + snmp_destination: 192.168.1.42:162 + snmp_version: V4 +""", 'snmp_version unsupported. Must be one of V2c, V3'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public + port: 9464 + snmp_destination: 192.168.1.42:162 +""", re.escape('Missing SNMP version (snmp_version)')), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + port: 9464 + auth_protocol: wah + snmp_destination: 192.168.1.42:162 + snmp_version: V3 +""", 'auth_protocol unsupported. Must be one of MD5, SHA'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: weewah + snmp_destination: 192.168.1.42:162 + snmp_version: V3 +""", 'privacy_protocol unsupported. Must be one of DES, AES'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + snmp_destination: 192.168.1.42:162 + snmp_version: V3 +""", 'Must provide an engine_id for SNMP V3 notifications'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public + port: 9464 + snmp_destination: 192.168.1.42 + snmp_version: V2c +""", re.escape('SNMP destination (snmp_destination) type (IPv4) is invalid. Must be either: IPv4:Port, Name:Port')), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + engine_id: bogus + snmp_destination: 192.168.1.42:162 + snmp_version: V3 +""", 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + port: 9464 + auth_protocol: SHA + engine_id: 8000C53F0000000000 + snmp_version: V3 +""", re.escape('SNMP destination (snmp_destination) must be provided')), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + engine_id: 8000C53F0000000000 + snmp_destination: my.imaginary.snmp-host + snmp_version: V3 +""", re.escape('SNMP destination (snmp_destination) is invalid: DNS lookup failed')), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + engine_id: 8000C53F0000000000 + snmp_destination: 10.79.32.10:fred + snmp_version: V3 +""", re.escape('SNMP destination (snmp_destination) is invalid: Port must be numeric')), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + engine_id: 8000C53 + snmp_destination: 10.79.32.10:162 + snmp_version: V3 +""", 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + engine_id: 8000C53DOH! + snmp_destination: 10.79.32.10:162 + snmp_version: V3 +""", 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + engine_id: 8000C53FCA7344403DC611EC9B985254002537A6C53FCA7344403DC6112537A60 + snmp_destination: 10.79.32.10:162 + snmp_version: V3 +""", 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret + port: 9464 + auth_protocol: SHA + privacy_protocol: AES + engine_id: 8000C53F00000 + snmp_destination: 10.79.32.10:162 + snmp_version: V3 +""", 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + ]) +def test_service_spec_validation_error(y, error_match): + data = yaml.safe_load(y) + with pytest.raises(SpecValidationError) as err: + specObj = ServiceSpec.from_json(data) + assert err.match(error_match) diff --git a/src/python-common/ceph/tests/test_utils.py b/src/python-common/ceph/tests/test_utils.py new file mode 100644 index 000000000..8a94ac400 --- /dev/null +++ b/src/python-common/ceph/tests/test_utils.py @@ -0,0 +1,75 @@ +import pytest + +from ceph.deployment.utils import is_ipv6, unwrap_ipv6, wrap_ipv6, valid_addr +from typing import NamedTuple + + +def test_is_ipv6(): + for good in ("[::1]", "::1", + "fff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"): + assert is_ipv6(good) + for bad in ("127.0.0.1", + "ffff:ffff:ffff:ffff:ffff:ffff:ffff:fffg", + "1:2:3:4:5:6:7:8:9", "fd00::1::1", "[fg::1]"): + assert not is_ipv6(bad) + + +def test_unwrap_ipv6(): + def unwrap_test(address, expected): + assert unwrap_ipv6(address) == expected + + tests = [ + ('::1', '::1'), ('[::1]', '::1'), + ('[fde4:8dba:82e1:0:5054:ff:fe6a:357]', 'fde4:8dba:82e1:0:5054:ff:fe6a:357'), + ('can actually be any string', 'can actually be any string'), + ('[but needs to be stripped] ', '[but needs to be stripped] ')] + for address, expected in tests: + unwrap_test(address, expected) + + +def test_wrap_ipv6(): + def wrap_test(address, expected): + assert wrap_ipv6(address) == expected + + tests = [ + ('::1', '[::1]'), ('[::1]', '[::1]'), + ('fde4:8dba:82e1:0:5054:ff:fe6a:357', '[fde4:8dba:82e1:0:5054:ff:fe6a:357]'), + ('myhost.example.com', 'myhost.example.com'), ('192.168.0.1', '192.168.0.1'), + ('', ''), ('fd00::1::1', 'fd00::1::1')] + for address, expected in tests: + wrap_test(address, expected) + + +class Address(NamedTuple): + addr: str + status: bool + description: str + + +@pytest.mark.parametrize('addr_object', [ + Address('www.ibm.com', True, 'Name'), + Address('www.google.com:162', True, 'Name:Port'), + Address('my.big.domain.name.for.big.people', False, 'DNS lookup failed'), + Address('192.168.122.1', True, 'IPv4'), + Address('[192.168.122.1]', False, 'IPv4 address wrapped in brackets is invalid'), + Address('10.40003.200', False, 'Invalid partial IPv4 address'), + Address('10.7.5', False, 'Invalid partial IPv4 address'), + Address('10.7', False, 'Invalid partial IPv4 address'), + Address('192.168.122.5:7090', True, 'IPv4:Port'), + Address('fe80::7561:c8fb:d3d7:1fa4', True, 'IPv6'), + Address('[fe80::7561:c8fb:d3d7:1fa4]:9464', True, 'IPv6:Port'), + Address('[fe80::7561:c8fb:d3d7:1fa4]', True, 'IPv6'), + Address('[fe80::7561:c8fb:d3d7:1fa4', False, + 'Address has incorrect/incomplete use of enclosing brackets'), + Address('fe80::7561:c8fb:d3d7:1fa4]', False, + 'Address has incorrect/incomplete use of enclosing brackets'), + Address('fred.knockinson.org', False, 'DNS lookup failed'), + Address('tumbleweed.pickles.gov.au', False, 'DNS lookup failed'), + Address('192.168.122.5:00PS', False, 'Port must be numeric'), + Address('[fe80::7561:c8fb:d3d7:1fa4]:DOH', False, 'Port must be numeric') +]) +def test_valid_addr(addr_object: Address): + + valid, description = valid_addr(addr_object.addr) + assert valid == addr_object.status + assert description == addr_object.description diff --git a/src/python-common/ceph/tests/utils.py b/src/python-common/ceph/tests/utils.py new file mode 100644 index 000000000..04b8a4e38 --- /dev/null +++ b/src/python-common/ceph/tests/utils.py @@ -0,0 +1,46 @@ +from ceph.deployment.inventory import Devices, Device + +try: + from typing import Any, List +except ImportError: + pass # for type checking + + +def _mk_device(rotational=True, + locked=False, + size="394.27 GB", + vendor='Vendor', + model='Model'): + return [Device( + path='??', + sys_api={ + "rotational": '1' if rotational else '0', + "vendor": vendor, + "human_readable_size": size, + "partitions": {}, + "locked": int(locked), + "sectorsize": "512", + "removable": "0", + "path": "??", + "support_discard": "", + "model": model, + "ro": "0", + "nr_requests": "128", + "size": 423347879936 # ignore coversion from human_readable_size + }, + available=not locked, + rejected_reasons=['locked'] if locked else [], + lvs=[], + device_id="Model-Vendor-foobar" + )] + + +def _mk_inventory(devices): + # type: (Any) -> List[Device] + devs = [] + for dev_, name in zip(devices, map(chr, range(ord('a'), ord('z')))): + dev = Device.from_json(dev_.to_json()) + dev.path = '/dev/sd' + name + dev.sys_api = dict(dev_.sys_api, path='/dev/sd' + name) + devs.append(dev) + return Devices(devices=devs).devices diff --git a/src/python-common/ceph/utils.py b/src/python-common/ceph/utils.py new file mode 100644 index 000000000..643be0658 --- /dev/null +++ b/src/python-common/ceph/utils.py @@ -0,0 +1,123 @@ +import datetime +import re +import string + +from typing import Optional + + +def datetime_now() -> datetime.datetime: + """ + Return the current local date and time. + :return: Returns an aware datetime object of the current date + and time. + """ + return datetime.datetime.now(tz=datetime.timezone.utc) + + +def datetime_to_str(dt: datetime.datetime) -> str: + """ + Convert a datetime object into a ISO 8601 string, e.g. + '2019-04-24T17:06:53.039991Z'. + :param dt: The datetime object to process. + :return: Return a string representing the date in + ISO 8601 (timezone=UTC). + """ + return dt.astimezone(tz=datetime.timezone.utc).strftime( + '%Y-%m-%dT%H:%M:%S.%fZ') + + +def str_to_datetime(string: str) -> datetime.datetime: + """ + Convert an ISO 8601 string into a datetime object. + The following formats are supported: + + - 2020-03-03T09:21:43.636153304Z + - 2020-03-03T15:52:30.136257504-0600 + - 2020-03-03T15:52:30.136257504 + + :param string: The string to parse. + :return: Returns an aware datetime object of the given date + and time string. + :raises: :exc:`~exceptions.ValueError` for an unknown + datetime string. + """ + fmts = [ + '%Y-%m-%dT%H:%M:%S.%f', + '%Y-%m-%dT%H:%M:%S.%f%z' + ] + + # In *all* cases, the 9 digit second precision is too much for + # Python's strptime. Shorten it to 6 digits. + p = re.compile(r'(\.[\d]{6})[\d]*') + string = p.sub(r'\1', string) + + # Replace trailing Z with -0000, since (on Python 3.6.8) it + # won't parse. + if string and string[-1] == 'Z': + string = string[:-1] + '-0000' + + for fmt in fmts: + try: + dt = datetime.datetime.strptime(string, fmt) + # Make sure the datetime object is aware (timezone is set). + # If not, then assume the time is in UTC. + if dt.tzinfo is None: + dt = dt.replace(tzinfo=datetime.timezone.utc) + return dt + except ValueError: + pass + + raise ValueError("Time data {} does not match one of the formats {}".format( + string, str(fmts))) + + +def parse_timedelta(delta: str) -> Optional[datetime.timedelta]: + """ + Returns a timedelta object represents a duration, the difference + between two dates or times. + + >>> parse_timedelta('foo') + + >>> parse_timedelta('2d') == datetime.timedelta(days=2) + True + + >>> parse_timedelta("4w") == datetime.timedelta(days=28) + True + + >>> parse_timedelta("5s") == datetime.timedelta(seconds=5) + True + + >>> parse_timedelta("-5s") == datetime.timedelta(days=-1, seconds=86395) + True + + :param delta: The string to process, e.g. '2h', '10d', '30s'. + :return: The `datetime.timedelta` object or `None` in case of + a parsing error. + """ + parts = re.match(r'(?P<seconds>-?\d+)s|' + r'(?P<minutes>-?\d+)m|' + r'(?P<hours>-?\d+)h|' + r'(?P<days>-?\d+)d|' + r'(?P<weeks>-?\d+)w$', + delta, + re.IGNORECASE) + if not parts: + return None + parts = parts.groupdict() + args = {name: int(param) for name, param in parts.items() if param} + return datetime.timedelta(**args) + + +def is_hex(s: str, strict: bool = True) -> bool: + """Simple check that a string contains only hex chars""" + try: + int(s, 16) + except ValueError: + return False + + # s is multiple chars, but we should catch a '+/-' prefix too. + if strict: + if s[0] not in string.hexdigits: + return False + + return True diff --git a/src/python-common/requirements-lint.txt b/src/python-common/requirements-lint.txt new file mode 100644 index 000000000..2a7142182 --- /dev/null +++ b/src/python-common/requirements-lint.txt @@ -0,0 +1,2 @@ +flake8==3.7.8 +rstcheck==3.3.1 diff --git a/src/python-common/requirements.txt b/src/python-common/requirements.txt new file mode 100644 index 000000000..432fcd5e3 --- /dev/null +++ b/src/python-common/requirements.txt @@ -0,0 +1,7 @@ +pytest >=2.1.3,<5; python_version < '3.5' +mock; python_version < '3.3' +mypy==0.790; python_version >= '3' +pytest-mypy; python_version >= '3' +pytest >= 2.1.3; python_version >= '3' +pyyaml +typing-extensions; python_version < '3.8' diff --git a/src/python-common/setup.py b/src/python-common/setup.py new file mode 100644 index 000000000..43a46eb10 --- /dev/null +++ b/src/python-common/setup.py @@ -0,0 +1,32 @@ +from setuptools import setup, find_packages + + +with open("README.rst", "r") as fh: + long_description = fh.read() + + +setup( + name='ceph', + version='1.0.0', + packages=find_packages(), + author='', + author_email='dev@ceph.io', + description='Ceph common library', + long_description=long_description, + license='LGPLv2+', + keywords='ceph', + url="https://github.com/ceph/ceph", + zip_safe = False, + install_requires=( + 'pyyaml', + ), + classifiers = [ + 'Intended Audience :: Developer', + 'Operating System :: POSIX :: Linux', + 'License :: OSI Approved :: GNU Lesser General Public License v2 or later (LGPLv2+)', + 'Programming Language :: Python', + 'Programming Language :: Python :: 2.7', + 'Programming Language :: Python :: 3.5', + 'Programming Language :: Python :: 3.6', + ] +) diff --git a/src/python-common/tox.ini b/src/python-common/tox.ini new file mode 100644 index 000000000..5869b6ae2 --- /dev/null +++ b/src/python-common/tox.ini @@ -0,0 +1,27 @@ +[tox] +envlist = py3, lint +skip_missing_interpreters = true + +[testenv:py3] +deps= + -rrequirements.txt +commands= + pytest --doctest-modules ceph/deployment/service_spec.py ceph/utils.py + pytest {posargs} + mypy --config-file=../mypy.ini -p ceph + +[tool:pytest] +norecursedirs = .* _* virtualenv + +[flake8] +max-line-length = 100 +exclude = + __pycache__ + +[testenv:lint] +deps = + -rrequirements-lint.txt +commands = + flake8 {posargs:ceph} + rstcheck --report info --debug README.rst + |