Checkin new helpers.

hooks/charmhelpers/contrib/storage/linux/ceph.py

@@ -0,0 +1,358 @@
+#
+# 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 os
+import shutil
+import json
+import time
+
+from subprocess import (
+    check_call,
+    check_output,
+    CalledProcessError
+)
+
+from charmhelpers.core.hookenv import (
+    relation_get,
+    relation_ids,
+    related_units,
+    log,
+    INFO,
+    WARNING,
+    ERROR
+)
+
+from charmhelpers.core.host import (
+    mount,
+    mounts,
+    service_start,
+    service_stop,
+    service_running,
+    umount,
+)
+
+from charmhelpers.fetch import (
+    apt_install,
+)
+
+KEYRING = '/etc/ceph/ceph.client.{}.keyring'
+KEYFILE = '/etc/ceph/ceph.client.{}.key'
+
+CEPH_CONF = """[global]
+ auth supported = {auth}
+ keyring = {keyring}
+ mon host = {mon_hosts}
+"""
+
+
+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])
+    except CalledProcessError:
+        return False
+    else:
+        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 pool_exists(service, name):
+    ''' Check to see if a RADOS pool already exists '''
+    try:
+        out = check_output(['rados', '--id', service, 'lspools'])
+    except CalledProcessError:
+        return False
+    else:
+        return name in out
+
+
+def get_osds():
+    '''
+    Return a list of all Ceph Object Storage Daemons
+    currently in the cluster
+    '''
+    return json.loads(check_output(['ceph', 'osd', 'ls', '--format=json']))
+
+
+def create_pool(service, name, replicas=2):
+    ''' Create a new RADOS pool '''
+    if pool_exists(service, name):
+        log("Ceph pool {} already exists, skipping creation".format(name),
+            level=WARNING)
+        return
+    # Calculate the number of placement groups based
+    # on upstream recommended best practices.
+    pgnum = (len(get_osds()) * 100 / replicas)
+    cmd = [
+        'ceph', '--id', service,
+        'osd', 'pool', 'create',
+        name, pgnum
+    ]
+    check_call(cmd)
+    cmd = [
+        'ceph', '--id', service,
+        'osd', 'set', name,
+        'size', replicas
+    ]
+    check_call(cmd)
+
+
+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('ceph: Created new 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('ceph: Keyfile exists at %s.' % keyfile, level=WARNING)
+        return
+    with open(keyfile, 'w') as fd:
+        fd.write(key)
+    log('ceph: Created new keyfile at %s.' % keyfile, level=INFO)
+
+
+def get_ceph_nodes():
+    ''' Query named relation 'ceph' to detemine current nodes '''
+    hosts = []
+    for r_id in relation_ids('ceph'):
+        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):
+    ''' 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))))
+    modprobe('rbd')
+
+
+def image_mapped(name):
+    ''' Determine whether a RADOS block device is mapped locally '''
+    try:
+        out = check_output(['rbd', 'showmapped'])
+    except CalledProcessError:
+        return False
+    else:
+        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('ceph: gave up waiting on block device %s' % blk_device,
+                level=ERROR)
+            raise IOError(e_noent, os.strerror(e_noent), blk_device)
+        log('ceph: waiting for block device %s to appear' % blk_device,
+            level=INFO)
+        count += 1
+        time.sleep(1)
+    else:
+        log('ceph: 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)
+
+
+# TODO: re-use
+def modprobe(module):
+    ''' Load a kernel module and configure for auto-load on reboot '''
+    log('ceph: Loading kernel module', level=INFO)
+    cmd = ['modprobe', module]
+    check_call(cmd)
+    with open('/etc/modules', 'r+') as modules:
+        if module not in modules.read():
+            modules.write(module)
+
+
+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=[]):
+    """
+    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('ceph: Creating new pool {}.'.format(pool))
+        create_pool(service, pool)
+
+    if not rbd_exists(service, pool, rbd_img):
+        log('ceph: Creating RBD image ({}).'.format(rbd_img))
+        create_rbd_image(service, pool, rbd_img, sizemb)
+
+    if not image_mapped(rbd_img):
+        log('ceph: Mapping RBD Image {} as a Block Device.'.format(rbd_img))
+        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('ceph: Stopping services {} prior to migrating data.'
+                    .format(svc))
+                service_stop(svc)
+
+        place_data_on_block_device(blk_device, mount_point)
+
+        for svc in system_services:
+            log('ceph: Starting service {} after migrating data.'
+                .format(svc))
+            service_start(svc)
+
+
+def ensure_ceph_keyring(service, user=None, group=None):
+    '''
+    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('ceph'):
+        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

hooks/charmhelpers/fetch/__init__.py

