diff --git a/charm-helpers-tests.yaml b/charm-helpers-tests.yaml
index 395acee..48b12f6 100644
--- a/charm-helpers-tests.yaml
+++ b/charm-helpers-tests.yaml
@@ -1,4 +1,4 @@
-branch: lp:~gnuoy/charm-helpers/cisco-vpp/
+branch: lp:charm-helpers
destination: tests/charmhelpers
include:
- contrib.amulet
diff --git a/tests/basic_deployment.py b/tests/basic_deployment.py
index 343f447..04197bf 100644
--- a/tests/basic_deployment.py
+++ b/tests/basic_deployment.py
@@ -54,14 +54,10 @@ class OVSODLBasicDeployment(OpenStackAmuletDeployment):
},
{
'name': 'odl-controller',
- 'location': 'lp:~sdn-charmers/charms/trusty/odl-controller/'
- 'trunk',
'constraints': {'mem': '8G'},
},
{
'name': 'neutron-api-odl',
- 'location': 'lp:~openstack-charmers/charms/trusty'
- '/neutron-api-odl/trunk',
},
{'name': 'mysql'},
{'name': 'rabbitmq-server'},
@@ -123,10 +119,8 @@ class OVSODLBasicDeployment(OpenStackAmuletDeployment):
mysql = {
'dataset-size': '50%',
}
- # TODO(jamespage): These need to be configurable in some way
odl_controller = {
- 'install-url': 'http://10.245.161.162/swift/v1/opendaylight/'
- 'distribution-karaf-0.2.3-Helium-SR3.tar.gz',
+ 'install-url': os.environ.get('AMULET_ODL_LOCATION'),
'http-proxy': os.environ.get('AMULET_HTTP_PROXY'),
'https-proxy': os.environ.get('AMULET_HTTP_PROXY'),
}
diff --git a/tests/charmhelpers/contrib/amulet/deployment.py b/tests/charmhelpers/contrib/amulet/deployment.py
index 367d6b4..d451698 100644
--- a/tests/charmhelpers/contrib/amulet/deployment.py
+++ b/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."""
diff --git a/tests/charmhelpers/contrib/amulet/utils.py b/tests/charmhelpers/contrib/amulet/utils.py
index f61c2e8..2591a9b 100644
--- a/tests/charmhelpers/contrib/amulet/utils.py
+++ b/tests/charmhelpers/contrib/amulet/utils.py
@@ -14,14 +14,25 @@
# You should have received a copy of the GNU Lesser General Public License
# along with charm-helpers. If not, see .
-import ConfigParser
import io
+import json
import logging
+import os
import re
+import socket
+import subprocess
import sys
import time
+import uuid
+import amulet
+import distro_info
import six
+from six.moves import configparser
+if six.PY3:
+ from urllib import parse as urlparse
+else:
+ import urlparse
class AmuletUtils(object):
@@ -33,6 +44,7 @@ class AmuletUtils(object):
def __init__(self, log_level=logging.ERROR):
self.log = self.get_logger(level=log_level)
+ self.ubuntu_releases = self.get_ubuntu_releases()
def get_logger(self, name="amulet-logger", level=logging.DEBUG):
"""Get a logger object that will log to stdout."""
@@ -70,12 +82,44 @@ class AmuletUtils(object):
else:
return False
- def validate_services(self, commands):
- """Validate services.
+ def get_ubuntu_release_from_sentry(self, sentry_unit):
+ """Get Ubuntu release codename from sentry unit.
- Verify the specified services are running on the corresponding
+ :param sentry_unit: amulet sentry/service unit pointer
+ :returns: list of strings - release codename, failure message
+ """
+ msg = None
+ cmd = 'lsb_release -cs'
+ release, code = sentry_unit.run(cmd)
+ if code == 0:
+ self.log.debug('{} lsb_release: {}'.format(
+ sentry_unit.info['unit_name'], release))
+ else:
+ msg = ('{} `{}` returned {} '
+ '{}'.format(sentry_unit.info['unit_name'],
+ cmd, release, code))
+ if release not in self.ubuntu_releases:
+ msg = ("Release ({}) not found in Ubuntu releases "
+ "({})".format(release, self.ubuntu_releases))
+ return release, msg
+
+ def validate_services(self, commands):
+ """Validate that lists of commands succeed on service units. Can be
+ used to verify system services are running on the corresponding
service units.
- """
+
+ :param commands: dict with sentry keys and arbitrary command list vals
+ :returns: None if successful, Failure string message otherwise
+ """
+ self.log.debug('Checking status of system services...')
+
+ # /!\ 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 '
+ 'validate_services_by_name instead of validate_services '
+ 'due to init system differences.')
+
for k, v in six.iteritems(commands):
for cmd in v:
output, code = k.run(cmd)
@@ -86,6 +130,45 @@ class AmuletUtils(object):
return "command `{}` returned {}".format(cmd, str(code))
return None
+ def validate_services_by_name(self, sentry_services):
+ """Validate system service status by service name, automatically
+ detecting init system based on Ubuntu release codename.
+
+ :param sentry_services: dict with sentry keys and svc list values
+ :returns: None if successful, Failure string message otherwise
+ """
+ self.log.debug('Checking status of system services...')
+
+ # Point at which systemd became a thing
+ systemd_switch = self.ubuntu_releases.index('vivid')
+
+ for sentry_unit, services_list in six.iteritems(sentry_services):
+ # Get lsb_release codename from unit
+ release, ret = self.get_ubuntu_release_from_sentry(sentry_unit)
+ if ret:
+ return ret
+
+ for service_name in services_list:
+ if (self.ubuntu_releases.index(release) >= systemd_switch or
+ service_name in ['rabbitmq-server', 'apache2']):
+ # init is systemd (or regular sysv)
+ cmd = 'sudo service {} status'.format(service_name)
+ output, code = sentry_unit.run(cmd)
+ service_running = code == 0
+ elif self.ubuntu_releases.index(release) < systemd_switch:
+ # init is upstart
+ cmd = 'sudo status {}'.format(service_name)
+ output, code = sentry_unit.run(cmd)
+ service_running = code == 0 and "start/running" in output
+
+ self.log.debug('{} `{}` returned '
+ '{}'.format(sentry_unit.info['unit_name'],
+ cmd, code))
+ if not service_running:
+ return u"command `{}` returned {} {}".format(
+ cmd, output, str(code))
+ return None
+
def _get_config(self, unit, filename):
"""Get a ConfigParser object for parsing a unit's config file."""
file_contents = unit.file_contents(filename)
@@ -93,7 +176,7 @@ class AmuletUtils(object):
# NOTE(beisner): by default, ConfigParser does not handle options
# with no value, such as the flags used in the mysql my.cnf file.
# https://bugs.python.org/issue7005
- config = ConfigParser.ConfigParser(allow_no_value=True)
+ config = configparser.ConfigParser(allow_no_value=True)
config.readfp(io.StringIO(file_contents))
return config
@@ -103,7 +186,15 @@ class AmuletUtils(object):
Verify that the specified section of the config file contains
the expected option key:value pairs.
+
+ Compare expected dictionary data vs actual dictionary data.
+ The values in the 'expected' dictionary can be strings, bools, ints,
+ longs, or can be a function that evaluates a variable and returns a
+ bool.
"""
+ self.log.debug('Validating config file data ({} in {} on {})'
+ '...'.format(section, config_file,
+ sentry_unit.info['unit_name']))
config = self._get_config(sentry_unit, config_file)
if section != 'DEFAULT' and not config.has_section(section):
@@ -112,9 +203,20 @@ class AmuletUtils(object):
for k in expected.keys():
if not config.has_option(section, k):
return "section [{}] is missing option {}".format(section, k)
- if config.get(section, k) != expected[k]:
+
+ actual = config.get(section, k)
+ v = expected[k]
+ if (isinstance(v, six.string_types) or
+ isinstance(v, bool) or
+ isinstance(v, six.integer_types)):
+ # handle explicit values
+ if actual != v:
+ return "section [{}] {}:{} != expected {}:{}".format(
+ section, k, actual, k, expected[k])
+ # handle function pointers, such as not_null or valid_ip
+ elif not v(actual):
return "section [{}] {}:{} != expected {}:{}".format(
- section, k, config.get(section, k), k, expected[k])
+ section, k, actual, k, expected[k])
return None
def _validate_dict_data(self, expected, actual):
@@ -122,7 +224,7 @@ class AmuletUtils(object):
Compare expected dictionary data vs actual dictionary data.
The values in the 'expected' dictionary can be strings, bools, ints,
- longs, or can be a function that evaluate a variable and returns a
+ longs, or can be a function that evaluates a variable and returns a
bool.
"""
self.log.debug('actual: {}'.format(repr(actual)))
@@ -133,8 +235,10 @@ class AmuletUtils(object):
if (isinstance(v, six.string_types) or
isinstance(v, bool) or
isinstance(v, six.integer_types)):
+ # handle explicit values
if v != actual[k]:
return "{}:{}".format(k, actual[k])
+ # handle function pointers, such as not_null or valid_ip
elif not v(actual[k]):
return "{}:{}".format(k, actual[k])
else:
@@ -167,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)):
@@ -202,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:
@@ -281,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)
@@ -300,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):
@@ -321,3 +501,318 @@ class AmuletUtils(object):
def endpoint_error(self, name, data):
return 'unexpected endpoint data in {} - {}'.format(name, data)
+
+ def get_ubuntu_releases(self):
+ """Return a list of all Ubuntu releases in order of release."""
+ _d = distro_info.UbuntuDistroInfo()
+ _release_list = _d.all
+ return _release_list
+
+ def file_to_url(self, file_rel_path):
+ """Convert a relative file path to a file URL."""
+ _abs_path = os.path.abspath(file_rel_path)
+ return urlparse.urlparse(_abs_path, scheme='file').geturl()
+
+ def check_commands_on_units(self, commands, sentry_units):
+ """Check that all commands in a list exit zero on all
+ sentry units in a list.
+
+ :param commands: list of bash commands
+ :param sentry_units: list of sentry unit pointers
+ :returns: None if successful; Failure message otherwise
+ """
+ self.log.debug('Checking exit codes for {} commands on {} '
+ 'sentry units...'.format(len(commands),
+ len(sentry_units)))
+ for sentry_unit in sentry_units:
+ for cmd in commands:
+ output, code = sentry_unit.run(cmd)
+ if code == 0:
+ self.log.debug('{} `{}` returned {} '
+ '(OK)'.format(sentry_unit.info['unit_name'],
+ cmd, code))
+ else:
+ return ('{} `{}` returned {} '
+ '{}'.format(sentry_unit.info['unit_name'],
+ cmd, code, output))
+ return None
+
+ def get_process_id_list(self, sentry_unit, process_name,
+ expect_success=True):
+ """Get a list of process ID(s) from a single sentry juju unit
+ for a single process name.
+
+ :param sentry_unit: Amulet sentry instance (juju unit)
+ :param process_name: Process name
+ :param expect_success: If False, expect the PID to be missing,
+ raise if it is present.
+ :returns: List of process IDs
+ """
+ cmd = 'pidof -x {}'.format(process_name)
+ if not expect_success:
+ cmd += " || exit 0 && exit 1"
+ output, code = sentry_unit.run(cmd)
+ if code != 0:
+ msg = ('{} `{}` returned {} '
+ '{}'.format(sentry_unit.info['unit_name'],
+ cmd, code, output))
+ amulet.raise_status(amulet.FAIL, msg=msg)
+ return str(output).split()
+
+ def get_unit_process_ids(self, unit_processes, expect_success=True):
+ """Construct a dict containing unit sentries, process names, and
+ process IDs.
+
+ :param unit_processes: A dictionary of Amulet sentry instance
+ to list of process names.
+ :param expect_success: if False expect the processes to not be
+ running, raise if they are.
+ :returns: Dictionary of Amulet sentry instance to dictionary
+ of process names to PIDs.
+ """
+ pid_dict = {}
+ for sentry_unit, process_list in six.iteritems(unit_processes):
+ pid_dict[sentry_unit] = {}
+ for process in process_list:
+ pids = self.get_process_id_list(
+ sentry_unit, process, expect_success=expect_success)
+ pid_dict[sentry_unit].update({process: pids})
+ return pid_dict
+
+ def validate_unit_process_ids(self, expected, actual):
+ """Validate process id quantities for services on units."""
+ self.log.debug('Checking units for running processes...')
+ self.log.debug('Expected PIDs: {}'.format(expected))
+ self.log.debug('Actual PIDs: {}'.format(actual))
+
+ if len(actual) != len(expected):
+ return ('Unit count mismatch. expected, actual: {}, '
+ '{} '.format(len(expected), len(actual)))
+
+ for (e_sentry, e_proc_names) in six.iteritems(expected):
+ e_sentry_name = e_sentry.info['unit_name']
+ if e_sentry in actual.keys():
+ a_proc_names = actual[e_sentry]
+ else:
+ return ('Expected sentry ({}) not found in actual dict data.'
+ '{}'.format(e_sentry_name, e_sentry))
+
+ if len(e_proc_names.keys()) != len(a_proc_names.keys()):
+ return ('Process name count mismatch. expected, actual: {}, '
+ '{}'.format(len(expected), len(actual)))
+
+ for (e_proc_name, e_pids_length), (a_proc_name, a_pids) in \
+ zip(e_proc_names.items(), a_proc_names.items()):
+ if e_proc_name != a_proc_name:
+ return ('Process name mismatch. expected, actual: {}, '
+ '{}'.format(e_proc_name, a_proc_name))
+
+ a_pids_length = len(a_pids)
+ fail_msg = ('PID count mismatch. {} ({}) expected, actual: '
+ '{}, {} ({})'.format(e_sentry_name, e_proc_name,
+ e_pids_length, a_pids_length,
+ a_pids))
+
+ # If expected is not bool, ensure PID quantities match
+ if not isinstance(e_pids_length, bool) and \
+ a_pids_length != e_pids_length:
+ return fail_msg
+ # If expected is bool True, ensure 1 or more PIDs exist
+ elif isinstance(e_pids_length, bool) and \
+ e_pids_length is True and a_pids_length < 1:
+ return fail_msg
+ # If expected is bool False, ensure 0 PIDs exist
+ elif isinstance(e_pids_length, bool) and \
+ e_pids_length is False and a_pids_length != 0:
+ return fail_msg
+ else:
+ self.log.debug('PID check OK: {} {} {}: '
+ '{}'.format(e_sentry_name, e_proc_name,
+ e_pids_length, a_pids))
+ return None
+
+ def validate_list_of_identical_dicts(self, list_of_dicts):
+ """Check that all dicts within a list are identical."""
+ hashes = []
+ for _dict in list_of_dicts:
+ hashes.append(hash(frozenset(_dict.items())))
+
+ self.log.debug('Hashes: {}'.format(hashes))
+ if len(set(hashes)) == 1:
+ self.log.debug('Dicts within list are identical')
+ else:
+ return 'Dicts within list are not identical'
+
+ 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.
+
+ _check_output parameter is used for dependency injection.
+
+ @return action_id.
+ """
+ unit_id = unit_sentry.info["unit_name"]
+ command = ["juju", "action", "do", "--format=json", unit_id, action]
+ self.log.info("Running command: %s\n" % " ".join(command))
+ output = _check_output(command, universal_newlines=True)
+ data = json.loads(output)
+ action_id = data[u'Action queued with id']
+ return action_id
+
+ def wait_on_action(self, action_id, _check_output=subprocess.check_output):
+ """Wait for a given action, returning if it completed or not.
+
+ _check_output parameter is used for dependency injection.
+ """
+ command = ["juju", "action", "fetch", "--format=json", "--wait=0",
+ action_id]
+ 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"])
diff --git a/tests/charmhelpers/contrib/openstack/amulet/deployment.py b/tests/charmhelpers/contrib/openstack/amulet/deployment.py
index 503b814..d2ede32 100644
--- a/tests/charmhelpers/contrib/openstack/amulet/deployment.py
+++ b/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 .
+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,30 +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."""
- base_charms = ['mysql', 'mongodb']
+
+ 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:
- if svc.get('location'):
- continue
+ 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:
- if svc.get('location'):
- continue
+ else:
if svc['name'] in base_charms:
temp = 'lp:charms/{}/{}'
svc['location'] = temp.format(base_series,
@@ -70,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,
@@ -81,30 +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']
- # Openstack subordinate charms do not expose an origin option as that
- # is controlled by the principle
- ignore = ['neutron-openvswitch', 'cisco-vpp', 'odl-controller',
- 'openvswitch-odl', 'neutron-api-odl']
+ 'ceph-osd', 'ceph-radosgw', 'ceph-mon']
+
+ # 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.
@@ -116,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,
@@ -128,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):
@@ -147,9 +260,43 @@ class OpenStackAmuletDeployment(AmuletDeployment):
('utopic', 'juno'),
('vivid', 'kilo'),
('wily', 'liberty'),
+ ('xenial', 'mitaka'),
])
if self.openstack:
os_origin = self.openstack.split(':')[1]
return os_origin.split('%s-' % self.series)[1].split('/')[0]
else:
return releases[self.series]
+
+ def get_ceph_expected_pools(self, radosgw=False):
+ """Return a list of expected ceph pools in a ceph + cinder + glance
+ test scenario, based on OpenStack release and whether ceph radosgw
+ is flagged as present or not."""
+
+ if self._get_openstack_release() >= self.trusty_kilo:
+ # Kilo or later
+ pools = [
+ 'rbd',
+ 'cinder',
+ 'glance'
+ ]
+ else:
+ # Juno or earlier
+ pools = [
+ 'data',
+ 'metadata',
+ 'rbd',
+ 'cinder',
+ 'glance'
+ ]
+
+ if radosgw:
+ pools.extend([
+ '.rgw.root',
+ '.rgw.control',
+ '.rgw',
+ '.rgw.gc',
+ '.users.uid'
+ ])
+
+ return pools
diff --git a/tests/charmhelpers/contrib/openstack/amulet/utils.py b/tests/charmhelpers/contrib/openstack/amulet/utils.py
index 9c3d918..2995124 100644
--- a/tests/charmhelpers/contrib/openstack/amulet/utils.py
+++ b/tests/charmhelpers/contrib/openstack/amulet/utils.py
@@ -14,16 +14,26 @@
# You should have received a copy of the GNU Lesser General Public License
# along with charm-helpers. If not, see .
+import amulet
+import json
import logging
import os
+import re
+import six
import time
import urllib
+import cinderclient.v1.client as cinder_client
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
+from keystoneclient.auth.identity import v3 as keystone_id_v3
+from keystoneclient import session as keystone_session
+from keystoneclient.v3 import client as keystone_client_v3
-import six
+import novaclient.v1_1.client as nova_client
+import pika
+import swiftclient
from charmhelpers.contrib.amulet.utils import (
AmuletUtils
@@ -37,7 +47,7 @@ class OpenStackAmuletUtils(AmuletUtils):
"""OpenStack amulet utilities.
This class inherits from AmuletUtils and has additional support
- that is specifically for use by OpenStack charms.
+ that is specifically for use by OpenStack charm tests.
"""
def __init__(self, log_level=ERROR):
@@ -51,6 +61,8 @@ class OpenStackAmuletUtils(AmuletUtils):
Validate actual endpoint data vs expected endpoint data. The ports
are used to find the matching endpoint.
"""
+ self.log.debug('Validating endpoint data...')
+ self.log.debug('actual: {}'.format(repr(endpoints)))
found = False
for ep in endpoints:
self.log.debug('endpoint: {}'.format(repr(ep)))
@@ -77,6 +89,7 @@ class OpenStackAmuletUtils(AmuletUtils):
Validate a list of actual service catalog endpoints vs a list of
expected service catalog endpoints.
"""
+ self.log.debug('Validating service catalog endpoint data...')
self.log.debug('actual: {}'.format(repr(actual)))
for k, v in six.iteritems(expected):
if k in actual:
@@ -93,6 +106,7 @@ class OpenStackAmuletUtils(AmuletUtils):
Validate a list of actual tenant data vs list of expected tenant
data.
"""
+ self.log.debug('Validating tenant data...')
self.log.debug('actual: {}'.format(repr(actual)))
for e in expected:
found = False
@@ -114,6 +128,7 @@ class OpenStackAmuletUtils(AmuletUtils):
Validate a list of actual role data vs a list of expected role
data.
"""
+ self.log.debug('Validating role data...')
self.log.debug('actual: {}'.format(repr(actual)))
for e in expected:
found = False
@@ -128,20 +143,26 @@ class OpenStackAmuletUtils(AmuletUtils):
return "role {} does not exist".format(e['name'])
return ret
- def validate_user_data(self, expected, actual):
+ def validate_user_data(self, expected, actual, api_version=None):
"""Validate user data.
Validate a list of actual user data vs a list of expected user
data.
"""
+ self.log.debug('Validating user data...')
self.log.debug('actual: {}'.format(repr(actual)))
for e in expected:
found = False
for act in actual:
- a = {'enabled': act.enabled, 'name': act.name,
- 'email': act.email, 'tenantId': act.tenantId,
- 'id': act.id}
- if e['name'] == a['name']:
+ if e['name'] == act.name:
+ a = {'enabled': act.enabled, 'name': act.name,
+ 'email': act.email, 'id': act.id}
+ if api_version == 2:
+ a['tenantId'] = act.tenantId
+ else:
+ a['default_project_id'] = getattr(act,
+ 'default_project_id',
+ 'none')
found = True
ret = self._validate_dict_data(e, a)
if ret:
@@ -155,26 +176,55 @@ class OpenStackAmuletUtils(AmuletUtils):
Validate a list of actual flavors vs a list of expected flavors.
"""
+ self.log.debug('Validating flavor data...')
self.log.debug('actual: {}'.format(repr(actual)))
act = [a.name for a in actual]
return self._validate_list_data(expected, act)
def tenant_exists(self, keystone, tenant):
"""Return True if tenant exists."""
+ self.log.debug('Checking if tenant exists ({})...'.format(tenant))
return tenant in [t.name for t in keystone.tenants.list()]
+ def authenticate_cinder_admin(self, keystone_sentry, username,
+ password, tenant):
+ """Authenticates admin user with cinder."""
+ # NOTE(beisner): cinder python client doesn't accept tokens.
+ service_ip = \
+ keystone_sentry.relation('shared-db',
+ 'mysql:shared-db')['private-address']
+ ept = "http://{}:5000/v2.0".format(service_ip.strip().decode('utf-8'))
+ return cinder_client.Client(username, password, tenant, ept)
+
def authenticate_keystone_admin(self, keystone_sentry, user, password,
- tenant):
+ tenant=None, api_version=None,
+ keystone_ip=None):
"""Authenticates admin user with the keystone admin endpoint."""
+ self.log.debug('Authenticating keystone admin...')
unit = keystone_sentry
- service_ip = unit.relation('shared-db',
- 'mysql:shared-db')['private-address']
- ep = "http://{}:35357/v2.0".format(service_ip.strip().decode('utf-8'))
- return keystone_client.Client(username=user, password=password,
- tenant_name=tenant, auth_url=ep)
+ if not keystone_ip:
+ keystone_ip = unit.relation('shared-db',
+ 'mysql:shared-db')['private-address']
+ base_ep = "http://{}:35357".format(keystone_ip.strip().decode('utf-8'))
+ if not api_version or api_version == 2:
+ ep = base_ep + "/v2.0"
+ return keystone_client.Client(username=user, password=password,
+ tenant_name=tenant, auth_url=ep)
+ else:
+ ep = base_ep + "/v3"
+ auth = keystone_id_v3.Password(
+ user_domain_name='admin_domain',
+ username=user,
+ password=password,
+ domain_name='admin_domain',
+ auth_url=ep,
+ )
+ sess = keystone_session.Session(auth=auth)
+ return keystone_client_v3.Client(session=sess)
def authenticate_keystone_user(self, keystone, user, password, tenant):
"""Authenticates a regular user with the keystone public endpoint."""
+ self.log.debug('Authenticating keystone user ({})...'.format(user))
ep = keystone.service_catalog.url_for(service_type='identity',
endpoint_type='publicURL')
return keystone_client.Client(username=user, password=password,
@@ -182,19 +232,49 @@ class OpenStackAmuletUtils(AmuletUtils):
def authenticate_glance_admin(self, keystone):
"""Authenticates admin user with glance."""
+ self.log.debug('Authenticating glance admin...')
ep = keystone.service_catalog.url_for(service_type='image',
endpoint_type='adminURL')
return glance_client.Client(ep, token=keystone.auth_token)
+ def authenticate_heat_admin(self, keystone):
+ """Authenticates the admin user with heat."""
+ self.log.debug('Authenticating heat admin...')
+ ep = keystone.service_catalog.url_for(service_type='orchestration',
+ endpoint_type='publicURL')
+ return heat_client.Client(endpoint=ep, token=keystone.auth_token)
+
def authenticate_nova_user(self, keystone, user, password, tenant):
"""Authenticates a regular user with nova-api."""
+ self.log.debug('Authenticating nova user ({})...'.format(user))
ep = keystone.service_catalog.url_for(service_type='identity',
endpoint_type='publicURL')
return nova_client.Client(username=user, api_key=password,
project_id=tenant, auth_url=ep)
+ def authenticate_swift_user(self, keystone, user, password, tenant):
+ """Authenticates a regular user with swift api."""
+ self.log.debug('Authenticating swift user ({})...'.format(user))
+ ep = keystone.service_catalog.url_for(service_type='identity',
+ endpoint_type='publicURL')
+ return swiftclient.Connection(authurl=ep,
+ user=user,
+ key=password,
+ tenant_name=tenant,
+ auth_version='2.0')
+
def create_cirros_image(self, glance, image_name):
- """Download the latest cirros image and upload it to glance."""
+ """Download the latest cirros image and upload it to glance,
+ validate and return a resource pointer.
+
+ :param glance: pointer to authenticated glance connection
+ :param image_name: display name for new image
+ :returns: glance image pointer
+ """
+ self.log.debug('Creating glance cirros image '
+ '({})...'.format(image_name))
+
+ # Download cirros image
http_proxy = os.getenv('AMULET_HTTP_PROXY')
self.log.debug('AMULET_HTTP_PROXY: {}'.format(http_proxy))
if http_proxy:
@@ -203,57 +283,67 @@ class OpenStackAmuletUtils(AmuletUtils):
else:
opener = urllib.FancyURLopener()
- f = opener.open("http://download.cirros-cloud.net/version/released")
+ f = opener.open('http://download.cirros-cloud.net/version/released')
version = f.read().strip()
- cirros_img = "cirros-{}-x86_64-disk.img".format(version)
+ cirros_img = 'cirros-{}-x86_64-disk.img'.format(version)
local_path = os.path.join('tests', cirros_img)
if not os.path.exists(local_path):
- cirros_url = "http://{}/{}/{}".format("download.cirros-cloud.net",
+ cirros_url = 'http://{}/{}/{}'.format('download.cirros-cloud.net',
version, cirros_img)
opener.retrieve(cirros_url, local_path)
f.close()
+ # Create glance image
with open(local_path) as f:
image = glance.images.create(name=image_name, is_public=True,
disk_format='qcow2',
container_format='bare', data=f)
- count = 1
- status = image.status
- while status != 'active' and count < 10:
- time.sleep(3)
- image = glance.images.get(image.id)
- status = image.status
- self.log.debug('image status: {}'.format(status))
- count += 1
- if status != 'active':
- self.log.error('image creation timed out')
- return None
+ # Wait for image to reach active status
+ img_id = image.id
+ ret = self.resource_reaches_status(glance.images, img_id,
+ expected_stat='active',
+ msg='Image status wait')
+ if not ret:
+ msg = 'Glance image failed to reach expected state.'
+ amulet.raise_status(amulet.FAIL, msg=msg)
+
+ # Re-validate new image
+ self.log.debug('Validating image attributes...')
+ val_img_name = glance.images.get(img_id).name
+ val_img_stat = glance.images.get(img_id).status
+ val_img_pub = glance.images.get(img_id).is_public
+ val_img_cfmt = glance.images.get(img_id).container_format
+ val_img_dfmt = glance.images.get(img_id).disk_format
+ msg_attr = ('Image attributes - name:{} public:{} id:{} stat:{} '
+ 'container fmt:{} disk fmt:{}'.format(
+ val_img_name, val_img_pub, img_id,
+ val_img_stat, val_img_cfmt, val_img_dfmt))
+
+ if val_img_name == image_name and val_img_stat == 'active' \
+ and val_img_pub is True and val_img_cfmt == 'bare' \
+ and val_img_dfmt == 'qcow2':
+ self.log.debug(msg_attr)
+ else:
+ msg = ('Volume validation failed, {}'.format(msg_attr))
+ amulet.raise_status(amulet.FAIL, msg=msg)
return image
def delete_image(self, glance, image):
"""Delete the specified image."""
- num_before = len(list(glance.images.list()))
- glance.images.delete(image)
- count = 1
- num_after = len(list(glance.images.list()))
- while num_after != (num_before - 1) and count < 10:
- time.sleep(3)
- num_after = len(list(glance.images.list()))
- self.log.debug('number of images: {}'.format(num_after))
- count += 1
-
- if num_after != (num_before - 1):
- self.log.error('image deletion timed out')
- return False
-
- return True
+ # /!\ DEPRECATION WARNING
+ self.log.warn('/!\\ DEPRECATION WARNING: use '
+ 'delete_resource instead of delete_image.')
+ self.log.debug('Deleting glance image ({})...'.format(image))
+ return self.delete_resource(glance.images, image, msg='glance image')
def create_instance(self, nova, image_name, instance_name, flavor):
"""Create the specified instance."""
+ self.log.debug('Creating instance '
+ '({}|{}|{})'.format(instance_name, image_name, flavor))
image = nova.images.find(name=image_name)
flavor = nova.flavors.find(name=flavor)
instance = nova.servers.create(name=instance_name, image=image,
@@ -276,19 +366,644 @@ class OpenStackAmuletUtils(AmuletUtils):
def delete_instance(self, nova, instance):
"""Delete the specified instance."""
- num_before = len(list(nova.servers.list()))
- nova.servers.delete(instance)
- count = 1
- num_after = len(list(nova.servers.list()))
- while num_after != (num_before - 1) and count < 10:
- time.sleep(3)
- num_after = len(list(nova.servers.list()))
- self.log.debug('number of instances: {}'.format(num_after))
- count += 1
+ # /!\ DEPRECATION WARNING
+ self.log.warn('/!\\ DEPRECATION WARNING: use '
+ 'delete_resource instead of delete_instance.')
+ self.log.debug('Deleting instance ({})...'.format(instance))
+ return self.delete_resource(nova.servers, instance,
+ msg='nova instance')
- if num_after != (num_before - 1):
- self.log.error('instance deletion timed out')
+ def create_or_get_keypair(self, nova, keypair_name="testkey"):
+ """Create a new keypair, or return pointer if it already exists."""
+ try:
+ _keypair = nova.keypairs.get(keypair_name)
+ self.log.debug('Keypair ({}) already exists, '
+ 'using it.'.format(keypair_name))
+ return _keypair
+ except:
+ self.log.debug('Keypair ({}) does not exist, '
+ 'creating it.'.format(keypair_name))
+
+ _keypair = nova.keypairs.create(name=keypair_name)
+ return _keypair
+
+ def create_cinder_volume(self, cinder, vol_name="demo-vol", vol_size=1,
+ img_id=None, src_vol_id=None, snap_id=None):
+ """Create cinder volume, optionally from a glance image, OR
+ optionally as a clone of an existing volume, OR optionally
+ from a snapshot. Wait for the new volume status to reach
+ the expected status, validate and return a resource pointer.
+
+ :param vol_name: cinder volume display name
+ :param vol_size: size in gigabytes
+ :param img_id: optional glance image id
+ :param src_vol_id: optional source volume id to clone
+ :param snap_id: optional snapshot id to use
+ :returns: cinder volume pointer
+ """
+ # Handle parameter input and avoid impossible combinations
+ if img_id and not src_vol_id and not snap_id:
+ # Create volume from image
+ self.log.debug('Creating cinder volume from glance image...')
+ bootable = 'true'
+ elif src_vol_id and not img_id and not snap_id:
+ # Clone an existing volume
+ self.log.debug('Cloning cinder volume...')
+ bootable = cinder.volumes.get(src_vol_id).bootable
+ elif snap_id and not src_vol_id and not img_id:
+ # Create volume from snapshot
+ self.log.debug('Creating cinder volume from snapshot...')
+ snap = cinder.volume_snapshots.find(id=snap_id)
+ vol_size = snap.size
+ snap_vol_id = cinder.volume_snapshots.get(snap_id).volume_id
+ bootable = cinder.volumes.get(snap_vol_id).bootable
+ elif not img_id and not src_vol_id and not snap_id:
+ # Create volume
+ self.log.debug('Creating cinder volume...')
+ bootable = 'false'
+ else:
+ # Impossible combination of parameters
+ msg = ('Invalid method use - name:{} size:{} img_id:{} '
+ 'src_vol_id:{} snap_id:{}'.format(vol_name, vol_size,
+ img_id, src_vol_id,
+ snap_id))
+ amulet.raise_status(amulet.FAIL, msg=msg)
+
+ # Create new volume
+ try:
+ vol_new = cinder.volumes.create(display_name=vol_name,
+ imageRef=img_id,
+ size=vol_size,
+ source_volid=src_vol_id,
+ snapshot_id=snap_id)
+ vol_id = vol_new.id
+ except Exception as e:
+ msg = 'Failed to create volume: {}'.format(e)
+ amulet.raise_status(amulet.FAIL, msg=msg)
+
+ # Wait for volume to reach available status
+ ret = self.resource_reaches_status(cinder.volumes, vol_id,
+ expected_stat="available",
+ msg="Volume status wait")
+ if not ret:
+ msg = 'Cinder volume failed to reach expected state.'
+ amulet.raise_status(amulet.FAIL, msg=msg)
+
+ # Re-validate new volume
+ self.log.debug('Validating volume attributes...')
+ val_vol_name = cinder.volumes.get(vol_id).display_name
+ val_vol_boot = cinder.volumes.get(vol_id).bootable
+ val_vol_stat = cinder.volumes.get(vol_id).status
+ val_vol_size = cinder.volumes.get(vol_id).size
+ msg_attr = ('Volume attributes - name:{} id:{} stat:{} boot:'
+ '{} size:{}'.format(val_vol_name, vol_id,
+ val_vol_stat, val_vol_boot,
+ val_vol_size))
+
+ if val_vol_boot == bootable and val_vol_stat == 'available' \
+ and val_vol_name == vol_name and val_vol_size == vol_size:
+ self.log.debug(msg_attr)
+ else:
+ msg = ('Volume validation failed, {}'.format(msg_attr))
+ amulet.raise_status(amulet.FAIL, msg=msg)
+
+ return vol_new
+
+ def delete_resource(self, resource, resource_id,
+ msg="resource", max_wait=120):
+ """Delete one openstack resource, such as one instance, keypair,
+ image, volume, stack, etc., and confirm deletion within max wait time.
+
+ :param resource: pointer to os resource type, ex:glance_client.images
+ :param resource_id: unique name or id for the openstack resource
+ :param msg: text to identify purpose in logging
+ :param max_wait: maximum wait time in seconds
+ :returns: True if successful, otherwise False
+ """
+ self.log.debug('Deleting OpenStack resource '
+ '{} ({})'.format(resource_id, msg))
+ num_before = len(list(resource.list()))
+ resource.delete(resource_id)
+
+ tries = 0
+ num_after = len(list(resource.list()))
+ while num_after != (num_before - 1) and tries < (max_wait / 4):
+ self.log.debug('{} delete check: '
+ '{} [{}:{}] {}'.format(msg, tries,
+ num_before,
+ num_after,
+ resource_id))
+ time.sleep(4)
+ num_after = len(list(resource.list()))
+ tries += 1
+
+ self.log.debug('{}: expected, actual count = {}, '
+ '{}'.format(msg, num_before - 1, num_after))
+
+ if num_after == (num_before - 1):
+ return True
+ else:
+ self.log.error('{} delete timed out'.format(msg))
return False
- return True
+ def resource_reaches_status(self, resource, resource_id,
+ expected_stat='available',
+ msg='resource', max_wait=120):
+ """Wait for an openstack resources status to reach an
+ expected status within a specified time. Useful to confirm that
+ nova instances, cinder vols, snapshots, glance images, heat stacks
+ and other resources eventually reach the expected status.
+
+ :param resource: pointer to os resource type, ex: heat_client.stacks
+ :param resource_id: unique id for the openstack resource
+ :param expected_stat: status to expect resource to reach
+ :param msg: text to identify purpose in logging
+ :param max_wait: maximum wait time in seconds
+ :returns: True if successful, False if status is not reached
+ """
+
+ tries = 0
+ resource_stat = resource.get(resource_id).status
+ while resource_stat != expected_stat and tries < (max_wait / 4):
+ self.log.debug('{} status check: '
+ '{} [{}:{}] {}'.format(msg, tries,
+ resource_stat,
+ expected_stat,
+ resource_id))
+ time.sleep(4)
+ resource_stat = resource.get(resource_id).status
+ tries += 1
+
+ self.log.debug('{}: expected, actual status = {}, '
+ '{}'.format(msg, resource_stat, expected_stat))
+
+ if resource_stat == expected_stat:
+ return True
+ else:
+ self.log.debug('{} never reached expected status: '
+ '{}'.format(resource_id, expected_stat))
+ return False
+
+ def get_ceph_osd_id_cmd(self, index):
+ """Produce a shell command that will return a ceph-osd id."""
+ return ("`initctl list | grep 'ceph-osd ' | "
+ "awk 'NR=={} {{ print $2 }}' | "
+ "grep -o '[0-9]*'`".format(index + 1))
+
+ def get_ceph_pools(self, sentry_unit):
+ """Return a dict of ceph pools from a single ceph unit, with
+ pool name as keys, pool id as vals."""
+ pools = {}
+ cmd = 'sudo ceph osd lspools'
+ output, code = sentry_unit.run(cmd)
+ if code != 0:
+ msg = ('{} `{}` returned {} '
+ '{}'.format(sentry_unit.info['unit_name'],
+ cmd, code, output))
+ amulet.raise_status(amulet.FAIL, msg=msg)
+
+ # Example output: 0 data,1 metadata,2 rbd,3 cinder,4 glance,
+ for pool in str(output).split(','):
+ pool_id_name = pool.split(' ')
+ if len(pool_id_name) == 2:
+ pool_id = pool_id_name[0]
+ pool_name = pool_id_name[1]
+ pools[pool_name] = int(pool_id)
+
+ self.log.debug('Pools on {}: {}'.format(sentry_unit.info['unit_name'],
+ pools))
+ return pools
+
+ def get_ceph_df(self, sentry_unit):
+ """Return dict of ceph df json output, including ceph pool state.
+
+ :param sentry_unit: Pointer to amulet sentry instance (juju unit)
+ :returns: Dict of ceph df output
+ """
+ cmd = 'sudo ceph df --format=json'
+ output, code = sentry_unit.run(cmd)
+ if code != 0:
+ msg = ('{} `{}` returned {} '
+ '{}'.format(sentry_unit.info['unit_name'],
+ cmd, code, output))
+ amulet.raise_status(amulet.FAIL, msg=msg)
+ return json.loads(output)
+
+ def get_ceph_pool_sample(self, sentry_unit, pool_id=0):
+ """Take a sample of attributes of a ceph pool, returning ceph
+ pool name, object count and disk space used for the specified
+ pool ID number.
+
+ :param sentry_unit: Pointer to amulet sentry instance (juju unit)
+ :param pool_id: Ceph pool ID
+ :returns: List of pool name, object count, kb disk space used
+ """
+ df = self.get_ceph_df(sentry_unit)
+ pool_name = df['pools'][pool_id]['name']
+ obj_count = df['pools'][pool_id]['stats']['objects']
+ kb_used = df['pools'][pool_id]['stats']['kb_used']
+ self.log.debug('Ceph {} pool (ID {}): {} objects, '
+ '{} kb used'.format(pool_name, pool_id,
+ obj_count, kb_used))
+ return pool_name, obj_count, kb_used
+
+ def validate_ceph_pool_samples(self, samples, sample_type="resource pool"):
+ """Validate ceph pool samples taken over time, such as pool
+ object counts or pool kb used, before adding, after adding, and
+ after deleting items which affect those pool attributes. The
+ 2nd element is expected to be greater than the 1st; 3rd is expected
+ to be less than the 2nd.
+
+ :param samples: List containing 3 data samples
+ :param sample_type: String for logging and usage context
+ :returns: None if successful, Failure message otherwise
+ """
+ original, created, deleted = range(3)
+ if samples[created] <= samples[original] or \
+ samples[deleted] >= samples[created]:
+ return ('Ceph {} samples ({}) '
+ 'unexpected.'.format(sample_type, samples))
+ else:
+ 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)