Merge lp:~stub/charms/trusty/cassandra/spike into lp:charms/trusty/cassandra

[Stuart Bishop]

Rework the Cassandra charm to rely on leadership, and maintain the unit and service status while we are at it.

By using leadership, we increase reliability and are able to remove a large amount of code that was needed to cope with flaky coordination.

[Stuart Bishop (stub)]

The charmhelpers coordinator module has not landed and can be reviewed at https://code.launchpad.net/~stub/charm-helpers/coordinator/+merge/261267

[Stuart Bishop (stub)]

I've also taken the opportunity to update the default versions. With 2.1.7, the 2.1 series is finally the recommended version and DataStax has released 4.7 (now based on Cassandra 2.1).

[Tim Van Steenburgh (tvansteenburgh)]

Hey stub, there are some failures in the tests. You may already be aware, but I wanted to point it out b/c the test runner is, for reasons unknown to me at this point, exiting 0 in spite of the failures.

Check out the console logs here: http://reports.vapour.ws/charm-test-details/charm-bundle-test-parent-259

review: Needs Fixing

[Stuart Bishop (stub)]

On 26 June 2015 at 22:37, Tim Van Steenburgh <email address hidden> wrote:

> Check out the console logs here: http://reports.vapour.ws/charm-test-details/charm-bundle-test-parent-259

These seem to be for an earlier revision of the charm. In particular,
the Test21Deployment tests should be failing because that TestCase no
longer exists, yet they are being run. This broke with my work on the
23rd, so I've fixed the Makefile and tests.yaml and pushed.

--
Stuart Bishop <email address hidden>

[Stuart Bishop (stub)]

I have fixed the error codes bug, so failing tests will be reported as failed in Jenkins.

[Stuart Bishop (stub)]

Another run at http://reports.vapour.ws/charm-test-details/charm-bundle-test-parent-263

http://reports.vapour.ws/all-bundle-and-charm-results/charm-bundle-test-parent-263/charm/charm-testing-hp/3 shows test_external_mount failing at 11:07, when amulet fails to read the contents of a remote directory. However, http://data.vapour.ws/charm-test/charm-bundle-test-hp-120-all-machines-log clearly shows the directory being created at 11:04:34 (the migration starts at 11:04:29).

[Stuart Bishop (stub)]

http://reports.vapour.ws/charm-test-details/charm-bundle-test-parent-264 has a complete successful run on AWS. The other clouds are having problems with provisioning nodes.

[Stuart Bishop (stub)]

I've also opened Bug #1470024 on the sentry.directory_listing failure, which is breaking the external storage test on hpcloud (I added extra debugging, attempting to list the contents of the root directory before attempting the cassandra specific directory and demonstrating this is an Amulet/Juju/Cloud failure rather than a charm failure).

[lp:~stub/charms/trusty/cassandra/spike]

362. By Tim Van Steenburgh on 2015-07-01

[stub] Rework the Cassandra charm to rely on leadership, and maintain
the unit and service status while we are at it.

By using leadership, we increase reliability and are able to remove a
large amount of code that was needed to cope with flaky coordination.

[Tim Van Steenburgh (tvansteenburgh)]

LGTM, merging on the merit of tests passing on AWS. Other failures are clearly tooling-related, not problems in the charm. Thanks for filing the bug against amulet.

review: Approve

=== modified file 'Makefile'
--- Makefile	2015-04-28 16:20:05 +0000
+++ Makefile	2015-06-30 07:36:26 +0000
@@ -34,6 +34,10 @@
 PIP=.venv3/bin/pip3.4 -q
 NOSETESTS=.venv3/bin/nosetests-3.4 -sv
 
+# Set pipefail so we can get sane error codes while tagging test output
+# with ts(1)
+SHELL=bash -o pipefail
+
 deps: packages venv3
 
 lint: deps
@@ -41,16 +45,17 @@
 	free --human
 	charm proof $(CHARM_DIR)
 	flake8 \
-	    --ignore=E402 \
+	    --ignore=E402,E265 \
 	    --exclude=charmhelpers,.venv2,.venv3 hooks tests testing
+	@echo OK: Lint free `date`
 
 unittest: lint
 	$(NOSETESTS) \
 	    tests.test_actions        --cover-package=actions \
 	    tests.test_helpers        --cover-package=helpers \
-	    tests.test_rollingrestart --cover-package=rollingrestart \
 	    tests.test_definitions    --cover-package=definitions \
 	    --with-coverage --cover-branches
+	@echo OK: Unit tests pass `date`
 
 test: unittest
 	AMULET_TIMEOUT=3600 \
@@ -62,11 +67,11 @@
 	AMULET_TIMEOUT=5400 \
 	$(NOSETESTS) tests.test_integration:Test1UnitDeployment 2>&1 | ts
 	
-21test: unittest Test21Deployment
-Test21Deployment: deps
+20test: unittest Test21Deployment
+Test20Deployment: deps
 	date
 	AMULET_TIMEOUT=5400 \
-	$(NOSETESTS) tests.test_integration:Test21Deployment 2>&1 | ts
+	$(NOSETESTS) tests.test_integration:Test20Deployment 2>&1 | ts
 	
 3test: unittest Test3UnitDeployment
 Test3UnitDeployment: deps
@@ -95,7 +100,6 @@
 	$(NOSETESTS) \
 	    tests.test_actions        --cover-package=actions \
 	    tests.test_helpers        --cover-package=helpers \
-	    tests.test_rollingrestart --cover-package=rollingrestart \
 	    tests.test_definitions    --cover-package=definitions \
 	    --with-coverage --cover-branches \
 	    --cover-html --cover-html-dir=coverage \

=== modified file 'README.md'
--- README.md	2015-03-07 09:27:02 +0000
+++ README.md	2015-06-30 07:36:26 +0000
@@ -14,9 +14,9 @@
 # Editions
 
 This charm supports Apache Cassandra 2.0, Apache Cassandra 2.1, and
-Datastax Enterprise 4.6. The default is Apache Cassandra 2.0.
+Datastax Enterprise 4.7. The default is Apache Cassandra 2.1.
 
-To use Apache Cassandra 2.1, specify the Apache Cassandra 2.1 archive source
+To use Apache Cassandra 2.0, specify the Apache Cassandra 2.0 archive source
 in the `install_sources` config setting when deploying.
 
 To use Datastax Enterprise, set the `edition` config setting to `dse`
@@ -83,23 +83,16 @@
 `install_keys` configuration item. Place the DataStax packages in a
 local archive to avoid downloading from datastax.com.
 
-The Cassandra Python driver and some dependencies are installed using
-`pip(1)`. Set the `http_proxy` config item to direct its downloads via
-a proxy server, such as squid or devpi.
-
 
 ## Oracle Java SE
 
-Cassandra recommends using Oracle Java SE 7. Unfortunately, this
+Cassandra recommends using Oracle Java SE 8. Unfortunately, this
 software is accessible only after accepting Oracle's click-through
 license making deployments using it much more cumbersome. You will need
-to download the Oracle Java SE 7 Server Runtime for Linux, and place the
+to download the Oracle Java SE 8 Server Runtime for Linux, and place the
 tarball at a URL accessible to your deployed units. The config item
 `private_jre_url` needs to be set to this URL.
 
-The recommended Java version is expected to change, as Java SE 7 public
-releases cease as of April 2015.
-
 
 # Usage
 

=== modified file 'charm-helpers.yaml'
--- charm-helpers.yaml	2015-04-03 15:42:21 +0000
+++ charm-helpers.yaml	2015-06-30 07:36:26 +0000
@@ -18,11 +18,10 @@
 #branch: lp:charm-helpers
 branch: lp:~stub/charm-helpers/integration
 include:
-    - __init__
+    - coordinator
     - core
     - fetch
     - contrib.charmsupport
     - contrib.templating.jinja
-    - contrib.peerstorage
     - contrib.network.ufw
-    - contrib.unison
+    - contrib.benchmark

=== modified file 'config.yaml'
--- config.yaml	2015-03-07 09:27:02 +0000
+++ config.yaml	2015-06-30 07:36:26 +0000
@@ -35,7 +35,7 @@
             If you are using Datastax Enterprise, you will need to
             override one defaults with your own username and password.
         default: |
-            - deb http://www.apache.org/dist/cassandra/debian 20x main
+            - deb http://www.apache.org/dist/cassandra/debian 21x main
         #   - deb http://debian.datastax.com/community stable main
         #   DSE requires you to register and add your username/password here.
         #   - deb http://un:[e-mail]/enterprise stable main
@@ -207,15 +207,6 @@
         default: 256
         description: Number of tokens per node.
             
-    # Not implemented.
-    # force_seed_nodes:
-    #     type: string
-    #     default: ""
-    #     description: >
-    #         A comma separated list of seed node ip addresses. This is
-    #         useful if the cluster being created in this juju environment
-    #         is part of a larger cluser and the seed nodes are remote.
-            
     # Topology of the service in the cluster.
     datacenter:
         type: string
@@ -312,11 +303,3 @@
             performance problems and even exaust the server heap. Adjust
             the thresholds here if you understand the dangers and want
             to scan more tombstones anyway.
-            
-    post_bootstrap_delay:
-        type: int
-        default: 120
-        description: >
-            Testing option - do not change. When adding multiple new
-            nodes to a cluster, you are supposed to wait two minutes
-            between nodes. I set this to zero to speed up test runs.

=== modified file 'hooks/actions.py'
--- hooks/actions.py	2015-04-12 16:20:50 +0000
+++ hooks/actions.py	2015-06-30 07:36:26 +0000
@@ -23,20 +23,22 @@
 import shlex
 import subprocess
 from textwrap import dedent
+import time
 import urllib.request
 
 from charmhelpers import fetch
 from charmhelpers.contrib.charmsupport import nrpe
 from charmhelpers.contrib.network import ufw
 from charmhelpers.contrib.templating import jinja
-from charmhelpers.contrib import unison
 from charmhelpers.core import hookenv, host
 from charmhelpers.core.fstab import Fstab
-from charmhelpers.core.hookenv import ERROR, WARNING
-
+from charmhelpers.core.hookenv import DEBUG, ERROR, WARNING
+
+import cassandra
+
+from coordinator import coordinator
 import helpers
 import relations
-import rollingrestart
 
 
 # These config keys cannot be changed after service deployment.
@@ -54,7 +56,6 @@
     'native_transport_port',
     'partitioner',
     'num_tokens',
-    'force_seed_nodes',
     'max_heap_size',
     'heap_newsize',
     'authorizer',
@@ -83,8 +84,7 @@
     'nagios_heapchk_warn_pct',
     'nagios_heapchk_crit_pct',
     'nagios_disk_warn_pct',
-    'nagios_disk_crit_pct',
-    'post_bootstrap_delay'])
+    'nagios_disk_crit_pct'])
 
 
 def action(func):
@@ -103,6 +103,17 @@
     return wrapper
 
 
+def leader_only(func):
+    '''Decorated function is only run on the leader.'''
+    @wraps(func)
+    def wrapper(*args, **kw):
+        if hookenv.is_leader():
+            return func(*args, **kw)
+        else:
+            return None
+    return wrapper
+
+
 @action
 def set_proxy():
     config = hookenv.config()
@@ -113,13 +124,14 @@
 
 @action
 def revert_unchangeable_config():
-    if hookenv.hook_name() == 'install':
-        # config.previous() only becomes meaningful after the install
-        # hook has run. During the first run on the unit hook, it
-        # reports everything has having None as the previous value.
+    config = hookenv.config()
+
+    # config.previous() only becomes meaningful after the install
+    # hook has run. During the first run on the unit hook, it
+    # reports everything has having None as the previous value.
+    if config._prev_dict is None:
         return
 
-    config = hookenv.config()
     for key in UNCHANGEABLE_KEYS:
         if config.changed(key):
             previous = config.previous(key)
@@ -231,48 +243,56 @@
     helpers.ensure_package_status(helpers.get_cassandra_packages())
 
 
-@action
-def install_oracle_jre():
-    if helpers.get_jre() != 'oracle':
-        return
-
+def _fetch_oracle_jre():
     config = hookenv.config()
     url = config.get('private_jre_url', None)
-    if not url:
-        hookenv.log('private_jre_url not set. Unable to continue.', ERROR)
-        raise SystemExit(1)
-
-    if config.get('retrieved_jre', None) != url:
-        filename = os.path.join('lib', url.split('/')[-1])
+    if url and config.get('retrieved_jre', None) != url:
+        filename = os.path.join(hookenv.charm_dir(),
+                                'lib', url.split('/')[-1])
         if not filename.endswith('-linux-x64.tar.gz'):
-            hookenv.log('Invalid JRE URL {}'.format(url), ERROR)
-            raise SystemExit(1)
+            helpers.status_set('blocked',
+                               'Invalid private_jre_url {}'.format(url))
+            raise SystemExit(0)
+        helpers.status_set(hookenv.status_get(),
+                           'Downloading Oracle JRE')
+        hookenv.log('Oracle JRE URL is {}'.format(url))
         urllib.request.urlretrieve(url, filename)
         config['retrieved_jre'] = url
 
-    pattern = 'lib/server-jre-?u*-linux-x64.tar.gz'
+    pattern = os.path.join(hookenv.charm_dir(),
+                           'lib', 'server-jre-?u*-linux-x64.tar.gz')
     tarballs = glob.glob(pattern)
-    if not tarballs:
-        hookenv.log('Oracle JRE tarball not found ({})'.format(pattern),
-                    ERROR)
-        # We could fallback to OpenJDK, but the user took the trouble
-        # to specify the Oracle JRE and it is recommended for Cassandra
-        # so lets hard fail instead.
-        raise SystemExit(1)
+    if not (url or tarballs):
+        helpers.status_set('blocked',
+                           'private_jre_url not set and no local tarballs.')
+        raise SystemExit(0)
+
+    elif not tarballs:
+        helpers.status_set('blocked',
+                           'Oracle JRE tarball not found ({})'.format(pattern))
+        raise SystemExit(0)
 
     # Latest tarball by filename/version num. Lets hope they don't hit
     # 99 (currently at 76).
     tarball = sorted(tarballs)[-1]
-
+    return tarball
+
+
+def _install_oracle_jre_tarball(tarball):
     # Same directory as webupd8 to avoid surprising people, but it could
     # be anything.
-    dest = '/usr/lib/jvm/java-7-oracle'
+    if 'jre-7u' in str(tarball):
+        dest = '/usr/lib/jvm/java-7-oracle'
+    else:
+        dest = '/usr/lib/jvm/java-8-oracle'
 
     if not os.path.isdir(dest):
         host.mkdir(dest)
 
     jre_exists = os.path.exists(os.path.join(dest, 'bin', 'java'))
 
+    config = hookenv.config()
+
     # Unpack the latest tarball if necessary.
     if config.get('oracle_jre_tarball', '') == tarball and jre_exists:
         hookenv.log('Already installed {}'.format(tarball))
@@ -293,6 +313,15 @@
 
 
 @action
+def install_oracle_jre():
+    if helpers.get_jre() != 'oracle':
+        return
+
+    tarball = _fetch_oracle_jre()
+    _install_oracle_jre_tarball(tarball)
+
+
+@action
 def emit_java_version():
     # Log the version for posterity. Could be useful since Oracle JRE
     # security updates are not automated.
@@ -355,82 +384,153 @@
     host.write_file(rackdc_path, rackdc_properties.encode('UTF-8'))
 
 
+def needs_reset_auth_keyspace_replication():
+    '''Guard for reset_auth_keyspace_replication.'''
+    num_nodes = helpers.num_nodes()
+    n = min(num_nodes, 3)
+    datacenter = hookenv.config()['datacenter']
+    with helpers.connect() as session:
+        strategy_opts = helpers.get_auth_keyspace_replication(session)
+        rf = int(strategy_opts.get(datacenter, -1))
+        hookenv.log('system_auth rf={!r}'.format(strategy_opts))
+        # If the node count has increased, we will change the rf.
+        # If the node count is decreasing, we do nothing as the service
+        # may be being destroyed.
+        if rf < n:
+            return True
+    return False
+
+
+@leader_only
 @action
[e-mail]('repair', needs_reset_auth_keyspace_replication)
 def reset_auth_keyspace_replication():
-    # This action only lowers the system_auth keyspace replication
-    # values when a node has been decommissioned. The replication settings
-    # are also updated during rolling restart, which takes care of when new
-    # nodes are added.
-    helpers.reset_auth_keyspace_replication()
+    # Cassandra requires you to manually set the replication factor of
+    # the system_auth keyspace, to ensure availability and redundancy.
+    # We replication factor in this service's DC can be no higher than
+    # the number of bootstrapped nodes. We also cap this at 3 to ensure
+    # we don't have too many seeds.
+    num_nodes = helpers.num_nodes()
+    n = min(num_nodes, 3)
+    datacenter = hookenv.config()['datacenter']
+    with helpers.connect() as session:
+        strategy_opts = helpers.get_auth_keyspace_replication(session)
+        rf = int(strategy_opts.get(datacenter, -1))
+        hookenv.log('system_auth rf={!r}'.format(strategy_opts))
+        if rf != n:
+            strategy_opts['class'] = 'NetworkTopologyStrategy'
+            strategy_opts[datacenter] = n
+            if 'replication_factor' in strategy_opts:
+                del strategy_opts['replication_factor']
+            helpers.set_auth_keyspace_replication(session, strategy_opts)
+            helpers.repair_auth_keyspace()
+            helpers.set_active()
 
 
 @action
 def store_unit_private_ip():
+    '''Store the unit's private ip address, so we can tell if it changes.'''
     hookenv.config()['unit_private_ip'] = hookenv.unit_private_ip()
 
 
-@action
-def set_unit_zero_bootstrapped():
-    '''Unit #0 is used as the first node in the cluster.
-
-    Unit #0 is implicitly flagged as bootstrap, and thus befores the
-    first node in the cluster and providing a seed for other nodes to
-    bootstrap off. We can change this when we have juju leadership,
-    making the leader the first node in the cluster. Until then, don't
-    attempt to create a multiunit service if you have removed Unit #0.
-    '''
-    relname = rollingrestart.get_peer_relation_name()
-    if helpers.unit_number() == 0 and hookenv.hook_name().startswith(relname):
-        helpers.set_bootstrapped(True)
-
-
-@action
-def maybe_schedule_restart():
-    '''Prepare for and schedule a rolling restart if necessary.'''
+def needs_restart():
+    '''Return True if Cassandra is not running or needs to be restarted.'''
+    if helpers.is_decommissioned():
+        # Decommissioned nodes are never restarted. They remain up
+        # telling everyone they are decommissioned.
+        helpers.status_set('blocked', 'Decommissioned node')
+        return False
+
     if not helpers.is_cassandra_running():
-        # Short circuit if Cassandra is not running to avoid log spam.
-        rollingrestart.request_restart()
-        return
-
-    if helpers.is_decommissioned():
-        hookenv.log("Decommissioned")
-        return
-
-    # If any of these config items changed, a restart is required.
+        if helpers.is_bootstrapped():
+            helpers.status_set('waiting', 'Waiting for permission to start')
+        else:
+            helpers.status_set('waiting',
+                               'Waiting for permission to bootstrap')
+        return True
+
     config = hookenv.config()
-    restart = False
-    for key in RESTART_REQUIRED_KEYS:
-        if config.changed(key):
-            hookenv.log('{} changed. Restart required.'.format(key))
-            restart = True
+
+    # If our IP address has changed, we need to restart.
+    if config.changed('unit_private_ip'):
+        hookenv.log('waiting',
+                    'IP address changed. Waiting for restart permission.')
+        return True
 
     # If the directory paths have changed, we need to migrate data
-    # during a restart. Directory config items have already been picked
-    # up in the previous check.
+    # during a restart.
     storage = relations.StorageRelation()
     if storage.needs_remount():
-        hookenv.log('Mountpoint changed. Restart and migration required.')
-        restart = True
+        helpers.status_set(hookenv.status_get(),
+                           'New mounts. Waiting for restart permission')
+        return True
 
-    # If our IP address has changed, we need to restart.
-    if config.changed('unit_private_ip'):
-        hookenv.log('Unit IP address changed. Restart required.')
-        restart = True
+    # If any of these config items changed, a restart is required.
+    for key in RESTART_REQUIRED_KEYS:
+        if config.changed(key):
+            hookenv.log('{} changed. Restart required.'.format(key))
+    for key in RESTART_REQUIRED_KEYS:
+        if config.changed(key):
+            helpers.status_set(hookenv.status_get(),
+                               'Config changes. '
+                               'Waiting for restart permission.')
+            return True
 
     # If we have new seeds, we should restart.
-    new_seeds = helpers.seed_ips()
-    config['configured_seeds'] = sorted(new_seeds)
-    if config.changed('configured_seeds'):
+    new_seeds = helpers.get_seed_ips()
+    if config.get('configured_seeds') != sorted(new_seeds):
         old_seeds = set(config.previous('configured_seeds') or [])
         changed = old_seeds.symmetric_difference(new_seeds)
         # We don't care about the local node in the changes.
         changed.discard(hookenv.unit_private_ip())
         if changed:
-            hookenv.log('New seeds {!r}. Restart required.'.format(new_seeds))
-            restart = True
-
-    if restart:
-        rollingrestart.request_restart()
+            helpers.status_set(hookenv.status_get(),
+                               'Updated seeds {!r}. '
+                               'Waiting for restart permission.'
+                               ''.format(new_seeds))
+            return True
+
+    hookenv.log('Restart not required')
+    return False
+
+
+@action
[e-mail]('restart', needs_restart)
+def maybe_restart():
+    '''Restart sequence.
+
+    If a restart is needed, shutdown Cassandra, perform all pending operations
+    that cannot be be done while Cassandra is live, and restart it.
+    '''
+    helpers.status_set('maintenance', 'Stopping Cassandra')
+    helpers.stop_cassandra()
+    helpers.remount_cassandra()
+    helpers.ensure_database_directories()
+    if helpers.peer_relid() and not helpers.is_bootstrapped():
+        helpers.status_set('maintenance', 'Bootstrapping')
+    else:
+        helpers.status_set('maintenance', 'Starting Cassandra')
+    helpers.start_cassandra()
+
+
+@action
+def post_bootstrap():
+    '''Maintain state on if the node has bootstrapped into the cluster.
+
+    Per documented procedure for adding new units to a cluster, wait 2
+    minutes if the unit has just bootstrapped to ensure other units
+    do not attempt bootstrap too soon.
+    '''
+    if not helpers.is_bootstrapped():
+        if coordinator.relid is not None:
+            helpers.status_set('maintenance', 'Post-bootstrap 2 minute delay')
+            hookenv.log('Post-bootstrap 2 minute delay')
+            time.sleep(120)  # Must wait 2 minutes between bootstrapping nodes.
+
+    # Unconditionally call this to publish the bootstrapped flag to
+    # the peer relation, as the first unit was bootstrapped before
+    # the peer relation existed.
+    helpers.set_bootstrapped()
 
 
 @action
@@ -443,66 +543,90 @@
     helpers.start_cassandra()
 
 
+@leader_only
 @action
-def ensure_unit_superuser():
-    helpers.ensure_unit_superuser()
+def create_unit_superusers():
+    # The leader creates and updates accounts for nodes, using the
+    # encrypted password they provide in relations.PeerRelation. We
+    # don't end up with unencrypted passwords leaving the unit, and we
+    # don't need to restart Cassandra in no-auth mode which is slow and
+    # I worry may cause issues interrupting the bootstrap.
+    if not coordinator.relid:
+        return  # No peer relation, no requests yet.
+
+    created_units = helpers.get_unit_superusers()
+    uncreated_units = [u for u in hookenv.related_units(coordinator.relid)
+                       if u not in created_units]
+    for peer in uncreated_units:
+        rel = hookenv.relation_get(unit=peer, rid=coordinator.relid)
+        username = rel.get('username')
+        pwhash = rel.get('pwhash')
+        if not username:
+            continue
+        hookenv.log('Creating {} account for {}'.format(username, peer))
+        with helpers.connect() as session:
+            helpers.ensure_user(session, username, pwhash, superuser=True)
+        created_units.add(peer)
+        helpers.set_unit_superusers(created_units)
 
 
 @action
 def reset_all_io_schedulers():
-    helpers.reset_all_io_schedulers()
+    dirs = helpers.get_all_database_directories()
+    dirs = (dirs['data_file_directories'] + [dirs['commitlog_directory']] +
+            [dirs['saved_caches_directory']])
+    config = hookenv.config()
+    for d in dirs:
+        if os.path.isdir(d):  # Directory may not exist yet.
+            helpers.set_io_scheduler(config['io_scheduler'], d)
+
+
+def _client_credentials(relid):
+    '''Return the client credentials used by relation relid.'''
+    relinfo = hookenv.relation_get(unit=hookenv.local_unit(), rid=relid)
+    username = relinfo.get('username')
+    password = relinfo.get('password')
+    if username is None or password is None:
+        for unit in hookenv.related_units(coordinator.relid):
+            try:
+                relinfo = hookenv.relation_get(unit=unit, rid=relid)
+                username = relinfo.get('username')
+                password = relinfo.get('password')
+                if username is not None and password is not None:
+                    return username, password
+            except subprocess.CalledProcessError:
+                pass  # Assume the remote unit has not joined yet.
+        return None, None
+    else:
+        return username, password
 
 
 def _publish_database_relation(relid, superuser):
-    # Due to Bug #1409763, this functionality is as action rather than a
-    # provided_data item.
-    #
     # The Casandra service needs to provide a common set of credentials
-    # to a client unit. Juju does not yet provide a leader so we
-    # need another mechanism for determine which unit will create the
-    # client's account, with the remaining units copying the lead unit's
-    # credentials. For the purposes of this charm, the first unit in
-    # order is considered the leader and creates the user with a random
-    # password. It then tickles the peer relation to ensure the other
-    # units get a hook fired and the opportunity to copy and publish
-    # these credentials. If the lowest numbered unit is removed before
-    # all of the other peers have copied its credentials, then the next
-    # lowest will have either already copied the credentials (and the
-    # remaining peers will use them), or the process starts again and
-    # it will generate new credentials.
-    node_list = list(rollingrestart.get_peers()) + [hookenv.local_unit()]
-    sorted_nodes = sorted(node_list, key=lambda unit: int(unit.split('/')[-1]))
-    first_node = sorted_nodes[0]
-
-    config = hookenv.config()
-
-    try:
-        relinfo = hookenv.relation_get(unit=first_node, rid=relid)
-    except subprocess.CalledProcessError:
-        if first_node == hookenv.local_unit():
-            raise
-        # relation-get may fail if the specified unit has not yet joined
-        # the peer relation, or has just departed. Try again later.
-        return
-
-    username = relinfo.get('username')
-    password = relinfo.get('password')
-    if hookenv.local_unit() == first_node:
-        # Lowest numbered unit, at least for now.
-        if 'username' not in relinfo:
-            # Credentials unset. Generate them.
-            username = 'juju_{}'.format(
-                relid.replace(':', '_').replace('-', '_'))
+    # to a client unit. The leader creates these, if none of the other
+    # units are found to have published them already (a previously elected
+    # leader may have done this). The leader then tickles the other units,
+    # firing a hook and giving them the opportunity to copy and publish
+    # these credentials.
+    username, password = _client_credentials(relid)
+    if username is None:
+        if hookenv.is_leader():
+            # Credentials not set. The leader must generate them. We use
+            # the service name so that database permissions remain valid
+            # even after the relation is dropped and recreated, or the
+            # juju environment rebuild and the database restored from
+            # backups.
+            username = 'juju_{}'.format(helpers.get_service_name(relid))
+            if superuser:
+                username += '_admin'
             password = host.pwgen()
-            # Wake the other peers, if any.
-            hookenv.relation_set(rollingrestart.get_peer_relation_id(),
-                                 ping=rollingrestart.utcnow_str())
-        # Create the account if necessary, and reset the password.
-        # We need to reset the password as another unit may have
-        # rudely changed it thinking they were the lowest numbered
-        # unit. Fix this behavior once juju provides real
-        # leadership.
-        helpers.ensure_user(username, password, superuser)
+            pwhash = helpers.encrypt_password(password)
+            with helpers.connect() as session:
+                helpers.ensure_user(session, username, pwhash, superuser)
+            # Wake the peers, if any.
+            helpers.leader_ping()
+        else:
+            return  # No credentials yet. Nothing to do.
 
     # Publish the information the client needs on the relation where
     # they can find it.
@@ -511,6 +635,7 @@
     #  - cluster_name, so clients can differentiate multiple clusters
     #  - datacenter + rack, so clients know what names they can use
     #    when altering keyspace replication settings.