@@ -0,0 +1,209 @@
+import importlib
+from yaml import safe_load
+from charmhelpers.core.host import (
+    lsb_release
+)
+from urlparse import (
+    urlparse,
+    urlunparse,
+)
+import subprocess
+from charmhelpers.core.hookenv import (
+    config,
+    log,
+)
+import apt_pkg
+
+CLOUD_ARCHIVE = """# Ubuntu Cloud Archive
+deb http://ubuntu-cloud.archive.canonical.com/ubuntu {} main
+"""
+PROPOSED_POCKET = """# Proposed
+deb http://archive.ubuntu.com/ubuntu {}-proposed main universe multiverse restricted
+"""
+
+
+def filter_installed_packages(packages):
+    """Returns a list of packages that require installation"""
+    apt_pkg.init()
+    cache = apt_pkg.Cache()
+    _pkgs = []
+    for package in packages:
+        try:
+            p = cache[package]
+            p.current_ver or _pkgs.append(package)
+        except KeyError:
+            log('Package {} has no installation candidate.'.format(package),
+                level='WARNING')
+            _pkgs.append(package)
+    return _pkgs
+
+
+def apt_install(packages, options=None, fatal=False):
+    """Install one or more packages"""
+    options = options or []
+    cmd = ['apt-get', '-y']
+    cmd.extend(options)
+    cmd.append('install')
+    if isinstance(packages, basestring):
+        cmd.append(packages)
+    else:
+        cmd.extend(packages)
+    log("Installing {} with options: {}".format(packages,
+                                                options))
+    if fatal:
+        subprocess.check_call(cmd)
+    else:
+        subprocess.call(cmd)
+
+
+def apt_update(fatal=False):
+    """Update local apt cache"""
+    cmd = ['apt-get', 'update']
+    if fatal:
+        subprocess.check_call(cmd)
+    else:
+        subprocess.call(cmd)
+
+
+def apt_purge(packages, fatal=False):
+    """Purge one or more packages"""
+    cmd = ['apt-get', '-y', 'purge']
+    if isinstance(packages, basestring):
+        cmd.append(packages)
+    else:
+        cmd.extend(packages)
+    log("Purging {}".format(packages))
+    if fatal:
+        subprocess.check_call(cmd)
+    else:
+        subprocess.call(cmd)
+
+
+def add_source(source, key=None):
+    if ((source.startswith('ppa:') or
+         source.startswith('http:'))):
+        subprocess.check_call(['add-apt-repository', '--yes', source])
+    elif source.startswith('cloud:'):
+        apt_install(filter_installed_packages(['ubuntu-cloud-keyring']),
+                    fatal=True)
+        pocket = source.split(':')[-1]
+        with open('/etc/apt/sources.list.d/cloud-archive.list', 'w') as apt:
+            apt.write(CLOUD_ARCHIVE.format(pocket))
+    elif source == 'proposed':
+        release = lsb_release()['DISTRIB_CODENAME']
+        with open('/etc/apt/sources.list.d/proposed.list', 'w') as apt:
+            apt.write(PROPOSED_POCKET.format(release))
+    if key:
+        subprocess.check_call(['apt-key', 'import', key])
+
+
+class SourceConfigError(Exception):
+    pass
+
+
+def configure_sources(update=False,
+                      sources_var='install_sources',
+                      keys_var='install_keys'):
+    """
+    Configure multiple sources from charm configuration
+
+    Example config:
+        install_sources:
+          - "ppa:foo"
+          - "http://example.com/repo precise main"
+        install_keys:
+          - null
+          - "a1b2c3d4"
+
+    Note that 'null' (a.k.a. None) should not be quoted.
+    """
+    sources = safe_load(config(sources_var))
+    keys = safe_load(config(keys_var))
+    if isinstance(sources, basestring) and isinstance(keys, basestring):
+        add_source(sources, keys)
+    else:
+        if not len(sources) == len(keys):
+            msg = 'Install sources and keys lists are different lengths'
+            raise SourceConfigError(msg)
+        for src_num in range(len(sources)):
+            add_source(sources[src_num], keys[src_num])
+    if update:
+        apt_update(fatal=True)
+
+# The order of this list is very important. Handlers should be listed in from
+# least- to most-specific URL matching.
+FETCH_HANDLERS = (
+    'charmhelpers.fetch.archiveurl.ArchiveUrlFetchHandler',
+    'charmhelpers.fetch.bzrurl.BzrUrlFetchHandler',
+)
+
+
+class UnhandledSource(Exception):
+    pass
+
+
+def install_remote(source):
+    """
+    Install a file tree from a remote source
+
+    The specified source should be a url of the form:
+        scheme://[host]/path[#[option=value][&...]]
+
+    Schemes supported are based on this modules submodules
+    Options supported are submodule-specific"""
+    # We ONLY check for True here because can_handle may return a string
+    # explaining why it can't handle a given source.
+    handlers = [h for h in plugins() if h.can_handle(source) is True]
+    installed_to = None
+    for handler in handlers:
+        try:
+            installed_to = handler.install(source)
+        except UnhandledSource:
+            pass
+    if not installed_to:
+        raise UnhandledSource("No handler found for source {}".format(source))
+    return installed_to
+
+
+def install_from_config(config_var_name):
+    charm_config = config()
+    source = charm_config[config_var_name]
+    return install_remote(source)
+
+
+class BaseFetchHandler(object):
+    """Base class for FetchHandler implementations in fetch plugins"""
+    def can_handle(self, source):
+        """Returns True if the source can be handled. Otherwise returns
+        a string explaining why it cannot"""
+        return "Wrong source type"
+
+    def install(self, source):
+        """Try to download and unpack the source. Return the path to the
+        unpacked files or raise UnhandledSource."""
+        raise UnhandledSource("Wrong source type {}".format(source))
+
+    def parse_url(self, url):
+        return urlparse(url)
+
+    def base_url(self, url):
+        """Return url without querystring or fragment"""
+        parts = list(self.parse_url(url))
+        parts[4:] = ['' for i in parts[4:]]
+        return urlunparse(parts)
+
+
+def plugins(fetch_handlers=None):
+    if not fetch_handlers:
+        fetch_handlers = FETCH_HANDLERS
+    plugin_list = []
+    for handler_name in fetch_handlers:
+        package, classname = handler_name.rsplit('.', 1)
+        try:
+            handler_class = getattr(importlib.import_module(package), classname)
+            plugin_list.append(handler_class())
+        except (ImportError, AttributeError):
+            # Skip missing plugins so that they can be ommitted from
+            # installation if desired
+            log("FetchHandler {} not found, skipping plugin".format(handler_name))
+    return plugin_list

