Add pause/resume actions and sync charm-helpers

Adds pause and resume unit to the charm such that the
charm stays paused during maintenance operations.

Change-Id: Iba0d39f9c5ca482a12fdb9f91d2890a715d3f8b7

actions.yaml

@@ -0,0 +1,4 @@
+pause:
+  description: Pause the ceph-radosgw unit.
+resume:
+  descrpition: Resume the ceph-radosgw unit.

actions/actions.py

@@ -0,0 +1,47 @@
+#!/usr/bin/python
+
+import os
+import sys
+
+sys.path.append('hooks/')
+from charmhelpers.core.hookenv import action_fail
+from utils import (
+    pause_unit_helper,
+    resume_unit_helper,
+    register_configs,
+)
+
+
+def pause(args):
+    """Pause the Ceilometer services.
+    @raises Exception should the service fail to stop.
+    """
+    pause_unit_helper(register_configs())
+
+
+def resume(args):
+    """Resume the Ceilometer services.
+    @raises Exception should the service fail to start."""
+    resume_unit_helper(register_configs())
+
+
+# A dictionary of all the defined actions to callables (which take
+# parsed arguments).
+ACTIONS = {"pause": pause, "resume": resume}
+
+
+def main(args):
+    action_name = os.path.basename(args[0])
+    try:
+        action = ACTIONS[action_name]
+    except KeyError:
+        return "Action %s undefined" % action_name
+    else:
+        try:
+            action(args)
+        except Exception as e:
+            action_fail(str(e))
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))

actions/pause

@@ -0,0 +1 @@
+actions.py

actions/resume

@@ -0,0 +1 @@
+actions.py

hooks/charmhelpers/contrib/openstack/templates/section-keystone-authtoken

@@ -1,20 +1,12 @@
 {% if auth_host -%}
-{% if api_version == '3' -%}
 [keystone_authtoken]
-auth_url = {{ service_protocol }}://{{ service_host }}:{{ service_port }}
+auth_uri = {{ service_protocol }}://{{ service_host }}:{{ service_port }}
+auth_url = {{ auth_protocol }}://{{ auth_host }}:{{ auth_port }}
+auth_plugin = password
+project_domain_id = default
+user_domain_id = default
 project_name = {{ admin_tenant_name }}
 username = {{ admin_user }}
 password = {{ admin_password }}
-project_domain_name = default
-user_domain_name = default
-auth_plugin = password
-{% else -%}
-[keystone_authtoken]
-identity_uri = {{ auth_protocol }}://{{ auth_host }}:{{ auth_port }}/{{ auth_admin_prefix }}
-auth_uri = {{ service_protocol }}://{{ service_host }}:{{ service_port }}/{{ service_admin_prefix }}
-admin_tenant_name = {{ admin_tenant_name }}
-admin_user = {{ admin_user }}
-admin_password = {{ admin_password }}
 signing_dir = {{ signing_dir }}
 {% endif -%}
-{% endif -%}

hooks/charmhelpers/contrib/openstack/templates/section-keystone-authtoken-legacy

@@ -0,0 +1,10 @@
+{% if auth_host -%}
+[keystone_authtoken]
+# Juno specific config (Bug #1557223)
+auth_uri = {{ service_protocol }}://{{ service_host }}:{{ service_port }}/{{ service_admin_prefix }}
+identity_uri = {{ auth_protocol }}://{{ auth_host }}:{{ auth_port }}
+admin_tenant_name = {{ admin_tenant_name }}
+admin_user = {{ admin_user }}
+admin_password = {{ admin_password }}
+signing_dir = {{ signing_dir }}
+{% endif -%}

hooks/hooks.py

