Coordinate package upgrades across cluster

Use the coordination module to manage package upgrades across the
cluster. To each this some of the setup was moved into a new
configure_rabbit_install method which handles setup is normally
run after an upgrade.

Change-Id: I8d244d96c83a5da164322faff873a72530ec9def

hooks/rabbit_utils.py

@@ -133,7 +133,8 @@ CRONJOB_CMD = ("{schedule} root timeout -k 10s -s SIGINT {timeout} "
                "{command} 2>&1 | logger -p local0.notice\n")
 
 COORD_KEY_RESTART = "restart"
-COORD_KEYS = [COORD_KEY_RESTART]
+COORD_KEY_PKG_UPGRADE = "pkg_upgrade"
+COORD_KEYS = [COORD_KEY_RESTART, COORD_KEY_PKG_UPGRADE]
 
 _named_passwd = '/var/lib/charm/{}/{}.passwd'
 _local_named_passwd = '/var/lib/charm/{}/{}.local_passwd'

hooks/rabbitmq_server_relations.py

@@ -186,8 +186,20 @@ def manage_restart(coordinate_restart=True):
         log("Restart not run")
 
 
+def coordinated_upgrade():
+    """Upgrade packages if lock is granted."""
+    serial = coordinator.Serial()
+    if serial.granted(rabbit.COORD_KEY_PKG_UPGRADE):
+        log("Package upgrade lock granted")
+        rabbit.install_or_upgrade_packages()
+        configure_rabbit_install()
+    else:
+        log("Package upgrade lock not granted")
+
+
 def check_coordinated_functions():
     """Run any functions that require coordination locks."""
+    coordinated_upgrade()
     manage_restart()
 
 
@@ -827,12 +839,34 @@ def config_changed(check_deferred_restarts=True):
     # result in an upgrade if applicable only if we change the 'source'
     # config option
     if rabbit.archive_upgrade_available():
-        # Avoid packge upgrade collissions
-        # Stopping and attempting to start rabbitmqs at the same time leads to
-        # failed restarts
-        rabbit.cluster_wait()
-        rabbit.install_or_upgrade_packages()
+        if rabbit.in_run_deferred_hooks_action():
+            log("In deferred hook action, package lock not needed")
+            rabbit.install_or_upgrade_packages()
+            configure_rabbit_install()
+        else:
+            serial = coordinator.Serial()
+            log("Requesting package lock")
+            serial.acquire(rabbit.COORD_KEY_PKG_UPGRADE)
+            # If lock is granted immediatly then upgrade will happen
+            # here, otherwise it will happen in a subsequent hook
+            # managed by the coodinator module.
+            coordinated_upgrade()
+    else:
+        configure_rabbit_install()
+        # Update cluster in case min-cluster-size has changed
+        for rid in relation_ids('cluster'):
+            for unit in related_units(rid):
+                cluster_changed(relation_id=rid, remote_unit=unit)
+        check_coordinated_functions()
 
+
+@rabbit.coordinated_restart_on_change(rabbit.restart_map(),
+                                      manage_restart)
+def configure_rabbit_install():
+    """Configure the rabbit installation environment.
+
+    NOTE: For setting up access for a client (vhosts etc) use `configure_amqp`
+    """
     if config('ssl') == 'off':
         open_port(5672)
         close_port(int(config('ssl_port')))

unit_tests/test_rabbit_utils.py

@@ -724,7 +724,8 @@ class UtilsTests(CharmTestCase):
             assess_cluster_status, status_set, clustered,
             get_deferred_restarts, get_deferred_hooks, is_cron_schedule_valid):
         self.coordinator.Serial().requested.side_effect = lambda x: {
-            'restart': True}[x]
+            'restart': True,
+            'pkg_upgrade': False}[x]
         is_cron_schedule_valid.return_value = True
         get_deferred_hooks.return_value = []
         get_deferred_restarts.return_value = []
@@ -740,6 +741,15 @@ class UtilsTests(CharmTestCase):
             'waiting',
             'Unit is ready and clustered. Waiting for restart lock(s)')
 
+        status_set.reset_mock()
+        self.coordinator.Serial().requested.side_effect = lambda x: {
+            'restart': False,
+            'pkg_upgrade': True}[x]
+        rabbit_utils.assess_status_func('test-config')()
+        status_set.assert_called_once_with(
+            'waiting',
+            'Unit is ready and clustered. Waiting for pkg_upgrade lock(s)')
+
     def test_pause_unit_helper(self):
         with mock.patch.object(rabbit_utils, '_pause_resume_helper') as prh:
             rabbit_utils.pause_unit_helper('random-config')