+    config = hookenv.config()
     hookenv.relation_set(relid,
                          username=username, password=password,
                          host=hookenv.unit_public_ip(),
@@ -546,53 +671,13 @@
 
 
 @action
-def emit_describe_cluster():
-    '''Spam 'nodetool describecluster' into the logs.'''
+def emit_cluster_info():
     helpers.emit_describe_cluster()
-
-
-@action
-def emit_auth_keyspace_status():
-    '''Spam 'nodetool status system_auth' into the logs.'''
     helpers.emit_auth_keyspace_status()
-
-
-@action
-def emit_netstats():
-    '''Spam 'nodetool netstats' into the logs.'''
     helpers.emit_netstats()
 
 
 @action
-def shutdown_before_joining_peers():
-    '''Shutdown the node before opening firewall ports for our peers.
-
-    When the peer relation is first joined, the node has already been
-    setup and is running as a standalone cluster. This is problematic
-    when the peer relation has been formed, as if it is left running
-    peers may conect to it and initiate replication before this node
-    has been properly reset and bootstrapped. To avoid this, we
-    shutdown the node before opening firewall ports to any peers. The
-    firewall can then be opened, and peers will not find a node here
-    until it starts its bootstrapping.
-    '''
-    relname = rollingrestart.get_peer_relation_name()
-    if hookenv.hook_name() == '{}-relation-joined'.format(relname):
-        if not helpers.is_bootstrapped():
-            helpers.stop_cassandra(immediate=True)
-
-
-@action
-def grant_ssh_access():
-    '''Grant SSH access to run nodetool on remote nodes.
-
-    This is easier than setting up remote JMX access, and more secure.
-    '''
-    unison.ssh_authorized_peers(rollingrestart.get_peer_relation_name(),
-                                'juju_ssh', ensure_local_user=True)
-
-
-@action
 def configure_firewall():
     '''Configure firewall rules using ufw.
 
@@ -632,13 +717,9 @@
 
     # Rules for peers
     for relinfo in hookenv.relations_of_type('cluster'):
-        for port in peer_ports:
-            desired_rules.add((relinfo['private-address'], 'any', port))
-
-    # External seeds also need access.
-    for seed_ip in helpers.seed_ips():
-        for port in peer_ports:
-            desired_rules.add((seed_ip, 'any', port))
+        if relinfo['private-address']:
+            for port in peer_ports:
+                desired_rules.add((relinfo['private-address'], 'any', port))
 
     previous_rules = set(tuple(rule) for rule in config.get('ufw_rules', []))
 
@@ -683,8 +764,7 @@
             description="Check Cassandra Heap",
             check_cmd="check_cassandra_heap.sh {} {} {}"
                       "".format(hookenv.unit_private_ip(), cassandra_heap_warn,
-                                cassandra_heap_crit)
-        )
+                                cassandra_heap_crit))
 
     cassandra_disk_warn = conf.get('nagios_disk_warn_pct')
     cassandra_disk_crit = conf.get('nagios_disk_crit_pct')
@@ -699,7 +779,116 @@
                 description="Check Cassandra Disk {}".format(disk),
                 check_cmd="check_disk -u GB -w {}% -c {}% -K 5% -p {}"
                           "".format(cassandra_disk_warn, cassandra_disk_crit,
-                                    disk)
-            )
-
+                                    disk))
     nrpe_compat.write()
+
+
+@leader_only
+@action
+def maintain_seeds():
+    '''The leader needs to maintain the list of seed nodes'''
+    seed_ips = helpers.get_seed_ips()
+    hookenv.log('Current seeds == {!r}'.format(seed_ips), DEBUG)
+
+    bootstrapped_ips = helpers.get_bootstrapped_ips()
+    hookenv.log('Bootstrapped == {!r}'.format(bootstrapped_ips), DEBUG)
+
+    # Remove any seeds that are no longer bootstrapped, such as dropped
+    # units.
+    seed_ips.intersection_update(bootstrapped_ips)
+
+    # Add more bootstrapped nodes, if necessary, to get to our maximum
+    # of 3 seeds.
+    potential_seed_ips = list(reversed(sorted(bootstrapped_ips)))
+    while len(seed_ips) < 3 and potential_seed_ips:
+        seed_ips.add(potential_seed_ips.pop())
+
+    # If there are no seeds or bootstrapped nodes, start with the leader. Us.
+    if len(seed_ips) == 0:
+        seed_ips.add(hookenv.unit_private_ip())
+
+    hookenv.log('Updated seeds == {!r}'.format(seed_ips), DEBUG)
+
+    hookenv.leader_set(seeds=','.join(sorted(seed_ips)))
+
+
+@leader_only
+@action
+def reset_default_password():
+    if hookenv.leader_get('default_admin_password_changed'):
+        hookenv.log('Default admin password already changed')
+        return
+
+    # Cassandra ships with well known credentials, rather than
+    # providing a tool to reset credentials. This is a huge security
+    # hole we must close.
+    try:
+        # We need a big timeout here, as the cassandra user actually
+        # springs into existence some time after Cassandra has started
+        # up and is accepting connections.
+        with helpers.connect('cassandra', 'cassandra',
+                             timeout=120, auth_timeout=120) as session:
+            # But before we close this security hole, we need to use these
+            # credentials to create a different admin account for the
+            # leader, allowing it to create accounts for other nodes as they
+            # join. The alternative is restarting Cassandra without
+            # authentication, which this charm will likely need to do in the
+            # future when we allow Cassandra services to be related together.
+            helpers.status_set('maintenance',
+                               'Creating initial superuser account')
+            username, password = helpers.superuser_credentials()
+            pwhash = helpers.encrypt_password(password)
+            helpers.ensure_user(session, username, pwhash, superuser=True)
+            helpers.set_unit_superusers([hookenv.local_unit()])
+
+            helpers.status_set('maintenance',
+                               'Changing default admin password')
+            helpers.query(session, 'ALTER USER cassandra WITH PASSWORD %s',
+                          cassandra.ConsistencyLevel.ALL, (host.pwgen(),))
+    except cassandra.AuthenticationFailed:
+        hookenv.log('Default superuser account already reset')
+        try:
+            with helpers.connect():
+                hookenv.log("Leader's superuser account already created")
+        except cassandra.AuthenticationFailed:
+            # We have no known superuser credentials. Create the account
+            # the hard, slow way. This will be the normal method
+            # of creating the service's initial account when we allow
+            # services to be related together.
+            helpers.create_unit_superuser_hard()
+
+    hookenv.leader_set(default_admin_password_changed=True)
+
+
+@action
+def set_active():
+    # If we got this far, the unit is active. Update the status if it is
+    # not already active. We don't do this unconditionally, as the charm
+    # may be active but doing stuff, like active but waiting for restart
+    # permission.
+    if hookenv.status_get() != 'active':
+        helpers.set_active()
+    else:
+        hookenv.log('Unit status already active', DEBUG)
+
+
+@action
+def request_unit_superuser():
+    relid = helpers.peer_relid()
+    if relid is None:
+        hookenv.log('Request deferred until peer relation exists')
+        return
+
+    relinfo = hookenv.relation_get(unit=hookenv.local_unit(),
+                                   rid=relid)
+    if relinfo and relinfo.get('username'):
+        # We must avoid blindly setting the pwhash on the relation,
+        # as we will likely get a different value everytime we
+        # encrypt the password due to the random salt.
+        hookenv.log('Superuser account request previously made')
+    else:
+        # Publish the requested superuser and hash to our peers.
+        username, password = helpers.superuser_credentials()
+        pwhash = helpers.encrypt_password(password)
+        hookenv.relation_set(relid, username=username, pwhash=pwhash)
+        hookenv.log('Requested superuser account creation')

=== added directory 'hooks/charmhelpers/contrib/benchmark'
=== added file 'hooks/charmhelpers/contrib/benchmark/__init__.py'
--- hooks/charmhelpers/contrib/benchmark/__init__.py	1970-01-01 00:00:00 +0000
+++ hooks/charmhelpers/contrib/benchmark/__init__.py	2015-06-30 07:36:26 +0000
@@ -0,0 +1,124 @@
+# 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/>.
+
+import subprocess
+import time
+import os
+from distutils.spawn import find_executable
+
+from charmhelpers.core.hookenv import (
+    in_relation_hook,
+    relation_ids,
+    relation_set,
+    relation_get,
+)
+
+
+def action_set(key, val):
+    if find_executable('action-set'):
+        action_cmd = ['action-set']
+
+        if isinstance(val, dict):
+            for k, v in iter(val.items()):
+                action_set('%s.%s' % (key, k), v)
+            return True
+
+        action_cmd.append('%s=%s' % (key, val))
+        subprocess.check_call(action_cmd)
+        return True
+    return False
+
+
+class Benchmark():
+    """
+    Helper class for the `benchmark` interface.
+
+    :param list actions: Define the actions that are also benchmarks
+
+    From inside the benchmark-relation-changed hook, you would
+    Benchmark(['memory', 'cpu', 'disk', 'smoke', 'custom'])
+
+    Examples:
+
+        siege = Benchmark(['siege'])
+        siege.start()
+        [... run siege ...]
+        # The higher the score, the better the benchmark
+        siege.set_composite_score(16.70, 'trans/sec', 'desc')
+        siege.finish()
+
+
+    """
+
+    required_keys = [
+        'hostname',
+        'port',
+        'graphite_port',
+        'graphite_endpoint',
+        'api_port'
+    ]
+
+    def __init__(self, benchmarks=None):
+        if in_relation_hook():
+            if benchmarks is not None:
+                for rid in sorted(relation_ids('benchmark')):
+                    relation_set(relation_id=rid, relation_settings={
+                        'benchmarks': ",".join(benchmarks)
+                    })
+
+            # Check the relation data
+            config = {}
+            for key in self.required_keys:
+                val = relation_get(key)
+                if val is not None:
+                    config[key] = val
+                else:
+                    # We don't have all of the required keys
+                    config = {}
+                    break
+
+            if len(config):
+                with open('/etc/benchmark.conf', 'w') as f:
+                    for key, val in iter(config.items()):
+                        f.write("%s=%s\n" % (key, val))
+
+    @staticmethod
+    def start():
+        action_set('meta.start', time.strftime('%Y-%m-%dT%H:%M:%SZ'))
+
+        """
+        If the collectd charm is also installed, tell it to send a snapshot
+        of the current profile data.
+        """
+        COLLECT_PROFILE_DATA = '/usr/local/bin/collect-profile-data'
+        if os.path.exists(COLLECT_PROFILE_DATA):
+            subprocess.check_output([COLLECT_PROFILE_DATA])
+
+    @staticmethod
+    def finish():
+        action_set('meta.stop', time.strftime('%Y-%m-%dT%H:%M:%SZ'))
+
+    @staticmethod
+    def set_composite_score(value, units, direction='asc'):
+        """
+        Set the composite score for a benchmark run. This is a single number
+        representative of the benchmark results. This could be the most
+        important metric, or an amalgamation of metric scores.
+        """
+        return action_set(
+            "meta.composite",
+            {'value': value, 'units': units, 'direction': direction}
+        )

=== modified file 'hooks/charmhelpers/contrib/charmsupport/nrpe.py'
--- hooks/charmhelpers/contrib/charmsupport/nrpe.py	2015-04-03 15:42:21 +0000
+++ hooks/charmhelpers/contrib/charmsupport/nrpe.py	2015-06-30 07:36:26 +0000
@@ -247,7 +247,9 @@
 
         service('restart', 'nagios-nrpe-server')
 
-        for rid in relation_ids("local-monitors"):
+        monitor_ids = relation_ids("local-monitors") + \
+            relation_ids("nrpe-external-master")
+        for rid in monitor_ids:
             relation_set(relation_id=rid, monitors=yaml.dump(monitors))
 
 

=== removed directory 'hooks/charmhelpers/contrib/peerstorage'
=== removed file 'hooks/charmhelpers/contrib/peerstorage/__init__.py'
--- hooks/charmhelpers/contrib/peerstorage/__init__.py	2015-01-26 13:07:31 +0000
+++ hooks/charmhelpers/contrib/peerstorage/__init__.py	1970-01-01 00:00:00 +0000
@@ -1,150 +0,0 @@
-# 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/>.
-
-import six
-from charmhelpers.core.hookenv import relation_id as current_relation_id
-from charmhelpers.core.hookenv import (
-    is_relation_made,
-    relation_ids,
-    relation_get,
-    local_unit,
-    relation_set,
-)
-
-
-"""
-This helper provides functions to support use of a peer relation
-for basic key/value storage, with the added benefit that all storage
-can be replicated across peer units.
-
-Requirement to use:
-
-To use this, the "peer_echo()" method has to be called form the peer
-relation's relation-changed hook:
-
[e-mail]("cluster-relation-changed") # Adapt the to your peer relation name
-def cluster_relation_changed():
-    peer_echo()
-
-Once this is done, you can use peer storage from anywhere:
-
[e-mail]("some-hook")
-def some_hook():
-    # You can store and retrieve key/values this way:
-    if is_relation_made("cluster"):  # from charmhelpers.core.hookenv
-        # There are peers available so we can work with peer storage
-        peer_store("mykey", "myvalue")
-        value = peer_retrieve("mykey")
-        print value
-    else:
-        print "No peers joind the relation, cannot share key/values :("
-"""
-
-
-def peer_retrieve(key, relation_name='cluster'):
-    """Retrieve a named key from peer relation `relation_name`."""
-    cluster_rels = relation_ids(relation_name)
-    if len(cluster_rels) > 0:
-        cluster_rid = cluster_rels[0]
-        return relation_get(attribute=key, rid=cluster_rid,
-                            unit=local_unit())
-    else:
-        raise ValueError('Unable to detect'
-                         'peer relation {}'.format(relation_name))
-
-
-def peer_retrieve_by_prefix(prefix, relation_name='cluster', delimiter='_',
-                            inc_list=None, exc_list=None):
-    """ Retrieve k/v pairs given a prefix and filter using {inc,exc}_list """
-    inc_list = inc_list if inc_list else []
-    exc_list = exc_list if exc_list else []
-    peerdb_settings = peer_retrieve('-', relation_name=relation_name)
-    matched = {}
-    if peerdb_settings is None:
-        return matched
-    for k, v in peerdb_settings.items():
-        full_prefix = prefix + delimiter
-        if k.startswith(full_prefix):
-            new_key = k.replace(full_prefix, '')
-            if new_key in exc_list:
-                continue
-            if new_key in inc_list or len(inc_list) == 0:
-                matched[new_key] = v
-    return matched
-
-
-def peer_store(key, value, relation_name='cluster'):
-    """Store the key/value pair on the named peer relation `relation_name`."""
-    cluster_rels = relation_ids(relation_name)
-    if len(cluster_rels) > 0:
-        cluster_rid = cluster_rels[0]
-        relation_set(relation_id=cluster_rid,
-                     relation_settings={key: value})
-    else:
-        raise ValueError('Unable to detect '
-                         'peer relation {}'.format(relation_name))
-
-
-def peer_echo(includes=None):
-    """Echo filtered attributes back onto the same relation for storage.
-
-    This is a requirement to use the peerstorage module - it needs to be called
-    from the peer relation's changed hook.
-    """
-    rdata = relation_get()
-    echo_data = {}
-    if includes is None:
-        echo_data = rdata.copy()
-        for ex in ['private-address', 'public-address']:
-            if ex in echo_data:
-                echo_data.pop(ex)
-    else:
-        for attribute, value in six.iteritems(rdata):
-            for include in includes:
-                if include in attribute:
-                    echo_data[attribute] = value
-    if len(echo_data) > 0:
-        relation_set(relation_settings=echo_data)
-
-
-def peer_store_and_set(relation_id=None, peer_relation_name='cluster',
-                       peer_store_fatal=False, relation_settings=None,
-                       delimiter='_', **kwargs):
-    """Store passed-in arguments both in argument relation and in peer storage.
-
-    It functions like doing relation_set() and peer_store() at the same time,
-    with the same data.
-
-    @param relation_id: the id of the relation to store the data on. Defaults
-                        to the current relation.
-    @param peer_store_fatal: Set to True, the function will raise an exception
-                             should the peer sotrage not be avialable."""
-
-    relation_settings = relation_settings if relation_settings else {}
-    relation_set(relation_id=relation_id,
-                 relation_settings=relation_settings,
-                 **kwargs)
-    if is_relation_made(peer_relation_name):
-        for key, value in six.iteritems(dict(list(kwargs.items()) +
-                                             list(relation_settings.items()))):
-            key_prefix = relation_id or current_relation_id()
-            peer_store(key_prefix + delimiter + key,
-                       value,
-                       relation_name=peer_relation_name)
-    else:
-        if peer_store_fatal:
-            raise ValueError('Unable to detect '
-                             'peer relation {}'.format(peer_relation_name))

=== removed directory 'hooks/charmhelpers/contrib/unison'
=== removed file 'hooks/charmhelpers/contrib/unison/__init__.py'
--- hooks/charmhelpers/contrib/unison/__init__.py	2015-04-03 15:42:21 +0000
+++ hooks/charmhelpers/contrib/unison/__init__.py	1970-01-01 00:00:00 +0000
@@ -1,298 +0,0 @@
-# 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/>.
-
-# Easy file synchronization among peer units using ssh + unison.
-#
-# From *both* peer relation -joined and -changed, add a call to
-# ssh_authorized_peers() describing the peer relation and the desired
-# user + group.  After all peer relations have settled, all hosts should
-# be able to connect to on another via key auth'd ssh as the specified user.
-#
-# Other hooks are then free to synchronize files and directories using
-# sync_to_peers().
-#
-# For a peer relation named 'cluster', for example:
-#
-# cluster-relation-joined:
-# ...
-# ssh_authorized_peers(peer_interface='cluster',
-#                      user='juju_ssh', group='juju_ssh',
-#                      ensure_user=True)
-# ...
-#
-# cluster-relation-changed:
-# ...
-# ssh_authorized_peers(peer_interface='cluster',
-#                      user='juju_ssh', group='juju_ssh',
-#                      ensure_user=True)
-# ...
-#
-# Hooks are now free to sync files as easily as:
-#
-# files = ['/etc/fstab', '/etc/apt.conf.d/']
-# sync_to_peers(peer_interface='cluster',
-#                user='juju_ssh, paths=[files])
-#
-# It is assumed the charm itself has setup permissions on each unit
-# such that 'juju_ssh' has read + write permissions.  Also assumed
-# that the calling charm takes care of leader delegation.
-#
-# Additionally files can be synchronized only to an specific unit:
-# sync_to_peer(slave_address, user='juju_ssh',
-#              paths=[files], verbose=False)
-
-import os
-import pwd
-
-from copy import copy
-from subprocess import check_call, check_output
-
-from charmhelpers.core.host import (
-    adduser,
-    add_user_to_group,
-    pwgen,
-)
-
-from charmhelpers.core.hookenv import (
-    log,
-    hook_name,
-    relation_ids,
-    related_units,
-    relation_set,
-    relation_get,
-    unit_private_ip,
-    INFO,
-    ERROR,
-)
-
-BASE_CMD = ['unison', '-auto', '-batch=true', '-confirmbigdel=false',
-            '-fastcheck=true', '-group=false', '-owner=false',
-            '-prefer=newer', '-times=true']
-
-
-def get_homedir(user):
-    try:
-        user = pwd.getpwnam(user)
-        return user.pw_dir
-    except KeyError:
-        log('Could not get homedir for user %s: user exists?' % (user), ERROR)
-        raise Exception
-
-
-def create_private_key(user, priv_key_path):
-    if not os.path.isfile(priv_key_path):
-        log('Generating new SSH key for user %s.' % user)
-        cmd = ['ssh-keygen', '-q', '-N', '', '-t', 'rsa', '-b', '2048',
-               '-f', priv_key_path]
-        check_call(cmd)
-    else:
-        log('SSH key already exists at %s.' % priv_key_path)
-    check_call(['chown', user, priv_key_path])
-    check_call(['chmod', '0600', priv_key_path])
-
-
-def create_public_key(user, priv_key_path, pub_key_path):
-    if not os.path.isfile(pub_key_path):
-        log('Generating missing ssh public key @ %s.' % pub_key_path)
-        cmd = ['ssh-keygen', '-y', '-f', priv_key_path]
-        p = check_output(cmd).strip()
-        with open(pub_key_path, 'wb') as out:
-            out.write(p)
-    check_call(['chown', user, pub_key_path])
-
-
-def get_keypair(user):
-    home_dir = get_homedir(user)
-    ssh_dir = os.path.join(home_dir, '.ssh')
-    priv_key = os.path.join(ssh_dir, 'id_rsa')
-    pub_key = '%s.pub' % priv_key
-
-    if not os.path.isdir(ssh_dir):
-        os.mkdir(ssh_dir)
-        check_call(['chown', '-R', user, ssh_dir])
-
-    create_private_key(user, priv_key)
-    create_public_key(user, priv_key, pub_key)
-
-    with open(priv_key, 'r') as p:
-        _priv = p.read().strip()
-
-    with open(pub_key, 'r') as p:
-        _pub = p.read().strip()
-
-    return (_priv, _pub)
-
-
-def write_authorized_keys(user, keys):
-    home_dir = get_homedir(user)
-    ssh_dir = os.path.join(home_dir, '.ssh')
-    auth_keys = os.path.join(ssh_dir, 'authorized_keys')
-    log('Syncing authorized_keys @ %s.' % auth_keys)
-    with open(auth_keys, 'w') as out:
-        for k in keys:
-            out.write('%s\n' % k)
-
-
-def write_known_hosts(user, hosts):
-    home_dir = get_homedir(user)
-    ssh_dir = os.path.join(home_dir, '.ssh')
-    known_hosts = os.path.join(ssh_dir, 'known_hosts')
-    khosts = []
-    for host in hosts:
-        cmd = ['ssh-keyscan', '-H', '-t', 'rsa', host]
-        remote_key = check_output(cmd, universal_newlines=True).strip()
-        khosts.append(remote_key)
-    log('Syncing known_hosts @ %s.' % known_hosts)
-    with open(known_hosts, 'w') as out:
-        for host in khosts:
-            out.write('%s\n' % host)
-
-
-def ensure_user(user, group=None):
-    adduser(user, pwgen())
-    if group:
-        add_user_to_group(user, group)
-
-
-def ssh_authorized_peers(peer_interface, user, group=None,
-                         ensure_local_user=False):
-    """
-    Main setup function, should be called from both peer -changed and -joined
-    hooks with the same parameters.
-    """
-    if ensure_local_user:
-        ensure_user(user, group)
-    priv_key, pub_key = get_keypair(user)
-    hook = hook_name()
-    if hook == '%s-relation-joined' % peer_interface:
-        relation_set(ssh_pub_key=pub_key)
-    elif hook == '%s-relation-changed' % peer_interface:
-        hosts = []
-        keys = []
-
-        for r_id in relation_ids(peer_interface):
-            for unit in related_units(r_id):
-                ssh_pub_key = relation_get('ssh_pub_key',
-                                           rid=r_id,
-                                           unit=unit)
-                priv_addr = relation_get('private-address',
-                                         rid=r_id,
-                                         unit=unit)
-                if ssh_pub_key:
-                    keys.append(ssh_pub_key)
-                    hosts.append(priv_addr)
-                else:
-                    log('ssh_authorized_peers(): ssh_pub_key '
-                        'missing for unit %s, skipping.' % unit)
-        write_authorized_keys(user, keys)
-        write_known_hosts(user, hosts)
-        authed_hosts = ':'.join(hosts)
-        relation_set(ssh_authorized_hosts=authed_hosts)
-
-
-def _run_as_user(user, gid=None):
-    try:
-        user = pwd.getpwnam(user)
-    except KeyError:
-        log('Invalid user: %s' % user)
-        raise Exception
-    uid = user.pw_uid
-    gid = gid or user.pw_gid
-    os.environ['HOME'] = user.pw_dir
-
-    def _inner():
-        os.setgid(gid)
-        os.setuid(uid)
-    return _inner
-
-
-def run_as_user(user, cmd, gid=None):
-    return check_output(cmd, preexec_fn=_run_as_user(user, gid), cwd='/')
-
-
-def collect_authed_hosts(peer_interface):
-    '''Iterate through the units on peer interface to find all that
-    have the calling host in its authorized hosts list'''
-    hosts = []
-    for r_id in (relation_ids(peer_interface) or []):
-        for unit in related_units(r_id):
-            private_addr = relation_get('private-address',
-                                        rid=r_id, unit=unit)
-            authed_hosts = relation_get('ssh_authorized_hosts',
-                                        rid=r_id, unit=unit)
-
-            if not authed_hosts:
-                log('Peer %s has not authorized *any* hosts yet, skipping.' %
-                    (unit), level=INFO)
-                continue
-
-            if unit_private_ip() in authed_hosts.split(':'):
-                hosts.append(private_addr)
-            else:
-                log('Peer %s has not authorized *this* host yet, skipping.' %
-                    (unit), level=INFO)
-    return hosts
-
-
-def sync_path_to_host(path, host, user, verbose=False, cmd=None, gid=None,
-                      fatal=False):
-    """Sync path to an specific peer host
-
-    Propagates exception if operation fails and fatal=True.
-    """
-    cmd = cmd or copy(BASE_CMD)
-    if not verbose:
-        cmd.append('-silent')
-
-    # removing trailing slash from directory paths, unison
-    # doesn't like these.
-    if path.endswith('/'):
-        path = path[:(len(path) - 1)]
-
-    cmd = cmd + [path, 'ssh://%s@%s/%s' % (user, host, path)]
-
-    try:
-        log('Syncing local path %s to %s@%s:%s' % (path, user, host, path))
-        run_as_user(user, cmd, gid)
-    except:
-        log('Error syncing remote files')
-        if fatal:
-            raise
-
-
-def sync_to_peer(host, user, paths=None, verbose=False, cmd=None, gid=None,
-                 fatal=False):
-    """Sync paths to an specific peer host
-
-    Propagates exception if any operation fails and fatal=True.
-    """
-    if paths:
-        for p in paths:
-            sync_path_to_host(p, host, user, verbose, cmd, gid, fatal)
-
-
-def sync_to_peers(peer_interface, user, paths=None, verbose=False, cmd=None,
-                  gid=None, fatal=False):
-    """Sync all hosts to an specific path
-
-    The type of group is integer, it allows user has permissions to
-    operate a directory have a different group id with the user id.
-
-    Propagates exception if any operation fails and fatal=True.
-    """
-    if paths:
-        for host in collect_authed_hosts(peer_interface):
-            sync_to_peer(host, user, paths, verbose, cmd, gid, fatal)

=== added file 'hooks/charmhelpers/coordinator.py'
--- hooks/charmhelpers/coordinator.py	1970-01-01 00:00:00 +0000
+++ hooks/charmhelpers/coordinator.py	2015-06-30 07:36:26 +0000
@@ -0,0 +1,607 @@
+# 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/>.
+'''
+The coordinator module allows you to use Juju's leadership feature to
+coordinate operations between units of a service.
+
+Behavior is defined in subclasses of coordinator.BaseCoordinator.
+One implementation is provided (coordinator.Serial), which allows an
+operation to be run on a single unit at a time, on a first come, first
+served basis. You can trivially define more complex behavior by
+subclassing BaseCoordinator or Serial.
+
+:author: Stuart Bishop <[e-mail]>
+
+
+Services Framework Usage
+========================
+
+Ensure a peer relation is defined in metadata.yaml. Instantiate a
+BaseCoordinator subclass before invoking ServiceManager.manage().
+Ensure that ServiceManager.manage() is wired up to the leader-elected,
+leader-settings-changed, peer relation-changed and peer
+relation-departed hooks in addition to any other hooks you need, or your
+service will deadlock.
+
+Ensure calls to acquire() are guarded, so that locks are only requested
+when they are really needed (and thus hooks only triggered when necessary).
+Failing to do this and calling acquire() unconditionally will put your unit
+into a hook loop. Calls to granted() do not need to be guarded.
+
+For example::
+
+    from charmhelpers.core import hookenv, services
+    from charmhelpers import coordinator
+
+    def maybe_restart(servicename):
+        serial = coordinator.Serial()
+        if needs_restart():
+            serial.acquire('restart')
+        if serial.granted('restart'):
+            hookenv.service_restart(servicename)
+
+    services = [dict(service='servicename',
+                     data_ready=[maybe_restart])]
+
+    if __name__ == '__main__':
+        _ = coordinator.Serial()  # Must instantiate before manager.manage()
+        manager = services.ServiceManager(services)
+        manager.manage()
+
+
+You can implement a similar pattern using a decorator. If the lock has
+not been granted, an attempt to acquire() it will be made if the guard
+function returns True. If the lock has been granted, the decorated function
+is run as normal::
+
+    from charmhelpers.core import hookenv, services
+    from charmhelpers import coordinator
+
+    serial = coordinator.Serial()  # Global, instatiated on module import.
+
+    def needs_restart():
+        [ ... Introspect state. Return True if restart is needed ... ]
+
+    @serial.require('restart', needs_restart)
+    def maybe_restart(servicename):
+        hookenv.service_restart(servicename)
+
+    services = [dict(service='servicename',
+                     data_ready=[maybe_restart])]
+
+    if __name__ == '__main__':
+        manager = services.ServiceManager(services)
+        manager.manage()
+
+
+Traditional Usage
+=================
+
+Ensure a peer relationis defined in metadata.yaml.
+
+If you are using charmhelpers.core.hookenv.Hooks, ensure that a
+BaseCoordinator subclass is instantiated before calling Hooks.execute.
+
+If you are not using charmhelpers.core.hookenv.Hooks, ensure
+that a BaseCoordinator subclass is instantiated and its handle()
+method called at the start of all your hooks.
+
+For example::
+
+    import sys
+    from charmhelpers.core import hookenv
+    from charmhelpers import coordinator
+
+    hooks = hookenv.Hooks()
+
+    def maybe_restart():
+        serial = coordinator.Serial()
+        if serial.granted('restart'):
+            hookenv.service_restart('myservice')
+
+    @hooks.hook
+    def config_changed():
+        update_config()
+        serial = coordinator.Serial()
+        if needs_restart():
+            serial.acquire('restart'):
+            maybe_restart()
+
+    # Cluster hooks must be wired up.
+    @hooks.hook('cluster-relation-changed', 'cluster-relation-departed')
+    def cluster_relation_changed():
+        maybe_restart()
+
+    # Leader hooks must be wired up.
+    @hooks.hook('leader-elected', 'leader-settings-changed')
+    def leader_settings_changed():
+        maybe_restart()
+
+    [ ... repeat for *all* other hooks you are using ... ]
+
+    if __name__ == '__main__':
+        _ = coordinator.Serial()  # Must instantiate before execute()
+        hooks.execute(sys.argv)
+
+
+You can also use the require decorator. If the lock has not been granted,
+an attempt to acquire() it will be made if the guard function returns True.
+If the lock has been granted, the decorated function is run as normal::
+
+    from charmhelpers.core import hookenv
+
+    hooks = hookenv.Hooks()
+    serial = coordinator.Serial()  # Must instantiate before execute()
+
+    @require('restart', needs_restart)
+    def maybe_restart():
+        hookenv.service_restart('myservice')
+
+    @hooks.hook('install', 'config-changed', 'upgrade-charm',
+                # Peer and leader hooks must be wired up.
+                'cluster-relation-changed', 'cluster-relation-departed',
+                'leader-elected', 'leader-settings-changed')
+    def default_hook():
+        [...]
+        maybe_restart()
+
+    if __name__ == '__main__':
+        hooks.execute()
+
+
+Details
+=======
+
+A simple API is provided similar to traditional locking APIs. A lock
+may be requested using the acquire() method, and the granted() method
+may be used do to check if a lock previously requested by acquire() has
+been granted. It doesn't matter how many times acquire() is called in a
+hook.
+
+Locks are released at the end of the hook they are acquired in. This may
+be the current hook if the unit is leader and the lock is free. It is
+more likely a future hook (probably leader-settings-changed, possibly
+the peer relation-changed or departed hook, potentially any hook).
+
+Whenever a charm needs to perform a coordinated action it will acquire()
+the lock and perform the action immediately if acquisition is
+successful. It will also need to perform the same action in every other
+hook if the lock has been granted.
+
+
+Grubby Details
+--------------
+
+Why do you need to be able to perform the same action in every hook?
+If the unit is the leader, then it may be able to grant its own lock
+and perform the action immediately in the source hook. If the unit is
+the leader and cannot immediately grant the lock, then its only
+guaranteed chance of acquiring the lock is in the peer relation-joined,
+relation-changed or peer relation-departed hooks when another unit has
+released it (the only channel to communicate to the leader is the peer
+relation). If the unit is not the leader, then it is unlikely the lock
+is granted in the source hook (a previous hook must have also made the
+request for this to happen). A non-leader is notified about the lock via
+leader settings. These changes may be visible in any hook, even before
+the leader-settings-changed hook has been invoked. Or the requesting
+unit may be promoted to leader after making a request, in which case the
+lock may be granted in leader-elected or in a future peer
+relation-changed or relation-departed hook.
+
+This could be simpler if leader-settings-changed was invoked on the
+leader. We could then never grant locks except in
+leader-settings-changed hooks giving one place for the operation to be
+performed. Unfortunately this is not the case with Juju 1.23 leadership.
+
+But of course, this doesn't really matter to most people as most people
+seem to prefer the Services Framework or similar reset-the-world
+approaches, rather than the twisty maze of attempting to deduce what
+should be done based on what hook happens to be running (which always
+seems to evolve into reset-the-world anyway when the charm grows beyond
+the trivial).
+
+I chose not to implement a callback model, where a callback was passed
+to acquire to be executed when the lock is granted, because the callback
+may become invalid between making the request and the lock being granted
+due to an upgrade-charm being run in the interim. And it would create
+restrictions, such no lambdas, callback defined at the top level of a
+module, etc. Still, we could implement it on top of what is here, eg.
+by adding a defer decorator that stores a pickle of itself to disk and
+have BaseCoordinator unpickle and execute them when the locks are granted.
+'''
+from datetime import datetime
+from functools import wraps
+import json
+import os.path
+
+from six import with_metaclass
+
+from charmhelpers.core import hookenv
+
+
+# We make BaseCoordinator and subclasses singletons, so that if we
+# need to spill to local storage then only a single instance does so,
+# rather than having multiple instances stomp over each other.
+class Singleton(type):
+    _instances = {}
+
+    def __call__(cls, *args, **kwargs):
+        if cls not in cls._instances:
+            cls._instances[cls] = super(Singleton, cls).__call__(*args,
+                                                                 **kwargs)
+        return cls._instances[cls]
+
+
+class BaseCoordinator(with_metaclass(Singleton, object)):
+    relid = None  # Peer relation-id, set by __init__
+    relname = None
+
+    grants = None  # self.grants[unit][lock] == timestamp
+    requests = None  # self.requests[unit][lock] == timestamp
+
+    def __init__(self, relation_key='coordinator', peer_relation_name=None):
+        '''Instatiate a Coordinator.
+
+        Data is stored on the peer relation and in leadership storage
+        under the provided relation_key.
+
+        The peer relation is identified by peer_relation_name, and defaults
+        to the first one found in metadata.yaml.
+        '''
+        # Most initialization is deferred, since invoking hook tools from
+        # the constructor makes testing hard.
+        self.key = relation_key
+        self.relname = peer_relation_name
+        hookenv.atstart(self.initialize)
+
+        # Ensure that handle() is called, without placing that burden on
+        # the charm author. They still need to do this manually if they
+        # are not using a hook framework.
+        hookenv.atstart(self.handle)
+
+    def initialize(self):
+        if self.requests is not None:
+            return  # Already initialized.
+
+        assert hookenv.has_juju_version('1.23'), 'Needs Juju 1.23+'
+
+        if self.relname is None:
+            self.relname = _implicit_peer_relation_name()
+
+        relids = hookenv.relation_ids(self.relname)
+        if relids:
+            self.relid = sorted(relids)[0]
+
+        # Load our state, from leadership, the peer relationship, and maybe
+        # local state as a fallback. Populates self.requests and self.grants.
+        self._load_state()
+        self._emit_state()
+
+        # Save our state if the hook completes successfully.
+        hookenv.atexit(self._save_state)
+
+        # Schedule release of granted locks for the end of the hook.
+        # This needs to be the last of our atexit callbacks to ensure
+        # it will be run first when the hook is complete, because there
+        # is no point mutating our state after it has been saved.
+        hookenv.atexit(self._release_granted)
+
+    def acquire(self, lock):
+        '''Acquire the named lock, non-blocking.
+
+        The lock may be granted immediately, or in a future hook.
+
+        Returns True if the lock has been granted. The lock will be
+        automatically released at the end of the hook in which it is
+        granted.
+
+        Do not mindlessly call this method, as it triggers a cascade of
+        hooks. For example, if you call acquire() every time in your
+        peer relation-changed hook you will end up with an infinite loop
+        of hooks. It should almost always be guarded by some condition.
+        '''
+        unit = hookenv.local_unit()
+        ts = self.requests[unit].get(lock)
+        if not ts:
+            # If there is no outstanding request on the peer relation,
+            # create one.
+            self.requests.setdefault(lock, {})
+            self.requests[unit][lock] = _timestamp()
+            self.msg('Requested {}'.format(lock))
+
+        # If the leader has granted the lock, yay.
+        if self.granted(lock):
+            self.msg('Acquired {}'.format(lock))
+            return True
+
+        # If the unit making the request also happens to be the
+        # leader, it must handle the request now. Even though the
+        # request has been stored on the peer relation, the peer
+        # relation-changed hook will not be triggered.
+        if hookenv.is_leader():
+            return self.grant(lock, unit)
+
+        return False  # Can't acquire lock, yet. Maybe next hook.
+
+    def granted(self, lock):
+        '''Return True if a previously requested lock has been granted'''
+        unit = hookenv.local_unit()
+        ts = self.requests[unit].get(lock)
+        if ts and self.grants.get(unit, {}).get(lock) == ts:
+            return True
+        return False
+
+    def requested(self, lock):
+        '''Return True if we are in the queue for the lock'''
+        return lock in self.requests[hookenv.local_unit()]
+
+    def request_timestamp(self, lock):
+        '''Return the timestamp of our outstanding request for lock, or None.
+
+        Returns a datetime.datetime() UTC timestamp, with no tzinfo attribute.
+        '''
+        ts = self.requests[hookenv.local_unit()].get(lock, None)
+        if ts is not None:
+            return datetime.strptime(ts, _timestamp_format)
+
+    def handle(self):
+        if not hookenv.is_leader():
+            return  # Only the leader can grant requests.
+
+        self.msg('Leader handling coordinator requests')
+
+        # Clear our grants that have been released.
+        for unit in self.grants.keys():
+            for lock, grant_ts in list(self.grants[unit].items()):
+                req_ts = self.requests.get(unit, {}).get(lock)
+                if req_ts != grant_ts:
+                    # The request timestamp does not match the granted
+                    # timestamp. Several hooks on 'unit' may have run
+                    # before the leader got a chance to make a decision,
+                    # and 'unit' may have released its lock and attempted
+                    # to reacquire it. This will change the timestamp,
+                    # and we correctly revoke the old grant putting it
+                    # to the end of the queue.
+                    ts = datetime.strptime(self.grants[unit][lock],
+                                           _timestamp_format)
+                    del self.grants[unit][lock]
+                    self.released(unit, lock, ts)
+
+        # Grant locks
+        for unit in self.requests.keys():
+            for lock in self.requests[unit]:
+                self.grant(lock, unit)
+
+    def grant(self, lock, unit):
+        '''Maybe grant the lock to a unit.
+
+        The decision to grant the lock or not is made for $lock
+        by a corresponding method grant_$lock, which you may define
+        in a subclass. If no such method is defined, the default_grant
+        method is used. See Serial.default_grant() for details.
+        '''
+        if not hookenv.is_leader():
+            return False  # Not the leader, so we cannot grant.
+
+        # Set of units already granted the lock.
+        granted = set()
+        for u in self.grants:
+            if lock in self.grants[u]:
+                granted.add(u)
+        if unit in granted:
+            return True  # Already granted.
+
+        # Ordered list of units waiting for the lock.
+        reqs = set()
+        for u in self.requests:
+            if u in granted:
+                continue  # In the granted set. Not wanted in the req list.
+            for l, ts in self.requests[u].items():
+                if l == lock:
+                    reqs.add((ts, u))
+        queue = [t[1] for t in sorted(reqs)]
+        if unit not in queue:
+            return False  # Unit has not requested the lock.
+
+        # Locate custom logic, or fallback to the default.
+        grant_func = getattr(self, 'grant_{}'.format(lock), self.default_grant)
+
+        if grant_func(lock, unit, granted, queue):
+            # Grant the lock.
+            self.msg('Leader grants {} to {}'.format(lock, unit))
+            self.grants.setdefault(unit, {})[lock] = self.requests[unit][lock]
+            return True
+
+        return False
+
+    def released(self, unit, lock, timestamp):
+        '''Called on the leader when it has released a lock.
+
+        By default, does nothing but log messages. Override if you
+        need to perform additional housekeeping when a lock is released,
+        for example recording timestamps.
+        '''
+        interval = _utcnow() - timestamp
+        self.msg('Leader released {} from {}, held {}'.format(lock, unit,
+                                                              interval))
+
+    def require(self, lock, guard_func, *guard_args, **guard_kw):
+        """Decorate a function to be run only when a lock is acquired.
+
+        The lock is requested if the guard function returns True.
+
+        The decorated function is called if the lock has been granted.
+        """
+        def decorator(f):
+            @wraps(f)
+            def wrapper(*args, **kw):
+                if self.granted(lock):
+                    self.msg('Granted {}'.format(lock))
+                    return f(*args, **kw)
+                if guard_func(*guard_args, **guard_kw) and self.acquire(lock):
+                    return f(*args, **kw)
+                return None
+            return wrapper
+        return decorator
+
+    def msg(self, msg):
+        '''Emit a message. Override to customize log spam.'''
+        hookenv.log('coordinator.{} {}'.format(self._name(), msg),
+                    level=hookenv.INFO)
+
+    def _name(self):
+        return self.__class__.__name__
+
+    def _load_state(self):
+        self.msg('Loading state'.format(self._name()))
+
+        # All responses must be stored in the leadership settings.
+        # The leader cannot use local state, as a different unit may
+        # be leader next time. Which is fine, as the leadership
+        # settings are always available.
+        self.grants = json.loads(hookenv.leader_get(self.key) or '{}')
+
+        local_unit = hookenv.local_unit()
+
+        # All requests must be stored on the peer relation. This is
+        # the only channel units have to communicate with the leader.
+        # Even the leader needs to store its requests here, as a
+        # different unit may be leader by the time the request can be
+        # granted.
+        if self.relid is None:
+            # The peer relation is not available. Maybe we are early in
+            # the units's lifecycle. Maybe this unit is standalone.
+            # Fallback to using local state.
+            self.msg('No peer relation. Loading local state')
+            self.requests = {local_unit: self._load_local_state()}
+        else:
+            self.requests = self._load_peer_state()
+            if local_unit not in self.requests:
+                # The peer relation has just been joined. Update any state
+                # loaded from our peers with our local state.
+                self.msg('New peer relation. Merging local state')
+                self.requests[local_unit] = self._load_local_state()
+
+    def _emit_state(self):
+        # Emit this units lock status.
+        for lock in sorted(self.requests[hookenv.local_unit()].keys()):
+            if self.granted(lock):
+                self.msg('Granted {}'.format(lock))
+            else:
+                self.msg('Waiting on {}'.format(lock))
+
+    def _save_state(self):
+        self.msg('Publishing state'.format(self._name()))
+        if hookenv.is_leader():
+            # sort_keys to ensure stability.
+            raw = json.dumps(self.grants, sort_keys=True)
+            hookenv.leader_set({self.key: raw})
+
+        local_unit = hookenv.local_unit()
+
+        if self.relid is None:
+            # No peer relation yet. Fallback to local state.
+            self.msg('No peer relation. Saving local state')
+            self._save_local_state(self.requests[local_unit])
+        else:
+            # sort_keys to ensure stability.
+            raw = json.dumps(self.requests[local_unit], sort_keys=True)
+            hookenv.relation_set(self.relid, relation_settings={self.key: raw})
+
+    def _load_peer_state(self):
+        requests = {}
+        units = set(hookenv.related_units(self.relid))
+        units.add(hookenv.local_unit())
+        for unit in units:
+            raw = hookenv.relation_get(self.key, unit, self.relid)
+            if raw:
+                requests[unit] = json.loads(raw)
+        return requests
+
+    def _local_state_filename(self):
+        # Include the class name. We allow multiple BaseCoordinator
+        # subclasses to be instantiated, and they are singletons, so
+        # this avoids conflicts (unless someone creates and uses two
+        # BaseCoordinator subclasses with the same class name, so don't
+        # do that).
+        return '.charmhelpers.coordinator.{}'.format(self._name())
+
+    def _load_local_state(self):
+        fn = self._local_state_filename()
+        if os.path.exists(fn):
+            with open(fn, 'r') as f:
+                return json.load(f)
+        return {}
+
+    def _save_local_state(self, state):
+        fn = self._local_state_filename()
+        with open(fn, 'w') as f:
+            json.dump(state, f)
+
+    def _release_granted(self):
+        # At the end of every hook, release all locks granted to
+        # this unit. If a hook neglects to make use of what it
+        # requested, it will just have to make the request again.
+        # Implicit release is the only way this will work, as
+        # if the unit is standalone there may be no future triggers
+        # called to do a manual release.
+        unit = hookenv.local_unit()
+        for lock in list(self.requests[unit].keys()):
+            if self.granted(lock):
+                self.msg('Released local {} lock'.format(lock))
+                del self.requests[unit][lock]
+
+
+class Serial(BaseCoordinator):
+    def default_grant(self, lock, unit, granted, queue):
+        '''Default logic to grant a lock to a unit. Unless overridden,
+        only one unit may hold the lock and it will be granted to the
+        earliest queued request.
+
+        To define custom logic for $lock, create a subclass and
+        define a grant_$lock method.
+
+        `unit` is the unit name making the request.
+
+        `granted` is the set of units already granted the lock. It will
+        never include `unit`. It may be empty.
+
+        `queue` is the list of units waiting for the lock, ordered by time
+        of request. It will always include `unit`, but `unit` is not
+        necessarily first.
+
+        Returns True if the lock should be granted to `unit`.
+        '''
+        return unit == queue[0] and not granted
+
+
+def _implicit_peer_relation_name():
+    md = hookenv.metadata()
+    assert 'peers' in md, 'No peer relations in metadata.yaml'
+    return sorted(md['peers'].keys())[0]
+
+
+# A human readable, sortable UTC timestamp format.
+_timestamp_format = '%Y-%m-%d %H:%M:%S.%fZ'
+
+
+def _utcnow():  # pragma: no cover
+    # This wrapper exists as mocking datetime methods is problematic.
+    return datetime.utcnow()
+
+
+def _timestamp():
+    return _utcnow().strftime(_timestamp_format)

=== modified file 'hooks/charmhelpers/core/hookenv.py'
--- hooks/charmhelpers/core/hookenv.py	2015-04-03 15:42:21 +0000
+++ hooks/charmhelpers/core/hookenv.py	2015-06-30 07:36:26 +0000
@@ -20,12 +20,17 @@
 # Authors:
 #  Charm Helpers Developers <[e-mail]>
 
+from __future__ import print_function
+from distutils.version import LooseVersion
 from functools import wraps
+import glob
 import os
 import json
 import yaml
 import subprocess
 import sys
+import errno
+import tempfile
 from subprocess import CalledProcessError
 
 import six
@@ -90,7 +95,18 @@
     if not isinstance(message, six.string_types):
         message = repr(message)
     command += [message]
-    subprocess.call(command)
+    # Missing juju-log should not cause failures in unit tests
+    # Send log output to stderr
+    try:
+        subprocess.call(command)
+    except OSError as e:
+        if e.errno == errno.ENOENT:
+            if level:
+                message = "{}: {}".format(level, message)
+            message = "juju-log: {}".format(message)
+            print(message, file=sys.stderr)
+        else:
+            raise
 
 
 class Serializable(UserDict):
@@ -228,6 +244,7 @@
         self.path = os.path.join(charm_dir(), Config.CONFIG_FILE_NAME)
         if os.path.exists(self.path):
             self.load_previous()
+        atexit(self._implicit_save)
 
     def __getitem__(self, key):
         """For regular dict lookups, check the current juju config first,
@@ -307,6 +324,10 @@
         with open(self.path, 'w') as f:
             json.dump(self, f)
 
+    def _implicit_save(self):
+        if self.implicit_save:
+            self.save()
+
 
 @cached
 def config(scope=None):
@@ -349,18 +370,49 @@
     """Set relation information for the current unit"""
     relation_settings = relation_settings if relation_settings else {}
     relation_cmd_line = ['relation-set']
+    accepts_file = "--file" in subprocess.check_output(
+        relation_cmd_line + ["--help"], universal_newlines=True)
     if relation_id is not None:
         relation_cmd_line.extend(('-r', relation_id))
-    for k, v in (list(relation_settings.items()) + list(kwargs.items())):
-        if v is None:
-            relation_cmd_line.append('{}='.format(k))
-        else:
-            relation_cmd_line.append('{}={}'.format(k, v))
-    subprocess.check_call(relation_cmd_line)
+    settings = relation_settings.copy()
+    settings.update(kwargs)
+    for key, value in settings.items():
+        # Force value to be a string: it always should, but some call
+        # sites pass in things like dicts or numbers.
+        if value is not None:
+            settings[key] = "{}".format(value)
+    if accepts_file:
+        # --file was introduced in Juju 1.23.2. Use it by default if
+        # available, since otherwise we'll break if the relation data is
+        # too big. Ideally we should tell relation-set to read the data from
+        # stdin, but that feature is broken in 1.23.2: Bug #1454678.
+        with tempfile.NamedTemporaryFile(delete=False) as settings_file:
+            settings_file.write(yaml.safe_dump(settings).encode("utf-8"))
+        subprocess.check_call(
+            relation_cmd_line + ["--file", settings_file.name])
+        os.remove(settings_file.name)
+    else:
+        for key, value in settings.items():
+            if value is None:
+                relation_cmd_line.append('{}='.format(key))
+            else:
+                relation_cmd_line.append('{}={}'.format(key, value))
+        subprocess.check_call(relation_cmd_line)
     # Flush cache of any relation-gets for local unit
     flush(local_unit())
 
 
+def relation_clear(r_id=None):
+    ''' Clears any relation data already set on relation r_id '''
+    settings = relation_get(rid=r_id,
+                            unit=local_unit())
+    for setting in settings:
+        if setting not in ['public-address', 'private-address']:
+            settings[setting] = None
+    relation_set(relation_id=r_id,
+                 **settings)
+
+
 @cached
 def relation_ids(reltype=None):
     """A list of relation_ids"""
@@ -542,10 +594,14 @@
             hooks.execute(sys.argv)
     """
 
-    def __init__(self, config_save=True):
+    def __init__(self, config_save=None):
         super(Hooks, self).__init__()
         self._hooks = {}
-        self._config_save = config_save
+
+        # For unknown reasons, we allow the Hooks constructor to override
+        # config().implicit_save.
+        if config_save is not None:
+            config().implicit_save = config_save
 
     def register(self, name, function):
         """Register a hook"""
@@ -553,13 +609,16 @@
 
     def execute(self, args):
         """Execute a registered hook based on args[0]"""
+        _run_atstart()
         hook_name = os.path.basename(args[0])
         if hook_name in self._hooks:
-            self._hooks[hook_name]()
-            if self._config_save:
-                cfg = config()
-                if cfg.implicit_save:
-                    cfg.save()
+            try:
+                self._hooks[hook_name]()
+            except SystemExit as x:
+                if x.code is None or x.code == 0:
+                    _run_atexit()
+                raise
+            _run_atexit()
         else:
             raise UnregisteredHookError(hook_name)
 
@@ -606,3 +665,160 @@
 
     The results set by action_set are preserved."""
     subprocess.check_call(['action-fail', message])
+
+
+def status_set(workload_state, message):
+    """Set the workload state with a message
+
+    Use status-set to set the workload state with a message which is visible
+    to the user via juju status. If the status-set command is not found then
+    assume this is juju < 1.23 and juju-log the message unstead.
+
+    workload_state -- valid juju workload state.
+    message        -- status update message
+    """
+    valid_states = ['maintenance', 'blocked', 'waiting', 'active']
+    if workload_state not in valid_states:
+        raise ValueError(
+            '{!r} is not a valid workload state'.format(workload_state)
+        )
+    cmd = ['status-set', workload_state, message]
+    try:
+        ret = subprocess.call(cmd)
+        if ret == 0:
+            return
+    except OSError as e:
+        if e.errno != errno.ENOENT:
+            raise
+    log_message = 'status-set failed: {} {}'.format(workload_state,
+                                                    message)
+    log(log_message, level='INFO')
+
+
+def status_get():
+    """Retrieve the previously set juju workload state
+
+    If the status-set command is not found then assume this is juju < 1.23 and
+    return 'unknown'
+    """
+    cmd = ['status-get']
+    try:
+        raw_status = subprocess.check_output(cmd, universal_newlines=True)
+        status = raw_status.rstrip()
+        return status
+    except OSError as e:
+        if e.errno == errno.ENOENT:
+            return 'unknown'
+        else:
+            raise
+
+
+def translate_exc(from_exc, to_exc):
+    def inner_translate_exc1(f):
+        def inner_translate_exc2(*args, **kwargs):
+            try:
+                return f(*args, **kwargs)
+            except from_exc:
+                raise to_exc
+
+        return inner_translate_exc2
+
+    return inner_translate_exc1
+
+
+@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
+def is_leader():
+    """Does the current unit hold the juju leadership
+
+    Uses juju to determine whether the current unit is the leader of its peers
+    """
+    cmd = ['is-leader', '--format=json']
+    return json.loads(subprocess.check_output(cmd).decode('UTF-8'))
+
+
+@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
+def leader_get(attribute=None):
+    """Juju leader get value(s)"""
+    cmd = ['leader-get', '--format=json'] + [attribute or '-']
+    return json.loads(subprocess.check_output(cmd).decode('UTF-8'))
+
+
+@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
+def leader_set(settings=None, **kwargs):
+    """Juju leader set value(s)"""
+    # Don't log secrets.
+    # log("Juju leader-set '%s'" % (settings), level=DEBUG)
+    cmd = ['leader-set']
+    settings = settings or {}
+    settings.update(kwargs)
+    for k, v in settings.items():
+        if v is None:
+            cmd.append('{}='.format(k))
+        else:
+            cmd.append('{}={}'.format(k, v))
+    subprocess.check_call(cmd)
+
+
+@cached
+def juju_version():
+    """Full version string (eg. '1.23.3.1-trusty-amd64')"""
+    # Per https://bugs.launchpad.net/juju-core/+bug/1455368/comments/1
+    jujud = glob.glob('/var/lib/juju/tools/machine-*/jujud')[0]
+    return subprocess.check_output([jujud, 'version'],
+                                   universal_newlines=True).strip()
+
+
+@cached
+def has_juju_version(minimum_version):
+    """Return True if the Juju version is at least the provided version"""
+    return LooseVersion(juju_version()) >= LooseVersion(minimum_version)
+
+
+_atexit = []
+_atstart = []
+
+
+def atstart(callback, *args, **kwargs):
+    '''Schedule a callback to run before the main hook.
+
+    Callbacks are run in the order they were added.
+
+    This is useful for modules and classes to perform initialization
+    and inject behavior. In particular:
+        - Run common code before all of your hooks, such as logging
+          the hook name or interesting relation data.
+        - Defer object or module initialization that requires a hook
+          context until we know there actually is a hook context,
+          making testing easier.
+        - Rather than requiring charm authors to include boilerplate to
+          invoke your helper's behavior, have it run automatically if
+          your object is instantiated or module imported.
+
+    This is not at all useful after your hook framework as been launched.
+    '''
+    global _atstart
+    _atstart.append((callback, args, kwargs))
+
+
+def atexit(callback, *args, **kwargs):
+    '''Schedule a callback to run on successful hook completion.
+
+    Callbacks are run in the reverse order that they were added.'''
+    _atexit.append((callback, args, kwargs))
+
+
+def _run_atstart():
+    '''Hook frameworks must invoke this before running the main hook body.'''
+    global _atstart
+    for callback, args, kwargs in _atstart:
+        callback(*args, **kwargs)
+    del _atstart[:]
+
+
+def _run_atexit():
+    '''Hook frameworks must invoke this after the main hook body has
+    successfully completed. Do not invoke it if the hook fails.'''
+    global _atexit
+    for callback, args, kwargs in reversed(_atexit):
+        callback(*args, **kwargs)
+    del _atexit[:]

=== modified file 'hooks/charmhelpers/core/host.py'
--- hooks/charmhelpers/core/host.py	2015-04-03 15:42:21 +0000
+++ hooks/charmhelpers/core/host.py	2015-06-30 07:36:26 +0000
@@ -413,7 +413,7 @@
     the pkgcache argument is None. Be sure to add charmhelpers.fetch if
     you call this function, or pass an apt_pkg.Cache() instance.
     '''
-    from apt import apt_pkg
+    import apt_pkg
     if not pkgcache:
         from charmhelpers.fetch import apt_cache
         pkgcache = apt_cache()

=== modified file 'hooks/charmhelpers/core/services/base.py'
--- hooks/charmhelpers/core/services/base.py	2015-01-26 13:07:31 +0000
+++ hooks/charmhelpers/core/services/base.py	2015-06-30 07:36:26 +0000
@@ -15,8 +15,8 @@
 # along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
-import re
 import json
+from inspect import getargspec
 from collections import Iterable, OrderedDict
 
 from charmhelpers.core import host
@@ -128,15 +128,18 @@
         """
         Handle the current hook by doing The Right Thing with the registered services.
         """
-        hook_name = hookenv.hook_name()
-        if hook_name == 'stop':
-            self.stop_services()
-        else:
-            self.provide_data()
-            self.reconfigure_services()
-        cfg = hookenv.config()
-        if cfg.implicit_save:
-            cfg.save()
+        hookenv._run_atstart()
+        try:
+            hook_name = hookenv.hook_name()
+            if hook_name == 'stop':
+                self.stop_services()
+            else:
+                self.reconfigure_services()
+                self.provide_data()
+        except SystemExit as x:
+            if x.code is None or x.code == 0:
+                hookenv._run_atexit()
+        hookenv._run_atexit()
 
     def provide_data(self):
         """
@@ -145,15 +148,36 @@
         A provider must have a `name` attribute, which indicates which relation
         to set data on, and a `provide_data()` method, which returns a dict of
         data to set.
+
+        The `provide_data()` method can optionally accept two parameters:
+
+          * ``remote_service`` The name of the remote service that the data will
+            be provided to.  The `provide_data()` method will be called once
+            for each connected service (not unit).  This allows the method to
+            tailor its data to the given service.
+          * ``service_ready`` Whether or not the service definition had all of
+            its requirements met, and thus the ``data_ready`` callbacks run.
+
+        Note that the ``provided_data`` methods are now called **after** the
+        ``data_ready`` callbacks are run.  This gives the ``data_ready`` callbacks
+        a chance to generate any data necessary for the providing to the remote
+        services.
         """
-        hook_name = hookenv.hook_name()
-        for service in self.services.values():
+        for service_name, service in self.services.items():
+            service_ready = self.is_ready(service_name)
             for provider in service.get('provided_data', []):
-                if re.match(r'{}-relation-(joined|changed)'.format(provider.name), hook_name):
-                    data = provider.provide_data()
-                    _ready = provider._is_ready(data) if hasattr(provider, '_is_ready') else data
-                    if _ready:
-                        hookenv.relation_set(None, data)
+                for relid in hookenv.relation_ids(provider.name):
+                    units = hookenv.related_units(relid)
+                    if not units:
+                        continue
+                    remote_service = units[0].split('/')[0]
+                    argspec = getargspec(provider.provide_data)
+                    if len(argspec.args) > 1:
+                        data = provider.provide_data(remote_service, service_ready)
+                    else:
+                        data = provider.provide_data()
+                    if data:
+                        hookenv.relation_set(relid, data)
 
     def reconfigure_services(self, *service_names):
         """

=== modified file 'hooks/charmhelpers/core/strutils.py'
--- hooks/charmhelpers/core/strutils.py	2015-02-27 05:36:25 +0000
+++ hooks/charmhelpers/core/strutils.py	2015-06-30 07:36:26 +0000
@@ -33,9 +33,9 @@
 
     value = value.strip().lower()
 
-    if value in ['y', 'yes', 'true', 't']:
+    if value in ['y', 'yes', 'true', 't', 'on']:
         return True
-    elif value in ['n', 'no', 'false', 'f']:
+    elif value in ['n', 'no', 'false', 'f', 'off']:
         return False
 
     msg = "Unable to interpret string value '%s' as boolean" % (value)

=== modified file 'hooks/charmhelpers/fetch/giturl.py'
--- hooks/charmhelpers/fetch/giturl.py	2015-02-18 14:24:32 +0000
+++ hooks/charmhelpers/fetch/giturl.py	2015-06-30 07:36:26 +0000
@@ -45,14 +45,16 @@
         else:
             return True
 
-    def clone(self, source, dest, branch):
+    def clone(self, source, dest, branch, depth=None):
         if not self.can_handle(source):
             raise UnhandledSource("Cannot handle {}".format(source))
 
-        repo = Repo.clone_from(source, dest)
-        repo.git.checkout(branch)
+        if depth:
+            Repo.clone_from(source, dest, branch=branch, depth=depth)
+        else:
+            Repo.clone_from(source, dest, branch=branch)
 
-    def install(self, source, branch="master", dest=None):
+    def install(self, source, branch="master", dest=None, depth=None):
         url_parts = self.parse_url(source)
         branch_name = url_parts.path.strip("/").split("/")[-1]
         if dest:
@@ -63,7 +65,7 @@
         if not os.path.exists(dest_dir):
             mkdir(dest_dir, perms=0o755)
         try:
-            self.clone(source, dest_dir, branch)
+            self.clone(source, dest_dir, branch, depth)
         except GitCommandError as e:
             raise UnhandledSource(e.message)
         except OSError as e:

=== modified symlink 'hooks/cluster-relation-changed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/cluster-relation-changed	1970-01-01 00:00:00 +0000
+++ hooks/cluster-relation-changed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified symlink 'hooks/cluster-relation-departed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/cluster-relation-departed	1970-01-01 00:00:00 +0000
+++ hooks/cluster-relation-departed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== removed symlink 'hooks/cluster-relation-joined'
=== target was u'hooks.py'
=== modified symlink 'hooks/config-changed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/config-changed	1970-01-01 00:00:00 +0000
+++ hooks/config-changed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== added file 'hooks/coordinator.py'
--- hooks/coordinator.py	1970-01-01 00:00:00 +0000
+++ hooks/coordinator.py	2015-06-30 07:36:26 +0000
@@ -0,0 +1,35 @@
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+from charmhelpers.coordinator import BaseCoordinator
+
+
+class CassandraCoordinator(BaseCoordinator):
+    def default_grant(self, lock, unit, granted, queue):
+        '''Grant locks to only one unit at a time, regardless of its name.
+
+        This lets us keep separate locks like repair and restart,
+        while ensuring the operations do not occur on different nodes
+        at the same time.
+        '''
+        # Return True if this unit has already been granted a lock.
+        if self.grants.get(unit):
+            return True
+
+        # Otherwise, return True if the unit is first in the queue.
+        return queue[0] == unit and not granted
+
+
+coordinator = CassandraCoordinator()

=== modified symlink 'hooks/data-relation-changed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/data-relation-changed	1970-01-01 00:00:00 +0000
+++ hooks/data-relation-changed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified symlink 'hooks/data-relation-departed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/data-relation-departed	1970-01-01 00:00:00 +0000
+++ hooks/data-relation-departed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified symlink 'hooks/database-admin-relation-changed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/database-admin-relation-changed	1970-01-01 00:00:00 +0000
+++ hooks/database-admin-relation-changed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified symlink 'hooks/database-relation-changed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/database-relation-changed	1970-01-01 00:00:00 +0000
+++ hooks/database-relation-changed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified file 'hooks/definitions.py'
--- hooks/definitions.py	2015-04-13 16:39:20 +0000
+++ hooks/definitions.py	2015-06-30 07:36:26 +0000
@@ -15,13 +15,11 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from charmhelpers.core import hookenv
-from charmhelpers.core.hookenv import ERROR
 from charmhelpers.core import services
 
 import actions
 import helpers
 import relations
-import rollingrestart
 
 
 def get_service_definitions():
@@ -35,32 +33,13 @@
     config = hookenv.config()
 
     return [
-        # Actions done before or while the Cassandra service is running.
-        dict(service=helpers.get_cassandra_service(),
-
-             # Open access to client and replication ports. Client
-             # protocols require password authentication. Access to
-             # the unauthenticated replication ports is protected via
-             # ufw firewall rules. We do not open the JMX port, although
-             # we could since it is similarly protected by ufw.
-             ports=[config['rpc_port'],               # Thrift clients
-                    config['native_transport_port'],  # Native clients.
-                    config['storage_port'],           # Plaintext replication
-                    config['ssl_storage_port']],      # Encrypted replication.
-
-             required_data=[relations.StorageRelation()],
-             provided_data=[relations.StorageRelation()],
+        # Prepare for the Cassandra service.
+        dict(service='install',
              data_ready=[actions.set_proxy,
                          actions.preinstall,
                          actions.emit_meminfo,
                          actions.revert_unchangeable_config,
                          actions.store_unit_private_ip,
-                         actions.set_unit_zero_bootstrapped,
-                         actions.shutdown_before_joining_peers,
-                         # Must open ports before attempting bind to the
-                         # public ip address.
-                         actions.configure_firewall,
-                         actions.grant_ssh_access,
                          actions.add_implicit_package_signing_keys,
                          actions.configure_sources,
                          actions.swapoff,
@@ -68,65 +47,75 @@
                          actions.install_oracle_jre,
                          actions.install_cassandra_packages,
                          actions.emit_java_version,
-                         actions.ensure_cassandra_package_status,
+                         actions.ensure_cassandra_package_status],
+             start=[], stop=[]),
+
+        # Get Cassandra running.
+        dict(service=helpers.get_cassandra_service(),
+
+             # Open access to client and replication ports. Client
+             # protocols require password authentication. Access to
+             # the unauthenticated replication ports is protected via
+             # ufw firewall rules. We do not open the JMX port, although
+             # we could since it is similarly protected by ufw.
+             ports=[config['rpc_port'],               # Thrift clients
+                    config['native_transport_port'],  # Native clients.
+                    config['storage_port'],           # Plaintext replication
+                    config['ssl_storage_port']],      # Encrypted replication.
+
+             required_data=[relations.StorageRelation(),
+                            relations.PeerRelation()],
+             provided_data=[relations.StorageRelation()],
+             data_ready=[actions.configure_firewall,
+                         actions.maintain_seeds,
                          actions.configure_cassandra_yaml,
                          actions.configure_cassandra_env,
                          actions.configure_cassandra_rackdc,
                          actions.reset_all_io_schedulers,
-                         actions.nrpe_external_master_relation,
-                         actions.maybe_schedule_restart],
+                         actions.maybe_restart,
+                         actions.request_unit_superuser,
+                         actions.reset_default_password],
              start=[services.open_ports],
              stop=[actions.stop_cassandra, services.close_ports]),
 
-        # Rolling restart. This service will call the restart hook when
-        # it is this units turn to restart. This is also where we do
-        # actions done while Cassandra is not running, and where we do
-        # actions that should only be done by one node at a time.
-        rollingrestart.make_service([helpers.stop_cassandra,
-                                     helpers.remount_cassandra,
-                                     helpers.ensure_database_directories,
-                                     helpers.pre_bootstrap,
-                                     helpers.start_cassandra,
-                                     helpers.post_bootstrap,
-                                     helpers.wait_for_agreed_schema,
-                                     helpers.wait_for_normality,
-                                     helpers.emit_describe_cluster,
-                                     helpers.reset_default_password,
-                                     helpers.ensure_unit_superuser,
-                                     helpers.reset_auth_keyspace_replication]),
-
         # Actions that must be done while Cassandra is running.
         dict(service='post',
              required_data=[RequiresLiveNode()],
-             data_ready=[actions.publish_database_relations,
+             data_ready=[actions.post_bootstrap,
+                         actions.create_unit_superusers,
+                         actions.reset_auth_keyspace_replication,
+                         actions.publish_database_relations,
                          actions.publish_database_admin_relations,
                          actions.install_maintenance_crontab,
-                         actions.emit_describe_cluster,
-                         actions.emit_auth_keyspace_status,
-                         actions.emit_netstats],
+                         actions.nrpe_external_master_relation,
+                         actions.emit_cluster_info,
+                         actions.set_active],
              start=[], stop=[])]
 
 
 class RequiresLiveNode:
     def __bool__(self):
-        return self.is_live()
+        is_live = self.is_live()
+        hookenv.log('Requirement RequiresLiveNode: {}'.format(is_live),
+                    hookenv.DEBUG)
+        return is_live
 
     def is_live(self):
+        if helpers.is_decommissioned():
+            hookenv.log('Node is decommissioned')
+            return False
+
         if helpers.is_cassandra_running():
-            if helpers.is_decommissioned():
-                # Node is decommissioned and will refuse to talk.
-                hookenv.log('Node is decommissioned')
-                return False
-            try:
-                with helpers.connect():
-                    hookenv.log("Node live and authentication working")
-                    return True
-            except Exception as x:
-                hookenv.log(
-                    'Unable to connect as superuser: {}'.format(str(x)),
-                    ERROR)
-                return False
-        return False
+            hookenv.log('Cassandra is running')
+            if hookenv.local_unit() in helpers.get_unit_superusers():
+                hookenv.log('Credentials created')
+                return True
+            else:
+                hookenv.log('Credentials have not been created')
+                return False
+        else:
+            hookenv.log('Cassandra is not running')
+            return False
 
 
 def get_service_manager():

=== modified file 'hooks/helpers.py'
--- hooks/helpers.py	2015-04-17 10:06:34 +0000
+++ hooks/helpers.py	2015-06-30 07:36:26 +0000
@@ -13,14 +13,12 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
 import configparser
 from contextlib import contextmanager
 from datetime import timedelta
 import errno
 from functools import wraps
 import io
-from itertools import chain
 import json
 import os.path
 import re
@@ -37,13 +35,11 @@
 import cassandra.query
 import yaml
 
-from charmhelpers.contrib import unison
 from charmhelpers.core import hookenv, host
 from charmhelpers.core.hookenv import DEBUG, ERROR, WARNING
 from charmhelpers import fetch
 
-import relations
-import rollingrestart
+from coordinator import coordinator
 
 
 RESTART_TIMEOUT = 600
@@ -101,10 +97,9 @@
     if hookenv.config('extra_packages'):
         packages.extend(hookenv.config('extra_packages').split())
     packages = fetch.filter_installed_packages(packages)
-    # if 'ntp' in packages:
-    #     fetch.apt_install(['ntp'], fatal=True)  # With autostart
-    #     packages.remove('ntp')
     if packages:
+        # The DSE packages are huge, so this might take some time.
+        status_set('maintenance', 'Installing packages')
         with autostart_disabled(['cassandra']):
             fetch.apt_install(packages, fatal=True)
 
@@ -128,26 +123,13 @@
     dpkg.communicate(input=''.join(selections).encode('US-ASCII'))
 
 
-def seed_ips():
+def get_seed_ips():
     '''Return the set of seed ip addresses.
 
-    If this is the only unit in the service, then the local unit's IP
-    address is returned as the only seed.
-
-    If there is more than one unit in the service, then the IP addresses
-    of three lowest numbered peers (if bootstrapped) are returned.
+    We use ip addresses rather than unit names, as we may need to use
+    external seed ips at some point.
     '''
-    peers = rollingrestart.get_peers()
-    if not peers:
-        return set([hookenv.unit_private_ip()])
-
-    # The three lowest numbered units are seeds & may include this unit.
-    seeds = sorted(list(peers) + [hookenv.local_unit()],
-                   key=lambda x: int(x.split('/')[-1]))[:3]
-
-    relid = rollingrestart.get_peer_relation_id()
-    return set(hookenv.relation_get('private-address', seed, relid)
-               for seed in seeds)
+    return set((hookenv.leader_get('seeds') or '').split(','))
 
 
 def actual_seed_ips():
@@ -163,6 +145,7 @@
     Entries in the config file may be absolute, relative to
     /var/lib/cassandra, or relative to the mountpoint.
     '''
+    import relations
     storage = relations.StorageRelation()
     if storage.mountpoint:
         root = os.path.join(storage.mountpoint, 'cassandra')
@@ -177,9 +160,6 @@
 
     Returns the absolute path.
     '''
-    # Guard against changing perms on a running db. Although probably
-    # harmless, it causes shutil.chown() to fail.
-    assert not is_cassandra_running()
     absdir = get_database_directory(config_path)
 
     # Work around Bug #1427150 by ensuring components of the path are
@@ -322,9 +302,7 @@
 
 def get_cassandra_version():
     if get_cassandra_edition() == 'dse':
-        # When we support multiple versions, we will need to map
-        # DataStax versions to Cassandra versions.
-        return '2.0' if get_package_version('dse-full') else None
+        return '2.1' if get_package_version('dse-full') else None
     return get_package_version('cassandra')
 
 
@@ -352,8 +330,6 @@
     edition = get_cassandra_edition()
     if edition == 'dse':
         pid_file = "/var/run/dse/dse.pid"
-    # elif apt_pkg.version_compare(get_cassandra_version(), "2.0") < 0:
-    #     pid_file = "/var/run/cassandra.pid"
     else:
         pid_file = "/var/run/cassandra/cassandra.pid"
     return pid_file
@@ -377,20 +353,20 @@
         # agreement.
         pass
     else:
+        # NB. OpenJDK 8 not available in trusty.
         packages.add('openjdk-7-jre-headless')
 
     return packages
 
 
 @logged
-def stop_cassandra(immediate=False):
+def stop_cassandra():
     if is_cassandra_running():
-        if not immediate:
-            # If there are cluster operations in progress, wait until
-            # they are complete before restarting. This might take days.
-            wait_for_normality()
+        hookenv.log('Shutting down Cassandra')
         host.service_stop(get_cassandra_service())
-    assert not is_cassandra_running()
+    if is_cassandra_running():
+        hookenv.status_set('blocked', 'Cassandra failed to shut down')
+        raise SystemExit(0)
 
 
 @logged
@@ -398,9 +374,19 @@
     if is_cassandra_running():
         return
 
-    actual_seeds = actual_seed_ips()
+    actual_seeds = sorted(actual_seed_ips())
     assert actual_seeds, 'Attempting to start cassandra with empty seed list'
-    hookenv.log('Starting Cassandra with seeds {!r}'.format(actual_seeds))
+    hookenv.config()['configured_seeds'] = actual_seeds
+
+    if is_bootstrapped():
+        status_set('maintenance',
+                   'Starting Cassandra with seeds {!r}'
+                   .format(','.join(actual_seeds)))
+    else:
+        status_set('maintenance',
+                   'Bootstrapping with seeds {}'
+                   .format(','.join(actual_seeds)))
+
     host.service_start(get_cassandra_service())
 
     # Wait for Cassandra to actually start, or abort.
@@ -409,35 +395,8 @@
         if is_cassandra_running():
             return
         time.sleep(1)
-    hookenv.log('Cassandra failed to start.', ERROR)
-    raise SystemExit(1)
-
-
-def is_responding(ip, port=None, timeout=5):
-    if port is None:
-        port = hookenv.config()['storage_port']
-    try:
-        subprocess.check_output(['nc', '-zw', str(timeout), ip, str(port)],
-                                stderr=subprocess.STDOUT)
-        hookenv.log('{} listening on port {}'.format(ip, port), DEBUG)
-        return True
-    except subprocess.TimeoutExpired:
-        hookenv.log('{} not listening on port {}'.format(ip, port), DEBUG)
-        return False
-
-
-@logged
-def are_all_nodes_responding():
-    all_contactable = True
-    for ip in node_ips():
-        if ip == hookenv.unit_private_ip():
-            pass
-        elif is_responding(ip, timeout=5):
-            hookenv.log('{} is responding'.format(ip))
-        else:
-            all_contactable = False
-            hookenv.log('{} is not responding'.format(ip))
-    return all_contactable
+    status_set('blocked', 'Cassandra failed to start')
+    raise SystemExit(0)
 
 
 @logged
@@ -451,8 +410,10 @@
 def remount_cassandra():
     '''If a new mountpoint is ready, migrate data across to it.'''
     assert not is_cassandra_running()  # Guard against data loss.
+    import relations
     storage = relations.StorageRelation()
     if storage.needs_remount():
+        status_set('maintenance', 'Migrating data to new mountpoint')
         hookenv.config()['bootstrapped_into_cluster'] = False
         if storage.mountpoint is None:
             hookenv.log('External storage AND DATA gone. '
@@ -468,6 +429,9 @@
 @logged
 def ensure_database_directories():
     '''Ensure that directories Cassandra expects to store its data in exist.'''
+    # Guard against changing perms on a running db. Although probably
+    # harmless, it causes shutil.chown() to fail.
+    assert not is_cassandra_running()
     db_dirs = get_all_database_directories()
     unpacked_db_dirs = (db_dirs['data_file_directories'] +
                         [db_dirs['commitlog_directory']] +
@@ -476,28 +440,7 @@
         ensure_database_directory(db_dir)
 
 
-@logged
-def reset_default_password():
-    config = hookenv.config()
-    if config.get('default_admin_password_changed', False):
-        hookenv.log('Default admin password changed in an earlier hook')
-        return
-
-    # If we can connect using the default superuser password
-    # 'cassandra', change it to something random. We do this in a loop
-    # to ensure the hook that calls this is idempotent.
-    try:
-        with connect('cassandra', 'cassandra') as session:
-            hookenv.log('Changing default admin password')
-            query(session, 'ALTER USER cassandra WITH PASSWORD %s',
-                  ConsistencyLevel.QUORUM, (host.pwgen(),))
-    except cassandra.AuthenticationFailed:
-        hookenv.log('Default admin password already changed')
-
-    config['default_admin_password_changed'] = True
-
-
-CONNECT_TIMEOUT = 300
+CONNECT_TIMEOUT = 10
 
 
 @contextmanager
@@ -568,72 +511,45 @@
                 raise
 
 
-def ensure_user(username, password, superuser=False):
+def encrypt_password(password):
+    return bcrypt.hashpw(password, bcrypt.gensalt())
+
+
+@logged
+def ensure_user(session, username, encrypted_password, superuser=False):
     '''Create the DB user if it doesn't already exist & reset the password.'''
     if superuser:
         hookenv.log('Creating SUPERUSER {}'.format(username))
-        sup = 'SUPERUSER'
     else:
         hookenv.log('Creating user {}'.format(username))
-        sup = 'NOSUPERUSER'
-    with connect() as session:
-        query(session,
-              'CREATE USER IF NOT EXISTS %s '
-              'WITH PASSWORD %s {}'.format(sup),
-              ConsistencyLevel.QUORUM, (username, password,))
-        query(session, 'ALTER USER %s WITH PASSWORD %s {}'.format(sup),
-              ConsistencyLevel.QUORUM, (username, password,))
-
-
-@logged
-def ensure_unit_superuser():
-    '''If the unit's superuser account is not working, recreate it.'''
-    try:
-        with connect(auth_timeout=10):
-            hookenv.log('Unit superuser account already setup', DEBUG)
-            return
-    except cassandra.AuthenticationFailed:
-        pass
-
-    create_unit_superuser()  # Doesn't exist or can't access, so create it.
-
-    with connect():
-        hookenv.log('Unit superuser password reset successful')
-
-
-@logged
-def create_unit_superuser():
+    query(session,
+          'INSERT INTO system_auth.users (name, super) VALUES (%s, %s)',
+          ConsistencyLevel.ALL, (username, superuser))
+    query(session,
+          'INSERT INTO system_auth.credentials (username, salted_hash) '
+          'VALUES (%s, %s)',
+          ConsistencyLevel.ALL, (username, encrypted_password))
+
+
+@logged
+def create_unit_superuser_hard():
     '''Create or recreate the unit's superuser account.
 
-    As there may be no known superuser credentials to use, we restart
-    the node using the AllowAllAuthenticator and insert our user
-    directly into the system_auth keyspace.
+    This method is used when there are no known superuser credentials
+    to use. We restart the node using the AllowAllAuthenticator and
+    insert our credentials directly into the system_auth keyspace.
     '''
     username, password = superuser_credentials()
+    pwhash = encrypt_password(password)
     hookenv.log('Creating unit superuser {}'.format(username))
 
     # Restart cassandra without authentication & listening on localhost.
-    wait_for_normality()
     reconfigure_and_restart_cassandra(
         dict(authenticator='AllowAllAuthenticator', rpc_address='localhost'))
-    wait_for_normality()
-    emit_cluster_info()
     for _ in backoff('superuser creation'):
         try:
             with connect() as session:
-                pwhash = bcrypt.hashpw(password,
-                                       bcrypt.gensalt())  # Cassandra 2.1
-                statement = dedent('''\
-                    INSERT INTO system_auth.users (name, super)
-                    VALUES (%s, TRUE)
-                    ''')
-                query(session, statement, ConsistencyLevel.QUORUM, (username,))
-                statement = dedent('''\
-                    INSERT INTO system_auth.credentials (username, salted_hash)
-                    VALUES (%s, %s)
-                    ''')
-                query(session, statement,
-                      ConsistencyLevel.QUORUM, (username, pwhash))
+                ensure_user(session, username, pwhash, superuser=True)
                 break
         except Exception as x:
             print(str(x))
@@ -641,7 +557,6 @@
     # Restart Cassandra with regular config.
     nodetool('flush')  # Ensure our backdoor updates are flushed.
     reconfigure_and_restart_cassandra()
-    wait_for_normality()
 
 
 def get_cqlshrc_path():
@@ -703,12 +618,8 @@
     sys.stdout.flush()
 
 
-def nodetool(*cmd, ip=None, timeout=120):
+def nodetool(*cmd, timeout=120):
     cmd = ['nodetool'] + [str(i) for i in cmd]
-    if ip is not None:
-        assert ip in unison.collect_authed_hosts(
-            rollingrestart.get_peer_relation_name())
-        cmd = ['sudo', '-Hsu', 'juju_ssh', 'ssh', ip] + cmd
     i = 0
     until = time.time() + timeout
     for _ in backoff('nodetool to work'):
@@ -734,55 +645,16 @@
         except subprocess.CalledProcessError as x:
             if i > 1:
                 emit(x.output.expandtabs())  # Expand tabs for juju debug-log.
+            if not is_cassandra_running():
+                status_set('blocked',
+                           'Cassandra has unexpectedly shutdown')
+                raise SystemExit(0)
             if time.time() >= until:
                 raise
 
 
-def node_ips():
-    '''IP addresses of all nodes in the Cassandra cluster.'''
-    if is_bootstrapped() or num_peers() == 0:
-        # If bootstrapped, or standalone, we trust our local info.
-        raw = nodetool('status', 'system_auth')
-    else:
-        # If not bootstrapped, we query a seed.
-        authed_ips = unison.collect_authed_hosts(
-            rollingrestart.get_peer_relation_name())
-        seeds = [ip for ip in seed_ips() if ip in authed_ips]
-        if not seeds:
-            return set()  # Not bootstrapped, and nobody else is either.
-        raw = nodetool('status', 'system_auth', ip=seeds[0])
-
-    ips = set()
-    for line in raw.splitlines():
-        match = re.search(r'^(\w)([NLJM])\s+([\d\.]+)\s', line)
-        if match is not None and (match.group(1) == 'U'
-                                  or match.group(2) != 'L'):
-            ips.add(match.group(3))
-    return ips
-
-
-def up_node_ips():
-    '''IP addresses of nodes that are up.'''
-    raw = nodetool('status', 'system_auth')
-    ips = set()
-    for line in raw.splitlines():
-        match = re.search(r'^(\w)([NLJM])\s+([\d\.]+)\s', line)
-        if match is not None and match.group(1) == 'U':  # Up
-            ips.add(match.group(3))
-    return ips
-
-
 def num_nodes():
-    '''Number of nodes in the Cassandra cluster.
-
-    This is not necessarily the same as the number of peers, as nodes
-    may be decommissioned.
-    '''
-    return len(node_ips())
-
-
-def num_peers():
-    return len(rollingrestart.get_peers())
+    return len(get_bootstrapped_ips())
 
 
 def read_cassandra_yaml():
@@ -819,7 +691,8 @@
                           'storage_port', 'ssl_storage_port']
     cassandra_yaml.update((k, config[k]) for k in simple_config_keys)
 
-    seeds = ','.join(seeds or seed_ips())  # Don't include whitespace!
+    seeds = ','.join(seeds or get_seed_ips())  # Don't include whitespace!
+    hookenv.log('Configuring seeds as {!r}'.format(seeds), DEBUG)
     assert seeds, 'Attempting to configure cassandra with empty seed list'
     cassandra_yaml['seed_provider'][0]['parameters'][0]['seeds'] = seeds
 
@@ -892,36 +765,6 @@
         return False
 
 
-@logged
-def reset_all_io_schedulers():
-    dirs = get_all_database_directories()
-    dirs = (dirs['data_file_directories'] + [dirs['commitlog_directory']] +
-            [dirs['saved_caches_directory']])
-    config = hookenv.config()
-    for d in dirs:
-        if os.path.isdir(d):  # Directory may not exist yet.
-            set_io_scheduler(config['io_scheduler'], d)
-
-
-@logged
-def reset_auth_keyspace_replication():
-    # Cassandra requires you to manually set the replication factor of
-    # the system_auth keyspace, to ensure availability and redundancy.
-    n = max(min(num_nodes(), 3), 1)  # Cap at 1-3 replicas.
-    datacenter = hookenv.config()['datacenter']
-    with connect() as session:
-        strategy_opts = get_auth_keyspace_replication(session)
-        rf = int(strategy_opts.get(datacenter, -1))
-        hookenv.log('system_auth rf={!r}'.format(strategy_opts))
-        if rf != n:
-            strategy_opts['class'] = 'NetworkTopologyStrategy'
-            strategy_opts[datacenter] = n
-            if 'replication_factor' in strategy_opts:
-                del strategy_opts['replication_factor']
-            set_auth_keyspace_replication(session, strategy_opts)
-            repair_auth_keyspace()
-
-
 def get_auth_keyspace_replication(session):
     statement = dedent('''\
         SELECT strategy_options FROM system.schema_keyspaces
@@ -931,321 +774,96 @@
     return json.loads(r[0][0])
 
 
+@logged
 def set_auth_keyspace_replication(session, settings):
-    wait_for_normality()
-    hookenv.log('Updating system_auth rf={!r}'.format(settings))
+    # Live operation, so keep status the same.
+    status_set(hookenv.status_get(),
+               'Updating system_auth rf to {!r}'.format(settings))
     statement = 'ALTER KEYSPACE system_auth WITH REPLICATION = %s'
-    query(session, statement, ConsistencyLevel.QUORUM, (settings,))
+    query(session, statement, ConsistencyLevel.ALL, (settings,))
 
 
 @logged
 def repair_auth_keyspace():
-    # First, wait for schema agreement. Attempting to repair a keyspace
-    # with inconsistent replication settings will fail.
-    wait_for_agreed_schema()
     # Repair takes a long time, and may need to be retried due to 'snapshot
     # creation' errors, but should certainly complete within an hour since
     # the keyspace is tiny.
+    status_set(hookenv.status_get(),
+               'Repairing system_auth keyspace')
     nodetool('repair', 'system_auth', timeout=3600)
 
 
-def non_system_keyspaces():
-    # If there are only system keyspaces defined, there is no data we
-    # may want to preserve and we can safely proceed with the bootstrap.
-    # This should always be the case, but it is worth checking for weird
-    # situations such as reusing an existing external mount without
-    # clearing its data.
-    dfds = get_all_database_directories()['data_file_directories']
-    keyspaces = set(chain(*[os.listdir(dfd) for dfd in dfds]))
-    hookenv.log('keyspaces={!r}'.format(keyspaces), DEBUG)
-    return keyspaces - set(['system', 'system_auth', 'system_traces',
-                            'dse_system'])
-
-
-def nuke_local_database():
-    '''Destroy the local database, entirely, so this node may bootstrap.
-
-    This needs to be lower level than just removing selected keyspaces
-    such as 'system', as commitlogs and other crumbs can also cause the
-    bootstrap process to fail disasterously.
-    '''
-    # This function is dangerous enough to warrent a guard.
-    assert not is_bootstrapped()
-    assert unit_number() != 0
-
-    keyspaces = non_system_keyspaces()
-    if keyspaces:
-        hookenv.log('Non-system keyspaces {!r} detected. '
-                    'Unable to bootstrap.'.format(keyspaces), ERROR)
-        raise SystemExit(1)
-
-    dirs = get_all_database_directories()
-    nuke_directory_contents(dirs['saved_caches_directory'])
-    nuke_directory_contents(dirs['commitlog_directory'])
-    for dfd in dirs['data_file_directories']:
-        nuke_directory_contents(dfd)
-
-
-def nuke_directory_contents(d):
-    '''Remove the contents of directory d, leaving d in place.
-
-    We don't remove the top level directory as it may be a mount
-    or symlink we cannot recreate.
-    '''
-    for name in os.listdir(d):
-        path = os.path.join(d, name)
-        if os.path.isdir(path):
-            shutil.rmtree(path)
-        else:
-            os.remove(path)
-
-
-def unit_number(unit=None):
-    if unit is None:
-        unit = hookenv.local_unit()
-    return int(unit.split('/')[-1])
-
-
 def is_bootstrapped(unit=None):
     '''Return True if the node has already bootstrapped into the cluster.'''
-    if unit is None:
-        unit = hookenv.local_unit()
-    if unit.endswith('/0'):
-        return True
-    relid = rollingrestart.get_peer_relation_id()
-    if not relid:
+    if unit is None or unit == hookenv.local_unit():
+        return hookenv.config().get('bootstrapped', False)
+    elif coordinator.relid:
+        return bool(hookenv.relation_get(rid=coordinator.relid,
+                                         unit=unit).get('bootstrapped'))
+    else:
         return False
-    return hookenv.relation_get('bootstrapped', unit, relid) == '1'
-
-
-def set_bootstrapped(flag):
-    relid = rollingrestart.get_peer_relation_id()
-    if bool(flag) is not is_bootstrapped():
-        hookenv.log('Setting bootstrapped to {}'.format(flag))
-
-    if flag:
-        hookenv.relation_set(relid, bootstrapped="1")
-    else:
-        hookenv.relation_set(relid, bootstrapped=None)
-
-
-def bootstrapped_peers():
-    '''Return the ordered list of bootstrapped peers'''
-    peer_relid = rollingrestart.get_peer_relation_id()
-    return [peer for peer in rollingrestart.get_peers()
-            if hookenv.relation_get('bootstrapped', peer, peer_relid) == '1']
-
-
-def unbootstrapped_peers():
-    '''Return the ordered list of unbootstrapped peers'''
-    peer_relid = rollingrestart.get_peer_relation_id()
-    return [peer for peer in rollingrestart.get_peers()
-            if hookenv.relation_get('bootstrapped', peer, peer_relid) != '1']
-
-
-@logged
-def pre_bootstrap():
-    """If we are about to bootstrap a node, prepare.
-
-    Only the first node in the cluster is not bootstrapped. All other
-    nodes added to the cluster need to bootstrap. To bootstrap, the local
-    node needs to be shutdown, the database needs to be completely reset,
-    and the node restarted with a valid seed_node.
-
-    Until juju gains the necessary features, the best we can do is assume
-    that Unit 0 should be the initial seed node. If the service has been
-    running for some time and never had a second unit added, it is almost
-    certain that it is Unit 0 that contains data we want to keep. This
-    assumption is false if you have removed the only unit in the service
-    at some point, so don't do that.
-    """
-    if is_bootstrapped():
-        hookenv.log("Already bootstrapped")
-        return
-
-    if num_peers() == 0:
-        hookenv.log("No peers, no cluster, no bootstrapping")
-        return
-
-    if not seed_ips():
-        hookenv.log("No seeds available. Deferring bootstrap.")
-        raise rollingrestart.DeferRestart()
-
-    # Don't attempt to bootstrap until all lower numbered units have
-    # bootstrapped. We need to do this as the rollingrestart algorithm
-    # fails during initial cluster setup, where two or more newly joined
-    # units may restart (and bootstrap) at the same time and exceed
-    # Cassandra's limits on the number of nodes that may bootstrap
-    # simultaneously. In addition, this ensures we wait until there is
-    # at least one seed bootstrapped, as the three first units will be
-    # seeds.
-    for peer in unbootstrapped_peers():
-        if unit_number(peer) < unit_number():
-            hookenv.log("{} is not bootstrapped. Deferring.".format(peer))
-            raise rollingrestart.DeferRestart()
-
-    # Don't attempt to bootstrap until all bootstrapped peers have
-    # authorized us.
-    peer_relname = rollingrestart.get_peer_relation_name()
-    peer_relid = rollingrestart.get_peer_relation_id()
-    authed_ips = unison.collect_authed_hosts(peer_relname)
-    for peer in bootstrapped_peers():
-        peer_ip = hookenv.relation_get('private-address', peer, peer_relid)
-        if peer_ip not in authed_ips:
-            hookenv.log("{} has not authorized us. Deferring.".format(peer))
-            raise rollingrestart.DeferRestart()
-
-    # Bootstrap fail if we haven't yet opened our ports to all
-    # the bootstrapped nodes. This is the case if we have not yet
-    # joined the peer relationship with the node's unit.
-    missing = node_ips() - peer_ips() - set([hookenv.unit_private_ip()])
-    if missing:
-        hookenv.log("Not yet in a peer relationship with {!r}. "
-                    "Deferring bootstrap".format(missing))
-        raise rollingrestart.DeferRestart()
-
-    # Bootstrap will fail if all the nodes that contain data that needs
-    # to move to the new node are down. If you are storing data in
-    # rf==1, bootstrap can fail if a single node is not contactable. As
-    # far as this charm is concerned, we should not attempt bootstrap
-    # until all the nodes in the cluster and contactable. This also
-    # ensures that peers have run their relation-changed hook and
-    # opened their firewall ports to the new unit.
-    if not are_all_nodes_responding():
-        raise rollingrestart.DeferRestart()
-
-    hookenv.log('Joining cluster and need to bootstrap.')
-
-    nuke_local_database()
-
-
-@logged
-def post_bootstrap():
-    '''Maintain state on if the node has bootstrapped into the cluster.
-
-    Per documented procedure for adding new units to a cluster, wait 2
-    minutes if the unit has just bootstrapped to ensure no other.
+
+
+def set_bootstrapped():
+    # We need to store this flag in two locations. The peer relation,
+    # so peers can see it, and local state, for when we haven't joined
+    # the peer relation yet. actions.publish_bootstrapped_flag()
+    # calls this method again when necessary to ensure that state is
+    # propagated # if/when the peer relation is joined.
+    config = hookenv.config()
+    config['bootstrapped'] = True
+    if coordinator.relid is not None:
+        hookenv.relation_set(coordinator.relid, bootstrapped="1")
+    if config.changed('bootstrapped'):
+        hookenv.log('Bootstrapped')
+    else:
+        hookenv.log('Already bootstrapped')
+
+
+def get_bootstrapped():
+    units = [hookenv.local_unit()]
+    if coordinator.relid is not None:
+        units.extend(hookenv.related_units(coordinator.relid))
+    return set([unit for unit in units if is_bootstrapped(unit)])
+
+
+def get_bootstrapped_ips():
+    return set([unit_to_ip(unit) for unit in get_bootstrapped()])
+
+
+def unit_to_ip(unit):
+    if unit is None or unit == hookenv.local_unit():
+        return hookenv.unit_private_ip()
+    elif coordinator.relid:
+        return hookenv.relation_get(rid=coordinator.relid,
+                                    unit=unit).get('private-address')
+    else:
+        return None
+
+
+def get_node_status():
+    '''Return the Cassandra node status.
+
+    May be NORMAL, JOINING, DECOMMISSIONED etc., or None if we can't tell.
     '''
-    if num_peers() == 0:
-        # There is no cluster (just us), so we are not bootstrapped into
-        # the cluster.
-        return
-
-    if not is_bootstrapped():
-        for _ in backoff('bootstrap to finish'):
-            if not is_cassandra_running():
-                hookenv.log('Bootstrap failed. Retrying')
-                start_cassandra()
-                break
-            if is_all_normal():
-                hookenv.log('Cluster settled after bootstrap.')
-                config = hookenv.config()
-                hookenv.log('Waiting {}s.'.format(
-                    config['post_bootstrap_delay']))
-                time.sleep(config['post_bootstrap_delay'])
-                if is_cassandra_running():
-                    set_bootstrapped(True)
-                    break
-
-
-def is_schema_agreed():
-    '''Return True if all the nodes that are up agree on a schema.'''
-    up_ips = set(up_node_ips())
-    # Always include ourself since we may be joining just now.
-    up_ips.add(hookenv.unit_private_ip())
-    raw = nodetool('describecluster')
-    # The output of nodetool describe cluster is almost yaml,
-    # so we use that tool once we fix the tabs.
-    description = yaml.load(raw.expandtabs())
-    versions = description['Cluster Information']['Schema versions'] or {}
-
-    for schema, schema_ips in versions.items():
-        schema_ips = set(schema_ips)
-        if up_ips.issubset(schema_ips):
-            hookenv.log('{!r} agree on schema'.format(up_ips), DEBUG)
-            return True
-    hookenv.log('{!r} do not agree on schema'.format(up_ips), DEBUG)
+    if not is_cassandra_running():
+        return None
+    raw = nodetool('netstats')
+    m = re.search(r'(?m)^Mode:\s+(\w+)$', raw)
+    if m is None:
+        return None
+    return m.group(1).upper()
+
+
+def is_decommissioned():
+    status = get_node_status()
+    if status in ('DECOMMISSIONED', 'LEAVING'):
+        hookenv.log('This node is {}'.format(status), WARNING)
+        return True
     return False
 
 
 @logged
-def wait_for_agreed_schema():
-    for _ in backoff('schema agreement'):
-        if is_schema_agreed():
-            return
-
-
-def peer_ips():
-    ips = set()
-    relid = rollingrestart.get_peer_relation_id()
-    if relid is not None:
-        for unit in hookenv.related_units(relid):
-            ip = hookenv.relation_get('private-address', unit, relid)
-            ips.add(ip)
-    return ips
-
-
-def is_all_normal(timeout=120):
-    '''All nodes in the cluster report status Normal.
-
-    Returns false if a node is joining, leaving or moving in the ring.
-    '''
-    is_all_normal = True
-    try:
-        raw = nodetool('status', 'system_auth', timeout=timeout)
-    except subprocess.TimeoutExpired:
-        return False
-    node_status_re = re.compile('^(.)([NLJM])\s+([\d\.]+)\s')
-    for line in raw.splitlines():
-        match = node_status_re.search(line)
-        if match is not None:
-            updown, mode, address = match.groups()
-            if updown == 'D':
-                # 'Down' is unfortunately just informative. During service
-                # teardown, nodes will disappear without decommissioning
-                # leaving these entries.
-                if address == hookenv.unit_private_ip():
-                    is_all_normal = False
-                    hookenv.log('Node {} (this node) is down'.format(address))
-                else:
-                    hookenv.log('Node {} is down'.format(address))
-            elif updown == '?':  # CASSANDRA-8791
-                is_all_normal = False
-                hookenv.log('Node {} is transitioning'.format(address))
-
-            if mode == 'L':
-                hookenv.log('Node {} is leaving the cluster'.format(address))
-                is_all_normal = False
-            elif mode == 'J':
-                hookenv.log('Node {} is joining the cluster'.format(address))
-                is_all_normal = False
-            elif mode == 'M':
-                hookenv.log('Node {} is moving ring position'.format(address))
-                is_all_normal = False
-    return is_all_normal
-
-
-@logged
-def wait_for_normality():
-    for _ in backoff('cluster operators to complete'):
-        if is_all_normal():
-            return
-
-
-def is_decommissioned():
-    if not is_cassandra_running():
-        return False  # Decommissioned nodes are not shut down.
-
-    for _ in backoff('stable node mode'):
-        raw = nodetool('netstats')
-        if 'Mode: DECOMMISSIONED' in raw:
-            hookenv.log('This node is DECOMMISSIONED', WARNING)
-            return True
-        elif 'Mode: NORMAL' in raw:
-            return False
-
-
-@logged
 def emit_describe_cluster():
     '''Run nodetool describecluster for the logs.'''
     nodetool('describecluster')  # Implicit emit
@@ -1309,3 +927,62 @@
 # FOR CHARMHELPERS. This should be a constant in nrpe.py
 def local_plugins_dir():
     return '/usr/local/lib/nagios/plugins'
+
+
+def leader_ping():
+    '''Make a change in the leader settings, waking the non-leaders.'''
+    assert hookenv.is_leader()
+    last = int(hookenv.leader_get('ping') or 0)
+    hookenv.leader_set(ping=str(last + 1))
+
+
+def get_unit_superusers():
+    '''Return the set of units that have had their superuser accounts created.
+    '''
+    raw = hookenv.leader_get('superusers')
+    return set(json.loads(raw or '[]'))
+
+
+def set_unit_superusers(superusers):
+    hookenv.leader_set(superusers=json.dumps(sorted(superusers)))
+
+
+def status_set(state, message):
+    '''Set the unit status and log a message.'''
+    hookenv.status_set(state, message)
+    hookenv.log('{} unit state: {}'.format(state, message))
+
+
+def service_status_set(state, message):
+    '''Set the service status and log a message.'''
+    subprocess.check_call(['status-set', '--service', state, message])
+    hookenv.log('{} service state: {}'.format(state, message))
+
+
+def get_service_name(relid):
+    '''Return the service name for the other end of relid.'''
+    units = hookenv.related_units(relid)
+    if units:
+        return units[0].split('/', 1)[0]
+    else:
+        return None
+
+
+def peer_relid():
+    return coordinator.relid
+
+
+@logged
+def set_active():
+    '''Set happy state'''
+    if hookenv.unit_private_ip() in get_seed_ips():
+        msg = 'Live seed'
+    else:
+        msg = 'Live node'
+    status_set('active', msg)
+
+    if hookenv.is_leader():
+        n = num_nodes()
+        if n == 1:
+            n = 'Single'
+        service_status_set('active', '{} node cluster'.format(n))

=== modified file 'hooks/hooks.py' (properties changed: +x to -x)
--- hooks/hooks.py	2015-04-28 16:20:05 +0000
+++ hooks/hooks.py	2015-06-30 07:36:26 +0000
@@ -1,5 +1,19 @@
 #!/usr/bin/python3
-
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 from charmhelpers import fetch
 from charmhelpers.core import hookenv
 
@@ -18,20 +32,21 @@
 
 
 def default_hook():
+    if not hookenv.has_juju_version('1.24'):
+        hookenv.status_set('blocked', 'Requires Juju 1.24 or higher')
+        # Error state, since we don't have 1.24 to give a nice blocked state.
+        raise SystemExit(1)
+
     # These need to be imported after bootstrap() or required Python
     # packages may not have been installed.
     import definitions
-    from loglog import loglog
 
     # Only useful for debugging, or perhaps have this enabled with a config
     # option?
-    loglog('/var/log/cassandra/system.log', prefix='C*: ')
+    ## from loglog import loglog
+    ## loglog('/var/log/cassandra/system.log', prefix='C*: ')
 
     hookenv.log('*** {} Hook Start'.format(hookenv.hook_name()))
     sm = definitions.get_service_manager()
     sm.manage()
     hookenv.log('*** {} Hook Done'.format(hookenv.hook_name()))
-
-if __name__ == '__main__':
-    bootstrap()
-    default_hook()

=== modified symlink 'hooks/install' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/install	1970-01-01 00:00:00 +0000
+++ hooks/install	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== added file 'hooks/leader-elected'
--- hooks/leader-elected	1970-01-01 00:00:00 +0000
+++ hooks/leader-elected	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== added file 'hooks/leader-settings-changed'
--- hooks/leader-settings-changed	1970-01-01 00:00:00 +0000
+++ hooks/leader-settings-changed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified symlink 'hooks/nrpe-external-master-relation-changed' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/nrpe-external-master-relation-changed	1970-01-01 00:00:00 +0000
+++ hooks/nrpe-external-master-relation-changed	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified file 'hooks/relations.py'
--- hooks/relations.py	2015-02-02 15:52:50 +0000
+++ hooks/relations.py	2015-06-30 07:36:26 +0000
@@ -22,8 +22,22 @@
 from charmhelpers.core.hookenv import log, WARNING
 from charmhelpers.core.services.helpers import RelationContext
 
-
-# FOR CHARMHELPERS
+from coordinator import coordinator
+
+
+class PeerRelation(RelationContext):
+    interface = 'cassandra-cluster'
+    name = 'cluster'
+
+    def is_ready(self):
+        # All units except the leader need to wait until the peer
+        # relation is available.
+        if coordinator.relid is not None or hookenv.is_leader():
+            return True
+        return False
+
+
+# FOR CHARMHELPERS (if we can integrate Juju 1.24 storage too)
 class StorageRelation(RelationContext):
     '''Wait for the block storage mount to become available.
 
@@ -80,7 +94,9 @@
                 return False
         return True
 
-    def provide_data(self):
+    def provide_data(self, remote_service, service_ready):
+        hookenv.log('Requesting mountpoint {} from {}'
+                    .format(self._requested_mountpoint, remote_service))
         return dict(mountpoint=self._requested_mountpoint)
 
     def needs_remount(self):

=== removed file 'hooks/rollingrestart.py'
--- hooks/rollingrestart.py	2015-02-14 04:58:55 +0000
+++ hooks/rollingrestart.py	1970-01-01 00:00:00 +0000
@@ -1,257 +0,0 @@
-# Copyright 2015 Canonical Ltd.
-#
-# This file is part of the Cassandra Charm for Juju.
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License version 3, as
-# published by the Free Software Foundation.
-#
-# This program is distributed in the hope that it will be useful, but
-# WITHOUT ANY WARRANTY; without even the implied warranties of
-# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
-# PURPOSE.  See the GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-from datetime import datetime
-import os.path
-
-from charmhelpers.contrib import peerstorage
-from charmhelpers.core import hookenv, host
-from charmhelpers.core.hookenv import DEBUG
-
-
-def request_restart():
-    '''Request a rolling restart.'''
-    # We store the local restart request as a flag on the filesystem,
-    # for rolling_restart() to deal with later. This allows
-    # rolling_restart to know that this is a new entry to be queued for
-    # a future hook to handle, or an existing entry in the queue that
-    # may be restarted this hook.
-    flag = os.path.join(hookenv.charm_dir(), '.needs-restart')
-    if os.path.exists(flag):
-        return
-    host.write_file(flag, utcnow_str().encode('US-ASCII'))
-
-
-def is_waiting_for_restart():
-    '''Is there an outstanding rolling restart request.'''
-    flag = os.path.join(hookenv.charm_dir(), '.needs-restart')
-    return os.path.exists(flag)
-
-
-def cancel_restart():
-    '''Cancel the rolling restart request, if any, and dequeue.'''
-    _enqueue(False)
-    flag = os.path.join(hookenv.charm_dir(), '.needs-restart')
-    if os.path.exists(flag):
-        os.remove(flag)
-
-
-class DeferRestart(Exception):
-    '''Defer the restart sequence until a future hook.
-
-    If a restart callback raises this exception, no further callbacks
-    are made and the restart request is moved to the end of the queue.
-    The restart will be triggered again in a future hook.
-    '''
-
-
-def get_restart_queue():
-    '''The sorted list of units waiting for a rolling restart.
-
-    The local unit will not be included until after rolling_restart
-    has been called.
-    '''
-    relid = get_peer_relation_id()
-    if relid is None:
-        return []
-
-    all_units = set(get_peers())
-    all_units.add(hookenv.local_unit())
-
-    queue = []
-
-    # Iterate over all units, retrieving restartrequest flags.
-    # We can't use the peerstorage helpers here, as that will only
-    # retrieve data that has been echoed and may be out of date if
-    # the necessary relation-changed hooks have not yet been invoked.
-    for unit in all_units:
-        relinfo = hookenv.relation_get(unit=unit, rid=relid)
-        request = relinfo.get('rollingrestart_{}'.format(unit))
-        if request is not None:
-            queue.append((request, unit))
-    queue.sort()
-    return [unit for _, unit in queue]
-
-
-def _enqueue(flag):
-    '''Add or remove an entry from peerstorage queue.'''
-    relid = get_peer_relation_id()
-    if not relid:
-        # No peer relation, no problems. If there is no peer relation,
-        # there are no peers to coordinate with.
-        return
-
-    unit = hookenv.local_unit()
-    queue = get_restart_queue()
-    if flag and unit in queue:
-        return  # Already queued.
-
-    key = _peerstorage_key()
-    value = utcnow_str() if flag else None
-    hookenv.log('Restart queue {} = {}'.format(key, value))
-    hookenv.relation_set(relid, {key: value})
-
-
-def rolling_restart(restart_hooks, prerequisites=()):
-    '''To ensure availability, only restart one unit at a time.
-
-    Returns:
-        True if the queued restart has occurred and restart_hook called.
-        True if no restart request has been queued.
-        False if we are still waiting on a queued restart.
-
-    restart_hooks is a list of callables that takes no parameters.
-
-    For best results, call rolling_restart() unconditionally at the
-    beginning and end of every hook.
-
-    At a minimum, rolling_restart() must be called every peer
-    relation-changed hook, and after any non-peer hook calls
-    request_restart(), or the system will deadlock.
-    '''
-    _peer_echo()  # Always call this, or peerstorage will fail.
-
-    if not is_waiting_for_restart():
-        cancel_restart()  # Clear any errant queue entries.
-        return True
-
-    for prerequisite in prerequisites:
-        if not prerequisite():
-            hookenv.log('Prerequisite {} not met. Restart deferred.')
-            _enqueue(False)
-            _enqueue(True)  # Rejoin at the end of the queue.
-            return False
-
-    def _restart():
-        for restart_hook in restart_hooks:
-            restart_hook()
-        hookenv.log('Restart complete.')
-        cancel_restart()
-        hookenv.log('Restart queue entries purged.')
-        return True
-
-    try:
-        # If there are no peers, restart the service now since there is
-        # nobody to coordinate with.
-        if len(get_peers()) == 0:
-            hookenv.log('Restart request with no peers. Restarting.')
-            return _restart()
-
-        local_unit = hookenv.local_unit()
-        queue = get_restart_queue()
-
-        # If we are not in the restart queue, join it and restart later.
-        # If there are peers, we cannot restart in the same hook we made
-        # the request or we will race with other units trying to restart.
-        if local_unit not in queue:
-            hookenv.log('Joining rolling restart queue')
-            _enqueue(True)
-            return False
-
-        # We joined the restart queue in a previous hook, but we still have
-        # to wait until a peer relation-changed hook before we can do a
-        # restart. If we did the restart in any old hook, we might not see
-        # higher priority requests from peers.
-        peer_relname = get_peer_relation_name()
-        changed_hook = '{}-relation-changed'.format(peer_relname)
-        if local_unit == queue[0] and hookenv.hook_name() == changed_hook:
-            hookenv.log('First in rolling restart queue. Restarting.')
-            return _restart()
-
-        queue_str = ', '.join(queue)
-        hookenv.log('Waiting in rolling restart queue ({})'.format(queue_str))
-        return False
-    except DeferRestart:
-        _enqueue(False)
-        _enqueue(True)  # Rejoin at the end of the queue.
-        return False
-
-
-def _peerstorage_key():
-    return 'rollingrestart_{}'.format(hookenv.local_unit())
-
-
-def _peer_echo():
-    peer_relname = get_peer_relation_name()
-    changed_hook = '{}-relation-changed'.format(peer_relname)
-
-    # If we are in the peer relation-changed hook, we must invoke
-    # peerstorage.peer_echo() or peerstorage will fail.
-    if hookenv.hook_name() == changed_hook:
-        # We only want to echo the restart entry for the remote unit, as
-        # any other information it has for other peers may be stale.
-        includes = ['rollingrestart_{}'.format(hookenv.remote_unit())]
-        hookenv.log('peerstorage.peer_echo for rolling restart.')
-        hookenv.log('peer_echo(includes={!r})'.format(includes), DEBUG)
-        # This method only works from the peer relation-changed hook.
-        peerstorage.peer_echo(includes)
-
-    # If a peer is departing, clean out any rubbish it left in the
-    # peerstorage.
-    departed_hook = '{}-relation-departed'.format(peer_relname)
-    if hookenv.hook_name() == departed_hook:
-        key = 'rollingrestart_{}'.format(hookenv.remote_unit())
-        peerstorage.peer_store(key, None, peer_relname)
-
-
-def get_peer_relation_name():
-    # Make this the charmhelpers.contrib.peerstorage default?
-    # Although it is normal to have a single peer relation, it is
-    # possible to have many. We use the first in alphabetical order.
-    md = hookenv.metadata()
-    assert 'peers' in md, 'No peer relations in metadata.yaml'
-    return sorted(md['peers'].keys())[0]
-
-
-def get_peer_relation_id():
-    for relid in hookenv.relation_ids(get_peer_relation_name()):
-        return relid
-    return None
-
-
-def get_peers():
-    """Return the sorted list of peer units.
-
-    Peers only. Does not include the local unit.
-    """
-    for relid in hookenv.relation_ids(get_peer_relation_name()):
-        return sorted(hookenv.related_units(relid),
-                      key=lambda x: int(x.split('/')[-1]))
-    return []
-
-
-def utcnow():
-    return datetime.utcnow()
-
-
-def utcnow_str():
-    return utcnow().strftime('%Y-%m-%d %H:%M:%S.%fZ')
-
-
-def make_service(restart_hooks, prerequisites=()):
-    """Create a service dictionary for rolling restart.
-
-    This needs to be a separate service, rather than a data_ready or
-    provided_data item on an existing service, because the rolling_restart()
-    function must be called for all hooks to avoid deadlocking the system.
-    It would only be safe to list rolling_restart() as a data_ready item in
-    a service with no requirements.
-    """
-    def data_ready(servicename):
-        rolling_restart(restart_hooks, prerequisites)
-
-    return dict(service='rollingrestart', data_ready=data_ready,
-                stop=[], start=[])

=== modified symlink 'hooks/stop' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/stop	1970-01-01 00:00:00 +0000
+++ hooks/stop	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified symlink 'hooks/upgrade-charm' (properties changed: -x to +x)
=== target was u'hooks.py'
--- hooks/upgrade-charm	1970-01-01 00:00:00 +0000
+++ hooks/upgrade-charm	2015-06-30 07:36:26 +0000
@@ -0,0 +1,20 @@
+#!/usr/bin/python3
+# Copyright 2015 Canonical Ltd.
+#
+# This file is part of the Cassandra Charm for Juju.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3, as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import hooks
+if __name__ == '__main__':
+    hooks.bootstrap()
+    hooks.default_hook()

=== modified file 'templates/cassandra_maintenance_cron.tmpl'
--- templates/cassandra_maintenance_cron.tmpl	2015-01-23 19:04:15 +0000
+++ templates/cassandra_maintenance_cron.tmpl	2015-06-30 07:36:26 +0000
@@ -1,4 +1,4 @@
 # Cassandra maintenance
 # Staggered weekly repairs
 # m h  dom mon dow user  command
-{{minute}} {{hour}} * * {{dow}} cassandra run-one-until-success nodetool repair -pr && run-one-until-success nodetool cleanup
+{{minute}} {{hour}} * * {{dow}} cassandra run-one-until-success nodetool repair -pr

=== modified file 'tests/test_actions.py'
--- tests/test_actions.py	2015-04-10 14:42:53 +0000
+++ tests/test_actions.py	2015-06-30 07:36:26 +0000
@@ -17,7 +17,6 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import errno
-import functools
 from itertools import repeat
 import os.path
 import re
@@ -29,17 +28,16 @@
 from unittest.mock import ANY, call, patch, sentinel
 import yaml
 
+import cassandra
 from charmhelpers.core import hookenv
 
 from tests.base import TestCaseBase
 import actions
+from coordinator import coordinator
 import helpers
 
 
-patch = functools.partial(patch, autospec=True)  # autospec=True as default.
-
-
-class TestsActions(TestCaseBase):
+class TestActions(TestCaseBase):
     def test_action_wrapper(self):
         @actions.action
         def somefunc(*args, **kw):
@@ -61,40 +59,38 @@
             '** Action catch-fire/somefunc (foo)')
 
     def test_revert_unchangeable_config(self):
-        hookenv.hook_name.return_value = 'config-changed'
         config = hookenv.config()
 
         self.assertIn('datacenter', actions.UNCHANGEABLE_KEYS)
 
+        # In the first hook, revert does nothing as there is nothing to
+        # revert too.
         config['datacenter'] = 'mission_control'
+        self.assertTrue(config.changed('datacenter'))
+        actions.revert_unchangeable_config('')
+        self.assertEqual(config['datacenter'], 'mission_control')
+
         config.save()
         config.load_previous()
         config['datacenter'] = 'orbital_1'
 
-        self.assertTrue(config.changed('datacenter'))
-
         actions.revert_unchangeable_config('')
         self.assertEqual(config['datacenter'], 'mission_control')  # Reverted
 
-        hookenv.log.assert_any_call(ANY, hookenv.ERROR)
-
-    def test_revert_unchangeable_config_install(self):
-        hookenv.hook_name.return_value = 'install'
-        config = hookenv.config()
-
-        self.assertIn('datacenter', actions.UNCHANGEABLE_KEYS)
-
-        config['datacenter'] = 'mission_control'
-        config.save()
-        config.load_previous()
-        config['datacenter'] = 'orbital_1'
-
-        self.assertTrue(config.changed('datacenter'))
-
-        # In the install hook, revert_unchangeable_config() does
-        # nothing.
-        actions.revert_unchangeable_config('')
-        self.assertEqual(config['datacenter'], 'orbital_1')
+        hookenv.log.assert_any_call(ANY, hookenv.ERROR)  # Logged the problem.
+
+    @patch('charmhelpers.core.hookenv.is_leader')
+    def test_leader_only(self, is_leader):
+
+        @actions.leader_only
+        def f(*args, **kw):
+            return args, kw
+
+        is_leader.return_value = False
+        self.assertIsNone(f(1, foo='bar'))
+
+        is_leader.return_value = True
+        self.assertEqual(f(1, foo='bar'), ((1,), dict(foo='bar')))
 
     def test_set_proxy(self):
         # NB. Environment is already mocked.
@@ -315,6 +311,123 @@
         # handles this.
         self.assertFalse(check_call.called)
 
+    @patch('actions._install_oracle_jre_tarball')
+    @patch('actions._fetch_oracle_jre')
+    def test_install_oracle_jre(self, fetch, install_tarball):
+        fetch.return_value = sentinel.tarball
+
+        actions.install_oracle_jre('')
+        self.assertFalse(fetch.called)
+        self.assertFalse(install_tarball.called)
+
+        hookenv.config()['jre'] = 'oracle'
+        actions.install_oracle_jre('')
+        fetch.assert_called_once_with()
+        install_tarball.assert_called_once_with(sentinel.tarball)
+
+    @patch('helpers.status_set')
+    @patch('urllib.request')
+    def test_fetch_oracle_jre(self, req, status_set):
+        config = hookenv.config()
+        url = 'https://foo.example.com/server-jre-7u42-linux-x64.tar.gz'
+        expected_tarball = os.path.join(hookenv.charm_dir(), 'lib',
+                                        'server-jre-7u42-linux-x64.tar.gz')
+        config['private_jre_url'] = url
+
+        # Create a dummy tarball, since the mock urlretrieve won't.
+        os.makedirs(os.path.dirname(expected_tarball))
+        with open(expected_tarball, 'w'):
+            pass  # Empty file
+
+        self.assertEqual(actions._fetch_oracle_jre(), expected_tarball)
+        req.urlretrieve.assert_called_once_with(url, expected_tarball)
+
+    def test_fetch_oracle_jre_local(self):
+        # Create an existing tarball. If it is found, it will be used
+        # without needing to specify a remote url or actually download
+        # anything.
+        expected_tarball = os.path.join(hookenv.charm_dir(), 'lib',
+                                        'server-jre-7u42-linux-x64.tar.gz')
+        os.makedirs(os.path.dirname(expected_tarball))
+        with open(expected_tarball, 'w'):
+            pass  # Empty file
+
+        self.assertEqual(actions._fetch_oracle_jre(), expected_tarball)
+
+    @patch('helpers.status_set')
+    def test_fetch_oracle_jre_notfound(self, status_set):
+        with self.assertRaises(SystemExit) as x:
+            actions._fetch_oracle_jre()
+            self.assertEqual(x.code, 0)
+            status_set.assert_called_once_with('blocked', ANY)
+
+    @patch('subprocess.check_call')
+    @patch('charmhelpers.core.host.mkdir')
+    @patch('os.path.isdir')
+    def test_install_oracle_jre_tarball(self, isdir, mkdir, check_call):
+        isdir.return_value = False
+
+        dest = '/usr/lib/jvm/java-8-oracle'
+
+        actions._install_oracle_jre_tarball(sentinel.tarball)
+        mkdir.assert_called_once_with(dest)
+        check_call.assert_has_calls([
+            call(['tar', '-xz', '-C', dest,
+                  '--strip-components=1', '-f', sentinel.tarball]),
+            call(['update-alternatives', '--install',
+                  '/usr/bin/java', 'java',
+                  os.path.join(dest, 'bin', 'java'), '1']),
+            call(['update-alternatives', '--set', 'java',
+                  os.path.join(dest, 'bin', 'java')]),
+            call(['update-alternatives', '--install',
+                  '/usr/bin/javac', 'javac',
+                  os.path.join(dest, 'bin', 'javac'), '1']),
+            call(['update-alternatives', '--set', 'javac',
+                  os.path.join(dest, 'bin', 'javac')])])
+
+    @patch('os.path.exists')
+    @patch('subprocess.check_call')
+    @patch('charmhelpers.core.host.mkdir')
+    @patch('os.path.isdir')
+    def test_install_oracle_jre_tarball_already(self, isdir,
+                                                mkdir, check_call, exists):
+        isdir.return_value = True
+        exists.return_value = True  # jre already installed
+
+        # Store the version previously installed.
+        hookenv.config()['oracle_jre_tarball'] = sentinel.tarball
+
+        dest = '/usr/lib/jvm/java-8-oracle'
+
+        actions._install_oracle_jre_tarball(sentinel.tarball)
+
+        self.assertFalse(mkdir.called)  # The jvm dir already existed.
+
+        exists.assert_called_once_with('/usr/lib/jvm/java-8-oracle/bin/java')
+
+        # update-alternatives done, but tarball not extracted.
+        check_call.assert_has_calls([
+            call(['update-alternatives', '--install',
+                  '/usr/bin/java', 'java',
+                  os.path.join(dest, 'bin', 'java'), '1']),
+            call(['update-alternatives', '--set', 'java',
+                  os.path.join(dest, 'bin', 'java')]),
+            call(['update-alternatives', '--install',
+                  '/usr/bin/javac', 'javac',
+                  os.path.join(dest, 'bin', 'javac'), '1']),
+            call(['update-alternatives', '--set', 'javac',
+                  os.path.join(dest, 'bin', 'javac')])])
+
+    @patch('subprocess.check_output')
+    def test_emit_java_version(self, check_output):
+        check_output.return_value = 'Line 1\nLine 2'
+        actions.emit_java_version('')
+        check_output.assert_called_once_with(['java', '-version'],
+                                             universal_newlines=True)
+        hookenv.log.assert_has_calls([call(ANY),
+                                      call('JRE: Line 1'),
+                                      call('JRE: Line 2')])
+
     @patch('helpers.configure_cassandra_yaml')
     def test_configure_cassandra_yaml(self, configure_cassandra_yaml):
         # actions.configure_cassandra_yaml is just a wrapper around the
@@ -399,10 +512,62 @@
                 self.assertEqual(f.read().strip(),
                                  'dc=test_dc\nrack=test_rack')
 
-    @patch('helpers.reset_auth_keyspace_replication')
-    def test_reset_auth_keyspace_replication(self, helpers_reset):
+    @patch('helpers.connect')
+    @patch('helpers.get_auth_keyspace_replication')
+    @patch('helpers.num_nodes')
+    def test_needs_reset_auth_keyspace_replication(self, num_nodes,
+                                                   get_auth_ks_rep,
+                                                   connect):
+        num_nodes.return_value = 4
+        connect().__enter__.return_value = sentinel.session
+        connect().__exit__.return_value = False
+        get_auth_ks_rep.return_value = {'another': '8'}
+        self.assertTrue(actions.needs_reset_auth_keyspace_replication())
+
+    @patch('helpers.connect')
+    @patch('helpers.get_auth_keyspace_replication')
+    @patch('helpers.num_nodes')
+    def test_needs_reset_auth_keyspace_replication_false(self, num_nodes,
+                                                         get_auth_ks_rep,
+                                                         connect):
+        config = hookenv.config()
+        config['datacenter'] = 'mydc'
+        connect().__enter__.return_value = sentinel.session
+        connect().__exit__.return_value = False
+
+        num_nodes.return_value = 4
+        get_auth_ks_rep.return_value = {'another': '8',
+                                        'mydc': '3'}
+        self.assertFalse(actions.needs_reset_auth_keyspace_replication())
+
+    @patch('helpers.set_active')
+    @patch('helpers.repair_auth_keyspace')
+    @patch('helpers.connect')
+    @patch('helpers.set_auth_keyspace_replication')
+    @patch('helpers.get_auth_keyspace_replication')
+    @patch('helpers.num_nodes')
+    @patch('charmhelpers.core.hookenv.is_leader')
+    def test_reset_auth_keyspace_replication(self, is_leader, num_nodes,
+                                             get_auth_ks_rep,
+                                             set_auth_ks_rep,
+                                             connect, repair, set_active):
+        is_leader.return_value = True
+        num_nodes.return_value = 4
+        coordinator.grants = {}
+        coordinator.requests = {hookenv.local_unit(): {}}
+        coordinator.grant('repair', hookenv.local_unit())
+        config = hookenv.config()
+        config['datacenter'] = 'mydc'
+        connect().__enter__.return_value = sentinel.session
+        connect().__exit__.return_value = False
+        get_auth_ks_rep.return_value = {'another': '8'}
+        self.assertTrue(actions.needs_reset_auth_keyspace_replication())
         actions.reset_auth_keyspace_replication('')
-        helpers_reset.assert_called_once_with()
+        set_auth_ks_rep.assert_called_once_with(
+            sentinel.session,
+            {'class': 'NetworkTopologyStrategy', 'another': '8', 'mydc': 3})
+        repair.assert_called_once_with()
+        set_active.assert_called_once_with()
 
     def test_store_unit_private_ip(self):
         hookenv.unit_private_ip.side_effect = None
@@ -410,181 +575,92 @@
         actions.store_unit_private_ip('')
         self.assertEqual(hookenv.config()['unit_private_ip'], sentinel.ip)
 
-    @patch('helpers.set_bootstrapped')
-    @patch('helpers.unit_number')
-    def test_set_unit_zero_bootstrapped(self, unit_number, set_bootstrapped):
-        unit_number.return_value = 0
-        hookenv.hook_name.return_value = 'cluster-whatever'
-        actions.set_unit_zero_bootstrapped('')
-        set_bootstrapped.assert_called_once_with(True)
-        set_bootstrapped.reset_mock()
-
-        # Noop if unit number is not 0.
-        unit_number.return_value = 2
-        actions.set_unit_zero_bootstrapped('')
-        self.assertFalse(set_bootstrapped.called)
-
-        # Noop if called from a non peer relation hook. Unfortunatly,
-        # attempting to set attributes on the peer relation fails before
-        # the peer relation has been joined.
-        unit_number.return_value = 0
-        hookenv.hook_name.return_value = 'install'
-        actions.set_unit_zero_bootstrapped('')
-        self.assertFalse(set_bootstrapped.called)
-
+    @patch('charmhelpers.core.host.service_start')
+    @patch('helpers.status_set')
+    @patch('helpers.actual_seed_ips')
+    @patch('helpers.get_seed_ips')
+    @patch('relations.StorageRelation.needs_remount')
+    @patch('helpers.is_bootstrapped')
     @patch('helpers.is_cassandra_running')
     @patch('helpers.is_decommissioned')
-    @patch('rollingrestart.request_restart')
-    def test_maybe_schedule_restart_decommissioned(self, request_restart,
-                                                   is_running, is_decom):
+    def test_needs_restart(self, is_decom, is_running, is_bootstrapped,
+                           needs_remount, seed_ips, actual_seeds,
+                           status_set, service_start):
+        is_decom.return_value = False
+        is_running.return_value = True
+        needs_remount.return_value = False
+        seed_ips.return_value = set(['1.2.3.4'])
+        actual_seeds.return_value = set(['1.2.3.4'])
+
+        config = hookenv.config()
+        config['configured_seeds'] = list(sorted(seed_ips()))
+        config.save()
+        config.load_previous()  # Ensure everything flagged as unchanged.
+
+        self.assertFalse(actions.needs_restart())
+
+        # Decommissioned nodes are not restarted.
         is_decom.return_value = True
-        is_running.return_value = True
-        actions.maybe_schedule_restart('')
-        self.assertFalse(request_restart.called)
-
-    @patch('helpers.is_decommissioned')
-    @patch('helpers.seed_ips')
-    @patch('helpers.is_cassandra_running')
-    @patch('relations.StorageRelation')
-    @patch('rollingrestart.request_restart')
-    def test_maybe_schedule_restart_need_remount(self, request_restart,
-                                                 storage_relation, is_running,
-                                                 seed_ips, is_decom):
-        config = hookenv.config()
-
-        is_decom.return_value = False
-        is_running.return_value = True
-
-        # Storage says we need to restart.
-        storage_relation().needs_remount.return_value = True
-
-        # No new seeds
-        seed_ips.return_value = set(['a'])
-        config['configured_seeds'] = sorted(seed_ips())
-        config.save()
-        config.load_previous()
-
-        # IP address is unchanged.
-        config['unit_private_ip'] = hookenv.unit_private_ip()
-
-        # Config items are unchanged.
-        config.save()
-        config.load_previous()
-
-        actions.maybe_schedule_restart('')
-        request_restart.assert_called_once_with()
-        hookenv.log.assert_any_call('Mountpoint changed. '
-                                    'Restart and migration required.')
-
-    @patch('helpers.is_decommissioned')
-    @patch('helpers.seed_ips')
-    @patch('helpers.is_cassandra_running')
-    @patch('relations.StorageRelation')
-    @patch('rollingrestart.request_restart')
-    def test_maybe_schedule_restart_unchanged(self, request_restart,
-                                              storage_relation, is_running,
-                                              seed_ips, is_decom):
-        config = hookenv.config()
-        config['configured_seeds'] = ['a']
-        config.save()
-        config.load_previous()
-
-        is_decom.return_value = False
-        is_running.return_value = True
-
-        # Storage says we do not need to restart.
-        storage_relation().needs_remount.return_value = False
-
-        # No new seeds
-        seed_ips.return_value = set(['a'])
-        config['configured_seeds'] = sorted(seed_ips())
-        config.save()
-        config.load_previous()
-
-        # IP address is unchanged.
-        config['unit_private_ip'] = hookenv.unit_private_ip()
-
-        # Config items are unchanged, except for ones that do not
-        # matter.
-        config.save()
-        config.load_previous()
-        config['package_status'] = 'new'
-        self.assertTrue(config.changed('package_status'))
-
-        actions.maybe_schedule_restart('')
-        self.assertFalse(request_restart.called)
-
-    @patch('helpers.is_decommissioned')
-    @patch('helpers.is_cassandra_running')
-    @patch('helpers.seed_ips')
-    @patch('relations.StorageRelation')
-    @patch('rollingrestart.request_restart')
-    def test_maybe_schedule_restart_config_changed(self, request_restart,
-                                                   storage_relation,
-                                                   seed_ips, is_running,
-                                                   is_decom):
-        config = hookenv.config()
-
-        is_decom.return_value = False
-        is_running.return_value = True
-
-        # Storage says we do not need to restart.
-        storage_relation().needs_remount.return_value = False
-
-        # IP address is unchanged.
-        config['unit_private_ip'] = hookenv.unit_private_ip()
-
-        # Config items are changed.
-        config.save()
-        config.load_previous()
-        config['package_status'] = 'new'
-        self.assertTrue(config.changed('package_status'))  # Doesn't matter.
-        config['max_heap_size'] = 'lots and lots'
-        self.assertTrue(config.changed('max_heap_size'))  # Requires restart.
-
-        actions.maybe_schedule_restart('')
-        request_restart.assert_called_once_with()
-        hookenv.log.assert_any_call('max_heap_size changed. Restart required.')
-
-    @patch('helpers.is_decommissioned')
-    @patch('helpers.seed_ips')
-    @patch('helpers.is_cassandra_running')
-    @patch('relations.StorageRelation')
-    @patch('rollingrestart.request_restart')
-    def test_maybe_schedule_restart_ip_changed(self, request_restart,
-                                               storage_relation, is_running,
-                                               seed_ips, is_decom):
-        is_decom.return_value = False
-        is_running.return_value = True
-
-        # Storage says we do not need to restart.
-        storage_relation().needs_remount.return_value = False
-
-        # No new seeds
-        seed_ips.return_value = set(['a'])
-        config = hookenv.config()
-        config['configured_seeds'] = sorted(seed_ips())
-        config.save()
-        config.load_previous()
-
-        # Config items are unchanged.
-        config = hookenv.config()
-        config.save()
-        config.load_previous()
-
-        actions.store_unit_private_ip('')  # IP address change
-
-        actions.maybe_schedule_restart('')
-        request_restart.assert_called_once_with()
-        hookenv.log.assert_any_call('Unit IP address changed. '
-                                    'Restart required.')
-
-    @patch('helpers.is_cassandra_running')
-    @patch('rollingrestart.request_restart')
-    def test_maybe_schedule_restart_down(self, request_restart, is_running):
+        self.assertFalse(actions.needs_restart())
+        is_decom.return_value = False
+        self.assertFalse(actions.needs_restart())
+
+        # Nodes not running need to be restarted.
         is_running.return_value = False
-        actions.maybe_schedule_restart('')
-        request_restart.assert_called_once_with()
+        self.assertTrue(actions.needs_restart())
+        is_running.return_value = True
+        self.assertFalse(actions.needs_restart())
+
+        # If we have a new mountpoint, we need to restart in order to
+        # migrate data.
+        needs_remount.return_value = True
+        self.assertTrue(actions.needs_restart())
+        needs_remount.return_value = False
+        self.assertFalse(actions.needs_restart())
+
+        # Certain changed config items trigger a restart.
+        config['max_heap_size'] = '512M'
+        self.assertTrue(actions.needs_restart())
+        config.save()
+        config.load_previous()
+        self.assertFalse(actions.needs_restart())
+
+        # A new IP address requires a restart.
+        config['unit_private_ip'] = 'new'
+        self.assertTrue(actions.needs_restart())
+        config.save()
+        config.load_previous()
+        self.assertFalse(actions.needs_restart())
+
+        # If the seeds have changed, we need to restart.
+        seed_ips.return_value = set(['9.8.7.6'])
+        actual_seeds.return_value = set(['9.8.7.6'])
+        self.assertTrue(actions.needs_restart())
+        is_running.side_effect = iter([False, True])
+        helpers.start_cassandra()
+        is_running.side_effect = None
+        is_running.return_value = True
+        self.assertFalse(actions.needs_restart())
+
+    @patch('charmhelpers.core.hookenv.is_leader')
+    @patch('helpers.is_bootstrapped')
+    @patch('helpers.ensure_database_directories')
+    @patch('helpers.remount_cassandra')
+    @patch('helpers.start_cassandra')
+    @patch('helpers.stop_cassandra')
+    @patch('helpers.status_set')
+    def test_maybe_restart(self, status_set, stop_cassandra, start_cassandra,
+                           remount, ensure_directories, is_bootstrapped,
+                           is_leader):
+        coordinator.grants = {}
+        coordinator.requests = {hookenv.local_unit(): {}}
+        coordinator.relid = 'cluster:1'
+        coordinator.grant('restart', hookenv.local_unit())
+        actions.maybe_restart('')
+        stop_cassandra.assert_called_once_with()
+        remount.assert_called_once_with()
+        ensure_directories.assert_called_once_with()
+        start_cassandra.assert_called_once_with()
 
     @patch('helpers.stop_cassandra')
     def test_stop_cassandra(self, helpers_stop_cassandra):
@@ -596,15 +672,29 @@
         actions.start_cassandra('ignored')
         helpers_start_cassandra.assert_called_once_with()
 
-    @patch('helpers.ensure_unit_superuser')
-    def test_ensure_unit_superuser(self, helpers_ensure_unit_superuser):
-        actions.ensure_unit_superuser('ignored')
-        helpers_ensure_unit_superuser.assert_called_once_with()
+    @patch('os.path.isdir')
+    @patch('helpers.get_all_database_directories')
+    @patch('helpers.set_io_scheduler')
+    def test_reset_all_io_schedulers(self, set_io_scheduler, dbdirs, isdir):
+        hookenv.config()['io_scheduler'] = sentinel.io_scheduler
+        dbdirs.return_value = dict(
+            data_file_directories=[sentinel.d1, sentinel.d2],
+            commitlog_directory=sentinel.cl,
+            saved_caches_directory=sentinel.sc)
+        isdir.return_value = True
+        actions.reset_all_io_schedulers('')
+        set_io_scheduler.assert_has_calls([
+            call(sentinel.io_scheduler, sentinel.d1),
+            call(sentinel.io_scheduler, sentinel.d2),
+            call(sentinel.io_scheduler, sentinel.cl),
+            call(sentinel.io_scheduler, sentinel.sc)],
+            any_order=True)