@@ -36,7 +36,6 @@ from charmhelpers.fetch import (
 )
 from charmhelpers.core.host import (
     lsb_release,
-    restart_on_change,
 )
 from charmhelpers.payload.execd import execd_preinstall
 from charmhelpers.core.host import (
@@ -54,21 +53,21 @@ from charmhelpers.contrib.openstack.ip import (
     canonical_url,
     PUBLIC, INTERNAL, ADMIN,
 )
-from charmhelpers.contrib.openstack.utils import (
-    set_os_workload_status,
-)
 from charmhelpers.contrib.storage.linux.ceph import (
     send_request_if_needed,
     is_request_complete,
 )
+from charmhelpers.contrib.openstack.utils import (
+    is_unit_paused_set,
+    pausable_restart_on_change as restart_on_change,
+)
 from utils import (
     enable_pocket,
     CEPHRG_HA_RES,
     register_configs,
-    REQUIRED_INTERFACES,
-    check_optional_relations,
     setup_ipv6,
     services,
+    assess_status,
 )
 from charmhelpers.contrib.charmsupport import nrpe
 
@@ -282,7 +281,8 @@ def mon_relation():
         key = relation_get('radosgw_key')
         if key:
             ceph.import_radosgw_key(key)
-            restart()  # TODO figure out a better way todo this
+            if not is_unit_paused_set():
+                restart()  # TODO figure out a better way todo this
     else:
         send_request_if_needed(rq, relation='mon')
 
@@ -339,7 +339,8 @@ def identity_joined(relid=None):
 def identity_changed(relid=None):
     identity_joined(relid)
     CONFIGS.write_all()
-    restart()
+    if not is_unit_paused_set():
+        restart()
 
 
 @hooks.hook('cluster-relation-joined')
@@ -454,5 +455,4 @@ if __name__ == '__main__':
         hooks.execute(sys.argv)
     except UnregisteredHookError as e:
         log('Unknown hook {} - skipping.'.format(e))
-    set_os_workload_status(CONFIGS, REQUIRED_INTERFACES,
-                           charm_func=check_optional_relations)
+    assess_status(CONFIGS)

hooks/utils.py

@@ -28,6 +28,9 @@ from charmhelpers.contrib.openstack import (
 from charmhelpers.contrib.openstack.utils import (
     os_release,
     set_os_workload_status,
+    make_assess_status_func,
+    pause_unit,
+    resume_unit,
 )
 from charmhelpers.contrib.hahelpers.cluster import get_hacluster_config
 from charmhelpers.core.host import (
@@ -177,3 +180,70 @@ def setup_ipv6():
                    'main')
         apt_update(fatal=True)
         apt_install('haproxy/trusty-backports', fatal=True)
+
+
+def assess_status(configs):
+    """Assess status of current unit
+    Decides what the state of the unit should be based on the current
+    configuration.
+    SIDE EFFECT: calls set_os_workload_status(...) which sets the workload
+    status of the unit.
+    Also calls status_set(...) directly if paused state isn't complete.
+    @param configs: a templating.OSConfigRenderer() object
+    @returns None - this function is executed for its side-effect
+    """
+    assess_status_func(configs)()
+
+
+def assess_status_func(configs):
+    """Helper function to create the function that will assess_status() for
+    the unit.
+    Uses charmhelpers.contrib.openstack.utils.make_assess_status_func() to
+    create the appropriate status function and then returns it.
+    Used directly by assess_status() and also for pausing and resuming
+    the unit.
+
+    NOTE(ajkavanagh) ports are not checked due to race hazards with services
+    that don't behave sychronously w.r.t their service scripts.  e.g.
+    apache2.
+    @param configs: a templating.OSConfigRenderer() object
+    @return f() -> None : a function that assesses the unit's workload status
+    """
+    return make_assess_status_func(
+        configs, REQUIRED_INTERFACES,
+        charm_func=check_optional_relations,
+        services=services(), ports=None)
+
+
+def pause_unit_helper(configs):
+    """Helper function to pause a unit, and then call assess_status(...) in
+    effect, so that the status is correctly updated.
+    Uses charmhelpers.contrib.openstack.utils.pause_unit() to do the work.
+    @param configs: a templating.OSConfigRenderer() object
+    @returns None - this function is executed for its side-effect
+    """
+    _pause_resume_helper(pause_unit, configs)
+
+
+def resume_unit_helper(configs):
+    """Helper function to resume a unit, and then call assess_status(...) in
+    effect, so that the status is correctly updated.
+    Uses charmhelpers.contrib.openstack.utils.resume_unit() to do the work.
+    @param configs: a templating.OSConfigRenderer() object
+    @returns None - this function is executed for its side-effect
+    """
+    _pause_resume_helper(resume_unit, configs)
+
+
+def _pause_resume_helper(f, configs):
+    """Helper function that uses the make_assess_status_func(...) from
+    charmhelpers.contrib.openstack.utils to create an assess_status(...)
+    function that can be used with the pause/resume of the unit
+    @param f: the function to be used with the assess_status(...) function
+    @returns None - this function is executed for its side-effect
+    """
+    # TODO(ajkavanagh) - ports= has been left off because of the race hazard
+    # that exists due to service_start()
+    f(assess_status_func(configs),
+      services=services(),
+      ports=None)

tests/basic_deployment.py

@@ -1,6 +1,9 @@
 #!/usr/bin/python
 
 import amulet
+import subprocess
+import json
+import time
 from charmhelpers.contrib.openstack.amulet.deployment import (
     OpenStackAmuletDeployment
 )
@@ -98,6 +101,32 @@ class CephRadosGwBasicDeployment(OpenStackAmuletDeployment):
                    'ceph-radosgw': radosgw_config}
         super(CephRadosGwBasicDeployment, self)._configure_services(configs)
 
+    def _run_action(self, unit_id, action, *args):
+        command = ["juju", "action", "do", "--format=json", unit_id, action]
+        command.extend(args)
+        print("Running command: %s\n" % " ".join(command))
+        output = subprocess.check_output(command)
+        output_json = output.decode(encoding="UTF-8")
+        data = json.loads(output_json)
+        action_id = data[u'Action queued with id']
+        return action_id
+
+    def _wait_on_action(self, action_id):
+        command = ["juju", "action", "fetch", "--format=json", action_id]
+        while True:
+            try:
+                output = subprocess.check_output(command)
+            except Exception as e:
+                print(e)
+                return False
+            output_json = output.decode(encoding="UTF-8")
+            data = json.loads(output_json)
+            if data[u"status"] == "completed":
+                return True
+            elif data[u"status"] == "failed":
+                return False
+            time.sleep(2)
+
     def _initialize_tests(self):
         """Perform final initialization before tests get run."""
         # Access the sentries for inspecting service units
@@ -491,6 +520,22 @@ class CephRadosGwBasicDeployment(OpenStackAmuletDeployment):
         if ret:
             amulet.raise_status(amulet.FAIL, msg=ret)
 
+    def test_910_pause_and_resume(self):
+        """The services can be paused and resumed. """
+        u.log.debug('Checking pause and resume actions...')
+        unit_name = "ceph-radosgw/0"
+        unit = self.d.sentry.unit[unit_name]
+
+        assert u.status_get(unit)[0] == "active"
+
+        action_id = self._run_action(unit_name, "pause")
+        assert self._wait_on_action(action_id), "Pause action failed."
+        assert u.status_get(unit)[0] == "maintenance"
+
+        action_id = self._run_action(unit_name, "resume")
+        assert self._wait_on_action(action_id), "Resume action failed."
+        assert u.status_get(unit)[0] == "active"
+        u.log.debug('OK')
     # Note(beisner): need to add basic object store functional checks.
 
     # FYI: No restart check as ceph services do not restart

tests/charmhelpers/contrib/amulet/utils.py

@@ -782,15 +782,20 @@ class AmuletUtils(object):
 
 # amulet juju action helpers:
     def run_action(self, unit_sentry, action,
-                   _check_output=subprocess.check_output):
+                   _check_output=subprocess.check_output,
+                   params=None):
         """Run the named action on a given unit sentry.
 
+        params a dict of parameters to use
         _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]
+        if params is not None:
+            for key, value in params.iteritems():
+                command.append("{}={}".format(key, value))
         self.log.info("Running command: %s\n" % " ".join(command))
         output = _check_output(command, universal_newlines=True)
         data = json.loads(output)

unit_tests/__init__.py

@@ -1,3 +1,4 @@
 import sys
 
+sys.path.append('actions/')
 sys.path.append('hooks/')

unit_tests/test_actions.py

@@ -0,0 +1,64 @@
+import mock
+from mock import patch
+
+from test_utils import CharmTestCase
+
+with patch('utils.register_configs') as configs:
+    configs.return_value = 'test-config'
+    import actions
+
+
+class PauseTestCase(CharmTestCase):
+
+    def setUp(self):
+        super(PauseTestCase, self).setUp(
+            actions, ["pause_unit_helper"])
+
+    def test_pauses_services(self):
+        actions.pause([])
+        self.pause_unit_helper.assert_called_once_with('test-config')
+
+
+class ResumeTestCase(CharmTestCase):
+
+    def setUp(self):
+        super(ResumeTestCase, self).setUp(
+            actions, ["resume_unit_helper"])
+
+    def test_pauses_services(self):
+        actions.resume([])
+        self.resume_unit_helper.assert_called_once_with('test-config')
+
+
+class MainTestCase(CharmTestCase):
+
+    def setUp(self):
+        super(MainTestCase, self).setUp(actions, ["action_fail"])
+
+    def test_invokes_action(self):
+        dummy_calls = []
+
+        def dummy_action(args):
+            dummy_calls.append(True)
+
+        with mock.patch.dict(actions.ACTIONS, {"foo": dummy_action}):
+            actions.main(["foo"])
+        self.assertEqual(dummy_calls, [True])
+
+    def test_unknown_action(self):
+        """Unknown actions aren't a traceback."""
+        exit_string = actions.main(["foo"])
+        self.assertEqual("Action foo undefined", exit_string)
+
+    def test_failing_action(self):
+        """Actions which traceback trigger action_fail() calls."""
+        dummy_calls = []
+
+        self.action_fail.side_effect = dummy_calls.append
+
+        def dummy_action(args):
+            raise ValueError("uh oh")
+
+        with mock.patch.dict(actions.ACTIONS, {"foo": dummy_action}):
+            actions.main(["foo"])
+        self.assertEqual(dummy_calls, ["uh oh"])

