OpenStack Block Storage (Cinder)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1056 lines
38 KiB

# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# Copyright 2011 Justin Santa Barbara
# All Rights Reserved.
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Utilities and helper functions for all Cinder code.
This file is for utilities useful in all of Cinder,
including cinder-manage, the api service, the scheduler,
Code related to volume drivers and connecting to volumes
should be placed in volume_utils instead.
from __future__ import annotations # Remove when only supporting python 3.9+
from collections import OrderedDict
import contextlib
import datetime
import functools
import inspect
import logging as py_logging
import math
import multiprocessing
import operator
import os
import pyclbr
import re
import shutil
import stat
import sys
import tempfile
import typing
from typing import Callable, Iterable, Iterator # noqa: H301
from typing import Optional, Type, Union # noqa: H301
import eventlet
from eventlet import tpool
from oslo_concurrency import lockutils
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
from oslo_utils import importutils
from oslo_utils import strutils
from oslo_utils import timeutils
import tenacity
from cinder import coordination
from cinder import exception
from cinder.i18n import _
LOG = logging.getLogger(__name__)
ISO_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
PERFECT_TIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
INFINITE_UNKNOWN_VALUES = ('infinite', 'unknown')
synchronized = lockutils.synchronized_with_prefix('cinder-')
synchronized_remove = lockutils.remove_external_lock_file_with_prefix(
def clean_volume_file_locks(volume_id, driver):
"""Remove file locks used by Cinder.
This doesn't take care of driver locks, those should be handled in driver's
delete_volume method.
for name in (volume_id + '-delete_volume', volume_id,
volume_id + '-detach_volume'):
except Exception as exc:
LOG.warning('Failed to cleanup volume lock %(name)s: %(exc)s',
{'name': name, 'exc': exc})
except Exception as exc:
LOG.warning('Failed to cleanup driver locks for volume %(id)s: '
'%(exc)s', {'id': volume_id, 'exc': exc})
def api_clean_volume_file_locks(volume_id):
coordination.synchronized_remove('attachment_update-' + volume_id + '-*')
def clean_snapshot_file_locks(snapshot_id, driver):
name = snapshot_id + '-delete_snapshot'
except Exception as exc:
LOG.warning('Failed to cleanup snapshot lock %(name)s: %(exc)s',
{'name': name, 'exc': exc})
except Exception as exc:
LOG.warning('Failed to cleanup driver locks for snapshot %(id)s: '
'%(exc)s', {'id': snapshot_id, 'exc': exc})
def as_int(obj: Union[int, float, str], quiet: bool = True) -> int:
# Try "2" -> 2
return int(obj)
except (ValueError, TypeError):
# Try "2.5" -> 2
return int(float(obj))
except (ValueError, TypeError):
# Eck, not sure what this is then.
if not quiet:
raise TypeError(_("Can not translate %s to integer.") % (obj))
obj = typing.cast(int, obj)
return obj
def check_exclusive_options(
**kwargs: Optional[Union[dict, str, bool]]) -> None:
"""Checks that only one of the provided options is actually not-none.
Iterates over all the kwargs passed in and checks that only one of said
arguments is not-none, if more than one is not-none then an exception will
be raised with the names of those arguments who were not-none.
if not kwargs:
pretty_keys = kwargs.pop("pretty_keys", True)
exclusive_options = {}
for (k, v) in kwargs.items():
if v is not None:
exclusive_options[k] = True
if len(exclusive_options) > 1:
# Change the format of the names from pythonic to
# something that is more readable.
# Ex: 'the_key' -> 'the key'
if pretty_keys:
tnames = [k.replace('_', ' ') for k in kwargs]
tnames = list(kwargs.keys())
names = ", ".join(sorted(tnames))
msg = (_("May specify only one of %s") % (names))
raise exception.InvalidInput(reason=msg)
def execute(*cmd: str, **kwargs: Union[bool, str]) -> tuple[str, str]:
"""Convenience wrapper around oslo's execute() method."""
if 'run_as_root' in kwargs and 'root_helper' not in kwargs:
kwargs['root_helper'] = get_root_helper()
return processutils.execute(*cmd, **kwargs)
def check_ssh_injection(cmd_list: list[str]) -> None:
ssh_injection_pattern: tuple[str, ...] = ('`', '$', '|', '||', ';', '&',
'&&', '>', '>>', '<')
# Check whether injection attacks exist
for arg in cmd_list:
arg = arg.strip()
# Check for matching quotes on the ends
is_quoted = re.match('^(?P<quote>[\'"])(?P<quoted>.*)(?P=quote)$', arg)
if is_quoted:
# Check for unescaped quotes within the quoted argument
quoted ='quoted')
if quoted:
if (re.match('[\'"]', quoted) or'[^\\\\][\'"]', quoted)):
raise exception.SSHInjectionThreat(command=cmd_list)
# We only allow spaces within quoted arguments, and that
# is the only special character allowed within quotes
if len(arg.split()) > 1:
raise exception.SSHInjectionThreat(command=cmd_list)
# Second, check whether danger character in command. So the shell
# special operator must be a single argument.
for c in ssh_injection_pattern:
if c not in arg:
result = arg.find(c)
if not result == -1:
if result == 0 or not arg[result - 1] == '\\':
raise exception.SSHInjectionThreat(command=cmd_list)
def check_metadata_properties(
metadata: Optional[dict[str, str]]) -> None:
"""Checks that the volume metadata properties are valid."""
if not metadata:
metadata = {}
if not isinstance(metadata, dict):
msg = _("Metadata should be a dict.")
raise exception.InvalidInput(msg)
for k, v in metadata.items():
check_string_length(k, "Metadata key: %s" % k, min_length=1)
check_string_length(v, "Value for metadata key: %s" % k)
except exception.InvalidInput as exc:
raise exception.InvalidVolumeMetadata(reason=exc)
# for backward compatibility
if len(k) > 255:
msg = _("Metadata property key %s greater than 255 "
"characters.") % k
raise exception.InvalidVolumeMetadataSize(reason=msg)
if len(v) > 255:
msg = _("Metadata property key %s value greater than "
"255 characters.") % k
raise exception.InvalidVolumeMetadataSize(reason=msg)
def last_completed_audit_period(unit: Optional[str] = None) -> \
tuple[Union[datetime.datetime, datetime.timedelta],
Union[datetime.datetime, datetime.timedelta]]:
"""This method gives you the most recently *completed* audit period.
units: string, one of 'hour', 'day', 'month', 'year'
Periods normally begin at the beginning (UTC) of the
period unit (So a 'day' period begins at midnight UTC,
a 'month' unit on the 1st, a 'year' on Jan, 1)
unit string may be appended with an optional offset
like so: 'day@18' This will begin the period at 18:00
UTC. 'month@15' starts a monthly period on the 15th,
and year@3 begins a yearly one on March 1st.
returns: 2 tuple of datetimes (begin, end)
The begin timestamp of this audit period is the same as the
end of the previous.
if not unit:
unit = CONF.volume_usage_audit_period
unit = typing.cast(str, unit)
offset: Union[str, int] = 0
if '@' in unit:
unit, offset = unit.split("@", 1)
offset = int(offset)
offset = typing.cast(int, offset)
rightnow = timeutils.utcnow()
if unit not in ('month', 'day', 'year', 'hour'):
raise ValueError('Time period must be hour, day, month or year')
if unit == 'month':
if offset == 0:
offset = 1
end = datetime.datetime(day=offset,
if end >= rightnow:
year = rightnow.year
if 1 >= rightnow.month:
year -= 1
month = 12 + (rightnow.month - 1)
month = rightnow.month - 1
end = datetime.datetime(day=offset,
year = end.year
if 1 >= end.month:
year -= 1
month = 12 + (end.month - 1)
month = end.month - 1
begin = datetime.datetime(day=offset, month=month, year=year)
elif unit == 'year':
if offset == 0:
offset = 1
end = datetime.datetime(day=1, month=offset, year=rightnow.year)
if end >= rightnow:
end = datetime.datetime(day=1,
year=rightnow.year - 1)
begin = datetime.datetime(day=1,
year=rightnow.year - 2)
begin = datetime.datetime(day=1,
year=rightnow.year - 1)
elif unit == 'day':
end = datetime.datetime(hour=offset,,
if end >= rightnow:
end = end - datetime.timedelta(days=1)
begin = end - datetime.timedelta(days=1)
elif unit == 'hour':
end = rightnow.replace(minute=offset, second=0, microsecond=0)
if end >= rightnow:
end = end - datetime.timedelta(hours=1)
begin = end - datetime.timedelta(hours=1)
return (begin, end)
def monkey_patch() -> None:
"""Patches decorators for all functions in a specified module.
If the CONF.monkey_patch set as True,
this function patches a decorator
for all functions in specified modules.
You can set decorators for each modules
using CONF.monkey_patch_modules.
The format is "Module path:Decorator function".
Example: '' \
Parameters of the decorator are as follows.
(See cinder.openstack.common.notifier.api.notify_decorator)
:param name: name of the function
:param function: object of the function
# If CONF.monkey_patch is not True, this function do nothing.
if not CONF.monkey_patch:
# Get list of modules and decorators
for module_and_decorator in CONF.monkey_patch_modules:
module, decorator_name = module_and_decorator.split(':')
# import decorator function
decorator = importutils.import_class(decorator_name)
# Retrieve module information using pyclbr
module_data = pyclbr.readmodule_ex(module)
for key in module_data.keys():
# set the decorator for the class methods
if isinstance(module_data[key], pyclbr.Class):
clz = importutils.import_class("%s.%s" % (module, key))
# On Python 3, unbound methods are regular functions
predicate = inspect.isfunction
for method, func in inspect.getmembers(clz, predicate):
clz, method,
decorator("%s.%s.%s" % (module, key, method), func))
# set the decorator for the function
elif isinstance(module_data[key], pyclbr.Function):
func = importutils.import_class("%s.%s" % (module, key))
setattr(sys.modules[module], key,
decorator("%s.%s" % (module, key), func))
def make_dev_path(dev: str,
partition: Optional[str] = None,
base: str = '/dev') -> str:
"""Return a path to a particular device.
>>> make_dev_path('xvdc')
>>> make_dev_path('xvdc', 1)
path = os.path.join(base, dev)
if partition:
path += str(partition)
return path
def robust_file_write(directory: str, filename: str, data: str) -> None:
"""Robust file write.
Use "write to temp file and rename" model for writing the
persistence file.
:param directory: Target directory to create a file.
:param filename: File name to store specified data.
:param data: String data.
tempname = None
dirfd = None
dirfd =, os.O_DIRECTORY)
# write data to temporary file
with tempfile.NamedTemporaryFile(prefix=filename,
delete=False) as tf:
tempname =
# Fsync the directory to ensure the fact of the existence of
# the temp file hits the disk.
# If destination file exists, it will be replaced silently.
os.rename(tempname, os.path.join(directory, filename))
# Fsync the directory to ensure the rename hits the disk.
except OSError:
with excutils.save_and_reraise_exception():
LOG.error("Failed to write persistence file: %(path)s.",
{'path': os.path.join(directory, filename)})
if tempname is not None:
if os.path.isfile(tempname):
if dirfd is not None:
def temporary_chown(path: str,
owner_uid: Optional[int] = None) -> Iterator[None]:
"""Temporarily chown a path.
:params owner_uid: UID of temporary owner (defaults to current user)
if == 'nt':
LOG.debug("Skipping chown for %s as this operation is "
"not available on Windows.", path)
if owner_uid is None:
owner_uid = os.getuid()
orig_uid = os.stat(path).st_uid
if orig_uid != owner_uid:
execute('chown', str(owner_uid), path, run_as_root=True)
if orig_uid != owner_uid:
execute('chown', str(orig_uid), path, run_as_root=True)
def tempdir(**kwargs) -> Iterator[str]:
tmpdir = tempfile.mkdtemp(**kwargs)
yield tmpdir
except OSError as e:
LOG.debug('Could not remove tmpdir: %s', str(e))
def get_root_helper() -> str:
return 'sudo cinder-rootwrap %s' % CONF.rootwrap_config
def get_file_mode(path: str) -> int:
"""This primarily exists to make unit testing easier."""
return stat.S_IMODE(os.stat(path).st_mode)
def get_file_gid(path: str) -> int:
"""This primarily exists to make unit testing easier."""
return os.stat(path).st_gid
def get_file_size(path: str) -> int:
"""Returns the file size."""
return os.stat(path).st_size
def _get_disk_of_partition(
devpath: str,
st: Optional[os.stat_result] = None) -> tuple[str, os.stat_result]:
"""Gets a disk device path and status from partition path.
Returns a disk device path from a partition device path, and stat for
the device. If devpath is not a partition, devpath is returned as it is.
For example, '/dev/sda' is returned for '/dev/sda1', and '/dev/disk1' is
for '/dev/disk1p1' ('p' is prepended to the partition number if the disk
name ends with numbers).
diskpath = re.sub(r'(?:(?<=\d)p)?\d+$', '', devpath)
if diskpath != devpath:
st_disk = os.stat(diskpath)
if stat.S_ISBLK(st_disk.st_mode):
return (diskpath, st_disk)
except OSError:
# devpath is not a partition
if st is None:
st = os.stat(devpath)
return (devpath, st)
def get_bool_param(param_string: str,
params: dict,
default: bool = False) -> bool:
param = params.get(param_string, default)
if not strutils.is_valid_boolstr(param):
msg = _("Value '%(param)s' for '%(param_string)s' is not "
"a boolean.") % {'param': param, 'param_string': param_string}
raise exception.InvalidParameterValue(err=msg)
return strutils.bool_from_string(param, strict=True)
def get_blkdev_major_minor(path: str,
lookup_for_file: bool = True) -> Optional[str]:
"""Get 'major:minor' number of block device.
Get the device's 'major:minor' number of a block device to control
I/O ratelimit of the specified path.
If lookup_for_file is True and the path is a regular file, lookup a disk
device which the file lies on and returns the result for the device.
st = os.stat(path)
if stat.S_ISBLK(st.st_mode):
path, st = _get_disk_of_partition(path, st)
return '%d:%d' % (os.major(st.st_rdev), os.minor(st.st_rdev))
elif stat.S_ISCHR(st.st_mode):
# No I/O ratelimit control is provided for character devices
return None
elif lookup_for_file:
# lookup the mounted disk which the file lies on
out, _err = execute('df', path)
devpath = out.split("\n")[1].split()[0]
if devpath[0] != '/':
# the file is on a network file system
return None
return get_blkdev_major_minor(devpath, False)
msg = _("Unable to get a block device for file \'%s\'") % path
raise exception.CinderException(msg)
def check_string_length(value: str, name: str, min_length: int = 0,
max_length: Optional[int] = None,
allow_all_spaces: bool = True) -> None:
"""Check the length of specified string.
:param value: the value of the string
:param name: the name of the string
:param min_length: the min_length of the string
:param max_length: the max_length of the string
strutils.check_string_length(value, name=name,
except(ValueError, TypeError) as exc:
raise exception.InvalidInput(reason=exc)
if not allow_all_spaces and value.isspace():
msg = _('%(name)s cannot be all spaces.')
raise exception.InvalidInput(reason=msg)
def is_blk_device(dev: str) -> bool:
if stat.S_ISBLK(os.stat(dev).st_mode):
return True
return False
except Exception:
LOG.debug('Path %s not found in is_blk_device check', dev)
return False
class ComparableMixin(object):
def _compare(self, other: object, method: Callable):
return method(self._cmpkey(), other._cmpkey()) # type: ignore
except (AttributeError, TypeError):
# _cmpkey not implemented, or return different type,
# so I can't compare with "other".
return NotImplemented
def __lt__(self, other: object):
return self._compare(other, lambda s, o: s < o)
def __le__(self, other: object):
return self._compare(other, lambda s, o: s <= o)
def __eq__(self, other: object):
return self._compare(other, lambda s, o: s == o)
def __ge__(self, other: object):
return self._compare(other, lambda s, o: s >= o)
def __gt__(self, other: object):
return self._compare(other, lambda s, o: s > o)
def __ne__(self, other: object):
return self._compare(other, lambda s, o: s != o)
class retry_if_exit_code(tenacity.retry_if_exception):
"""Retry on ProcessExecutionError specific exit codes."""
def __init__(self, codes): = (codes,) if isinstance(codes, int) else codes
super(retry_if_exit_code, self).__init__(self._check_exit_code)
def _check_exit_code(self, exc):
return (exc and isinstance(exc, processutils.ProcessExecutionError) and
exc.exit_code in
def retry(retry_param: Union[None,
tuple[Type[Exception], ...],
tuple[int, ...]],
interval: float = 1,
retries: int = 3,
backoff_rate: float = 2,
wait_random: bool = False,
retry=tenacity.retry_if_exception_type) -> Callable:
if retries < 1:
raise ValueError('Retries must be greater than or '
'equal to 1 (received: %s). ' % retries)
if wait_random:
wait = tenacity.wait_random_exponential(multiplier=interval)
wait = tenacity.wait_exponential(
multiplier=interval, min=0, exp_base=backoff_rate)
def _decorator(f: Callable) -> Callable:
def _wrapper(*args, **kwargs):
r = tenacity.Retrying(
before_sleep=tenacity.before_sleep_log(LOG, logging.DEBUG),
after=tenacity.after_log(LOG, logging.DEBUG),
return r(f, *args, **kwargs)
return _wrapper
return _decorator
def convert_str(text: Union[str, bytes]) -> str:
"""Convert to native string.
Convert bytes and Unicode strings to native strings:
* convert to bytes on Python 2:
encode Unicode using encodeutils.safe_encode()
* convert to Unicode on Python 3: decode bytes from UTF-8
if isinstance(text, bytes):
return text.decode('utf-8')
return text
def build_or_str(elements: Union[None, str, Iterable[str]],
str_format: Optional[str] = None) -> str:
"""Builds a string of elements joined by 'or'.
Will join strings with the 'or' word and if a str_format is provided it
will be used to format the resulted joined string.
If there are no elements an empty string will be returned.
:param elements: Elements we want to join.
:type elements: String or iterable of strings.
:param str_format: String to use to format the response.
:type str_format: String.
if not elements:
return ''
if not isinstance(elements, str):
elements = _(' or ').join(elements)
elements = typing.cast(str, elements)
if str_format:
return str_format % elements
return elements
def calculate_capacity_factors(total_capacity: float,
free_capacity: float,
provisioned_capacity: float,
thin_provisioning_support: bool,
max_over_subscription_ratio: float,
reserved_percentage: int,
thin: bool) -> dict:
"""Create the various capacity factors of the a particular backend.
Based off of definition of terms
Description of factors calculated where units of gb are Gibibytes.
reserved_capacity - The amount of space reserved from the total_capacity
as reported by the backend.
total_reserved_available_capacity - The total capacity minus reserved
total_available_capacity - The total capacity available to cinder
calculated from total_reserved_available_capacity (for thick) OR
for thin total_reserved_available_capacity max_over_subscription_ratio
calculated_free_capacity - total_available_capacity - provisioned_capacity
virtual_free_capacity - The calculated free capacity available to cinder
to allocate new storage.
For thin: calculated_free_capacity
For thick: the reported free_capacity can be less than the calculated
capacity, so we use free_capacity - reserved_capacity.
free_percent - the percentage of the virtual_free_capacity and
total_available_capacity is left over
provisioned_ratio - The ratio of provisioned storage to
:param total_capacity: The reported total capacity in the backend.
:type total_capacity: float
:param free_capacity: The free space/capacity as reported by the backend.
:type free_capacity: float
:param provisioned_capacity: as reported by backend or volume manager from
:type provisioned_capacity: float
:param thin_provisioning_support: Is thin provisioning supported?
:type thin_provisioning_support: bool
:param max_over_subscription_ratio: as reported by the backend
:type max_over_subscription_ratio: float
:param reserved_percentage: the % amount to reserve as unavailable. 0-100
:type reserved_percentage: int, 0-100
:param thin: calculate based on thin provisioning if enabled by
:type thin: bool
:return: A dictionary of all of the capacity factors.
:rtype: dict
total = float(total_capacity)
reserved = float(reserved_percentage) / 100
reserved_capacity = math.floor(total * reserved)
total_reserved_available = total - reserved_capacity
if thin and thin_provisioning_support:
total_available_capacity = (
total_reserved_available * max_over_subscription_ratio
calculated_free = total_available_capacity - provisioned_capacity
virtual_free = calculated_free
provisioned_type = 'thin'
# Calculate how much free space is left after taking into
# account the reserved space.
total_available_capacity = total_reserved_available
calculated_free = total_available_capacity - provisioned_capacity
virtual_free = calculated_free
if free_capacity < calculated_free:
virtual_free = free_capacity
provisioned_type = 'thick'
if total_available_capacity:
provisioned_ratio = provisioned_capacity / total_available_capacity
free_percent = (virtual_free / total_available_capacity) * 100
provisioned_ratio = 0
free_percent = 0
def _limit(x):
"""Limit our floating points to 2 decimal places."""
return round(x, 2)
return {
"total_capacity": total,
"free_capacity": free_capacity,
"reserved_capacity": reserved_capacity,
"total_reserved_available_capacity": _limit(total_reserved_available),
"max_over_subscription_ratio": (
max_over_subscription_ratio if provisioned_type == 'thin' else None
"total_available_capacity": _limit(total_available_capacity),
"provisioned_capacity": provisioned_capacity,
"calculated_free_capacity": _limit(calculated_free),
"virtual_free_capacity": _limit(virtual_free),
"free_percent": _limit(free_percent),
"provisioned_ratio": _limit(provisioned_ratio),
"provisioned_type": provisioned_type
def calculate_virtual_free_capacity(total_capacity: float,
free_capacity: float,
provisioned_capacity: float,
thin_provisioning_support: bool,
max_over_subscription_ratio: float,
reserved_percentage: int,
thin: bool) -> float:
"""Calculate the virtual free capacity based on multiple factors.
:param total_capacity: total_capacity_gb of a host_state or pool.
:param free_capacity: free_capacity_gb of a host_state or pool.
:param provisioned_capacity: provisioned_capacity_gb of a host_state
or pool.
:param thin_provisioning_support: thin_provisioning_support of
a host_state or a pool.
:param max_over_subscription_ratio: max_over_subscription_ratio of
a host_state or a pool
:param reserved_percentage: reserved_percentage of a host_state or
a pool.
:param thin: whether volume to be provisioned is thin
:returns: the calculated virtual free capacity.
factors = calculate_capacity_factors(
return factors["virtual_free_capacity"]
def calculate_max_over_subscription_ratio(
capability: dict,
global_max_over_subscription_ratio: float) -> float:
# provisioned_capacity_gb is the apparent total capacity of
# all the volumes created on a backend, which is greater than
# or equal to allocated_capacity_gb, which is the apparent
# total capacity of all the volumes created on a backend
# in Cinder. Using allocated_capacity_gb as the default of
# provisioned_capacity_gb if it is not set.
allocated_capacity_gb = capability.get('allocated_capacity_gb', 0)
provisioned_capacity_gb = capability.get('provisioned_capacity_gb',
thin_provisioning_support = capability.get('thin_provisioning_support',
total_capacity_gb = capability.get('total_capacity_gb', 0)
free_capacity_gb = capability.get('free_capacity_gb', 0)
pool_name = capability.get('pool_name',
# If thin provisioning is not supported the capacity filter will not use
# the value we return, no matter what it is.
if not thin_provisioning_support:
LOG.debug("Trying to retrieve max_over_subscription_ratio from a "
"service that does not support thin provisioning")
return 1.0
# Again, if total or free capacity is infinite or unknown, the capacity
# filter will not use the max_over_subscription_ratio at all. So, does
# not matter what we return here.
if ((total_capacity_gb in INFINITE_UNKNOWN_VALUES) or
(free_capacity_gb in INFINITE_UNKNOWN_VALUES)):
return 1.0
max_over_subscription_ratio = (capability.get(
'max_over_subscription_ratio') or global_max_over_subscription_ratio)
# We only calculate the automatic max_over_subscription_ratio (mosr)
# when the global or driver conf is set auto and while
# provisioned_capacity_gb is not 0. When auto is set and
# provisioned_capacity_gb is 0, we use the default value 20.0.
if max_over_subscription_ratio == 'auto':
if provisioned_capacity_gb != 0:
used_capacity = total_capacity_gb - free_capacity_gb
LOG.debug("Calculating max_over_subscription_ratio for "
"pool %s: provisioned_capacity_gb=%s, "
pool_name, provisioned_capacity_gb, used_capacity)
max_over_subscription_ratio = 1 + (
float(provisioned_capacity_gb) / (used_capacity + 1))
max_over_subscription_ratio = INITIAL_AUTO_MOSR"Auto max_over_subscription_ratio for pool %s is "
"%s", pool_name, max_over_subscription_ratio)
max_over_subscription_ratio = float(max_over_subscription_ratio)
return max_over_subscription_ratio
def validate_dictionary_string_length(specs: dict) -> None:
"""Check the length of each key and value of dictionary."""
if not isinstance(specs, dict):
msg = _('specs must be a dictionary.')
raise exception.InvalidInput(reason=msg)
for key, value in specs.items():
if key is not None:
check_string_length(key, 'Key "%s"' % key,
min_length=1, max_length=255)
if value is not None:
check_string_length(value, 'Value for key "%s"' % key,
min_length=0, max_length=255)
def service_expired_time(
with_timezone: Optional[bool] = False) -> datetime.datetime:
return (timeutils.utcnow(with_timezone=with_timezone) -
class DoNothing(str):
"""Class that literrally does nothing.
We inherit from str in case it's called with json.dumps.
def __call__(self, *args, **kwargs):
return self
def __getattr__(self, name):
return self
DO_NOTHING = DoNothing()
def notifications_enabled(conf):
"""Check if oslo notifications are enabled."""
notifications_driver = set(conf.oslo_messaging_notifications.driver)
return notifications_driver and notifications_driver != {'noop'}
def if_notifications_enabled(f: Callable) -> Callable:
"""Calls decorated method only if notifications are enabled."""
def wrapped(*args, **kwargs):
if notifications_enabled(CONF):
return f(*args, **kwargs)
return wrapped
def get_log_method(level_string: str) -> int:
level_string = level_string or ''
upper_level_string = level_string.upper()
if upper_level_string not in LOG_LEVELS:
raise exception.InvalidInput(
reason=_('%s is not a valid log level.') % level_string)
return getattr(logging, upper_level_string)
def set_log_levels(prefix: str, level_string: str) -> None:
level = get_log_method(level_string)
prefix = prefix or ''
for k, v in logging.get_loggers().items():
if k and k.startswith(prefix):
def get_log_levels(prefix: str) -> dict:
prefix = prefix or ''
return {k: py_logging.getLevelName(v.logger.getEffectiveLevel())
for k, v in logging.get_loggers().items()
if k and k.startswith(prefix)}
def paths_normcase_equal(path_a: str, path_b: str) -> bool:
return os.path.normcase(path_a) == os.path.normcase(path_b)
def create_ordereddict(adict: dict) -> OrderedDict:
"""Given a dict, return a sorted OrderedDict."""
return OrderedDict(sorted(adict.items(),
class Semaphore(object):
"""Custom semaphore to workaround eventlet issues with multiprocessing."""
def __init__(self, limit):
self.limit = limit
self.semaphore = multiprocessing.Semaphore(limit)
def __enter__(self):
# Eventlet does not work with multiprocessing's Semaphore, so we have
# to execute it in a native thread to avoid getting blocked when trying
# to acquire the semaphore.
return tpool.execute(self.semaphore.__enter__)
def __exit__(self, *args):
# Don't use native thread for exit, as it will only add overhead
return self.semaphore.__exit__(*args)
def semaphore_factory(limit: int,
concurrent_processes: int) -> Union[eventlet.Semaphore,
"""Get a semaphore to limit concurrent operations.
The semaphore depends on the limit we want to set and the concurrent
processes that need to be limited.
# Limit of 0 is no limit, so we won't use a semaphore
if limit:
# If we only have 1 process we can use eventlet's Semaphore
if concurrent_processes == 1:
return eventlet.Semaphore(limit)
# Use our own Sempahore for interprocess because eventlet blocks with
# the standard one
return Semaphore(limit)
return contextlib.suppress()
def limit_operations(func: Callable) -> Callable:
"""Decorator to limit the number of concurrent operations.
This method decorator expects to have a _semaphore attribute holding an
initialized semaphore in the self instance object.
We can get the appropriate semaphore with the semaphore_factory method.
def wrapper(self, *args, **kwargs):
with self._semaphore:
return func(self, *args, **kwargs)
return wrapper