Merge lp:~gnuoy/charms/trusty/cinder-ceph/1453940-stable into lp:~openstack-charmers-archive/charms/precise/cinder-ceph/trunk

[Liam Young]

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #9552 cinder-ceph for gnuoy mp271261
    UNIT OK: passed

Build: http://10.245.162.77:8080/job/charm_unit_test/9552/

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #10411 cinder-ceph for gnuoy mp271261
    LINT OK: passed

Build: http://10.245.162.77:8080/job/charm_lint_check/10411/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #6551 cinder-ceph for gnuoy mp271261
    AMULET FAIL: amulet-test failed

AMULET Results (max last 2 lines):
make: *** [functional_test] Error 1
ERROR:root:Make target returned non-zero.

Full amulet test output: http://paste.ubuntu.com/12514387/
Build: http://10.245.162.77:8080/job/charm_amulet_test/6551/

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #10436 cinder-ceph for gnuoy mp271261
    LINT OK: passed

Build: http://10.245.162.77:8080/job/charm_lint_check/10436/

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #9629 cinder-ceph for gnuoy mp271261
    UNIT OK: passed

Build: http://10.245.162.77:8080/job/charm_unit_test/9629/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #6578 cinder-ceph for gnuoy mp271261
    AMULET FAIL: amulet-test failed

AMULET Results (max last 2 lines):
make: *** [functional_test] Error 1
ERROR:root:Make target returned non-zero.

Full amulet test output: http://paste.ubuntu.com/12521251/
Build: http://10.245.162.77:8080/job/charm_amulet_test/6578/

[Ryan Beisner (1chb1n)]

To save you the time of grepping logs, that amulet fail is real.

00:17:33.288 2015-09-22 13:59:57,026 test_201_ceph_cinderceph_ceph_relation DEBUG: Checking ceph:client to cinder-ceph:ceph relation data...

...

00:17:33.289 Broker request invalid or failed: None

review: Needs Fixing

[Liam Young (gnuoy)]

OSCI failure caused by ceph having not landed yet.

[lp:~gnuoy/charms/trusty/cinder-ceph/1453940-stable]

38. By Liam Young on 2015-09-22

Empty commit to trigger osci

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #9652 cinder-ceph for gnuoy mp271261
    UNIT OK: passed

Build: http://10.245.162.77:8080/job/charm_unit_test/9652/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #6585 cinder-ceph for gnuoy mp271261
    AMULET FAIL: amulet-test failed

AMULET Results (max last 2 lines):
make: *** [functional_test] Error 1
ERROR:root:Make target returned non-zero.

Full amulet test output: http://paste.ubuntu.com/12521979/
Build: http://10.245.162.77:8080/job/charm_amulet_test/6585/

[lp:~gnuoy/charms/trusty/cinder-ceph/1453940-stable]

39. By Liam Young on 2015-09-22

Empty commit to trigger osci

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #9654 cinder-ceph for gnuoy mp271261
    UNIT OK: passed

Build: http://10.245.162.77:8080/job/charm_unit_test/9654/

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #10463 cinder-ceph for gnuoy mp271261
    LINT OK: passed

Build: http://10.245.162.77:8080/job/charm_lint_check/10463/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #6606 cinder-ceph for gnuoy mp271261
    AMULET FAIL: amulet-test failed

AMULET Results (max last 2 lines):
make: *** [functional_test] Error 1
ERROR:root:Make target returned non-zero.

Full amulet test output: http://paste.ubuntu.com/12526301/
Build: http://10.245.162.77:8080/job/charm_amulet_test/6606/

[lp:~gnuoy/charms/trusty/cinder-ceph/1453940-stable]

40. By Liam Young on 2015-09-23

Empty commit to kick osci

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #9712 cinder-ceph for gnuoy mp271261
    UNIT OK: passed

Build: http://10.245.162.77:8080/job/charm_unit_test/9712/

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #10524 cinder-ceph for gnuoy mp271261
    LINT OK: passed