unit_tests/test_ceph_radosgw_utils.py

@@ -0,0 +1,56 @@
+import utils
+from mock import patch, MagicMock
+
+from test_utils import CharmTestCase
+
+TO_PATCH = [
+]
+
+
+class CephRadosGWUtilTests(CharmTestCase):
+    def setUp(self):
+        super(CephRadosGWUtilTests, self).setUp(utils, TO_PATCH)
+
+    def test_assess_status(self):
+        with patch.object(utils, 'assess_status_func') as asf:
+            callee = MagicMock()
+            asf.return_value = callee
+            utils.assess_status('test-config')
+            asf.assert_called_once_with('test-config')
+            callee.assert_called_once_with()
+
+    @patch.object(utils, 'check_optional_relations')
+    @patch.object(utils, 'REQUIRED_INTERFACES')
+    @patch.object(utils, 'services')
+    @patch.object(utils, 'make_assess_status_func')
+    def test_assess_status_func(self,
+                                make_assess_status_func,
+                                services,
+                                REQUIRED_INTERFACES,
+                                check_optional_relations):
+        services.return_value = 's1'
+        utils.assess_status_func('test-config')
+        # ports=None whilst port checks are disabled.
+        make_assess_status_func.assert_called_once_with(
+            'test-config', REQUIRED_INTERFACES,
+            charm_func=check_optional_relations,
+            services='s1', ports=None)
+
+    def test_pause_unit_helper(self):
+        with patch.object(utils, '_pause_resume_helper') as prh:
+            utils.pause_unit_helper('random-config')
+            prh.assert_called_once_with(utils.pause_unit, 'random-config')
+        with patch.object(utils, '_pause_resume_helper') as prh:
+            utils.resume_unit_helper('random-config')
+            prh.assert_called_once_with(utils.resume_unit, 'random-config')
+
+    @patch.object(utils, 'services')
+    def test_pause_resume_helper(self, services):
+        f = MagicMock()
+        services.return_value = 's1'
+        with patch.object(utils, 'assess_status_func') as asf:
+            asf.return_value = 'assessor'
+            utils._pause_resume_helper(f, 'some-config')
+            asf.assert_called_once_with('some-config')
+            # ports=None whilst port checks are disabled.
+            f.assert_called_once_with('assessor', services='s1', ports=None)