hooks/charmhelpers/fetch/archiveurl.py

@@ -0,0 +1,48 @@
+import os
+import urllib2
+from charmhelpers.fetch import (
+    BaseFetchHandler,
+    UnhandledSource
+)
+from charmhelpers.payload.archive import (
+    get_archive_handler,
+    extract,
+)
+from charmhelpers.core.host import mkdir
+
+
+class ArchiveUrlFetchHandler(BaseFetchHandler):
+    """Handler for archives via generic URLs"""
+    def can_handle(self, source):
+        url_parts = self.parse_url(source)
+        if url_parts.scheme not in ('http', 'https', 'ftp', 'file'):
+            return "Wrong source type"
+        if get_archive_handler(self.base_url(source)):
+            return True
+        return False
+
+    def download(self, source, dest):
+        # propogate all exceptions
+        # URLError, OSError, etc
+        response = urllib2.urlopen(source)
+        try:
+            with open(dest, 'w') as dest_file:
+                dest_file.write(response.read())
+        except Exception as e:
+            if os.path.isfile(dest):
+                os.unlink(dest)
+            raise e
+
+    def install(self, source):
+        url_parts = self.parse_url(source)
+        dest_dir = os.path.join(os.environ.get('CHARM_DIR'), 'fetched')
+        if not os.path.exists(dest_dir):
+            mkdir(dest_dir, perms=0755)
+        dld_file = os.path.join(dest_dir, os.path.basename(url_parts.path))
+        try:
+            self.download(source, dld_file)
+        except urllib2.URLError as e:
+            raise UnhandledSource(e.reason)
+        except OSError as e:
+            raise UnhandledSource(e.strerror)
+        return extract(dld_file)

hooks/charmhelpers/fetch/bzrurl.py

@@ -0,0 +1,49 @@
+import os
+from charmhelpers.fetch import (
+    BaseFetchHandler,
+    UnhandledSource
+)
+from charmhelpers.core.host import mkdir
+
+try:
+    from bzrlib.branch import Branch
+except ImportError:
+    from charmhelpers.fetch import apt_install
+    apt_install("python-bzrlib")
+    from bzrlib.branch import Branch
+
+class BzrUrlFetchHandler(BaseFetchHandler):
+    """Handler for bazaar branches via generic and lp URLs"""
+    def can_handle(self, source):
+        url_parts = self.parse_url(source)
+        if url_parts.scheme not in ('bzr+ssh', 'lp'):
+            return False
+        else:
+            return True
+
+    def branch(self, source, dest):
+        url_parts = self.parse_url(source)
+        # If we use lp:branchname scheme we need to load plugins
+        if not self.can_handle(source):
+            raise UnhandledSource("Cannot handle {}".format(source))
+        if url_parts.scheme == "lp":
+            from bzrlib.plugin import load_plugins
+            load_plugins()
+        try:
+            remote_branch = Branch.open(source)
+            remote_branch.bzrdir.sprout(dest).open_branch()
+        except Exception as e:
+            raise e
+
+    def install(self, source):
+        url_parts = self.parse_url(source)
+        branch_name = url_parts.path.strip("/").split("/")[-1]
+        dest_dir = os.path.join(os.environ.get('CHARM_DIR'), "fetched", branch_name)
+        if not os.path.exists(dest_dir):
+            mkdir(dest_dir, perms=0755)
+        try:
+            self.branch(source, dest_dir)
+        except OSError as e:
+            raise UnhandledSource(e.strerror)
+        return dest_dir
+