[beisner, r=corey.bryant] Sync charm-helpers and enable amulet tests 018 and 020

hooks/charmhelpers/cli/__init__.py

@@ -20,7 +20,7 @@ import sys
 
 from six.moves import zip
 
-from charmhelpers.core import unitdata
+import charmhelpers.core.unitdata
 
 
 class OutputFormatter(object):
@@ -163,8 +163,8 @@ class CommandLine(object):
         if getattr(arguments.func, '_cli_no_output', False):
             output = ''
         self.formatter.format_output(output, arguments.format)
-        if unitdata._KV:
-            unitdata._KV.flush()
+        if charmhelpers.core.unitdata._KV:
+            charmhelpers.core.unitdata._KV.flush()
 
 
 cmdline = CommandLine()

hooks/charmhelpers/contrib/charmsupport/nrpe.py

@@ -148,6 +148,13 @@ define service {{
         self.description = description
         self.check_cmd = self._locate_cmd(check_cmd)
 
+    def _get_check_filename(self):
+        return os.path.join(NRPE.nrpe_confdir, '{}.cfg'.format(self.command))
+
+    def _get_service_filename(self, hostname):
+        return os.path.join(NRPE.nagios_exportdir,
+                            'service__{}_{}.cfg'.format(hostname, self.command))
+
     def _locate_cmd(self, check_cmd):
         search_path = (
             '/usr/lib/nagios/plugins',
@@ -163,9 +170,21 @@ define service {{
         log('Check command not found: {}'.format(parts[0]))
         return ''
 
+    def _remove_service_files(self):
+        if not os.path.exists(NRPE.nagios_exportdir):
+            return
+        for f in os.listdir(NRPE.nagios_exportdir):
+            if f.endswith('_{}.cfg'.format(self.command)):
+                os.remove(os.path.join(NRPE.nagios_exportdir, f))
+
+    def remove(self, hostname):
+        nrpe_check_file = self._get_check_filename()
+        if os.path.exists(nrpe_check_file):
+            os.remove(nrpe_check_file)
+        self._remove_service_files()
+
     def write(self, nagios_context, hostname, nagios_servicegroups):
-        nrpe_check_file = '/etc/nagios/nrpe.d/{}.cfg'.format(
-            self.command)
+        nrpe_check_file = self._get_check_filename()
         with open(nrpe_check_file, 'w') as nrpe_check_config:
             nrpe_check_config.write("# check {}\n".format(self.shortname))
             nrpe_check_config.write("command[{}]={}\n".format(
@@ -180,9 +199,7 @@ define service {{
 
     def write_service_config(self, nagios_context, hostname,
                              nagios_servicegroups):
-        for f in os.listdir(NRPE.nagios_exportdir):
-            if re.search('.*{}.cfg'.format(self.command), f):
-                os.remove(os.path.join(NRPE.nagios_exportdir, f))
+        self._remove_service_files()
 
         templ_vars = {
             'nagios_hostname': hostname,
@@ -192,8 +209,7 @@ define service {{
             'command': self.command,
         }
         nrpe_service_text = Check.service_template.format(**templ_vars)
-        nrpe_service_file = '{}/service__{}_{}.cfg'.format(
-            NRPE.nagios_exportdir, hostname, self.command)
+        nrpe_service_file = self._get_service_filename(hostname)
         with open(nrpe_service_file, 'w') as nrpe_service_config:
             nrpe_service_config.write(str(nrpe_service_text))
 
@@ -218,12 +234,32 @@ class NRPE(object):
         if hostname:
             self.hostname = hostname
         else:
-            self.hostname = "{}-{}".format(self.nagios_context, self.unit_name)
+            nagios_hostname = get_nagios_hostname()
+            if nagios_hostname:
+                self.hostname = nagios_hostname
+            else:
+                self.hostname = "{}-{}".format(self.nagios_context, self.unit_name)
         self.checks = []
 
     def add_check(self, *args, **kwargs):
         self.checks.append(Check(*args, **kwargs))
 
+    def remove_check(self, *args, **kwargs):
+        if kwargs.get('shortname') is None:
+            raise ValueError('shortname of check must be specified')
+
+        # Use sensible defaults if they're not specified - these are not
+        # actually used during removal, but they're required for constructing
+        # the Check object; check_disk is chosen because it's part of the
+        # nagios-plugins-basic package.
+        if kwargs.get('check_cmd') is None:
+            kwargs['check_cmd'] = 'check_disk'
+        if kwargs.get('description') is None:
+            kwargs['description'] = ''
+
+        check = Check(*args, **kwargs)
+        check.remove(self.hostname)
+
     def write(self):
         try:
             nagios_uid = pwd.getpwnam('nagios').pw_uid
@@ -260,7 +296,7 @@ def get_nagios_hostcontext(relation_name='nrpe-external-master'):
     :param str relation_name: Name of relation nrpe sub joined to
     """
     for rel in relations_of_type(relation_name):
-        if 'nagios_hostname' in rel:
+        if 'nagios_host_context' in rel:
             return rel['nagios_host_context']
 
 
@@ -301,11 +337,13 @@ def add_init_service_checks(nrpe, services, unit_name):
         upstart_init = '/etc/init/%s.conf' % svc
         sysv_init = '/etc/init.d/%s' % svc
         if os.path.exists(upstart_init):
-            nrpe.add_check(
-                shortname=svc,
-                description='process check {%s}' % unit_name,
-                check_cmd='check_upstart_job %s' % svc
-            )
+            # Don't add a check for these services from neutron-gateway
+            if svc not in ['ext-port', 'os-charm-phy-nic-mtu']:
+                nrpe.add_check(
+                    shortname=svc,
+                    description='process check {%s}' % unit_name,
+                    check_cmd='check_upstart_job %s' % svc
+                )
         elif os.path.exists(sysv_init):
             cronpath = '/etc/cron.d/nagios-service-check-%s' % svc
             cron_file = ('*/5 * * * * root '

hooks/charmhelpers/contrib/network/ip.py

@@ -23,7 +23,7 @@ import socket
 from functools import partial
 
 from charmhelpers.core.hookenv import unit_get
-from charmhelpers.fetch import apt_install
+from charmhelpers.fetch import apt_install, apt_update
 from charmhelpers.core.hookenv import (
     log,
     WARNING,
@@ -32,13 +32,15 @@ from charmhelpers.core.hookenv import (
 try:
     import netifaces
 except ImportError:
-    apt_install('python-netifaces')
+    apt_update(fatal=True)
+    apt_install('python-netifaces', fatal=True)
     import netifaces
 
 try:
     import netaddr
 except ImportError:
-    apt_install('python-netaddr')
+    apt_update(fatal=True)
+    apt_install('python-netaddr', fatal=True)
     import netaddr
 
 
@@ -51,7 +53,7 @@ def _validate_cidr(network):
 
 
 def no_ip_found_error_out(network):
-    errmsg = ("No IP address found in network: %s" % network)
+    errmsg = ("No IP address found in network(s): %s" % network)
     raise ValueError(errmsg)
 
 
@@ -59,7 +61,7 @@ def get_address_in_network(network, fallback=None, fatal=False):
     """Get an IPv4 or IPv6 address within the network from the host.
 
     :param network (str): CIDR presentation format. For example,
-        '192.168.1.0/24'.
+        '192.168.1.0/24'. Supports multiple networks as a space-delimited list.
     :param fallback (str): If no address is found, return fallback.
     :param fatal (boolean): If no address is found, fallback is not
         set and fatal is True then exit(1).
@@ -73,24 +75,26 @@ def get_address_in_network(network, fallback=None, fatal=False):
         else:
             return None
 
-    _validate_cidr(network)
-    network = netaddr.IPNetwork(network)
-    for iface in netifaces.interfaces():
-        addresses = netifaces.ifaddresses(iface)
-        if network.version == 4 and netifaces.AF_INET in addresses:
-            addr = addresses[netifaces.AF_INET][0]['addr']
-            netmask = addresses[netifaces.AF_INET][0]['netmask']
-            cidr = netaddr.IPNetwork("%s/%s" % (addr, netmask))
-            if cidr in network:
-                return str(cidr.ip)
+    networks = network.split() or [network]
+    for network in networks:
+        _validate_cidr(network)
+        network = netaddr.IPNetwork(network)
+        for iface in netifaces.interfaces():
+            addresses = netifaces.ifaddresses(iface)
+            if network.version == 4 and netifaces.AF_INET in addresses:
+                addr = addresses[netifaces.AF_INET][0]['addr']
+                netmask = addresses[netifaces.AF_INET][0]['netmask']
+                cidr = netaddr.IPNetwork("%s/%s" % (addr, netmask))
+                if cidr in network:
+                    return str(cidr.ip)
 
-        if network.version == 6 and netifaces.AF_INET6 in addresses:
-            for addr in addresses[netifaces.AF_INET6]:
-                if not addr['addr'].startswith('fe80'):
-                    cidr = netaddr.IPNetwork("%s/%s" % (addr['addr'],
-                                                        addr['netmask']))
-                    if cidr in network:
-                        return str(cidr.ip)
+            if network.version == 6 and netifaces.AF_INET6 in addresses:
+                for addr in addresses[netifaces.AF_INET6]:
+                    if not addr['addr'].startswith('fe80'):
+                        cidr = netaddr.IPNetwork("%s/%s" % (addr['addr'],
+                                                            addr['netmask']))
+                        if cidr in network:
+                            return str(cidr.ip)
 
     if fallback is not None:
         return fallback

hooks/charmhelpers/core/hookenv.py

@@ -490,6 +490,19 @@ def relation_types():
     return rel_types
 
 
+@cached
+def peer_relation_id():
+    '''Get the peers relation id if a peers relation has been joined, else None.'''
+    md = metadata()
+    section = md.get('peers')
+    if section:
+        for key in section:
+            relids = relation_ids(key)
+            if relids:
+                return relids[0]
+    return None
+
+
 @cached
 def relation_to_interface(relation_name):
     """
@@ -504,12 +517,12 @@ def relation_to_interface(relation_name):
 def relation_to_role_and_interface(relation_name):
     """
     Given the name of a relation, return the role and the name of the interface
-    that relation uses (where role is one of ``provides``, ``requires``, or ``peer``).
+    that relation uses (where role is one of ``provides``, ``requires``, or ``peers``).
 
     :returns: A tuple containing ``(role, interface)``, or ``(None, None)``.
     """
     _metadata = metadata()
-    for role in ('provides', 'requires', 'peer'):
+    for role in ('provides', 'requires', 'peers'):
         interface = _metadata.get(role, {}).get(relation_name, {}).get('interface')
         if interface:
             return role, interface
@@ -521,7 +534,7 @@ def role_and_interface_to_relations(role, interface_name):
     """
     Given a role and interface name, return a list of relation names for the
     current charm that use that interface under that role (where role is one
-    of ``provides``, ``requires``, or ``peer``).
+    of ``provides``, ``requires``, or ``peers``).
 
     :returns: A list of relation names.
     """
@@ -542,7 +555,7 @@ def interface_to_relations(interface_name):
     :returns: A list of relation names.
     """
     results = []
-    for role in ('provides', 'requires', 'peer'):
+    for role in ('provides', 'requires', 'peers'):
         results.extend(role_and_interface_to_relations(role, interface_name))
     return results
 
@@ -623,6 +636,38 @@ def unit_private_ip():
     return unit_get('private-address')
 
 
+@cached
+def storage_get(attribute=None, storage_id=None):
+    """Get storage attributes"""
+    _args = ['storage-get', '--format=json']
+    if storage_id:
+        _args.extend(('-s', storage_id))
+    if attribute:
+        _args.append(attribute)
+    try:
+        return json.loads(subprocess.check_output(_args).decode('UTF-8'))
+    except ValueError:
+        return None
+
+
+@cached
+def storage_list(storage_name=None):
+    """List the storage IDs for the unit"""
+    _args = ['storage-list', '--format=json']
+    if storage_name:
+        _args.append(storage_name)
+    try:
+        return json.loads(subprocess.check_output(_args).decode('UTF-8'))
+    except ValueError:
+        return None
+    except OSError as e:
+        import errno
+        if e.errno == errno.ENOENT:
+            # storage-list does not exist
+            return []
+        raise
+
+
 class UnregisteredHookError(Exception):
     """Raised when an undefined hook is called"""
     pass
@@ -788,6 +833,7 @@ def status_get():
 
 def translate_exc(from_exc, to_exc):
     def inner_translate_exc1(f):
+        @wraps(f)
         def inner_translate_exc2(*args, **kwargs):
             try:
                 return f(*args, **kwargs)
@@ -832,6 +878,40 @@ def leader_set(settings=None, **kwargs):
     subprocess.check_call(cmd)
 
 
+@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
+def payload_register(ptype, klass, pid):
+    """ is used while a hook is running to let Juju know that a
+        payload has been started."""
+    cmd = ['payload-register']
+    for x in [ptype, klass, pid]:
+        cmd.append(x)
+    subprocess.check_call(cmd)
+
+
+@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
+def payload_unregister(klass, pid):
+    """ is used while a hook is running to let Juju know
+    that a payload has been manually stopped. The <class> and <id> provided
+    must match a payload that has been previously registered with juju using
+    payload-register."""
+    cmd = ['payload-unregister']
+    for x in [klass, pid]:
+        cmd.append(x)
+    subprocess.check_call(cmd)
+
+
+@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
+def payload_status_set(klass, pid, status):
+    """is used to update the current status of a registered payload.
+    The <class> and <id> provided must match a payload that has been previously
+    registered with juju using payload-register. The <status> must be one of the
+    follow: starting, started, stopping, stopped"""
+    cmd = ['payload-status-set']
+    for x in [klass, pid, status]:
+        cmd.append(x)
+    subprocess.check_call(cmd)
+
+
 @cached
 def juju_version():
     """Full version string (eg. '1.23.3.1-trusty-amd64')"""

hooks/charmhelpers/core/host.py

@@ -63,55 +63,86 @@ def service_reload(service_name, restart_on_failure=False):
     return service_result
 
 
-def service_pause(service_name, init_dir=None):
+def service_pause(service_name, init_dir="/etc/init", initd_dir="/etc/init.d"):
     """Pause a system service.
 
     Stop it, and prevent it from starting again at boot."""
-    if init_dir is None:
-        init_dir = "/etc/init"
-    stopped = service_stop(service_name)
-    # XXX: Support systemd too
-    override_path = os.path.join(
-        init_dir, '{}.override'.format(service_name))
-    with open(override_path, 'w') as fh:
-        fh.write("manual\n")
+    stopped = True
+    if service_running(service_name):
+        stopped = service_stop(service_name)
+    upstart_file = os.path.join(init_dir, "{}.conf".format(service_name))
+    sysv_file = os.path.join(initd_dir, service_name)
+    if init_is_systemd():
+        service('disable', service_name)
+    elif os.path.exists(upstart_file):
+        override_path = os.path.join(
+            init_dir, '{}.override'.format(service_name))
+        with open(override_path, 'w') as fh:
+            fh.write("manual\n")
+    elif os.path.exists(sysv_file):
+        subprocess.check_call(["update-rc.d", service_name, "disable"])
+    else:
+        raise ValueError(
+            "Unable to detect {0} as SystemD, Upstart {1} or"
+            " SysV {2}".format(
+                service_name, upstart_file, sysv_file))
     return stopped
 
 
-def service_resume(service_name, init_dir=None):
+def service_resume(service_name, init_dir="/etc/init",
+                   initd_dir="/etc/init.d"):
     """Resume a system service.
 
     Reenable starting again at boot. Start the service"""
-    # XXX: Support systemd too
-    if init_dir is None:
-        init_dir = "/etc/init"
-    override_path = os.path.join(
-        init_dir, '{}.override'.format(service_name))
-    if os.path.exists(override_path):
-        os.unlink(override_path)
-    started = service_start(service_name)
+    upstart_file = os.path.join(init_dir, "{}.conf".format(service_name))
+    sysv_file = os.path.join(initd_dir, service_name)
+    if init_is_systemd():
+        service('enable', service_name)
+    elif os.path.exists(upstart_file):
+        override_path = os.path.join(
+            init_dir, '{}.override'.format(service_name))
+        if os.path.exists(override_path):
+            os.unlink(override_path)
+    elif os.path.exists(sysv_file):
+        subprocess.check_call(["update-rc.d", service_name, "enable"])
+    else:
+        raise ValueError(
+            "Unable to detect {0} as SystemD, Upstart {1} or"
+            " SysV {2}".format(
+                service_name, upstart_file, sysv_file))
+
+    started = service_running(service_name)
+    if not started:
+        started = service_start(service_name)
     return started
 
 
 def service(action, service_name):
     """Control a system service"""
-    cmd = ['service', service_name, action]
+    if init_is_systemd():
+        cmd = ['systemctl', action, service_name]
+    else:
+        cmd = ['service', service_name, action]
     return subprocess.call(cmd) == 0
 
 
-def service_running(service):
+def service_running(service_name):
     """Determine whether a system service is running"""
-    try:
-        output = subprocess.check_output(
-            ['service', service, 'status'],
-            stderr=subprocess.STDOUT).decode('UTF-8')
-    except subprocess.CalledProcessError:
-        return False
+    if init_is_systemd():
+        return service('is-active', service_name)
     else:
-        if ("start/running" in output or "is running" in output):
-            return True
-        else:
+        try:
+            output = subprocess.check_output(
+                ['service', service_name, 'status'],
+                stderr=subprocess.STDOUT).decode('UTF-8')
+        except subprocess.CalledProcessError:
             return False
+        else:
+            if ("start/running" in output or "is running" in output or
+                    "up and running" in output):
+                return True
+            else:
+                return False
 
 
 def service_available(service_name):
@@ -126,8 +157,29 @@ def service_available(service_name):
         return True
 
 
-def adduser(username, password=None, shell='/bin/bash', system_user=False):
-    """Add a user to the system"""
+SYSTEMD_SYSTEM = '/run/systemd/system'
+
+
+def init_is_systemd():
+    """Return True if the host system uses systemd, False otherwise."""
+    return os.path.isdir(SYSTEMD_SYSTEM)
+
+
+def adduser(username, password=None, shell='/bin/bash', system_user=False,
+            primary_group=None, secondary_groups=None):
+    """Add a user to the system.
+
+    Will log but otherwise succeed if the user already exists.
+
+    :param str username: Username to create
+    :param str password: Password for user; if ``None``, create a system user
+    :param str shell: The default shell for the user
+    :param bool system_user: Whether to create a login or system user
+    :param str primary_group: Primary group for user; defaults to username
+    :param list secondary_groups: Optional list of additional groups
+
+    :returns: The password database entry struct, as returned by `pwd.getpwnam`
+    """
     try:
         user_info = pwd.getpwnam(username)
         log('user {0} already exists!'.format(username))
@@ -142,6 +194,16 @@ def adduser(username, password=None, shell='/bin/bash', system_user=False):
                 '--shell', shell,
                 '--password', password,
             ])
+        if not primary_group:
+            try:
+                grp.getgrnam(username)
+                primary_group = username  # avoid "group exists" error
+            except KeyError:
+                pass
+        if primary_group:
+            cmd.extend(['-g', primary_group])
+        if secondary_groups:
+            cmd.extend(['-G', ','.join(secondary_groups)])
         cmd.append(username)
         subprocess.check_call(cmd)
         user_info = pwd.getpwnam(username)
@@ -239,14 +301,12 @@ def write_file(path, content, owner='root', group='root', perms=0o444):
 
 
 def fstab_remove(mp):
-    """Remove the given mountpoint entry from /etc/fstab
-    """
+    """Remove the given mountpoint entry from /etc/fstab"""
     return Fstab.remove_by_mountpoint(mp)
 
 
 def fstab_add(dev, mp, fs, options=None):
-    """Adds the given device entry to the /etc/fstab file
-    """
+    """Adds the given device entry to the /etc/fstab file"""
     return Fstab.add(dev, mp, fs, options=options)
 
 
@@ -302,8 +362,7 @@ def fstab_mount(mountpoint):
 
 
 def file_hash(path, hash_type='md5'):
-    """
-    Generate a hash checksum of the contents of 'path' or None if not found.
+    """Generate a hash checksum of the contents of 'path' or None if not found.
 
     :param str hash_type: Any hash alrgorithm supported by :mod:`hashlib`,
                           such as md5, sha1, sha256, sha512, etc.
@@ -318,10 +377,9 @@ def file_hash(path, hash_type='md5'):
 
 
 def path_hash(path):
-    """
-    Generate a hash checksum of all files matching 'path'. Standard wildcards
-    like '*' and '?' are supported, see documentation for the 'glob' module for
-    more information.
+    """Generate a hash checksum of all files matching 'path'. Standard
+    wildcards like '*' and '?' are supported, see documentation for the 'glob'
+    module for more information.
 
     :return: dict: A { filename: hash } dictionary for all matched files.
                    Empty if none found.
@@ -333,8 +391,7 @@ def path_hash(path):
 
 
 def check_hash(path, checksum, hash_type='md5'):
-    """
-    Validate a file using a cryptographic checksum.
+    """Validate a file using a cryptographic checksum.
 
     :param str checksum: Value of the checksum used to validate the file.
     :param str hash_type: Hash algorithm used to generate `checksum`.
@@ -349,6 +406,7 @@ def check_hash(path, checksum, hash_type='md5'):
 
 
 class ChecksumError(ValueError):
+    """A class derived from Value error to indicate the checksum failed."""
     pass
 
 
@@ -454,7 +512,7 @@ def get_bond_master(interface):
 
 
 def list_nics(nic_type=None):
-    '''Return a list of nics of given type(s)'''
+    """Return a list of nics of given type(s)"""
     if isinstance(nic_type, six.string_types):
         int_types = [nic_type]
     else:
@@ -496,12 +554,13 @@ def list_nics(nic_type=None):
 
 
 def set_nic_mtu(nic, mtu):
-    '''Set MTU on a network interface'''
+    """Set the Maximum Transmission Unit (MTU) on a network interface."""
     cmd = ['ip', 'link', 'set', nic, 'mtu', mtu]
     subprocess.check_call(cmd)
 
 
 def get_nic_mtu(nic):
+    """Return the Maximum Transmission Unit (MTU) for a network interface."""
     cmd = ['ip', 'addr', 'show', nic]
     ip_output = subprocess.check_output(cmd).decode('UTF-8').split('\n')
     mtu = ""
@@ -513,6 +572,7 @@ def get_nic_mtu(nic):
 
 
 def get_nic_hwaddr(nic):
+    """Return the Media Access Control (MAC) for a network interface."""
     cmd = ['ip', '-o', '-0', 'addr', 'show', nic]
     ip_output = subprocess.check_output(cmd).decode('UTF-8')
     hwaddr = ""
@@ -523,7 +583,7 @@ def get_nic_hwaddr(nic):
 
 
 def cmp_pkgrevno(package, revno, pkgcache=None):
-    '''Compare supplied revno with the revno of the installed package
+    """Compare supplied revno with the revno of the installed package
 
     *  1 => Installed revno is greater than supplied arg
     *  0 => Installed revno is the same as supplied arg
@@ -532,7 +592,7 @@ def cmp_pkgrevno(package, revno, pkgcache=None):
     This function imports apt_cache function from charmhelpers.fetch if
     the pkgcache argument is None. Be sure to add charmhelpers.fetch if
     you call this function, or pass an apt_pkg.Cache() instance.
-    '''
+    """
     import apt_pkg
     if not pkgcache:
         from charmhelpers.fetch import apt_cache
@@ -542,15 +602,30 @@ def cmp_pkgrevno(package, revno, pkgcache=None):
 
 
 @contextmanager
-def chdir(d):
+def chdir(directory):
+    """Change the current working directory to a different directory for a code
+    block and return the previous directory after the block exits. Useful to
+    run commands from a specificed directory.
+
+    :param str directory: The directory path to change to for this context.
+    """
     cur = os.getcwd()
     try:
-        yield os.chdir(d)
+        yield os.chdir(directory)
     finally:
         os.chdir(cur)
 
 
-def chownr(path, owner, group, follow_links=True):
+def chownr(path, owner, group, follow_links=True, chowntopdir=False):
+    """Recursively change user and group ownership of files and directories
+    in given path. Doesn't chown path itself by default, only its children.
+
+    :param str path: The string path to start changing ownership.
+    :param str owner: The owner string to use when looking up the uid.
+    :param str group: The group string to use when looking up the gid.
+    :param bool follow_links: Also Chown links if True
+    :param bool chowntopdir: Also chown path itself if True
+    """
     uid = pwd.getpwnam(owner).pw_uid
     gid = grp.getgrnam(group).gr_gid
     if follow_links:
@@ -558,6 +633,10 @@ def chownr(path, owner, group, follow_links=True):
     else:
         chown = os.lchown
 
+    if chowntopdir:
+        broken_symlink = os.path.lexists(path) and not os.path.exists(path)
+        if not broken_symlink:
+            chown(path, uid, gid)
     for root, dirs, files in os.walk(path):
         for name in dirs + files:
             full = os.path.join(root, name)
@@ -567,4 +646,28 @@ def chownr(path, owner, group, follow_links=True):
 
 
 def lchownr(path, owner, group):
+    """Recursively change user and group ownership of files and directories
+    in a given path, not following symbolic links. See the documentation for
+    'os.lchown' for more information.
+
+    :param str path: The string path to start changing ownership.
+    :param str owner: The owner string to use when looking up the uid.
+    :param str group: The group string to use when looking up the gid.
+    """
     chownr(path, owner, group, follow_links=False)
+
+
+def get_total_ram():
+    """The total amount of system RAM in bytes.
+
+    This is what is reported by the OS, and may be overcommitted when
+    there are multiple containers hosted on the same machine.
+    """
+    with open('/proc/meminfo', 'r') as f:
+        for line in f.readlines():
+            if line:
+                key, value, unit = line.split()
+                if key == 'MemTotal:':
+                    assert unit == 'kB', 'Unknown unit'
+                    return int(value) * 1024  # Classic, not KiB.
+        raise NotImplementedError()

hooks/charmhelpers/core/hugepage.py

@@ -25,11 +25,13 @@ from charmhelpers.core.host import (
     fstab_mount,
     mkdir,
 )
+from charmhelpers.core.strutils import bytes_from_string
+from subprocess import check_output
 
 
 def hugepage_support(user, group='hugetlb', nr_hugepages=256,
                      max_map_count=65536, mnt_point='/run/hugepages/kvm',
-                     pagesize='2MB', mount=True):
+                     pagesize='2MB', mount=True, set_shmmax=False):
     """Enable hugepages on system.
 
     Args:
@@ -44,11 +46,18 @@ def hugepage_support(user, group='hugetlb', nr_hugepages=256,
     group_info = add_group(group)
     gid = group_info.gr_gid
     add_user_to_group(user, group)
+    if max_map_count < 2 * nr_hugepages:
+        max_map_count = 2 * nr_hugepages
     sysctl_settings = {
         'vm.nr_hugepages': nr_hugepages,
         'vm.max_map_count': max_map_count,
         'vm.hugetlb_shm_group': gid,
     }
+    if set_shmmax:
+        shmmax_current = int(check_output(['sysctl', '-n', 'kernel.shmmax']))
+        shmmax_minsize = bytes_from_string(pagesize) * nr_hugepages
+        if shmmax_minsize > shmmax_current:
+            sysctl_settings['kernel.shmmax'] = shmmax_minsize
     sysctl.create(yaml.dump(sysctl_settings), '/etc/sysctl.d/10-hugepage.conf')
     mkdir(mnt_point, owner='root', group='root', perms=0o755, force=False)
     lfstab = fstab.Fstab()

hooks/charmhelpers/core/kernel.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# 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/>.
+
+__author__ = "Jorge Niedbalski <jorge.niedbalski@canonical.com>"
+
+from charmhelpers.core.hookenv import (
+    log,
+    INFO
+)
+
+from subprocess import check_call, check_output
+import re
+
+
+def modprobe(module, persist=True):
+    """Load a kernel module and configure for auto-load on reboot."""
+    cmd = ['modprobe', module]
+
+    log('Loading kernel module %s' % module, level=INFO)
+
+    check_call(cmd)
+    if persist:
+        with open('/etc/modules', 'r+') as modules:
+            if module not in modules.read():
+                modules.write(module)
+
+
+def rmmod(module, force=False):
+    """Remove a module from the linux kernel"""
+    cmd = ['rmmod']
+    if force:
+        cmd.append('-f')
+    cmd.append(module)
+    log('Removing kernel module %s' % module, level=INFO)
+    return check_call(cmd)
+
+
+def lsmod():
+    """Shows what kernel modules are currently loaded"""
+    return check_output(['lsmod'],
+                        universal_newlines=True)
+
+
+def is_module_loaded(module):
+    """Checks if a kernel module is already loaded"""
+    matches = re.findall('^%s[ ]+' % module, lsmod(), re.M)
+    return len(matches) > 0
+
+
+def update_initramfs(version='all'):
+    """Updates an initramfs image"""
+    return check_call(["update-initramfs", "-k", version, "-u"])

hooks/charmhelpers/core/services/helpers.py

@@ -243,33 +243,40 @@ class TemplateCallback(ManagerCallback):
     :param str source: The template source file, relative to
         `$CHARM_DIR/templates`
 
-    :param str target: The target to write the rendered template to
+    :param str target: The target to write the rendered template to (or None)
     :param str owner: The owner of the rendered file
     :param str group: The group of the rendered file
     :param int perms: The permissions of the rendered file
     :param partial on_change_action: functools partial to be executed when
                                      rendered file changes
+    :param jinja2 loader template_loader: A jinja2 template loader
+
+    :return str: The rendered template
     """
     def __init__(self, source, target,
                  owner='root', group='root', perms=0o444,
-                 on_change_action=None):
+                 on_change_action=None, template_loader=None):
         self.source = source
         self.target = target
         self.owner = owner
         self.group = group
         self.perms = perms
         self.on_change_action = on_change_action
+        self.template_loader = template_loader
 
     def __call__(self, manager, service_name, event_name):
         pre_checksum = ''
         if self.on_change_action and os.path.isfile(self.target):
             pre_checksum = host.file_hash(self.target)
         service = manager.get_service(service_name)
-        context = {}
+        context = {'ctx': {}}
         for ctx in service.get('required_data', []):
             context.update(ctx)
-        templating.render(self.source, self.target, context,
-                          self.owner, self.group, self.perms)
+            context['ctx'].update(ctx)
+
+        result = templating.render(self.source, self.target, context,
+                                   self.owner, self.group, self.perms,
+                                   template_loader=self.template_loader)
         if self.on_change_action:
             if pre_checksum == host.file_hash(self.target):
                 hookenv.log(
@@ -278,6 +285,8 @@ class TemplateCallback(ManagerCallback):
             else:
                 self.on_change_action()
 
+        return result
+
 
 # Convenience aliases for templates
 render_template = template = TemplateCallback

hooks/charmhelpers/core/strutils.py

@@ -18,6 +18,7 @@
 # along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
 
 import six
+import re
 
 
 def bool_from_string(value):
@@ -40,3 +41,32 @@ def bool_from_string(value):
 
     msg = "Unable to interpret string value '%s' as boolean" % (value)
     raise ValueError(msg)
+
+
+def bytes_from_string(value):
+    """Interpret human readable string value as bytes.
+
+    Returns int
+    """
+    BYTE_POWER = {
+        'K': 1,
+        'KB': 1,
+        'M': 2,
+        'MB': 2,
+        'G': 3,
+        'GB': 3,
+        'T': 4,
+        'TB': 4,
+        'P': 5,
+        'PB': 5,
+    }
+    if isinstance(value, six.string_types):
+        value = six.text_type(value)
+    else:
+        msg = "Unable to interpret non-string value '%s' as boolean" % (value)
+        raise ValueError(msg)
+    matches = re.match("([0-9]+)([a-zA-Z]+)", value)
+    if not matches:
+        msg = "Unable to interpret string value '%s' as bytes" % (value)
+        raise ValueError(msg)
+    return int(matches.group(1)) * (1024 ** BYTE_POWER[matches.group(2)])

hooks/charmhelpers/core/templating.py

@@ -21,13 +21,14 @@ from charmhelpers.core import hookenv
 
 
 def render(source, target, context, owner='root', group='root',
-           perms=0o444, templates_dir=None, encoding='UTF-8'):
+           perms=0o444, templates_dir=None, encoding='UTF-8', template_loader=None):
     """
     Render a template.
 
     The `source` path, if not absolute, is relative to the `templates_dir`.
 
-    The `target` path should be absolute.
+    The `target` path should be absolute.  It can also be `None`, in which
+    case no file will be written.
 
     The context should be a dict containing the values to be replaced in the
     template.
@@ -36,6 +37,9 @@ def render(source, target, context, owner='root', group='root',
 
     If omitted, `templates_dir` defaults to the `templates` folder in the charm.
 
+    The rendered template will be written to the file as well as being returned
+    as a string.
+
     Note: Using this requires python-jinja2; if it is not installed, calling
     this will attempt to use charmhelpers.fetch.apt_install to install it.
     """
@@ -52,17 +56,26 @@ def render(source, target, context, owner='root', group='root',
         apt_install('python-jinja2', fatal=True)
         from jinja2 import FileSystemLoader, Environment, exceptions
 
-    if templates_dir is None:
-        templates_dir = os.path.join(hookenv.charm_dir(), 'templates')
-    loader = Environment(loader=FileSystemLoader(templates_dir))
+    if template_loader:
+        template_env = Environment(loader=template_loader)
+    else:
+        if templates_dir is None:
+            templates_dir = os.path.join(hookenv.charm_dir(), 'templates')
+        template_env = Environment(loader=FileSystemLoader(templates_dir))
     try:
         source = source
-        template = loader.get_template(source)
+        template = template_env.get_template(source)
     except exceptions.TemplateNotFound as e:
         hookenv.log('Could not load template %s from %s.' %
                     (source, templates_dir),
                     level=hookenv.ERROR)
         raise e
     content = template.render(context)
-    host.mkdir(os.path.dirname(target), owner, group, perms=0o755)
-    host.write_file(target, content.encode(encoding), owner, group, perms)
+    if target is not None:
+        target_dir = os.path.dirname(target)
+        if not os.path.exists(target_dir):
+            # This is a terrible default directory permission, as the file
+            # or its siblings will often contain secrets.
+            host.mkdir(os.path.dirname(target), owner, group, perms=0o755)
+        host.write_file(target, content.encode(encoding), owner, group, perms)
+    return content

hooks/charmhelpers/fetch/__init__.py

@@ -98,6 +98,14 @@ CLOUD_ARCHIVE_POCKETS = {
     'liberty/proposed': 'trusty-proposed/liberty',
     'trusty-liberty/proposed': 'trusty-proposed/liberty',
     'trusty-proposed/liberty': 'trusty-proposed/liberty',
+    # Mitaka
+    'mitaka': 'trusty-updates/mitaka',
+    'trusty-mitaka': 'trusty-updates/mitaka',
+    'trusty-mitaka/updates': 'trusty-updates/mitaka',
+    'trusty-updates/mitaka': 'trusty-updates/mitaka',
+    'mitaka/proposed': 'trusty-proposed/mitaka',
+    'trusty-mitaka/proposed': 'trusty-proposed/mitaka',
+    'trusty-proposed/mitaka': 'trusty-proposed/mitaka',
 }
 
 # The order of this list is very important. Handlers should be listed in from
@@ -225,12 +233,12 @@ def apt_purge(packages, fatal=False):
 
 def apt_mark(packages, mark, fatal=False):
     """Flag one or more packages using apt-mark"""
+    log("Marking {} as {}".format(packages, mark))
     cmd = ['apt-mark', mark]
     if isinstance(packages, six.string_types):
         cmd.append(packages)
     else:
         cmd.extend(packages)
-    log("Holding {}".format(packages))
 
     if fatal:
         subprocess.check_call(cmd, universal_newlines=True)
@@ -411,7 +419,7 @@ def plugins(fetch_handlers=None):
                 importlib.import_module(package),
                 classname)
             plugin_list.append(handler_class())
-        except (ImportError, AttributeError):
+        except NotImplementedError:
             # Skip missing plugins so that they can be ommitted from
             # installation if desired
             log("FetchHandler {} not found, skipping plugin".format(

hooks/charmhelpers/fetch/archiveurl.py

@@ -108,7 +108,7 @@ class ArchiveUrlFetchHandler(BaseFetchHandler):
                 install_opener(opener)
         response = urlopen(source)
         try:
-            with open(dest, 'w') as dest_file:
+            with open(dest, 'wb') as dest_file:
                 dest_file.write(response.read())
         except Exception as e:
             if os.path.isfile(dest):

hooks/charmhelpers/fetch/bzrurl.py

@@ -15,60 +15,50 @@
 # along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+from subprocess import check_call
 from charmhelpers.fetch import (
     BaseFetchHandler,
-    UnhandledSource
+    UnhandledSource,
+    filter_installed_packages,
+    apt_install,
 )
 from charmhelpers.core.host import mkdir
 
-import six
-if six.PY3:
-    raise ImportError('bzrlib does not support Python3')
 
-try:
-    from bzrlib.branch import Branch
-    from bzrlib import bzrdir, workingtree, errors
-except ImportError:
-    from charmhelpers.fetch import apt_install
-    apt_install("python-bzrlib")
-    from bzrlib.branch import Branch
-    from bzrlib import bzrdir, workingtree, errors
+if filter_installed_packages(['bzr']) != []:
+    apt_install(['bzr'])
+    if filter_installed_packages(['bzr']) != []:
+        raise NotImplementedError('Unable to install bzr')
 
 
 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'):
+        if url_parts.scheme not in ('bzr+ssh', 'lp', ''):
             return False
+        elif not url_parts.scheme:
+            return os.path.exists(os.path.join(source, '.bzr'))
         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:
-            local_branch = bzrdir.BzrDir.create_branch_convenience(dest)
-        except errors.AlreadyControlDirError:
-            local_branch = Branch.open(dest)
-        try:
-            remote_branch = Branch.open(source)
-            remote_branch.push(local_branch)
-            tree = workingtree.WorkingTree.open(dest)
-            tree.update()
-        except Exception as e:
-            raise e
+        if os.path.exists(dest):
+            check_call(['bzr', 'pull', '--overwrite', '-d', dest, source])
+        else:
+            check_call(['bzr', 'branch', source, dest])
 
-    def install(self, source):
+    def install(self, source, dest=None):
         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 dest:
+            dest_dir = os.path.join(dest, branch_name)
+        else:
+            dest_dir = os.path.join(os.environ.get('CHARM_DIR'), "fetched",
+                                    branch_name)
+
         if not os.path.exists(dest_dir):
             mkdir(dest_dir, perms=0o755)
         try:

hooks/charmhelpers/fetch/giturl.py

@@ -15,24 +15,18 @@
 # along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
+from subprocess import check_call, CalledProcessError
 from charmhelpers.fetch import (
     BaseFetchHandler,
-    UnhandledSource
+    UnhandledSource,
+    filter_installed_packages,
+    apt_install,
 )
-from charmhelpers.core.host import mkdir
 
-import six
-if six.PY3:
-    raise ImportError('GitPython does not support Python 3')
-
-try:
-    from git import Repo
-except ImportError:
-    from charmhelpers.fetch import apt_install
-    apt_install("python-git")
-    from git import Repo
-
-from git.exc import GitCommandError  # noqa E402
+if filter_installed_packages(['git']) != []:
+    apt_install(['git'])
+    if filter_installed_packages(['git']) != []:
+        raise NotImplementedError('Unable to install git')
 
 
 class GitUrlFetchHandler(BaseFetchHandler):
@@ -40,19 +34,24 @@ class GitUrlFetchHandler(BaseFetchHandler):
     def can_handle(self, source):
         url_parts = self.parse_url(source)
         # TODO (mattyw) no support for ssh git@ yet
-        if url_parts.scheme not in ('http', 'https', 'git'):
+        if url_parts.scheme not in ('http', 'https', 'git', ''):
             return False
+        elif not url_parts.scheme:
+            return os.path.exists(os.path.join(source, '.git'))
         else:
             return True
 
-    def clone(self, source, dest, branch, depth=None):
+    def clone(self, source, dest, branch="master", depth=None):
         if not self.can_handle(source):
             raise UnhandledSource("Cannot handle {}".format(source))
 
-        if depth:
-            Repo.clone_from(source, dest, branch=branch, depth=depth)
+        if os.path.exists(dest):
+            cmd = ['git', '-C', dest, 'pull', source, branch]
         else:
-            Repo.clone_from(source, dest, branch=branch)
+            cmd = ['git', 'clone', source, dest, '--branch', branch]
+            if depth:
+                cmd.extend(['--depth', depth])
+        check_call(cmd)
 
     def install(self, source, branch="master", dest=None, depth=None):
         url_parts = self.parse_url(source)
@@ -62,11 +61,9 @@ class GitUrlFetchHandler(BaseFetchHandler):
         else:
             dest_dir = os.path.join(os.environ.get('CHARM_DIR'), "fetched",
                                     branch_name)
-        if not os.path.exists(dest_dir):
-            mkdir(dest_dir, perms=0o755)
         try:
             self.clone(source, dest_dir, branch, depth)
-        except GitCommandError as e:
+        except CalledProcessError as e:
             raise UnhandledSource(e)
         except OSError as e:
             raise UnhandledSource(e.strerror)

tests/charmhelpers/contrib/amulet/deployment.py

@@ -51,7 +51,8 @@ class AmuletDeployment(object):
         if 'units' not in this_service:
             this_service['units'] = 1
 
-        self.d.add(this_service['name'], units=this_service['units'])
+        self.d.add(this_service['name'], units=this_service['units'],
+                   constraints=this_service.get('constraints'))
 
         for svc in other_services:
             if 'location' in svc:
@@ -64,7 +65,8 @@ class AmuletDeployment(object):
             if 'units' not in svc:
                 svc['units'] = 1
 
-            self.d.add(svc['name'], charm=branch_location, units=svc['units'])
+            self.d.add(svc['name'], charm=branch_location, units=svc['units'],
+                       constraints=svc.get('constraints'))
 
     def _add_relations(self, relations):
         """Add all of the relations for the services."""

tests/charmhelpers/contrib/amulet/utils.py

@@ -19,9 +19,11 @@ import json
 import logging
 import os
 import re
+import socket
 import subprocess
 import sys
 import time
+import uuid
 
 import amulet
 import distro_info
@@ -114,7 +116,7 @@ class AmuletUtils(object):
         # /!\ DEPRECATION WARNING (beisner):
         # New and existing tests should be rewritten to use
         # validate_services_by_name() as it is aware of init systems.
-        self.log.warn('/!\\ DEPRECATION WARNING:  use '
+        self.log.warn('DEPRECATION WARNING:  use '
                       'validate_services_by_name instead of validate_services '
                       'due to init system differences.')
 
@@ -269,33 +271,52 @@ class AmuletUtils(object):
         """Get last modification time of directory."""
         return sentry_unit.directory_stat(directory)['mtime']
 
-    def _get_proc_start_time(self, sentry_unit, service, pgrep_full=False):
-        """Get process' start time.
+    def _get_proc_start_time(self, sentry_unit, service, pgrep_full=None):
+        """Get start time of a process based on the last modification time
+           of the /proc/pid directory.
 
-           Determine start time of the process based on the last modification
-           time of the /proc/pid directory. If pgrep_full is True, the process
-           name is matched against the full command line.
-           """
-        if pgrep_full:
-            cmd = 'pgrep -o -f {}'.format(service)
-        else:
-            cmd = 'pgrep -o {}'.format(service)
-        cmd = cmd + '  | grep  -v pgrep || exit 0'
-        cmd_out = sentry_unit.run(cmd)
-        self.log.debug('CMDout: ' + str(cmd_out))
-        if cmd_out[0]:
-            self.log.debug('Pid for %s %s' % (service, str(cmd_out[0])))
-            proc_dir = '/proc/{}'.format(cmd_out[0].strip())
-            return self._get_dir_mtime(sentry_unit, proc_dir)
+        :sentry_unit:  The sentry unit to check for the service on
+        :service:  service name to look for in process table
+        :pgrep_full:  [Deprecated] Use full command line search mode with pgrep
+        :returns:  epoch time of service process start
+        :param commands:  list of bash commands
+        :param sentry_units:  list of sentry unit pointers
+        :returns:  None if successful; Failure message otherwise
+        """
+        if pgrep_full is not None:
+            # /!\ DEPRECATION WARNING (beisner):
+            # No longer implemented, as pidof is now used instead of pgrep.
+            # https://bugs.launchpad.net/charm-helpers/+bug/1474030
+            self.log.warn('DEPRECATION WARNING:  pgrep_full bool is no '
+                          'longer implemented re: lp 1474030.')
+
+        pid_list = self.get_process_id_list(sentry_unit, service)
+        pid = pid_list[0]
+        proc_dir = '/proc/{}'.format(pid)
+        self.log.debug('Pid for {} on {}: {}'.format(
+            service, sentry_unit.info['unit_name'], pid))
+
+        return self._get_dir_mtime(sentry_unit, proc_dir)
 
     def service_restarted(self, sentry_unit, service, filename,
-                          pgrep_full=False, sleep_time=20):
+                          pgrep_full=None, sleep_time=20):
         """Check if service was restarted.
 
            Compare a service's start time vs a file's last modification time
            (such as a config file for that service) to determine if the service
            has been restarted.
            """
+        # /!\ DEPRECATION WARNING (beisner):
+        # This method is prone to races in that no before-time is known.
+        # Use validate_service_config_changed instead.
+
+        # NOTE(beisner) pgrep_full is no longer implemented, as pidof is now
+        # used instead of pgrep.  pgrep_full is still passed through to ensure
+        # deprecation WARNS.  lp1474030
+        self.log.warn('DEPRECATION WARNING:  use '
+                      'validate_service_config_changed instead of '
+                      'service_restarted due to known races.')
+
         time.sleep(sleep_time)
         if (self._get_proc_start_time(sentry_unit, service, pgrep_full) >=
                 self._get_file_mtime(sentry_unit, filename)):
@@ -304,78 +325,122 @@ class AmuletUtils(object):
             return False
 
     def service_restarted_since(self, sentry_unit, mtime, service,
-                                pgrep_full=False, sleep_time=20,
-                                retry_count=2):
+                                pgrep_full=None, sleep_time=20,
+                                retry_count=30, retry_sleep_time=10):
         """Check if service was been started after a given time.
 
         Args:
           sentry_unit (sentry): The sentry unit to check for the service on
           mtime (float): The epoch time to check against
           service (string): service name to look for in process table
-          pgrep_full (boolean): Use full command line search mode with pgrep
-          sleep_time (int): Seconds to sleep before looking for process
-          retry_count (int): If service is not found, how many times to retry
+          pgrep_full: [Deprecated] Use full command line search mode with pgrep
+          sleep_time (int): Initial sleep time (s) before looking for file
+          retry_sleep_time (int): Time (s) to sleep between retries
+          retry_count (int): If file is not found, how many times to retry
 
         Returns:
           bool: True if service found and its start time it newer than mtime,
                 False if service is older than mtime or if service was
                 not found.
         """
-        self.log.debug('Checking %s restarted since %s' % (service, mtime))
+        # NOTE(beisner) pgrep_full is no longer implemented, as pidof is now
+        # used instead of pgrep.  pgrep_full is still passed through to ensure
+        # deprecation WARNS.  lp1474030
+
+        unit_name = sentry_unit.info['unit_name']
+        self.log.debug('Checking that %s service restarted since %s on '
+                       '%s' % (service, mtime, unit_name))
         time.sleep(sleep_time)
-        proc_start_time = self._get_proc_start_time(sentry_unit, service,
-                                                    pgrep_full)
-        while retry_count > 0 and not proc_start_time:
-            self.log.debug('No pid file found for service %s, will retry %i '
-                           'more times' % (service, retry_count))
-            time.sleep(30)
-            proc_start_time = self._get_proc_start_time(sentry_unit, service,
-                                                        pgrep_full)
-            retry_count = retry_count - 1
+        proc_start_time = None
+        tries = 0
+        while tries <= retry_count and not proc_start_time:
+            try:
+                proc_start_time = self._get_proc_start_time(sentry_unit,
+                                                            service,
+                                                            pgrep_full)
+                self.log.debug('Attempt {} to get {} proc start time on {} '
+                               'OK'.format(tries, service, unit_name))
+            except IOError as e:
+                # NOTE(beisner) - race avoidance, proc may not exist yet.
+                # https://bugs.launchpad.net/charm-helpers/+bug/1474030
+                self.log.debug('Attempt {} to get {} proc start time on {} '
+                               'failed\n{}'.format(tries, service,
+                                                   unit_name, e))
+                time.sleep(retry_sleep_time)
+                tries += 1
 
         if not proc_start_time:
             self.log.warn('No proc start time found, assuming service did '
                           'not start')
             return False
         if proc_start_time >= mtime:
-            self.log.debug('proc start time is newer than provided mtime'
-                           '(%s >= %s)' % (proc_start_time, mtime))
+            self.log.debug('Proc start time is newer than provided mtime'
+                           '(%s >= %s) on %s (OK)' % (proc_start_time,
+                                                      mtime, unit_name))
             return True
         else:
-            self.log.warn('proc start time (%s) is older than provided mtime '
-                          '(%s), service did not restart' % (proc_start_time,
-                                                             mtime))
+            self.log.warn('Proc start time (%s) is older than provided mtime '
+                          '(%s) on %s, service did not '
+                          'restart' % (proc_start_time, mtime, unit_name))
             return False
 
     def config_updated_since(self, sentry_unit, filename, mtime,
-                             sleep_time=20):
+                             sleep_time=20, retry_count=30,
+                             retry_sleep_time=10):
         """Check if file was modified after a given time.
 
         Args:
           sentry_unit (sentry): The sentry unit to check the file mtime on
           filename (string): The file to check mtime of
           mtime (float): The epoch time to check against
-          sleep_time (int): Seconds to sleep before looking for process
+          sleep_time (int): Initial sleep time (s) before looking for file
+          retry_sleep_time (int): Time (s) to sleep between retries
+          retry_count (int): If file is not found, how many times to retry
 
         Returns:
           bool: True if file was modified more recently than mtime, False if
-                file was modified before mtime,
+                file was modified before mtime, or if file not found.
         """
-        self.log.debug('Checking %s updated since %s' % (filename, mtime))
+        unit_name = sentry_unit.info['unit_name']
+        self.log.debug('Checking that %s updated since %s on '
+                       '%s' % (filename, mtime, unit_name))
         time.sleep(sleep_time)
-        file_mtime = self._get_file_mtime(sentry_unit, filename)
+        file_mtime = None
+        tries = 0
+        while tries <= retry_count and not file_mtime:
+            try:
+                file_mtime = self._get_file_mtime(sentry_unit, filename)
+                self.log.debug('Attempt {} to get {} file mtime on {} '
+                               'OK'.format(tries, filename, unit_name))
+            except IOError as e:
+                # NOTE(beisner) - race avoidance, file may not exist yet.
+                # https://bugs.launchpad.net/charm-helpers/+bug/1474030
+                self.log.debug('Attempt {} to get {} file mtime on {} '
+                               'failed\n{}'.format(tries, filename,
+                                                   unit_name, e))
+                time.sleep(retry_sleep_time)
+                tries += 1
+
+        if not file_mtime:
+            self.log.warn('Could not determine file mtime, assuming '
+                          'file does not exist')
+            return False
+
         if file_mtime >= mtime:
             self.log.debug('File mtime is newer than provided mtime '
-                           '(%s >= %s)' % (file_mtime, mtime))
+                           '(%s >= %s) on %s (OK)' % (file_mtime,
+                                                      mtime, unit_name))
             return True
         else:
-            self.log.warn('File mtime %s is older than provided mtime %s'
-                          % (file_mtime, mtime))
+            self.log.warn('File mtime is older than provided mtime'
+                          '(%s < on %s) on %s' % (file_mtime,
+                                                  mtime, unit_name))
             return False
 
     def validate_service_config_changed(self, sentry_unit, mtime, service,
-                                        filename, pgrep_full=False,
-                                        sleep_time=20, retry_count=2):
+                                        filename, pgrep_full=None,
+                                        sleep_time=20, retry_count=30,
+                                        retry_sleep_time=10):
         """Check service and file were updated after mtime
 
         Args:
@@ -383,9 +448,10 @@ class AmuletUtils(object):
           mtime (float): The epoch time to check against
           service (string): service name to look for in process table
           filename (string): The file to check mtime of
-          pgrep_full (boolean): Use full command line search mode with pgrep
-          sleep_time (int): Seconds to sleep before looking for process
+          pgrep_full: [Deprecated] Use full command line search mode with pgrep
+          sleep_time (int): Initial sleep in seconds to pass to test helpers
           retry_count (int): If service is not found, how many times to retry
+          retry_sleep_time (int): Time in seconds to wait between retries
 
         Typical Usage:
             u = OpenStackAmuletUtils(ERROR)
@@ -402,15 +468,27 @@ class AmuletUtils(object):
                 mtime, False if service is older than mtime or if service was
                 not found or if filename was modified before mtime.
         """
-        self.log.debug('Checking %s restarted since %s' % (service, mtime))
-        time.sleep(sleep_time)
-        service_restart = self.service_restarted_since(sentry_unit, mtime,
-                                                       service,
-                                                       pgrep_full=pgrep_full,
-                                                       sleep_time=0,
-                                                       retry_count=retry_count)
-        config_update = self.config_updated_since(sentry_unit, filename, mtime,
-                                                  sleep_time=0)
+
+        # NOTE(beisner) pgrep_full is no longer implemented, as pidof is now
+        # used instead of pgrep.  pgrep_full is still passed through to ensure
+        # deprecation WARNS.  lp1474030
+
+        service_restart = self.service_restarted_since(
+            sentry_unit, mtime,
+            service,
+            pgrep_full=pgrep_full,
+            sleep_time=sleep_time,
+            retry_count=retry_count,
+            retry_sleep_time=retry_sleep_time)
+
+        config_update = self.config_updated_since(
+            sentry_unit,
+            filename,
+            mtime,
+            sleep_time=sleep_time,
+            retry_count=retry_count,
+            retry_sleep_time=retry_sleep_time)
+
         return service_restart and config_update
 
     def get_sentry_time(self, sentry_unit):
@@ -428,7 +506,6 @@ class AmuletUtils(object):
         """Return a list of all Ubuntu releases in order of release."""
         _d = distro_info.UbuntuDistroInfo()
         _release_list = _d.all
-        self.log.debug('Ubuntu release list: {}'.format(_release_list))
         return _release_list
 
     def file_to_url(self, file_rel_path):
@@ -568,6 +645,142 @@ class AmuletUtils(object):
 
         return None
 
+    def validate_sectionless_conf(self, file_contents, expected):
+        """A crude conf parser.  Useful to inspect configuration files which
+        do not have section headers (as would be necessary in order to use
+        the configparser).  Such as openstack-dashboard or rabbitmq confs."""
+        for line in file_contents.split('\n'):
+            if '=' in line:
+                args = line.split('=')
+                if len(args) <= 1:
+                    continue
+                key = args[0].strip()
+                value = args[1].strip()
+                if key in expected.keys():
+                    if expected[key] != value:
+                        msg = ('Config mismatch.  Expected, actual:  {}, '
+                               '{}'.format(expected[key], value))
+                        amulet.raise_status(amulet.FAIL, msg=msg)
+
+    def get_unit_hostnames(self, units):
+        """Return a dict of juju unit names to hostnames."""
+        host_names = {}
+        for unit in units:
+            host_names[unit.info['unit_name']] = \
+                str(unit.file_contents('/etc/hostname').strip())
+        self.log.debug('Unit host names: {}'.format(host_names))
+        return host_names
+
+    def run_cmd_unit(self, sentry_unit, cmd):
+        """Run a command on a unit, return the output and exit code."""
+        output, code = sentry_unit.run(cmd)
+        if code == 0:
+            self.log.debug('{} `{}` command returned {} '
+                           '(OK)'.format(sentry_unit.info['unit_name'],
+                                         cmd, code))
+        else:
+            msg = ('{} `{}` command returned {} '
+                   '{}'.format(sentry_unit.info['unit_name'],
+                               cmd, code, output))
+            amulet.raise_status(amulet.FAIL, msg=msg)
+        return str(output), code
+
+    def file_exists_on_unit(self, sentry_unit, file_name):
+        """Check if a file exists on a unit."""
+        try:
+            sentry_unit.file_stat(file_name)
+            return True
+        except IOError:
+            return False
+        except Exception as e:
+            msg = 'Error checking file {}: {}'.format(file_name, e)
+            amulet.raise_status(amulet.FAIL, msg=msg)
+
+    def file_contents_safe(self, sentry_unit, file_name,
+                           max_wait=60, fatal=False):
+        """Get file contents from a sentry unit.  Wrap amulet file_contents
+        with retry logic to address races where a file checks as existing,
+        but no longer exists by the time file_contents is called.
+        Return None if file not found. Optionally raise if fatal is True."""
+        unit_name = sentry_unit.info['unit_name']
+        file_contents = False
+        tries = 0
+        while not file_contents and tries < (max_wait / 4):
+            try:
+                file_contents = sentry_unit.file_contents(file_name)
+            except IOError:
+                self.log.debug('Attempt {} to open file {} from {} '
+                               'failed'.format(tries, file_name,
+                                               unit_name))
+                time.sleep(4)
+                tries += 1
+
+        if file_contents:
+            return file_contents
+        elif not fatal:
+            return None
+        elif fatal:
+            msg = 'Failed to get file contents from unit.'
+            amulet.raise_status(amulet.FAIL, msg)
+
+    def port_knock_tcp(self, host="localhost", port=22, timeout=15):
+        """Open a TCP socket to check for a listening sevice on a host.
+
+        :param host: host name or IP address, default to localhost
+        :param port: TCP port number, default to 22
+        :param timeout: Connect timeout, default to 15 seconds
+        :returns: True if successful, False if connect failed
+        """
+
+        # Resolve host name if possible
+        try:
+            connect_host = socket.gethostbyname(host)
+            host_human = "{} ({})".format(connect_host, host)
+        except socket.error as e:
+            self.log.warn('Unable to resolve address: '
+                          '{} ({}) Trying anyway!'.format(host, e))
+            connect_host = host
+            host_human = connect_host
+
+        # Attempt socket connection
+        try:
+            knock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            knock.settimeout(timeout)
+            knock.connect((connect_host, port))
+            knock.close()
+            self.log.debug('Socket connect OK for host '
+                           '{} on port {}.'.format(host_human, port))
+            return True
+        except socket.error as e:
+            self.log.debug('Socket connect FAIL for'
+                           ' {} port {} ({})'.format(host_human, port, e))
+            return False
+
+    def port_knock_units(self, sentry_units, port=22,
+                         timeout=15, expect_success=True):
+        """Open a TCP socket to check for a listening sevice on each
+        listed juju unit.
+
+        :param sentry_units: list of sentry unit pointers
+        :param port: TCP port number, default to 22
+        :param timeout: Connect timeout, default to 15 seconds
+        :expect_success: True by default, set False to invert logic
+        :returns: None if successful, Failure message otherwise
+        """
+        for unit in sentry_units:
+            host = unit.info['public-address']
+            connected = self.port_knock_tcp(host, port, timeout)
+            if not connected and expect_success:
+                return 'Socket connect failed.'
+            elif connected and not expect_success:
+                return 'Socket connected unexpectedly.'
+
+    def get_uuid_epoch_stamp(self):
+        """Returns a stamp string based on uuid4 and epoch time.  Useful in
+        generating test messages which need to be unique-ish."""
+        return '[{}-{}]'.format(uuid.uuid4(), time.time())
+
+# amulet juju action helpers:
     def run_action(self, unit_sentry, action,
                    _check_output=subprocess.check_output):
         """Run the named action on a given unit sentry.
@@ -594,3 +807,12 @@ class AmuletUtils(object):
         output = _check_output(command, universal_newlines=True)
         data = json.loads(output)
         return data.get(u"status") == "completed"
+
+    def status_get(self, unit):
+        """Return the current service status of this unit."""
+        raw_status, return_code = unit.run(
+            "status-get --format=json --include-data")
+        if return_code != 0:
+            return ("unknown", "")
+        status = json.loads(raw_status)
+        return (status["status"], status["message"])

tests/charmhelpers/contrib/openstack/amulet/deployment.py

@@ -14,12 +14,18 @@
 # 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/>.
 
+import logging
+import re
+import sys
 import six
 from collections import OrderedDict
 from charmhelpers.contrib.amulet.deployment import (
     AmuletDeployment
 )
 
+DEBUG = logging.DEBUG
+ERROR = logging.ERROR
+
 
 class OpenStackAmuletDeployment(AmuletDeployment):
     """OpenStack amulet deployment.
@@ -28,9 +34,12 @@ class OpenStackAmuletDeployment(AmuletDeployment):
        that is specifically for use by OpenStack charms.
        """
 
-    def __init__(self, series=None, openstack=None, source=None, stable=True):
+    def __init__(self, series=None, openstack=None, source=None,
+                 stable=True, log_level=DEBUG):
         """Initialize the deployment environment."""
         super(OpenStackAmuletDeployment, self).__init__(series)
+        self.log = self.get_logger(level=log_level)
+        self.log.info('OpenStackAmuletDeployment:  init')
         self.openstack = openstack
         self.source = source
         self.stable = stable
@@ -38,26 +47,55 @@ class OpenStackAmuletDeployment(AmuletDeployment):
         # out.
         self.current_next = "trusty"
 
+    def get_logger(self, name="deployment-logger", level=logging.DEBUG):
+        """Get a logger object that will log to stdout."""
+        log = logging
+        logger = log.getLogger(name)
+        fmt = log.Formatter("%(asctime)s %(funcName)s "
+                            "%(levelname)s: %(message)s")
+
+        handler = log.StreamHandler(stream=sys.stdout)
+        handler.setLevel(level)
+        handler.setFormatter(fmt)
+
+        logger.addHandler(handler)
+        logger.setLevel(level)
+
+        return logger
+
     def _determine_branch_locations(self, other_services):
         """Determine the branch locations for the other services.
 
            Determine if the local branch being tested is derived from its
            stable or next (dev) branch, and based on this, use the corresonding
            stable or next branches for the other_services."""
+
+        self.log.info('OpenStackAmuletDeployment:  determine branch locations')
+
+        # Charms outside the lp:~openstack-charmers namespace
         base_charms = ['mysql', 'mongodb', 'nrpe']
 
+        # Force these charms to current series even when using an older series.
+        # ie. Use trusty/nrpe even when series is precise, as the P charm
+        # does not possess the necessary external master config and hooks.
+        force_series_current = ['nrpe']
+
         if self.series in ['precise', 'trusty']:
             base_series = self.series
         else:
             base_series = self.current_next
 
-        if self.stable:
-            for svc in other_services:
+        for svc in other_services:
+            if svc['name'] in force_series_current:
+                base_series = self.current_next
+            # If a location has been explicitly set, use it
+            if svc.get('location'):
+                continue
+            if self.stable:
                 temp = 'lp:charms/{}/{}'
                 svc['location'] = temp.format(base_series,
                                               svc['name'])
-        else:
-            for svc in other_services:
+            else:
                 if svc['name'] in base_charms:
                     temp = 'lp:charms/{}/{}'
                     svc['location'] = temp.format(base_series,
@@ -66,10 +104,13 @@ class OpenStackAmuletDeployment(AmuletDeployment):
                     temp = 'lp:~openstack-charmers/charms/{}/{}/next'
                     svc['location'] = temp.format(self.current_next,
                                                   svc['name'])
+
         return other_services
 
     def _add_services(self, this_service, other_services):
         """Add services to the deployment and set openstack-origin/source."""
+        self.log.info('OpenStackAmuletDeployment:  adding services')
+
         other_services = self._determine_branch_locations(other_services)
 
         super(OpenStackAmuletDeployment, self)._add_services(this_service,
@@ -77,29 +118,103 @@ class OpenStackAmuletDeployment(AmuletDeployment):
 
         services = other_services
         services.append(this_service)
+
+        # Charms which should use the source config option
         use_source = ['mysql', 'mongodb', 'rabbitmq-server', 'ceph',
                       'ceph-osd', 'ceph-radosgw']
-        # Most OpenStack subordinate charms do not expose an origin option
-        # as that is controlled by the principle.
-        ignore = ['cinder-ceph', 'hacluster', 'neutron-openvswitch', 'nrpe']
+
+        # Charms which can not use openstack-origin, ie. many subordinates
+        no_origin = ['cinder-ceph', 'hacluster', 'neutron-openvswitch', 'nrpe',
+                     'openvswitch-odl', 'neutron-api-odl', 'odl-controller',
+                     'cinder-backup']
 
         if self.openstack:
             for svc in services:
-                if svc['name'] not in use_source + ignore:
+                if svc['name'] not in use_source + no_origin:
                     config = {'openstack-origin': self.openstack}
                     self.d.configure(svc['name'], config)
 
         if self.source:
             for svc in services:
-                if svc['name'] in use_source and svc['name'] not in ignore:
+                if svc['name'] in use_source and svc['name'] not in no_origin:
                     config = {'source': self.source}
                     self.d.configure(svc['name'], config)
 
     def _configure_services(self, configs):
         """Configure all of the services."""
+        self.log.info('OpenStackAmuletDeployment:  configure services')
         for service, config in six.iteritems(configs):
             self.d.configure(service, config)
 
+    def _auto_wait_for_status(self, message=None, exclude_services=None,
+                              include_only=None, timeout=1800):
+        """Wait for all units to have a specific extended status, except
+        for any defined as excluded.  Unless specified via message, any
+        status containing any case of 'ready' will be considered a match.
+
+        Examples of message usage:
+
+          Wait for all unit status to CONTAIN any case of 'ready' or 'ok':
+              message = re.compile('.*ready.*|.*ok.*', re.IGNORECASE)
+
+          Wait for all units to reach this status (exact match):
+              message = re.compile('^Unit is ready and clustered$')
+
+          Wait for all units to reach any one of these (exact match):
+              message = re.compile('Unit is ready|OK|Ready')
+
+          Wait for at least one unit to reach this status (exact match):
+              message = {'ready'}
+
+        See Amulet's sentry.wait_for_messages() for message usage detail.
+        https://github.com/juju/amulet/blob/master/amulet/sentry.py
+
+        :param message: Expected status match
+        :param exclude_services: List of juju service names to ignore,
+            not to be used in conjuction with include_only.
+        :param include_only: List of juju service names to exclusively check,
+            not to be used in conjuction with exclude_services.
+        :param timeout: Maximum time in seconds to wait for status match
+        :returns: None.  Raises if timeout is hit.
+        """
+        self.log.info('Waiting for extended status on units...')
+
+        all_services = self.d.services.keys()
+
+        if exclude_services and include_only:
+            raise ValueError('exclude_services can not be used '
+                             'with include_only')
+
+        if message:
+            if isinstance(message, re._pattern_type):
+                match = message.pattern
+            else:
+                match = message
+
+            self.log.debug('Custom extended status wait match: '
+                           '{}'.format(match))
+        else:
+            self.log.debug('Default extended status wait match:  contains '
+                           'READY (case-insensitive)')
+            message = re.compile('.*ready.*', re.IGNORECASE)
+
+        if exclude_services:
+            self.log.debug('Excluding services from extended status match: '
+                           '{}'.format(exclude_services))
+        else:
+            exclude_services = []
+
+        if include_only:
+            services = include_only
+        else:
+            services = list(set(all_services) - set(exclude_services))
+
+        self.log.debug('Waiting up to {}s for extended status on services: '
+                       '{}'.format(timeout, services))
+        service_messages = {service: message for service in services}
+        self.d.sentry.wait_for_messages(service_messages, timeout=timeout)
+        self.log.info('OK')
+
     def _get_openstack_release(self):
         """Get openstack release.
 
@@ -111,7 +226,8 @@ class OpenStackAmuletDeployment(AmuletDeployment):
          self.precise_havana, self.precise_icehouse,
          self.trusty_icehouse, self.trusty_juno, self.utopic_juno,
          self.trusty_kilo, self.vivid_kilo, self.trusty_liberty,
-         self.wily_liberty) = range(12)
+         self.wily_liberty, self.trusty_mitaka,
+         self.xenial_mitaka) = range(14)
 
         releases = {
             ('precise', None): self.precise_essex,
@@ -123,9 +239,11 @@ class OpenStackAmuletDeployment(AmuletDeployment):
             ('trusty', 'cloud:trusty-juno'): self.trusty_juno,
             ('trusty', 'cloud:trusty-kilo'): self.trusty_kilo,
             ('trusty', 'cloud:trusty-liberty'): self.trusty_liberty,
+            ('trusty', 'cloud:trusty-mitaka'): self.trusty_mitaka,
             ('utopic', None): self.utopic_juno,
             ('vivid', None): self.vivid_kilo,
-            ('wily', None): self.wily_liberty}
+            ('wily', None): self.wily_liberty,
+            ('xenial', None): self.xenial_mitaka}
         return releases[(self.series, self.openstack)]
 
     def _get_openstack_release_string(self):
@@ -142,6 +260,7 @@ class OpenStackAmuletDeployment(AmuletDeployment):
             ('utopic', 'juno'),
             ('vivid', 'kilo'),
             ('wily', 'liberty'),
+            ('xenial', 'mitaka'),
         ])
         if self.openstack:
             os_origin = self.openstack.split(':')[1]

tests/charmhelpers/contrib/openstack/amulet/utils.py

@@ -18,6 +18,7 @@ import amulet
 import json
 import logging
 import os
+import re
 import six
 import time
 import urllib
@@ -27,6 +28,7 @@ import glanceclient.v1.client as glance_client
 import heatclient.v1.client as heat_client
 import keystoneclient.v2_0 as keystone_client
 import novaclient.v1_1.client as nova_client
+import pika
 import swiftclient
 
 from charmhelpers.contrib.amulet.utils import (
@@ -602,3 +604,382 @@ class OpenStackAmuletUtils(AmuletUtils):
             self.log.debug('Ceph {} samples (OK): '
                            '{}'.format(sample_type, samples))
             return None
+
+    # rabbitmq/amqp specific helpers:
+
+    def rmq_wait_for_cluster(self, deployment, init_sleep=15, timeout=1200):
+        """Wait for rmq units extended status to show cluster readiness,
+        after an optional initial sleep period.  Initial sleep is likely
+        necessary to be effective following a config change, as status
+        message may not instantly update to non-ready."""
+
+        if init_sleep:
+            time.sleep(init_sleep)
+
+        message = re.compile('^Unit is ready and clustered$')
+        deployment._auto_wait_for_status(message=message,
+                                         timeout=timeout,
+                                         include_only=['rabbitmq-server'])
+
+    def add_rmq_test_user(self, sentry_units,
+                          username="testuser1", password="changeme"):
+        """Add a test user via the first rmq juju unit, check connection as
+        the new user against all sentry units.
+
+        :param sentry_units: list of sentry unit pointers
+        :param username: amqp user name, default to testuser1
+        :param password: amqp user password
+        :returns: None if successful.  Raise on error.
+        """
+        self.log.debug('Adding rmq user ({})...'.format(username))
+
+        # Check that user does not already exist
+        cmd_user_list = 'rabbitmqctl list_users'
+        output, _ = self.run_cmd_unit(sentry_units[0], cmd_user_list)
+        if username in output:
+            self.log.warning('User ({}) already exists, returning '
+                             'gracefully.'.format(username))
+            return
+
+        perms = '".*" ".*" ".*"'
+        cmds = ['rabbitmqctl add_user {} {}'.format(username, password),
+                'rabbitmqctl set_permissions {} {}'.format(username, perms)]
+
+        # Add user via first unit
+        for cmd in cmds:
+            output, _ = self.run_cmd_unit(sentry_units[0], cmd)
+
+        # Check connection against the other sentry_units
+        self.log.debug('Checking user connect against units...')
+        for sentry_unit in sentry_units:
+            connection = self.connect_amqp_by_unit(sentry_unit, ssl=False,
+                                                   username=username,
+                                                   password=password)
+            connection.close()
+
+    def delete_rmq_test_user(self, sentry_units, username="testuser1"):
+        """Delete a rabbitmq user via the first rmq juju unit.
+
+        :param sentry_units: list of sentry unit pointers
+        :param username: amqp user name, default to testuser1
+        :param password: amqp user password
+        :returns: None if successful or no such user.
+        """
+        self.log.debug('Deleting rmq user ({})...'.format(username))
+
+        # Check that the user exists
+        cmd_user_list = 'rabbitmqctl list_users'
+        output, _ = self.run_cmd_unit(sentry_units[0], cmd_user_list)
+
+        if username not in output:
+            self.log.warning('User ({}) does not exist, returning '
+                             'gracefully.'.format(username))
+            return
+
+        # Delete the user
+        cmd_user_del = 'rabbitmqctl delete_user {}'.format(username)
+        output, _ = self.run_cmd_unit(sentry_units[0], cmd_user_del)
+
+    def get_rmq_cluster_status(self, sentry_unit):
+        """Execute rabbitmq cluster status command on a unit and return
+        the full output.
+
+        :param unit: sentry unit
+        :returns: String containing console output of cluster status command
+        """
+        cmd = 'rabbitmqctl cluster_status'
+        output, _ = self.run_cmd_unit(sentry_unit, cmd)
+        self.log.debug('{} cluster_status:\n{}'.format(
+            sentry_unit.info['unit_name'], output))
+        return str(output)
+
+    def get_rmq_cluster_running_nodes(self, sentry_unit):
+        """Parse rabbitmqctl cluster_status output string, return list of
+        running rabbitmq cluster nodes.
+
+        :param unit: sentry unit
+        :returns: List containing node names of running nodes
+        """
+        # NOTE(beisner): rabbitmqctl cluster_status output is not
+        # json-parsable, do string chop foo, then json.loads that.
+        str_stat = self.get_rmq_cluster_status(sentry_unit)
+        if 'running_nodes' in str_stat:
+            pos_start = str_stat.find("{running_nodes,") + 15
+            pos_end = str_stat.find("]},", pos_start) + 1
+            str_run_nodes = str_stat[pos_start:pos_end].replace("'", '"')
+            run_nodes = json.loads(str_run_nodes)
+            return run_nodes
+        else:
+            return []
+
+    def validate_rmq_cluster_running_nodes(self, sentry_units):
+        """Check that all rmq unit hostnames are represented in the
+        cluster_status output of all units.
+
+        :param host_names: dict of juju unit names to host names
+        :param units: list of sentry unit pointers (all rmq units)
+        :returns: None if successful, otherwise return error message
+        """
+        host_names = self.get_unit_hostnames(sentry_units)
+        errors = []
+
+        # Query every unit for cluster_status running nodes
+        for query_unit in sentry_units:
+            query_unit_name = query_unit.info['unit_name']
+            running_nodes = self.get_rmq_cluster_running_nodes(query_unit)
+
+            # Confirm that every unit is represented in the queried unit's
+            # cluster_status running nodes output.
+            for validate_unit in sentry_units:
+                val_host_name = host_names[validate_unit.info['unit_name']]
+                val_node_name = 'rabbit@{}'.format(val_host_name)
+
+                if val_node_name not in running_nodes:
+                    errors.append('Cluster member check failed on {}: {} not '
+                                  'in {}\n'.format(query_unit_name,
+                                                   val_node_name,
+                                                   running_nodes))
+        if errors:
+            return ''.join(errors)
+
+    def rmq_ssl_is_enabled_on_unit(self, sentry_unit, port=None):
+        """Check a single juju rmq unit for ssl and port in the config file."""
+        host = sentry_unit.info['public-address']
+        unit_name = sentry_unit.info['unit_name']
+
+        conf_file = '/etc/rabbitmq/rabbitmq.config'
+        conf_contents = str(self.file_contents_safe(sentry_unit,
+                                                    conf_file, max_wait=16))
+        # Checks
+        conf_ssl = 'ssl' in conf_contents
+        conf_port = str(port) in conf_contents
+
+        # Port explicitly checked in config
+        if port and conf_port and conf_ssl:
+            self.log.debug('SSL is enabled  @{}:{} '
+                           '({})'.format(host, port, unit_name))
+            return True
+        elif port and not conf_port and conf_ssl:
+            self.log.debug('SSL is enabled @{} but not on port {} '
+                           '({})'.format(host, port, unit_name))
+            return False
+        # Port not checked (useful when checking that ssl is disabled)
+        elif not port and conf_ssl:
+            self.log.debug('SSL is enabled  @{}:{} '
+                           '({})'.format(host, port, unit_name))
+            return True
+        elif not conf_ssl:
+            self.log.debug('SSL not enabled @{}:{} '
+                           '({})'.format(host, port, unit_name))
+            return False
+        else:
+            msg = ('Unknown condition when checking SSL status @{}:{} '
+                   '({})'.format(host, port, unit_name))
+            amulet.raise_status(amulet.FAIL, msg)
+
+    def validate_rmq_ssl_enabled_units(self, sentry_units, port=None):
+        """Check that ssl is enabled on rmq juju sentry units.
+
+        :param sentry_units: list of all rmq sentry units
+        :param port: optional ssl port override to validate
+        :returns: None if successful, otherwise return error message
+        """
+        for sentry_unit in sentry_units:
+            if not self.rmq_ssl_is_enabled_on_unit(sentry_unit, port=port):
+                return ('Unexpected condition:  ssl is disabled on unit '
+                        '({})'.format(sentry_unit.info['unit_name']))
+        return None
+
+    def validate_rmq_ssl_disabled_units(self, sentry_units):
+        """Check that ssl is enabled on listed rmq juju sentry units.
+
+        :param sentry_units: list of all rmq sentry units
+        :returns: True if successful.  Raise on error.
+        """
+        for sentry_unit in sentry_units:
+            if self.rmq_ssl_is_enabled_on_unit(sentry_unit):
+                return ('Unexpected condition:  ssl is enabled on unit '
+                        '({})'.format(sentry_unit.info['unit_name']))
+        return None
+
+    def configure_rmq_ssl_on(self, sentry_units, deployment,
+                             port=None, max_wait=60):
+        """Turn ssl charm config option on, with optional non-default
+        ssl port specification.  Confirm that it is enabled on every
+        unit.
+
+        :param sentry_units: list of sentry units
+        :param deployment: amulet deployment object pointer
+        :param port: amqp port, use defaults if None
+        :param max_wait: maximum time to wait in seconds to confirm
+        :returns: None if successful.  Raise on error.
+        """
+        self.log.debug('Setting ssl charm config option:  on')
+
+        # Enable RMQ SSL
+        config = {'ssl': 'on'}
+        if port:
+            config['ssl_port'] = port
+
+        deployment.d.configure('rabbitmq-server', config)
+
+        # Wait for unit status
+        self.rmq_wait_for_cluster(deployment)
+
+        # Confirm
+        tries = 0
+        ret = self.validate_rmq_ssl_enabled_units(sentry_units, port=port)
+        while ret and tries < (max_wait / 4):
+            time.sleep(4)
+            self.log.debug('Attempt {}: {}'.format(tries, ret))
+            ret = self.validate_rmq_ssl_enabled_units(sentry_units, port=port)
+            tries += 1
+
+        if ret:
+            amulet.raise_status(amulet.FAIL, ret)
+
+    def configure_rmq_ssl_off(self, sentry_units, deployment, max_wait=60):
+        """Turn ssl charm config option off, confirm that it is disabled
+        on every unit.
+
+        :param sentry_units: list of sentry units
+        :param deployment: amulet deployment object pointer
+        :param max_wait: maximum time to wait in seconds to confirm
+        :returns: None if successful.  Raise on error.
+        """
+        self.log.debug('Setting ssl charm config option:  off')
+
+        # Disable RMQ SSL
+        config = {'ssl': 'off'}
+        deployment.d.configure('rabbitmq-server', config)
+
+        # Wait for unit status
+        self.rmq_wait_for_cluster(deployment)
+
+        # Confirm
+        tries = 0
+        ret = self.validate_rmq_ssl_disabled_units(sentry_units)
+        while ret and tries < (max_wait / 4):
+            time.sleep(4)
+            self.log.debug('Attempt {}: {}'.format(tries, ret))
+            ret = self.validate_rmq_ssl_disabled_units(sentry_units)
+            tries += 1
+
+        if ret:
+            amulet.raise_status(amulet.FAIL, ret)
+
+    def connect_amqp_by_unit(self, sentry_unit, ssl=False,
+                             port=None, fatal=True,
+                             username="testuser1", password="changeme"):
+        """Establish and return a pika amqp connection to the rabbitmq service
+        running on a rmq juju unit.
+
+        :param sentry_unit: sentry unit pointer
+        :param ssl: boolean, default to False
+        :param port: amqp port, use defaults if None
+        :param fatal: boolean, default to True (raises on connect error)
+        :param username: amqp user name, default to testuser1
+        :param password: amqp user password
+        :returns: pika amqp connection pointer or None if failed and non-fatal
+        """
+        host = sentry_unit.info['public-address']
+        unit_name = sentry_unit.info['unit_name']
+
+        # Default port logic if port is not specified
+        if ssl and not port:
+            port = 5671
+        elif not ssl and not port:
+            port = 5672
+
+        self.log.debug('Connecting to amqp on {}:{} ({}) as '
+                       '{}...'.format(host, port, unit_name, username))
+
+        try:
+            credentials = pika.PlainCredentials(username, password)
+            parameters = pika.ConnectionParameters(host=host, port=port,
+                                                   credentials=credentials,
+                                                   ssl=ssl,
+                                                   connection_attempts=3,
+                                                   retry_delay=5,
+                                                   socket_timeout=1)
+            connection = pika.BlockingConnection(parameters)
+            assert connection.server_properties['product'] == 'RabbitMQ'
+            self.log.debug('Connect OK')
+            return connection
+        except Exception as e:
+            msg = ('amqp connection failed to {}:{} as '
+                   '{} ({})'.format(host, port, username, str(e)))
+            if fatal:
+                amulet.raise_status(amulet.FAIL, msg)
+            else:
+                self.log.warn(msg)
+                return None
+
+    def publish_amqp_message_by_unit(self, sentry_unit, message,
+                                     queue="test", ssl=False,
+                                     username="testuser1",
+                                     password="changeme",
+                                     port=None):
+        """Publish an amqp message to a rmq juju unit.
+
+        :param sentry_unit: sentry unit pointer
+        :param message: amqp message string
+        :param queue: message queue, default to test
+        :param username: amqp user name, default to testuser1
+        :param password: amqp user password
+        :param ssl: boolean, default to False
+        :param port: amqp port, use defaults if None
+        :returns: None.  Raises exception if publish failed.
+        """
+        self.log.debug('Publishing message to {} queue:\n{}'.format(queue,
+                                                                    message))
+        connection = self.connect_amqp_by_unit(sentry_unit, ssl=ssl,
+                                               port=port,
+                                               username=username,
+                                               password=password)
+
+        # NOTE(beisner): extra debug here re: pika hang potential:
+        #   https://github.com/pika/pika/issues/297
+        #   https://groups.google.com/forum/#!topic/rabbitmq-users/Ja0iyfF0Szw
+        self.log.debug('Defining channel...')
+        channel = connection.channel()
+        self.log.debug('Declaring queue...')
+        channel.queue_declare(queue=queue, auto_delete=False, durable=True)
+        self.log.debug('Publishing message...')
+        channel.basic_publish(exchange='', routing_key=queue, body=message)
+        self.log.debug('Closing channel...')
+        channel.close()
+        self.log.debug('Closing connection...')
+        connection.close()
+
+    def get_amqp_message_by_unit(self, sentry_unit, queue="test",
+                                 username="testuser1",
+                                 password="changeme",
+                                 ssl=False, port=None):
+        """Get an amqp message from a rmq juju unit.
+
+        :param sentry_unit: sentry unit pointer
+        :param queue: message queue, default to test
+        :param username: amqp user name, default to testuser1
+        :param password: amqp user password
+        :param ssl: boolean, default to False
+        :param port: amqp port, use defaults if None
+        :returns: amqp message body as string.  Raise if get fails.
+        """
+        connection = self.connect_amqp_by_unit(sentry_unit, ssl=ssl,
+                                               port=port,
+                                               username=username,
+                                               password=password)
+        channel = connection.channel()
+        method_frame, _, body = channel.basic_get(queue)
+
+        if method_frame:
+            self.log.debug('Retreived message from {} queue:\n{}'.format(queue,
+                                                                         body))
+            channel.basic_ack(method_frame.delivery_tag)
+            channel.close()
+            connection.close()
+            return body
+        else:
+            msg = 'No message retrieved.'
+            amulet.raise_status(amulet.FAIL, msg)