diff options
Diffstat (limited to 'charms/trusty/ceilometer/charmhelpers/contrib/storage/linux/ceph.py')
| -rw-r--r-- | charms/trusty/ceilometer/charmhelpers/contrib/storage/linux/ceph.py | 1206 |
1 files changed, 0 insertions, 1206 deletions
diff --git a/charms/trusty/ceilometer/charmhelpers/contrib/storage/linux/ceph.py b/charms/trusty/ceilometer/charmhelpers/contrib/storage/linux/ceph.py deleted file mode 100644 index d008081..0000000 --- a/charms/trusty/ceilometer/charmhelpers/contrib/storage/linux/ceph.py +++ /dev/null @@ -1,1206 +0,0 @@ -# Copyright 2014-2015 Canonical Limited. -# -# This file is part of charm-helpers. -# -# charm-helpers is free software: you can redistribute it and/or modify -# it under the terms of the GNU Lesser General Public License version 3 as -# published by the Free Software Foundation. -# -# charm-helpers is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public License -# along with charm-helpers. If not, see <http://www.gnu.org/licenses/>. - -# -# Copyright 2012 Canonical Ltd. -# -# This file is sourced from lp:openstack-charm-helpers -# -# Authors: -# James Page <james.page@ubuntu.com> -# Adam Gandelman <adamg@ubuntu.com> -# -import bisect -import errno -import hashlib -import six - -import os -import shutil -import json -import time -import uuid - -from subprocess import ( - check_call, - check_output, - CalledProcessError, -) -from charmhelpers.core.hookenv import ( - local_unit, - relation_get, - relation_ids, - relation_set, - related_units, - log, - DEBUG, - INFO, - WARNING, - ERROR, -) -from charmhelpers.core.host import ( - mount, - mounts, - service_start, - service_stop, - service_running, - umount, -) -from charmhelpers.fetch import ( - apt_install, -) - -from charmhelpers.core.kernel import modprobe - -KEYRING = '/etc/ceph/ceph.client.{}.keyring' -KEYFILE = '/etc/ceph/ceph.client.{}.key' - -CEPH_CONF = """[global] -auth supported = {auth} -keyring = {keyring} -mon host = {mon_hosts} -log to syslog = {use_syslog} -err to syslog = {use_syslog} -clog to syslog = {use_syslog} -""" -# For 50 < osds < 240,000 OSDs (Roughly 1 Exabyte at 6T OSDs) -powers_of_two = [8192, 16384, 32768, 65536, 131072, 262144, 524288, 1048576, 2097152, 4194304, 8388608] - - -def validator(value, valid_type, valid_range=None): - """ - Used to validate these: http://docs.ceph.com/docs/master/rados/operations/pools/#set-pool-values - Example input: - validator(value=1, - valid_type=int, - valid_range=[0, 2]) - This says I'm testing value=1. It must be an int inclusive in [0,2] - - :param value: The value to validate - :param valid_type: The type that value should be. - :param valid_range: A range of values that value can assume. - :return: - """ - assert isinstance(value, valid_type), "{} is not a {}".format( - value, - valid_type) - if valid_range is not None: - assert isinstance(valid_range, list), \ - "valid_range must be a list, was given {}".format(valid_range) - # If we're dealing with strings - if valid_type is six.string_types: - assert value in valid_range, \ - "{} is not in the list {}".format(value, valid_range) - # Integer, float should have a min and max - else: - if len(valid_range) != 2: - raise ValueError( - "Invalid valid_range list of {} for {}. " - "List must be [min,max]".format(valid_range, value)) - assert value >= valid_range[0], \ - "{} is less than minimum allowed value of {}".format( - value, valid_range[0]) - assert value <= valid_range[1], \ - "{} is greater than maximum allowed value of {}".format( - value, valid_range[1]) - - -class PoolCreationError(Exception): - """ - A custom error to inform the caller that a pool creation failed. Provides an error message - """ - - def __init__(self, message): - super(PoolCreationError, self).__init__(message) - - -class Pool(object): - """ - An object oriented approach to Ceph pool creation. This base class is inherited by ReplicatedPool and ErasurePool. - Do not call create() on this base class as it will not do anything. Instantiate a child class and call create(). - """ - - def __init__(self, service, name): - self.service = service - self.name = name - - # Create the pool if it doesn't exist already - # To be implemented by subclasses - def create(self): - pass - - def add_cache_tier(self, cache_pool, mode): - """ - Adds a new cache tier to an existing pool. - :param cache_pool: six.string_types. The cache tier pool name to add. - :param mode: six.string_types. The caching mode to use for this pool. valid range = ["readonly", "writeback"] - :return: None - """ - # Check the input types and values - validator(value=cache_pool, valid_type=six.string_types) - validator(value=mode, valid_type=six.string_types, valid_range=["readonly", "writeback"]) - - check_call(['ceph', '--id', self.service, 'osd', 'tier', 'add', self.name, cache_pool]) - check_call(['ceph', '--id', self.service, 'osd', 'tier', 'cache-mode', cache_pool, mode]) - check_call(['ceph', '--id', self.service, 'osd', 'tier', 'set-overlay', self.name, cache_pool]) - check_call(['ceph', '--id', self.service, 'osd', 'pool', 'set', cache_pool, 'hit_set_type', 'bloom']) - - def remove_cache_tier(self, cache_pool): - """ - Removes a cache tier from Ceph. Flushes all dirty objects from writeback pools and waits for that to complete. - :param cache_pool: six.string_types. The cache tier pool name to remove. - :return: None - """ - # read-only is easy, writeback is much harder - mode = get_cache_mode(self.service, cache_pool) - version = ceph_version() - if mode == 'readonly': - check_call(['ceph', '--id', self.service, 'osd', 'tier', 'cache-mode', cache_pool, 'none']) - check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove', self.name, cache_pool]) - - elif mode == 'writeback': - pool_forward_cmd = ['ceph', '--id', self.service, 'osd', 'tier', - 'cache-mode', cache_pool, 'forward'] - if version >= '10.1': - # Jewel added a mandatory flag - pool_forward_cmd.append('--yes-i-really-mean-it') - - check_call(pool_forward_cmd) - # Flush the cache and wait for it to return - check_call(['rados', '--id', self.service, '-p', cache_pool, 'cache-flush-evict-all']) - check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove-overlay', self.name]) - check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove', self.name, cache_pool]) - - def get_pgs(self, pool_size): - """ - :param pool_size: int. pool_size is either the number of replicas for replicated pools or the K+M sum for - erasure coded pools - :return: int. The number of pgs to use. - """ - validator(value=pool_size, valid_type=int) - osd_list = get_osds(self.service) - if not osd_list: - # NOTE(james-page): Default to 200 for older ceph versions - # which don't support OSD query from cli - return 200 - - osd_list_length = len(osd_list) - # Calculate based on Ceph best practices - if osd_list_length < 5: - return 128 - elif 5 < osd_list_length < 10: - return 512 - elif 10 < osd_list_length < 50: - return 4096 - else: - estimate = (osd_list_length * 100) / pool_size - # Return the next nearest power of 2 - index = bisect.bisect_right(powers_of_two, estimate) - return powers_of_two[index] - - -class ReplicatedPool(Pool): - def __init__(self, service, name, pg_num=None, replicas=2): - super(ReplicatedPool, self).__init__(service=service, name=name) - self.replicas = replicas - if pg_num is None: - self.pg_num = self.get_pgs(self.replicas) - else: - self.pg_num = pg_num - - def create(self): - if not pool_exists(self.service, self.name): - # Create it - cmd = ['ceph', '--id', self.service, 'osd', 'pool', 'create', - self.name, str(self.pg_num)] - try: - check_call(cmd) - # Set the pool replica size - update_pool(client=self.service, - pool=self.name, - settings={'size': str(self.replicas)}) - except CalledProcessError: - raise - - -# Default jerasure erasure coded pool -class ErasurePool(Pool): - def __init__(self, service, name, erasure_code_profile="default"): - super(ErasurePool, self).__init__(service=service, name=name) - self.erasure_code_profile = erasure_code_profile - - def create(self): - if not pool_exists(self.service, self.name): - # Try to find the erasure profile information so we can properly size the pgs - erasure_profile = get_erasure_profile(service=self.service, name=self.erasure_code_profile) - - # Check for errors - if erasure_profile is None: - log(message='Failed to discover erasure_profile named={}'.format(self.erasure_code_profile), - level=ERROR) - raise PoolCreationError(message='unable to find erasure profile {}'.format(self.erasure_code_profile)) - if 'k' not in erasure_profile or 'm' not in erasure_profile: - # Error - log(message='Unable to find k (data chunks) or m (coding chunks) in {}'.format(erasure_profile), - level=ERROR) - raise PoolCreationError( - message='unable to find k (data chunks) or m (coding chunks) in {}'.format(erasure_profile)) - - pgs = self.get_pgs(int(erasure_profile['k']) + int(erasure_profile['m'])) - # Create it - cmd = ['ceph', '--id', self.service, 'osd', 'pool', 'create', self.name, str(pgs), str(pgs), - 'erasure', self.erasure_code_profile] - try: - check_call(cmd) - except CalledProcessError: - raise - - """Get an existing erasure code profile if it already exists. - Returns json formatted output""" - - -def get_mon_map(service): - """ - Returns the current monitor map. - :param service: six.string_types. The Ceph user name to run the command under - :return: json string. :raise: ValueError if the monmap fails to parse. - Also raises CalledProcessError if our ceph command fails - """ - try: - mon_status = check_output( - ['ceph', '--id', service, - 'mon_status', '--format=json']) - try: - return json.loads(mon_status) - except ValueError as v: - log("Unable to parse mon_status json: {}. Error: {}".format( - mon_status, v.message)) - raise - except CalledProcessError as e: - log("mon_status command failed with message: {}".format( - e.message)) - raise - - -def hash_monitor_names(service): - """ - Uses the get_mon_map() function to get information about the monitor - cluster. - Hash the name of each monitor. Return a sorted list of monitor hashes - in an ascending order. - :param service: six.string_types. The Ceph user name to run the command under - :rtype : dict. json dict of monitor name, ip address and rank - example: { - 'name': 'ip-172-31-13-165', - 'rank': 0, - 'addr': '172.31.13.165:6789/0'} - """ - try: - hash_list = [] - monitor_list = get_mon_map(service=service) - if monitor_list['monmap']['mons']: - for mon in monitor_list['monmap']['mons']: - hash_list.append( - hashlib.sha224(mon['name'].encode('utf-8')).hexdigest()) - return sorted(hash_list) - else: - return None - except (ValueError, CalledProcessError): - raise - - -def monitor_key_delete(service, key): - """ - Delete a key and value pair from the monitor cluster - :param service: six.string_types. The Ceph user name to run the command under - Deletes a key value pair on the monitor cluster. - :param key: six.string_types. The key to delete. - """ - try: - check_output( - ['ceph', '--id', service, - 'config-key', 'del', str(key)]) - except CalledProcessError as e: - log("Monitor config-key put failed with message: {}".format( - e.output)) - raise - - -def monitor_key_set(service, key, value): - """ - Sets a key value pair on the monitor cluster. - :param service: six.string_types. The Ceph user name to run the command under - :param key: six.string_types. The key to set. - :param value: The value to set. This will be converted to a string - before setting - """ - try: - check_output( - ['ceph', '--id', service, - 'config-key', 'put', str(key), str(value)]) - except CalledProcessError as e: - log("Monitor config-key put failed with message: {}".format( - e.output)) - raise - - -def monitor_key_get(service, key): - """ - Gets the value of an existing key in the monitor cluster. - :param service: six.string_types. The Ceph user name to run the command under - :param key: six.string_types. The key to search for. - :return: Returns the value of that key or None if not found. - """ - try: - output = check_output( - ['ceph', '--id', service, - 'config-key', 'get', str(key)]) - return output - except CalledProcessError as e: - log("Monitor config-key get failed with message: {}".format( - e.output)) - return None - - -def monitor_key_exists(service, key): - """ - Searches for the existence of a key in the monitor cluster. - :param service: six.string_types. The Ceph user name to run the command under - :param key: six.string_types. The key to search for - :return: Returns True if the key exists, False if not and raises an - exception if an unknown error occurs. :raise: CalledProcessError if - an unknown error occurs - """ - try: - check_call( - ['ceph', '--id', service, - 'config-key', 'exists', str(key)]) - # I can return true here regardless because Ceph returns - # ENOENT if the key wasn't found - return True - except CalledProcessError as e: - if e.returncode == errno.ENOENT: - return False - else: - log("Unknown error from ceph config-get exists: {} {}".format( - e.returncode, e.output)) - raise - - -def get_erasure_profile(service, name): - """ - :param service: six.string_types. The Ceph user name to run the command under - :param name: - :return: - """ - try: - out = check_output(['ceph', '--id', service, - 'osd', 'erasure-code-profile', 'get', - name, '--format=json']) - return json.loads(out) - except (CalledProcessError, OSError, ValueError): - return None - - -def pool_set(service, pool_name, key, value): - """ - Sets a value for a RADOS pool in ceph. - :param service: six.string_types. The Ceph user name to run the command under - :param pool_name: six.string_types - :param key: six.string_types - :param value: - :return: None. Can raise CalledProcessError - """ - cmd = ['ceph', '--id', service, 'osd', 'pool', 'set', pool_name, key, value] - try: - check_call(cmd) - except CalledProcessError: - raise - - -def snapshot_pool(service, pool_name, snapshot_name): - """ - Snapshots a RADOS pool in ceph. - :param service: six.string_types. The Ceph user name to run the command under - :param pool_name: six.string_types - :param snapshot_name: six.string_types - :return: None. Can raise CalledProcessError - """ - cmd = ['ceph', '--id', service, 'osd', 'pool', 'mksnap', pool_name, snapshot_name] - try: - check_call(cmd) - except CalledProcessError: - raise - - -def remove_pool_snapshot(service, pool_name, snapshot_name): - """ - Remove a snapshot from a RADOS pool in ceph. - :param service: six.string_types. The Ceph user name to run the command under - :param pool_name: six.string_types - :param snapshot_name: six.string_types - :return: None. Can raise CalledProcessError - """ - cmd = ['ceph', '--id', service, 'osd', 'pool', 'rmsnap', pool_name, snapshot_name] - try: - check_call(cmd) - except CalledProcessError: - raise - - -# max_bytes should be an int or long -def set_pool_quota(service, pool_name, max_bytes): - """ - :param service: six.string_types. The Ceph user name to run the command under - :param pool_name: six.string_types - :param max_bytes: int or long - :return: None. Can raise CalledProcessError - """ - # Set a byte quota on a RADOS pool in ceph. - cmd = ['ceph', '--id', service, 'osd', 'pool', 'set-quota', pool_name, - 'max_bytes', str(max_bytes)] - try: - check_call(cmd) - except CalledProcessError: - raise - - -def remove_pool_quota(service, pool_name): - """ - Set a byte quota on a RADOS pool in ceph. - :param service: six.string_types. The Ceph user name to run the command under - :param pool_name: six.string_types - :return: None. Can raise CalledProcessError - """ - cmd = ['ceph', '--id', service, 'osd', 'pool', 'set-quota', pool_name, 'max_bytes', '0'] - try: - check_call(cmd) - except CalledProcessError: - raise - - -def remove_erasure_profile(service, profile_name): - """ - Create a new erasure code profile if one does not already exist for it. Updates - the profile if it exists. Please see http://docs.ceph.com/docs/master/rados/operations/erasure-code-profile/ - for more details - :param service: six.string_types. The Ceph user name to run the command under - :param profile_name: six.string_types - :return: None. Can raise CalledProcessError - """ - cmd = ['ceph', '--id', service, 'osd', 'erasure-code-profile', 'rm', - profile_name] - try: - check_call(cmd) - except CalledProcessError: - raise - - -def create_erasure_profile(service, profile_name, erasure_plugin_name='jerasure', - failure_domain='host', - data_chunks=2, coding_chunks=1, - locality=None, durability_estimator=None): - """ - Create a new erasure code profile if one does not already exist for it. Updates - the profile if it exists. Please see http://docs.ceph.com/docs/master/rados/operations/erasure-code-profile/ - for more details - :param service: six.string_types. The Ceph user name to run the command under - :param profile_name: six.string_types - :param erasure_plugin_name: six.string_types - :param failure_domain: six.string_types. One of ['chassis', 'datacenter', 'host', 'osd', 'pdu', 'pod', 'rack', 'region', - 'room', 'root', 'row']) - :param data_chunks: int - :param coding_chunks: int - :param locality: int - :param durability_estimator: int - :return: None. Can raise CalledProcessError - """ - # Ensure this failure_domain is allowed by Ceph - validator(failure_domain, six.string_types, - ['chassis', 'datacenter', 'host', 'osd', 'pdu', 'pod', 'rack', 'region', 'room', 'root', 'row']) - - cmd = ['ceph', '--id', service, 'osd', 'erasure-code-profile', 'set', profile_name, - 'plugin=' + erasure_plugin_name, 'k=' + str(data_chunks), 'm=' + str(coding_chunks), - 'ruleset_failure_domain=' + failure_domain] - if locality is not None and durability_estimator is not None: - raise ValueError("create_erasure_profile should be called with k, m and one of l or c but not both.") - - # Add plugin specific information - if locality is not None: - # For local erasure codes - cmd.append('l=' + str(locality)) - if durability_estimator is not None: - # For Shec erasure codes - cmd.append('c=' + str(durability_estimator)) - - if erasure_profile_exists(service, profile_name): - cmd.append('--force') - - try: - check_call(cmd) - except CalledProcessError: - raise - - -def rename_pool(service, old_name, new_name): - """ - Rename a Ceph pool from old_name to new_name - :param service: six.string_types. The Ceph user name to run the command under - :param old_name: six.string_types - :param new_name: six.string_types - :return: None - """ - validator(value=old_name, valid_type=six.string_types) - validator(value=new_name, valid_type=six.string_types) - - cmd = ['ceph', '--id', service, 'osd', 'pool', 'rename', old_name, new_name] - check_call(cmd) - - -def erasure_profile_exists(service, name): - """ - Check to see if an Erasure code profile already exists. - :param service: six.string_types. The Ceph user name to run the command under - :param name: six.string_types - :return: int or None - """ - validator(value=name, valid_type=six.string_types) - try: - check_call(['ceph', '--id', service, - 'osd', 'erasure-code-profile', 'get', - name]) - return True - except CalledProcessError: - return False - - -def get_cache_mode(service, pool_name): - """ - Find the current caching mode of the pool_name given. - :param service: six.string_types. The Ceph user name to run the command under - :param pool_name: six.string_types - :return: int or None - """ - validator(value=service, valid_type=six.string_types) - validator(value=pool_name, valid_type=six.string_types) - out = check_output(['ceph', '--id', service, 'osd', 'dump', '--format=json']) - try: - osd_json = json.loads(out) - for pool in osd_json['pools']: - if pool['pool_name'] == pool_name: - return pool['cache_mode'] - return None - except ValueError: - raise - - -def pool_exists(service, name): - """Check to see if a RADOS pool already exists.""" - try: - out = check_output(['rados', '--id', service, - 'lspools']).decode('UTF-8') - except CalledProcessError: - return False - - return name in out.split() - - -def get_osds(service): - """Return a list of all Ceph Object Storage Daemons currently in the - cluster. - """ - version = ceph_version() - if version and version >= '0.56': - return json.loads(check_output(['ceph', '--id', service, - 'osd', 'ls', - '--format=json']).decode('UTF-8')) - - return None - - -def install(): - """Basic Ceph client installation.""" - ceph_dir = "/etc/ceph" - if not os.path.exists(ceph_dir): - os.mkdir(ceph_dir) - - apt_install('ceph-common', fatal=True) - - -def rbd_exists(service, pool, rbd_img): - """Check to see if a RADOS block device exists.""" - try: - out = check_output(['rbd', 'list', '--id', - service, '--pool', pool]).decode('UTF-8') - except CalledProcessError: - return False - - return rbd_img in out - - -def create_rbd_image(service, pool, image, sizemb): - """Create a new RADOS block device.""" - cmd = ['rbd', 'create', image, '--size', str(sizemb), '--id', service, - '--pool', pool] - check_call(cmd) - - -def update_pool(client, pool, settings): - cmd = ['ceph', '--id', client, 'osd', 'pool', 'set', pool] - for k, v in six.iteritems(settings): - cmd.append(k) - cmd.append(v) - - check_call(cmd) - - -def create_pool(service, name, replicas=3, pg_num=None): - """Create a new RADOS pool.""" - if pool_exists(service, name): - log("Ceph pool {} already exists, skipping creation".format(name), - level=WARNING) - return - - if not pg_num: - # Calculate the number of placement groups based - # on upstream recommended best practices. - osds = get_osds(service) - if osds: - pg_num = (len(osds) * 100 // replicas) - else: - # NOTE(james-page): Default to 200 for older ceph versions - # which don't support OSD query from cli - pg_num = 200 - - cmd = ['ceph', '--id', service, 'osd', 'pool', 'create', name, str(pg_num)] - check_call(cmd) - - update_pool(service, name, settings={'size': str(replicas)}) - - -def delete_pool(service, name): - """Delete a RADOS pool from ceph.""" - cmd = ['ceph', '--id', service, 'osd', 'pool', 'delete', name, - '--yes-i-really-really-mean-it'] - check_call(cmd) - - -def _keyfile_path(service): - return KEYFILE.format(service) - - -def _keyring_path(service): - return KEYRING.format(service) - - -def create_keyring(service, key): - """Create a new Ceph keyring containing key.""" - keyring = _keyring_path(service) - if os.path.exists(keyring): - log('Ceph keyring exists at %s.' % keyring, level=WARNING) - return - - cmd = ['ceph-authtool', keyring, '--create-keyring', - '--name=client.{}'.format(service), '--add-key={}'.format(key)] - check_call(cmd) - log('Created new ceph keyring at %s.' % keyring, level=DEBUG) - - -def delete_keyring(service): - """Delete an existing Ceph keyring.""" - keyring = _keyring_path(service) - if not os.path.exists(keyring): - log('Keyring does not exist at %s' % keyring, level=WARNING) - return - - os.remove(keyring) - log('Deleted ring at %s.' % keyring, level=INFO) - - -def create_key_file(service, key): - """Create a file containing key.""" - keyfile = _keyfile_path(service) - if os.path.exists(keyfile): - log('Keyfile exists at %s.' % keyfile, level=WARNING) - return - - with open(keyfile, 'w') as fd: - fd.write(key) - - log('Created new keyfile at %s.' % keyfile, level=INFO) - - -def get_ceph_nodes(relation='ceph'): - """Query named relation to determine current nodes.""" - hosts = [] - for r_id in relation_ids(relation): - for unit in related_units(r_id): - hosts.append(relation_get('private-address', unit=unit, rid=r_id)) - - return hosts - - -def configure(service, key, auth, use_syslog): - """Perform basic configuration of Ceph.""" - create_keyring(service, key) - create_key_file(service, key) - hosts = get_ceph_nodes() - with open('/etc/ceph/ceph.conf', 'w') as ceph_conf: - ceph_conf.write(CEPH_CONF.format(auth=auth, - keyring=_keyring_path(service), - mon_hosts=",".join(map(str, hosts)), - use_syslog=use_syslog)) - modprobe('rbd') - - -def image_mapped(name): - """Determine whether a RADOS block device is mapped locally.""" - try: - out = check_output(['rbd', 'showmapped']).decode('UTF-8') - except CalledProcessError: - return False - - return name in out - - -def map_block_storage(service, pool, image): - """Map a RADOS block device for local use.""" - cmd = [ - 'rbd', - 'map', - '{}/{}'.format(pool, image), - '--user', - service, - '--secret', - _keyfile_path(service), - ] - check_call(cmd) - - -def filesystem_mounted(fs): - """Determine whether a filesytems is already mounted.""" - return fs in [f for f, m in mounts()] - - -def make_filesystem(blk_device, fstype='ext4', timeout=10): - """Make a new filesystem on the specified block device.""" - count = 0 - e_noent = os.errno.ENOENT - while not os.path.exists(blk_device): - if count >= timeout: - log('Gave up waiting on block device %s' % blk_device, - level=ERROR) - raise IOError(e_noent, os.strerror(e_noent), blk_device) - - log('Waiting for block device %s to appear' % blk_device, - level=DEBUG) - count += 1 - time.sleep(1) - else: - log('Formatting block device %s as filesystem %s.' % - (blk_device, fstype), level=INFO) - check_call(['mkfs', '-t', fstype, blk_device]) - - -def place_data_on_block_device(blk_device, data_src_dst): - """Migrate data in data_src_dst to blk_device and then remount.""" - # mount block device into /mnt - mount(blk_device, '/mnt') - # copy data to /mnt - copy_files(data_src_dst, '/mnt') - # umount block device - umount('/mnt') - # Grab user/group ID's from original source - _dir = os.stat(data_src_dst) - uid = _dir.st_uid - gid = _dir.st_gid - # re-mount where the data should originally be - # TODO: persist is currently a NO-OP in core.host - mount(blk_device, data_src_dst, persist=True) - # ensure original ownership of new mount. - os.chown(data_src_dst, uid, gid) - - -def copy_files(src, dst, symlinks=False, ignore=None): - """Copy files from src to dst.""" - for item in os.listdir(src): - s = os.path.join(src, item) - d = os.path.join(dst, item) - if os.path.isdir(s): - shutil.copytree(s, d, symlinks, ignore) - else: - shutil.copy2(s, d) - - -def ensure_ceph_storage(service, pool, rbd_img, sizemb, mount_point, - blk_device, fstype, system_services=[], - replicas=3): - """NOTE: This function must only be called from a single service unit for - the same rbd_img otherwise data loss will occur. - - Ensures given pool and RBD image exists, is mapped to a block device, - and the device is formatted and mounted at the given mount_point. - - If formatting a device for the first time, data existing at mount_point - will be migrated to the RBD device before being re-mounted. - - All services listed in system_services will be stopped prior to data - migration and restarted when complete. - """ - # Ensure pool, RBD image, RBD mappings are in place. - if not pool_exists(service, pool): - log('Creating new pool {}.'.format(pool), level=INFO) - create_pool(service, pool, replicas=replicas) - - if not rbd_exists(service, pool, rbd_img): - log('Creating RBD image ({}).'.format(rbd_img), level=INFO) - create_rbd_image(service, pool, rbd_img, sizemb) - - if not image_mapped(rbd_img): - log('Mapping RBD Image {} as a Block Device.'.format(rbd_img), - level=INFO) - map_block_storage(service, pool, rbd_img) - - # make file system - # TODO: What happens if for whatever reason this is run again and - # the data is already in the rbd device and/or is mounted?? - # When it is mounted already, it will fail to make the fs - # XXX: This is really sketchy! Need to at least add an fstab entry - # otherwise this hook will blow away existing data if its executed - # after a reboot. - if not filesystem_mounted(mount_point): - make_filesystem(blk_device, fstype) - - for svc in system_services: - if service_running(svc): - log('Stopping services {} prior to migrating data.' - .format(svc), level=DEBUG) - service_stop(svc) - - place_data_on_block_device(blk_device, mount_point) - - for svc in system_services: - log('Starting service {} after migrating data.' - .format(svc), level=DEBUG) - service_start(svc) - - -def ensure_ceph_keyring(service, user=None, group=None, relation='ceph'): - """Ensures a ceph keyring is created for a named service and optionally - ensures user and group ownership. - - Returns False if no ceph key is available in relation state. - """ - key = None - for rid in relation_ids(relation): - for unit in related_units(rid): - key = relation_get('key', rid=rid, unit=unit) - if key: - break - - if not key: - return False - - create_keyring(service=service, key=key) - keyring = _keyring_path(service) - if user and group: - check_call(['chown', '%s.%s' % (user, group), keyring]) - - return True - - -def ceph_version(): - """Retrieve the local version of ceph.""" - if os.path.exists('/usr/bin/ceph'): - cmd = ['ceph', '-v'] - output = check_output(cmd).decode('US-ASCII') - output = output.split() - if len(output) > 3: - return output[2] - else: - return None - else: - return None - - -class CephBrokerRq(object): - """Ceph broker request. - - Multiple operations can be added to a request and sent to the Ceph broker - to be executed. - - Request is json-encoded for sending over the wire. - - The API is versioned and defaults to version 1. - """ - - def __init__(self, api_version=1, request_id=None): - self.api_version = api_version - if request_id: - self.request_id = request_id - else: - self.request_id = str(uuid.uuid1()) - self.ops = [] - - def add_op_create_pool(self, name, replica_count=3, pg_num=None): - """Adds an operation to create a pool. - - @param pg_num setting: optional setting. If not provided, this value - will be calculated by the broker based on how many OSDs are in the - cluster at the time of creation. Note that, if provided, this value - will be capped at the current available maximum. - """ - self.ops.append({'op': 'create-pool', 'name': name, - 'replicas': replica_count, 'pg_num': pg_num}) - - def set_ops(self, ops): - """Set request ops to provided value. - - Useful for injecting ops that come from a previous request - to allow comparisons to ensure validity. - """ - self.ops = ops - - @property - def request(self): - return json.dumps({'api-version': self.api_version, 'ops': self.ops, - 'request-id': self.request_id}) - - def _ops_equal(self, other): - if len(self.ops) == len(other.ops): - for req_no in range(0, len(self.ops)): - for key in ['replicas', 'name', 'op', 'pg_num']: - if self.ops[req_no].get(key) != other.ops[req_no].get(key): - return False - else: - return False - return True - - def __eq__(self, other): - if not isinstance(other, self.__class__): - return False - if self.api_version == other.api_version and \ - self._ops_equal(other): - return True - else: - return False - - def __ne__(self, other): - return not self.__eq__(other) - - -class CephBrokerRsp(object): - """Ceph broker response. - - Response is json-decoded and contents provided as methods/properties. - - The API is versioned and defaults to version 1. - """ - - def __init__(self, encoded_rsp): - self.api_version = None - self.rsp = json.loads(encoded_rsp) - - @property - def request_id(self): - return self.rsp.get('request-id') - - @property - def exit_code(self): - return self.rsp.get('exit-code') - - @property - def exit_msg(self): - return self.rsp.get('stderr') - - -# Ceph Broker Conversation: -# If a charm needs an action to be taken by ceph it can create a CephBrokerRq -# and send that request to ceph via the ceph relation. The CephBrokerRq has a -# unique id so that the client can identity which CephBrokerRsp is associated -# with the request. Ceph will also respond to each client unit individually -# creating a response key per client unit eg glance/0 will get a CephBrokerRsp -# via key broker-rsp-glance-0 -# -# To use this the charm can just do something like: -# -# from charmhelpers.contrib.storage.linux.ceph import ( -# send_request_if_needed, -# is_request_complete, -# CephBrokerRq, -# ) -# -# @hooks.hook('ceph-relation-changed') -# def ceph_changed(): -# rq = CephBrokerRq() -# rq.add_op_create_pool(name='poolname', replica_count=3) -# -# if is_request_complete(rq): -# <Request complete actions> -# else: -# send_request_if_needed(get_ceph_request()) -# -# CephBrokerRq and CephBrokerRsp are serialized into JSON. Below is an example -# of glance having sent a request to ceph which ceph has successfully processed -# 'ceph:8': { -# 'ceph/0': { -# 'auth': 'cephx', -# 'broker-rsp-glance-0': '{"request-id": "0bc7dc54", "exit-code": 0}', -# 'broker_rsp': '{"request-id": "0da543b8", "exit-code": 0}', -# 'ceph-public-address': '10.5.44.103', -# 'key': 'AQCLDttVuHXINhAAvI144CB09dYchhHyTUY9BQ==', -# 'private-address': '10.5.44.103', -# }, -# 'glance/0': { -# 'broker_req': ('{"api-version": 1, "request-id": "0bc7dc54", ' -# '"ops": [{"replicas": 3, "name": "glance", ' -# '"op": "create-pool"}]}'), -# 'private-address': '10.5.44.109', -# }, -# } - -def get_previous_request(rid): - """Return the last ceph broker request sent on a given relation - - @param rid: Relation id to query for request - """ - request = None - broker_req = relation_get(attribute='broker_req', rid=rid, - unit=local_unit()) - if broker_req: - request_data = json.loads(broker_req) - request = CephBrokerRq(api_version=request_data['api-version'], - request_id=request_data['request-id']) - request.set_ops(request_data['ops']) - - return request - - -def get_request_states(request, relation='ceph'): - """Return a dict of requests per relation id with their corresponding - completion state. - - This allows a charm, which has a request for ceph, to see whether there is - an equivalent request already being processed and if so what state that - request is in. - - @param request: A CephBrokerRq object - """ - complete = [] - requests = {} - for rid in relation_ids(relation): - complete = False - previous_request = get_previous_request(rid) - if request == previous_request: - sent = True - complete = is_request_complete_for_rid(previous_request, rid) - else: - sent = False - complete = False - - requests[rid] = { - 'sent': sent, - 'complete': complete, - } - - return requests - - -def is_request_sent(request, relation='ceph'): - """Check to see if a functionally equivalent request has already been sent - - Returns True if a similair request has been sent - - @param request: A CephBrokerRq object - """ - states = get_request_states(request, relation=relation) - for rid in states.keys(): - if not states[rid]['sent']: - return False - - return True - - -def is_request_complete(request, relation='ceph'): - """Check to see if a functionally equivalent request has already been - completed - - Returns True if a similair request has been completed - - @param request: A CephBrokerRq object - """ - states = get_request_states(request, relation=relation) - for rid in states.keys(): - if not states[rid]['complete']: - return False - - return True - - -def is_request_complete_for_rid(request, rid): - """Check if a given request has been completed on the given relation - - @param request: A CephBrokerRq object - @param rid: Relation ID - """ - broker_key = get_broker_rsp_key() - for unit in related_units(rid): - rdata = relation_get(rid=rid, unit=unit) - if rdata.get(broker_key): - rsp = CephBrokerRsp(rdata.get(broker_key)) - if rsp.request_id == request.request_id: - if not rsp.exit_code: - return True - else: - # The remote unit sent no reply targeted at this unit so either the - # remote ceph cluster does not support unit targeted replies or it - # has not processed our request yet. - if rdata.get('broker_rsp'): - request_data = json.loads(rdata['broker_rsp']) - if request_data.get('request-id'): - log('Ignoring legacy broker_rsp without unit key as remote ' - 'service supports unit specific replies', level=DEBUG) - else: - log('Using legacy broker_rsp as remote service does not ' - 'supports unit specific replies', level=DEBUG) - rsp = CephBrokerRsp(rdata['broker_rsp']) - if not rsp.exit_code: - return True - - return False - - -def get_broker_rsp_key(): - """Return broker response key for this unit - - This is the key that ceph is going to use to pass request status - information back to this unit - """ - return 'broker-rsp-' + local_unit().replace('/', '-') - - -def send_request_if_needed(request, relation='ceph'): - """Send broker request if an equivalent request has not already been sent - - @param request: A CephBrokerRq object - """ - if is_request_sent(request, relation=relation): - log('Request already sent but not complete, not sending new request', - level=DEBUG) - else: - for rid in relation_ids(relation): - log('Sending request {}'.format(request.request_id), level=DEBUG) - relation_set(relation_id=rid, broker_req=request.request) |