ceph/src/python-common/ceph/deployment/drive_selection/selector.py

   1 import logging
   2
   3 try:
   4     from typing import List, Optional
   5 except ImportError:
   6     pass
   7
   8 from ..inventory import Device
   9 from ..drive_group import DriveGroupSpec, DeviceSelection
  10
  11 from .filter import FilterGenerator
  12
  13 logger = logging.getLogger(__name__)
  14
  15
  16 class DriveSelection(object):
  17     def __init__(self,
  18                  spec,  # type: DriveGroupSpec
  19                  disks,  # type: List[Device]
  20                  ):
  21         self.disks = disks.copy()
  22         self.spec = spec
  23
  24         if self.spec.data_devices.paths:  # type: ignore
  25             # re: type: ignore there is *always* a path attribute assigned to DeviceSelection
  26             # it's just None if actual drivegroups are used
  27             self._data = self.spec.data_devices.paths  # type: ignore
  28             self._db = []  # type: List
  29             self._wal = []  # type: List
  30             self._journal = []  # type: List
  31         else:
  32             self._data = self.assign_devices(self.spec.data_devices)
  33             self._wal = self.assign_devices(self.spec.wal_devices)
  34             self._db = self.assign_devices(self.spec.db_devices)
  35             self._journal = self.assign_devices(self.spec.journal_devices)
  36
  37     def data_devices(self):
  38         # type: () -> List[Device]
  39         return self._data
  40
  41     def wal_devices(self):
  42         # type: () -> List[Device]
  43         return self._wal
  44
  45     def db_devices(self):
  46         # type: () -> List[Device]
  47         return self._db
  48
  49     def journal_devices(self):
  50         # type: () -> List[Device]
  51         return self._journal
  52
  53     @staticmethod
  54     def _limit_reached(device_filter, len_devices,
  55                        disk_path):
  56         # type: (DeviceSelection, int, str) -> bool
  57         """ Check for the <limit> property and apply logic
  58
  59         If a limit is set in 'device_attrs' we have to stop adding
  60         disks at some point.
  61
  62         If limit is set (>0) and len(devices) >= limit
  63
  64         :param int len_devices: Length of the already populated device set/list
  65         :param str disk_path: The disk identifier (for logging purposes)
  66         :return: True/False if the device should be added to the list of devices
  67         :rtype: bool
  68         """
  69         limit = device_filter.limit or 0
  70
  71         if limit > 0 and len_devices >= limit:
  72             logger.info("Refuse to add {} due to limit policy of <{}>".format(
  73                 disk_path, limit))
  74             return True
  75         return False
  76
  77     @staticmethod
  78     def _has_mandatory_idents(disk):
  79         # type: (Device) -> bool
  80         """ Check for mandatory identification fields
  81         """
  82         if disk.path:
  83             logger.debug("Found matching disk: {}".format(disk.path))
  84             return True
  85         else:
  86             raise Exception(
  87                 "Disk {} doesn't have a 'path' identifier".format(disk))
  88
  89     def assign_devices(self, device_filter):
  90         # type: (Optional[DeviceSelection]) -> List[Device]
  91         """ Assign drives based on used filters
  92
  93         Do not add disks when:
  94
  95         1) Filter didn't match
  96         2) Disk doesn't have a mandatory identification item (path)
  97         3) The set :limit was reached
  98
  99         After the disk was added we make sure not to re-assign this disk
 100         for another defined type[wal/db/journal devices]
 101
 102         return a sorted(by path) list of devices
 103         """
 104
 105         if not device_filter:
 106             logger.debug('device_filter is None')
 107             return []
 108
 109         if not self.spec.data_devices:
 110             logger.debug('data_devices is None')
 111             return []
 112
 113         devices = list()  # type: List[Device]
 114         for disk in self.disks:
 115             logger.debug("Processing disk {}".format(disk.path))
 116
 117             if not disk.available:
 118                 logger.debug(
 119                     "Ignoring disk {}. Disk is not available".format(disk.path))
 120                 continue
 121
 122             if not self._has_mandatory_idents(disk):
 123                 logger.debug(
 124                     "Ignoring disk {}. Missing mandatory idents".format(
 125                         disk.path))
 126                 continue
 127
 128             # break on this condition.
 129             if self._limit_reached(device_filter, len(devices), disk.path):
 130                 logger.debug("Ignoring disk {}. Limit reached".format(
 131                     disk.path))
 132                 break
 133
 134             if disk in devices:
 135                 continue
 136
 137             if not all(m.compare(disk) for m in FilterGenerator(device_filter)):
 138                 logger.debug(
 139                     "Ignoring disk {}. Filter did not match".format(
 140                         disk.path))
 141                 continue
 142
 143             logger.debug('Adding disk {}'.format(disk.path))
 144             devices.append(disk)
 145
 146         # This disk is already taken and must not be re-assigned.
 147         for taken_device in devices:
 148             if taken_device in self.disks:
 149                 self.disks.remove(taken_device)
 150
 151         return sorted([x for x in devices], key=lambda dev: dev.path)