Build: http://10.245.162.77:8080/job/charm_lint_check/10524/

[Liam Young (gnuoy)]

Since the latest revision of ceph is still not in the charm store I tests an amulet run manually with the latest ceph from stable. It LGTM http://paste.ubuntu.com/12529898/

[Chris Glass (tribaal)]

After further testing on a bastion, this should pass. In order to cut down on wait time I'm going to go ahead and merge this already.

+1

review: Approve

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #6666 cinder-ceph for gnuoy mp271261
    AMULET OK: passed

Build: http://10.245.162.77:8080/job/charm_amulet_test/6666/

=== modified file 'hooks/charmhelpers/cli/__init__.py'
--- hooks/charmhelpers/cli/__init__.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/cli/__init__.py	2015-09-23 07:31:24 +0000
@@ -152,15 +152,11 @@
         arguments = self.argument_parser.parse_args()
         argspec = inspect.getargspec(arguments.func)
         vargs = []
-        kwargs = {}
         for arg in argspec.args:
             vargs.append(getattr(arguments, arg))
         if argspec.varargs:
             vargs.extend(getattr(arguments, argspec.varargs))
-        if argspec.keywords:
-            for kwarg in argspec.keywords.items():
-                kwargs[kwarg] = getattr(arguments, kwarg)
-        output = arguments.func(*vargs, **kwargs)
+        output = arguments.func(*vargs)
         if getattr(arguments.func, '_cli_test_command', False):
             self.exit_code = 0 if output else 1
             output = ''

=== modified file 'hooks/charmhelpers/cli/commands.py'
--- hooks/charmhelpers/cli/commands.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/cli/commands.py	2015-09-23 07:31:24 +0000
@@ -26,7 +26,7 @@
 """
 Import the sub-modules which have decorated subcommands to register with chlp.
 """
-import host  # noqa
-import benchmark  # noqa
-import unitdata  # noqa
-from charmhelpers.core import hookenv  # noqa
+from . import host  # noqa
+from . import benchmark  # noqa
+from . import unitdata  # noqa
+from . import hookenv  # noqa

=== added file 'hooks/charmhelpers/cli/hookenv.py'
--- hooks/charmhelpers/cli/hookenv.py	1970-01-01 00:00:00 +0000
+++ hooks/charmhelpers/cli/hookenv.py	2015-09-23 07:31:24 +0000
@@ -0,0 +1,23 @@
+# 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/>.
+
+from . import cmdline
+from charmhelpers.core import hookenv
+
+
+cmdline.subcommand('relation-id')(hookenv.relation_id._wrapped)
+cmdline.subcommand('service-name')(hookenv.service_name)
+cmdline.subcommand('remote-service-name')(hookenv.remote_service_name._wrapped)

=== modified file 'hooks/charmhelpers/contrib/openstack/context.py'
--- hooks/charmhelpers/contrib/openstack/context.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/contrib/openstack/context.py	2015-09-23 07:31:24 +0000
@@ -483,13 +483,15 @@
 
         log('Generating template context for ceph', level=DEBUG)
         mon_hosts = []
-        auth = None
-        key = None
-        use_syslog = str(config('use-syslog')).lower()
+        ctxt = {
+            'use_syslog': str(config('use-syslog')).lower()
+        }
         for rid in relation_ids('ceph'):
             for unit in related_units(rid):
