cinder.utils module

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, etc.

Code related to volume drivers and connecting to volumes should be placed in volume_utils instead.

class ComparableMixin

Bases: object


Class that literrally does nothing.

We inherit from str in case it’s called with json.dumps.

class DoNothing

Bases: str

Class that literrally does nothing.

We inherit from str in case it’s called with json.dumps.

class Semaphore(limit)

Bases: object

Custom semaphore to workaround eventlet issues with multiprocessing.

as_int(obj: Union[int, float, str], quiet: bool = True)int
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.

  • elements (String or iterable of strings.) – Elements we want to join.

  • str_format (String.) – String to use to format the response.

calculate_max_over_subscription_ratio(capability: dict, global_max_over_subscription_ratio: float)float
calculate_virtual_free_capacity(total_capacity: float, free_capacity: float, provisioned_capacity: float, thin_provisioning_support: bool, max_over_subscription_ratio: float, reserved_percentage: float, thin: bool)float

Calculate the virtual free capacity based on thin provisioning support.

  • total_capacity – total_capacity_gb of a host_state or pool.

  • free_capacity – free_capacity_gb of a host_state or pool.

  • provisioned_capacity – provisioned_capacity_gb of a host_state or pool.

  • thin_provisioning_support – thin_provisioning_support of a host_state or a pool.

  • max_over_subscription_ratio – max_over_subscription_ratio of a host_state or a pool

  • reserved_percentage – reserved_percentage of a host_state or a pool.

  • thin – whether volume to be provisioned is thin


the calculated virtual free capacity.

check_exclusive_options(**kwargs: dict)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.

check_metadata_properties(metadata: Optional[Dict[str, str]])None

Checks that the volume metadata properties are valid.

check_ssh_injection(cmd_list: List[str])None
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.

  • value – the value of the string

  • name – the name of the string

  • min_length – the min_length of the string

  • max_length – the max_length of the string

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

create_ordereddict(adict: dict)collections.OrderedDict

Given a dict, return a sorted OrderedDict.

execute(*cmd: str, **kwargs: Union[bool, str])Tuple[str, str]

Convenience wrapper around oslo’s execute() method.

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.

get_bool_param(param_string: str, params: dict, default: bool = False)bool
get_file_gid(path: str)int

This primarily exists to make unit testing easier.

get_file_mode(path: str)int

This primarily exists to make unit testing easier.

get_file_size(path: str)int

Returns the file size.

get_log_levels(prefix: str)dict
get_log_method(level_string: str)int
if_notifications_enabled(f: Callable)Callable

Calls decorated method only if notifications are enabled.

is_blk_device(dev: str)bool
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.

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.

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)

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: ‘’ cinder.openstack.common.notifier.api.notify_decorator’

Parameters of the decorator are as follows. (See cinder.openstack.common.notifier.api.notify_decorator)

  • name – name of the function

  • function – object of the function


Check if oslo notifications are enabled.

paths_normcase_equal(path_a: str, path_b: str)bool
retry(retry_param: Optional[Type[Exception]], interval: int = 1, retries: int = 3, backoff_rate: int = 2, wait_random: bool = False, retry=<class 'tenacity.retry.retry_if_exception_type'>)Callable
class retry_if_exit_code(codes)

Bases: tenacity.retry.retry_if_exception

Retry on ProcessExecutionError specific exit codes.

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.

  • directory – Target directory to create a file.

  • filename – File name to store specified data.

  • data – String data.

semaphore_factory(limit: int, concurrent_processes: int)Union[eventlet.semaphore.Semaphore, cinder.utils.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.

service_expired_time(with_timezone: Optional[bool] = False)datetime.datetime
set_log_levels(prefix: str, level_string: str)None
temporary_chown(path: str, owner_uid: int = None)Iterator[None]

Temporarily chown a path.

Params owner_uid

UID of temporary owner (defaults to current user)

validate_dictionary_string_length(specs: dict)None

Check the length of each key and value of dictionary.