-                auth = relation_get('auth', rid=rid, unit=unit)
-                key = relation_get('key', rid=rid, unit=unit)
+                if not ctxt.get('auth'):
+                    ctxt['auth'] = relation_get('auth', rid=rid, unit=unit)
+                if not ctxt.get('key'):
+                    ctxt['key'] = relation_get('key', rid=rid, unit=unit)
                 ceph_pub_addr = relation_get('ceph-public-address', rid=rid,
                                              unit=unit)
                 unit_priv_addr = relation_get('private-address', rid=rid,
@@ -498,10 +500,7 @@
                 ceph_addr = format_ipv6_addr(ceph_addr) or ceph_addr
                 mon_hosts.append(ceph_addr)
 
-        ctxt = {'mon_hosts': ' '.join(sorted(mon_hosts)),
-                'auth': auth,
-                'key': key,
-                'use_syslog': use_syslog}
+        ctxt['mon_hosts'] = ' '.join(sorted(mon_hosts))
 
         if not os.path.isdir('/etc/ceph'):
             os.mkdir('/etc/ceph')

=== modified file 'hooks/charmhelpers/contrib/openstack/utils.py'
--- hooks/charmhelpers/contrib/openstack/utils.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/contrib/openstack/utils.py	2015-09-23 07:31:24 +0000
@@ -1,5 +1,3 @@
-#!/usr/bin/python
-
 # Copyright 2014-2015 Canonical Limited.
 #
 # This file is part of charm-helpers.
@@ -167,9 +165,9 @@
         error_out(e)
 
 
-def get_os_version_codename(codename):
+def get_os_version_codename(codename, version_map=OPENSTACK_CODENAMES):
     '''Determine OpenStack version number from codename.'''
-    for k, v in six.iteritems(OPENSTACK_CODENAMES):
+    for k, v in six.iteritems(version_map):
         if v == codename:
             return k
     e = 'Could not derive OpenStack version for '\
@@ -392,7 +390,11 @@
     import apt_pkg as apt
     src = config('openstack-origin')
     cur_vers = get_os_version_package(package)
-    available_vers = get_os_version_install_source(src)
+    if "swift" in package:
+        codename = get_os_codename_install_source(src)
+        available_vers = get_os_version_codename(codename, SWIFT_CODENAMES)
+    else:
+        available_vers = get_os_version_install_source(src)
     apt.init()
     return apt.version_compare(available_vers, cur_vers) == 1
 

=== modified file 'hooks/charmhelpers/contrib/storage/linux/ceph.py'
--- hooks/charmhelpers/contrib/storage/linux/ceph.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/contrib/storage/linux/ceph.py	2015-09-23 07:31:24 +0000
@@ -28,6 +28,7 @@
 import shutil
 import json
 import time
+import uuid
 
 from subprocess import (
     check_call,
@@ -35,8 +36,10 @@
     CalledProcessError,
 )
 from charmhelpers.core.hookenv import (
+    local_unit,
     relation_get,
     relation_ids,
+    relation_set,
     related_units,
     log,
     DEBUG,
@@ -411,17 +414,52 @@
 
     The API is versioned and defaults to version 1.
     """
-    def __init__(self, api_version=1):
+    def __init__(self, api_version=1, request_id=None):
         self.api_version = api_version
+        if request_id:
+            self.request_id = request_id
+        else:
+            self.request_id = str(uuid.uuid1())
         self.ops = []
 
     def add_op_create_pool(self, name, replica_count=3):
         self.ops.append({'op': 'create-pool', 'name': name,
                          'replicas': replica_count})
 
+    def set_ops(self, ops):
+        """Set request ops to provided value.
+
+        Useful for injecting ops that come from a previous request
+        to allow comparisons to ensure validity.
+        """
+        self.ops = ops
+
     @property
     def request(self):
-        return json.dumps({'api-version': self.api_version, 'ops': self.ops})
+        return json.dumps({'api-version': self.api_version, 'ops': self.ops,
+                           'request-id': self.request_id})
+
+    def _ops_equal(self, other):
+        if len(self.ops) == len(other.ops):
+            for req_no in range(0, len(self.ops)):
+                for key in ['replicas', 'name', 'op']:
+                    if self.ops[req_no][key] != other.ops[req_no][key]:
+                        return False
+        else:
+            return False
+        return True
+
+    def __eq__(self, other):
+        if not isinstance(other, self.__class__):
+            return False
+        if self.api_version == other.api_version and \
+                self._ops_equal(other):
+            return True
+        else:
+            return False
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
 
 
 class CephBrokerRsp(object):
@@ -431,14 +469,198 @@
 
     The API is versioned and defaults to version 1.
     """
+
     def __init__(self, encoded_rsp):
         self.api_version = None
         self.rsp = json.loads(encoded_rsp)
 
     @property
+    def request_id(self):
+        return self.rsp.get('request-id')
+
+    @property
     def exit_code(self):
         return self.rsp.get('exit-code')
 
     @property
     def exit_msg(self):
         return self.rsp.get('stderr')
+
+
+# Ceph Broker Conversation:
+# If a charm needs an action to be taken by ceph it can create a CephBrokerRq
+# and send that request to ceph via the ceph relation. The CephBrokerRq has a
+# unique id so that the client can identity which CephBrokerRsp is associated
+# with the request. Ceph will also respond to each client unit individually
+# creating a response key per client unit eg glance/0 will get a CephBrokerRsp
+# via key broker-rsp-glance-0
+#
+# To use this the charm can just do something like:
+#
+# from charmhelpers.contrib.storage.linux.ceph import (
+#     send_request_if_needed,
+#     is_request_complete,
+#     CephBrokerRq,
+# )
+#
+# @hooks.hook('ceph-relation-changed')
+# def ceph_changed():
+#     rq = CephBrokerRq()
+#     rq.add_op_create_pool(name='poolname', replica_count=3)
+#
+#     if is_request_complete(rq):
+#         <Request complete actions>
+#     else:
+#         send_request_if_needed(get_ceph_request())
+#
+# CephBrokerRq and CephBrokerRsp are serialized into JSON. Below is an example
+# of glance having sent a request to ceph which ceph has successfully processed
+#  'ceph:8': {
+#      'ceph/0': {
+#          'auth': 'cephx',
+#          'broker-rsp-glance-0': '{"request-id": "0bc7dc54", "exit-code": 0}',
+#          'broker_rsp': '{"request-id": "0da543b8", "exit-code": 0}',
+#          'ceph-public-address': '10.5.44.103',
+#          'key': 'AQCLDttVuHXINhAAvI144CB09dYchhHyTUY9BQ==',
+#          'private-address': '10.5.44.103',
+#      },
+#      'glance/0': {
+#          'broker_req': ('{"api-version": 1, "request-id": "0bc7dc54", '
+#                         '"ops": [{"replicas": 3, "name": "glance", '
+#                         '"op": "create-pool"}]}'),
+#          'private-address': '10.5.44.109',
+#      },
+#  }
+
+def get_previous_request(rid):
+    """Return the last ceph broker request sent on a given relation
+
+    @param rid: Relation id to query for request
+    """
+    request = None
+    broker_req = relation_get(attribute='broker_req', rid=rid,
+                              unit=local_unit())
+    if broker_req:
+        request_data = json.loads(broker_req)
+        request = CephBrokerRq(api_version=request_data['api-version'],
+                               request_id=request_data['request-id'])
+        request.set_ops(request_data['ops'])
+
+    return request
+
+
+def get_request_states(request):
+    """Return a dict of requests per relation id with their corresponding
+       completion state.
+
+    This allows a charm, which has a request for ceph, to see whether there is
+    an equivalent request already being processed and if so what state that
+    request is in.
+
+    @param request: A CephBrokerRq object
+    """
+    complete = []
+    requests = {}
+    for rid in relation_ids('ceph'):
+        complete = False
+        previous_request = get_previous_request(rid)
+        if request == previous_request:
+            sent = True
+            complete = is_request_complete_for_rid(previous_request, rid)
+        else:
+            sent = False
+            complete = False
+
+        requests[rid] = {
+            'sent': sent,
+            'complete': complete,
+        }
+
+    return requests
+
+
+def is_request_sent(request):
+    """Check to see if a functionally equivalent request has already been sent
+
+    Returns True if a similair request has been sent
+
+    @param request: A CephBrokerRq object
+    """
+    states = get_request_states(request)
+    for rid in states.keys():
+        if not states[rid]['sent']:
+            return False
+
+    return True
+
+
+def is_request_complete(request):
+    """Check to see if a functionally equivalent request has already been
+    completed
+
+    Returns True if a similair request has been completed
+
+    @param request: A CephBrokerRq object
+    """
+    states = get_request_states(request)
+    for rid in states.keys():
+        if not states[rid]['complete']:
+            return False
+
+    return True
+
+
+def is_request_complete_for_rid(request, rid):
+    """Check if a given request has been completed on the given relation
+
+    @param request: A CephBrokerRq object
+    @param rid: Relation ID
+    """
+    broker_key = get_broker_rsp_key()
+    for unit in related_units(rid):
+        rdata = relation_get(rid=rid, unit=unit)
+        if rdata.get(broker_key):
+            rsp = CephBrokerRsp(rdata.get(broker_key))
+            if rsp.request_id == request.request_id:
+                if not rsp.exit_code:
+                    return True
+        else:
+            # The remote unit sent no reply targeted at this unit so either the
+            # remote ceph cluster does not support unit targeted replies or it
+            # has not processed our request yet.
+            if rdata.get('broker_rsp'):
+                request_data = json.loads(rdata['broker_rsp'])
+                if request_data.get('request-id'):
+                    log('Ignoring legacy broker_rsp without unit key as remote '
+                        'service supports unit specific replies', level=DEBUG)
+                else:
+                    log('Using legacy broker_rsp as remote service does not '
+                        'supports unit specific replies', level=DEBUG)
+                    rsp = CephBrokerRsp(rdata['broker_rsp'])
+                    if not rsp.exit_code:
+                        return True
+
+    return False
+
+
+def get_broker_rsp_key():
+    """Return broker response key for this unit
+
+    This is the key that ceph is going to use to pass request status
+    information back to this unit
+    """
+    return 'broker-rsp-' + local_unit().replace('/', '-')
+
+
+def send_request_if_needed(request):
+    """Send broker request if an equivalent request has not already been sent
+
+    @param request: A CephBrokerRq object
+    """
+    if is_request_sent(request):
+        log('Request already sent but not complete, not sending new request',
+            level=DEBUG)
+    else:
+        for rid in relation_ids('ceph'):
+            log('Sending request {}'.format(request.request_id), level=DEBUG)
+            relation_set(relation_id=rid, broker_req=request.request)

=== modified file 'hooks/charmhelpers/contrib/storage/linux/utils.py'
--- hooks/charmhelpers/contrib/storage/linux/utils.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/contrib/storage/linux/utils.py	2015-09-23 07:31:24 +0000
@@ -43,9 +43,10 @@
 
     :param block_device: str: Full path of block device to clean.
     '''
+    # https://github.com/ceph/ceph/commit/fdd7f8d83afa25c4e09aaedd90ab93f3b64a677b
     # sometimes sgdisk exits non-zero; this is OK, dd will clean up
-    call(['sgdisk', '--zap-all', '--mbrtogpt',
-          '--clear', block_device])
+    call(['sgdisk', '--zap-all', '--', block_device])
+    call(['sgdisk', '--clear', '--mbrtogpt', '--', block_device])
     dev_end = check_output(['blockdev', '--getsz',
                             block_device]).decode('UTF-8')
     gpt_end = int(dev_end.split()[0]) - 100

=== modified file 'hooks/charmhelpers/core/hookenv.py'
--- hooks/charmhelpers/core/hookenv.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/core/hookenv.py	2015-09-23 07:31:24 +0000
@@ -34,23 +34,6 @@
 import tempfile
 from subprocess import CalledProcessError
 
-try:
-    from charmhelpers.cli import cmdline
-except ImportError as e:
-    # due to the anti-pattern of partially synching charmhelpers directly
-    # into charms, it's possible that charmhelpers.cli is not available;
-    # if that's the case, they don't really care about using the cli anyway,
-    # so mock it out
-    if str(e) == 'No module named cli':
-        class cmdline(object):
-            @classmethod
-            def subcommand(cls, *args, **kwargs):
-                def _wrap(func):
-                    return func
-                return _wrap
-    else:
-        raise
-
 import six
 if not six.PY3:
     from UserDict import UserDict
@@ -91,6 +74,7 @@
         res = func(*args, **kwargs)
         cache[key] = res
         return res
+    wrapper._wrapped = func
     return wrapper
 
 
@@ -190,7 +174,6 @@
     return os.environ.get('JUJU_RELATION', None)
 
 
[e-mail]()
 @cached
 def relation_id(relation_name=None, service_or_unit=None):
     """The relation ID for the current or a specified relation"""
@@ -216,13 +199,11 @@
     return os.environ.get('JUJU_REMOTE_UNIT', None)
 
 
[e-mail]()
 def service_name():
     """The name service group this unit belongs to"""
     return local_unit().split('/')[0]
 
 
[e-mail]()
 @cached
 def remote_service_name(relid=None):
     """The remote service name for a given relation-id (or the current relation)"""

=== modified file 'hooks/charmhelpers/core/host.py'
--- hooks/charmhelpers/core/host.py	2015-08-10 16:34:26 +0000
+++ hooks/charmhelpers/core/host.py	2015-09-23 07:31:24 +0000
@@ -72,7 +72,7 @@
     stopped = service_stop(service_name)
     # XXX: Support systemd too
     override_path = os.path.join(
-        init_dir, '{}.conf.override'.format(service_name))
+        init_dir, '{}.override'.format(service_name))
     with open(override_path, 'w') as fh:
         fh.write("manual\n")
     return stopped
@@ -86,7 +86,7 @@
     if init_dir is None:
         init_dir = "/etc/init"
     override_path = os.path.join(
-        init_dir, '{}.conf.override'.format(service_name))
+        init_dir, '{}.override'.format(service_name))
     if os.path.exists(override_path):
         os.unlink(override_path)
     started = service_start(service_name)

=== modified file 'hooks/cinder_hooks.py'
--- hooks/cinder_hooks.py	2015-09-16 12:55:55 +0000
+++ hooks/cinder_hooks.py	2015-09-23 07:31:24 +0000
@@ -17,12 +17,9 @@
     UnregisteredHookError,
     config,
     service_name,
-    relation_get,
     relation_set,
     relation_ids,
     log,
-    INFO,
-    ERROR,
 )
 from charmhelpers.fetch import apt_install, apt_update
 from charmhelpers.core.host import (
@@ -30,9 +27,10 @@
     service_restart,
 )
 from charmhelpers.contrib.storage.linux.ceph import (
+    send_request_if_needed,
+    is_request_complete,
     ensure_ceph_keyring,
     CephBrokerRq,
-    CephBrokerRsp,
     delete_keyring,
 )
 from charmhelpers.payload.execd import execd_preinstall
@@ -55,6 +53,14 @@
         os.mkdir('/etc/ceph')
 
 
+def get_ceph_request():
+    service = service_name()
+    rq = CephBrokerRq()
+    replicas = config('ceph-osd-replication-count')
+    rq.add_op_create_pool(name=service, replica_count=replicas)
+    return rq
+
+
 @hooks.hook('ceph-relation-changed')
 @restart_on_change(restart_map())
 def ceph_changed():
@@ -68,32 +74,17 @@
         log('Could not create ceph keyring: peer not ready?')
         return
 
-    settings = relation_get()
-    if settings and 'broker_rsp' in settings:
-        rsp = CephBrokerRsp(settings['broker_rsp'])
-        # Non-zero return code implies failure
-        if rsp.exit_code:
-            log("Ceph broker request failed (rc=%s, msg=%s)" %
-                (rsp.exit_code, rsp.exit_msg), level=ERROR)
-            return
-
-        log("Ceph broker request succeeded (rc=%s, msg=%s)" %
-            (rsp.exit_code, rsp.exit_msg), level=INFO)
+    if is_request_complete(get_ceph_request()):
+        log('Request complete')
         CONFIGS.write_all()
         set_ceph_env_variables(service=service)
         for rid in relation_ids('storage-backend'):
             storage_backend(rid)
-
         # Ensure that cinder-volume is restarted since only now can we
         # guarantee that ceph resources are ready.
         service_restart('cinder-volume')
     else:
-        rq = CephBrokerRq()
-        replicas = config('ceph-osd-replication-count')
-        rq.add_op_create_pool(name=service, replica_count=replicas)
-        for rid in relation_ids('ceph'):
-            relation_set(relation_id=rid, broker_req=rq.request)
-            log("Request(s) sent to Ceph broker (rid=%s)" % (rid))
+        send_request_if_needed(get_ceph_request())
 
 
 @hooks.hook('ceph-relation-broken')

=== modified file 'tests/basic_deployment.py'
--- tests/basic_deployment.py	2015-08-10 17:28:35 +0000
+++ tests/basic_deployment.py	2015-09-23 07:31:24 +0000
@@ -4,6 +4,7 @@
 Basic cinder-ceph functional test.
 """
 import amulet
+import json
 import time
 
 from charmhelpers.contrib.openstack.amulet.deployment import (
@@ -279,40 +280,83 @@
             amulet.raise_status(amulet.FAIL,
                                 msg='cinder endpoint: {}'.format(ret))
 
+    def validate_broker_req(self, unit, relation, expected):
+        rel_data = json.loads(unit.relation(
+            relation[0],
+            relation[1])['broker_req'])
+        if rel_data['api-version'] != expected['api-version']:
+            return "Broker request api mismatch"
+        for index in range(0, len(rel_data['ops'])):
+            actual_op = rel_data['ops'][index]
+            expected_op = expected['ops'][index]
+            for key in ['op', 'name', 'replicas']:
+                if actual_op[key] == expected_op[key]:
+                    u.log.debug("OK op {} key {}".format(index, key))
+                else:
+                    return "Mismatch, op: {} key: {}".format(index, key)
+        return None
+
+    def get_broker_request(self):
+        client_unit = self.cinder_ceph_sentry
+        broker_req = json.loads(client_unit.relation(
+            'ceph',
+            'ceph:client')['broker_req'])
+        return broker_req
+
+    def get_broker_response(self):
+        broker_request = self.get_broker_request()
+        response_key = "broker-rsp-cinder-ceph-0"
+        ceph_sentrys = [self.ceph0_sentry, self.ceph1_sentry, self.ceph2_sentry]
+        for sentry in ceph_sentrys:
+            relation_data = sentry.relation('client', 'cinder-ceph:ceph')
+            if relation_data.get(response_key):
+                broker_response = json.loads(relation_data[response_key])
+                if broker_request['request-id'] == broker_response['request-id']:
+                    return broker_response
+
     def test_200_cinderceph_ceph_ceph_relation(self):
         u.log.debug('Checking cinder-ceph:ceph to ceph:client '
                     'relation data...')
         unit = self.cinder_ceph_sentry
         relation = ['ceph', 'ceph:client']
 
-        req = ('{"api-version": 1, "ops": [{"replicas": 3, "name": '
-               '"cinder-ceph", "op": "create-pool"}]}')
-
+        req = {
+           "api-version": 1,
+           "ops": [{"replicas": 3, "name": "cinder-ceph", "op": "create-pool"}]
+        }
         expected = {
             'private-address': u.valid_ip,
-            'broker_req': req
+            'broker_req': u.not_null,
         }
         ret = u.validate_relation_data(unit, relation, expected)
         if ret:
             msg = u.relation_error('cinder-ceph ceph', ret)
             amulet.raise_status(amulet.FAIL, msg=msg)
+        ret = self.validate_broker_req(unit, relation, req)
+        if ret:
+            msg = u.relation_error('cinder-ceph ceph', ret)
+            amulet.raise_status(amulet.FAIL, msg=msg)
 
     def test_201_ceph_cinderceph_ceph_relation(self):
         u.log.debug('Checking ceph:client to cinder-ceph:ceph '
                     'relation data...')
-        unit = self.ceph0_sentry
+        response_key = "broker-rsp-cinder-ceph-0"
+        ceph_unit = self.ceph0_sentry
         relation = ['client', 'cinder-ceph:ceph']
         expected = {
             'key': u.not_null,
             'private-address': u.valid_ip,
-            'broker_rsp': '{"exit-code": 0}',
             'ceph-public-address': u.valid_ip,
-            'auth': 'none'
+            'auth': 'none',
         }
-        ret = u.validate_relation_data(unit, relation, expected)
+        ret = u.validate_relation_data(ceph_unit, relation, expected)
         if ret:
             msg = u.relation_error('cinder cinder-ceph storage-backend', ret)
             amulet.raise_status(amulet.FAIL, msg=msg)
+        broker_response = self.get_broker_response()
+        if not broker_response or broker_response['exit-code'] != 0:
+            msg='Broker request invalid or failed: {}'.format(broker_response)
+            amulet.raise_status(amulet.FAIL, msg=msg)
 
     def test_202_cinderceph_cinder_backend_relation(self):
         u.log.debug('Checking cinder-ceph:storage-backend to '

=== modified file 'unit_tests/test_cinder_hooks.py'
--- unit_tests/test_cinder_hooks.py	2015-09-16 12:55:55 +0000
+++ unit_tests/test_cinder_hooks.py	2015-09-23 07:31:24 +0000
@@ -18,11 +18,12 @@
     'register_configs',
     'restart_map',
     'set_ceph_env_variables',
+    'is_request_complete',
+    'send_request_if_needed',
     'CONFIGS',
     # charmhelpers.core.hookenv
     'config',
     'relation_ids',
-    'relation_get',
     'relation_set',
     'service_name',
     'service_restart',
@@ -69,9 +70,8 @@
     @patch('charmhelpers.core.hookenv.config')
     def test_ceph_changed(self, mock_config):
         '''It ensures ceph assets created on ceph changed'''
+        self.is_request_complete.return_value = True
         self.CONFIGS.complete_contexts.return_value = ['ceph']
-        rsp = json.dumps({'exit_code': 0})
-        self.relation_get.return_value = {'broker_rsp': rsp}
         self.service_name.return_value = 'cinder'
         self.ensure_ceph_keyring.return_value = True
         hooks.hooks.execute(['hooks/ceph-relation-changed'])
@@ -81,11 +81,27 @@
         self.assertTrue(self.CONFIGS.write_all.called)
         self.set_ceph_env_variables.assert_called_with(service='cinder')
 
+    @patch.object(hooks, 'get_ceph_request')
+    @patch('charmhelpers.core.hookenv.config')
+    def test_ceph_changed_newrq(self, mock_config, mock_get_ceph_request):
+        '''It ensures ceph assets created on ceph changed'''
+        mock_get_ceph_request.return_value = 'cephreq'
+        self.is_request_complete.return_value = False
+        self.CONFIGS.complete_contexts.return_value = ['ceph']
+        self.service_name.return_value = 'cinder'
+        self.ensure_ceph_keyring.return_value = True
+        hooks.hooks.execute(['hooks/ceph-relation-changed'])
+        self.ensure_ceph_keyring.assert_called_with(service='cinder',
+                                                    user='cinder',
+                                                    group='cinder')
+        self.send_request_if_needed.assert_called_with('cephreq')
+
     @patch('charmhelpers.core.hookenv.config')
     def test_ceph_changed_no_keys(self, mock_config):
         '''It ensures ceph assets created on ceph changed'''
         self.CONFIGS.complete_contexts.return_value = ['ceph']
         self.service_name.return_value = 'cinder'
+        self.is_request_complete.return_value = True
         self.ensure_ceph_keyring.return_value = False
         hooks.hooks.execute(['hooks/ceph-relation-changed'])
         # NOTE(jamespage): If ensure_ceph keyring fails, then