Merge lp:~stub/charms/precise/postgresql/fix-races into lp:charms/postgresql

[Stuart Bishop]

The new local provider is quite fast, and exposes race conditions in the PostgreSQL charm.

This branch reworks the replication peer relationship so the test suite runs more reliably. For example, election has been rewritten to cope with situations such as, when creating a new service of 3 units, units 1 and 2 may have already elected a master amongst themselves before unit 0 has joined the relation, so the assumption that the lowest numbered unit in a new replication peer relation is the master is false.

[lp:~stub/charms/precise/postgresql/fix-races]

75. By Stuart Bishop on 2013-08-23

Merged cleanups into fix-races.

[Mark Mims (mark-mims)]

review: Approve

=== modified file 'config.yaml'
--- config.yaml	2013-05-27 14:37:19 +0000
+++ config.yaml	2013-08-23 09:40:16 +0000
@@ -313,3 +313,10 @@
     type: string
     description: |
         Extra archives to add with add-apt-repository(1).
+  advisory_lock_restart_key:
+    default: 765
+    type: int
+    description: |
+        An advisory lock key used internally by the charm. You do not need
+        to change it unless it happens to conflict with an advisory lock key
+        being used by your applications.

=== modified file 'hooks/hooks.py'
--- hooks/hooks.py	2013-08-23 09:40:15 +0000
+++ hooks/hooks.py	2013-08-23 09:40:16 +0000
@@ -20,7 +20,7 @@
 
 from charmhelpers.core import hookenv, host
 from charmhelpers.core.hookenv import (
-    CRITICAL, ERROR, WARNING, INFO, DEBUG, log,
+    CRITICAL, ERROR, WARNING, INFO, DEBUG,
     )
 
 hooks = hookenv.Hooks()
@@ -28,16 +28,24 @@
 # jinja2 may not be importable until the install hook has installed the
 # required packages.
 def Template(*args, **kw):
+    """jinja2.Template with deferred jinja2 import"""
     from jinja2 import Template
     return Template(*args, **kw)
 
 
 def log(msg, lvl=INFO):
-    # Per Bug #1208787, log messages sent via juju-log are being lost.
-    # Spit messages out to a log file to work around the problem.
+    '''Log a message.
+
+    Per Bug #1208787, log messages sent via juju-log are being lost.
+    Spit messages out to a log file to work around the problem.
+    It is also rather nice to have the log messages we explicitly emit
+    in a separate log file, rather than just mashed up with all the
+    juju noise.
+    '''
     myname = hookenv.local_unit().replace('/', '-')
-    with open('/tmp/{}-debug.log'.format(myname), 'a') as f:
-        f.write('{}: {}\n'.format(lvl, msg))
+    ts = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime())
+    with open('/var/log/juju/{}-debug.log'.format(myname), 'a') as f:
+        f.write('{} {}: {}\n'.format(ts, lvl, msg))
     hookenv.log(msg, lvl)
 
 
@@ -49,6 +57,7 @@
         self.load()
 
     def load(self):
+        '''Load stored state from local disk.'''
         if os.path.exists(self._state_file):
             state = pickle.load(open(self._state_file, 'rb'))
         else:
@@ -58,6 +67,7 @@
         self.update(state)
 
     def save(self):
+        '''Store state to local disk.'''
         state = {}
         state.update(self)
         pickle.dump(state, open(self._state_file, 'wb'))
@@ -181,13 +191,13 @@
 
 
 def postgresql_autostart(enabled):
+    startup_file = os.path.join(postgresql_config_dir, 'start.conf')
     if enabled:
         log("Enabling PostgreSQL startup in {}".format(startup_file))
         mode = 'auto'
     else:
         log("Disabling PostgreSQL startup in {}".format(startup_file))
         mode = 'manual'
-    startup_file = os.path.join(postgresql_config_dir, 'start.conf')
     contents = Template(open("templates/start_conf.tmpl").read()).render(
         {'mode': mode})
     host.write_file(
@@ -209,6 +219,7 @@
 
 
 def postgresql_is_running():
+    '''Return true if PostgreSQL is running.'''
     # init script always return true (9.1), add extra check to make it useful
     status, output = commands.getstatusoutput("invoke-rc.d postgresql status")
     if status != 0:
@@ -219,72 +230,65 @@
 
 
 def postgresql_stop():
-    host.service_stop('postgresql')
-    return not postgresql_is_running()
+    '''Shutdown PostgreSQL.'''
+    success = host.service_stop('postgresql')
+    return not (success and postgresql_is_running())
 
 
 def postgresql_start():
-    host.service_start('postgresql')
-    return postgresql_is_running()
+    '''Start PostgreSQL if it is not already running.'''
+    success = host.service_start('postgresql')
+    return success and postgresql_is_running()
 
 
 def postgresql_restart():
+    '''Restart PostgreSQL, or start it if it is not already running.'''
     if postgresql_is_running():
-        # If the database is in backup mode, we don't want to restart
-        # PostgreSQL and abort the procedure. This may be another unit being
-        # cloned, or a filesystem level backup is being made. There is no
-        # timeout here, as backups can take hours or days. Instead, keep
-        # logging so admins know wtf is going on.
-        last_warning = time.time()
-        while postgresql_is_in_backup_mode():
-            if time.time() + 120 > last_warning:
-                log("In backup mode. PostgreSQL restart blocked.", WARNING)
-                log(
-                    "Run \"psql -U postgres -c 'SELECT pg_stop_backup()'\""
-                    "to cancel backup mode and forcefully unblock this hook.")
-                last_warning = time.time()
-            time.sleep(5)
-
-        return host.service_restart('postgresql')
+        with restart_lock(hookenv.local_unit(), True):
+            # 'service postgresql restart' fails; it only does a reload.
+            # success = host.service_restart('postgresql')
+            try:
+                run('pg_ctlcluster -force {version} {cluster_name} '
+                    'restart'.format(**config_data))
+                success = True
+            except subprocess.CalledProcessError as e:
+                success = False
     else:
-        return host.service_start('postgresql')
+        success = host.service_start('postgresql')
 
     # Store a copy of our known live configuration so
     # postgresql_reload_or_restart() can make good choices.
-    if 'saved_config' in local_state:
+    if success and 'saved_config' in local_state:
         local_state['live_config'] = local_state['saved_config']
         local_state.save()
 
-    return postgresql_is_running()
+    return success and postgresql_is_running()
 
 
 def postgresql_reload():
+    '''Make PostgreSQL reload its configuration.'''
     # reload returns a reliable exit status
     status, output = commands.getstatusoutput("invoke-rc.d postgresql reload")
     return (status == 0)
 
 
-def postgresql_reload_or_restart():
-    """Reload PostgreSQL configuration, restarting if necessary."""
-    # Pull in current values of settings that can only be changed on
-    # server restart.
+def requires_restart():
+    '''Check for configuration changes requiring a restart to take effect.'''
     if not postgresql_is_running():
-        return postgresql_restart()
+        return True
 
-    # Suck in the config last written to postgresql.conf.
     saved_config = local_state.get('saved_config', None)
     if not saved_config:
         # No record of postgresql.conf state, perhaps an upgrade.
         # Better restart.
-        return postgresql_restart()
+        return True
 
-    # Suck in our live config from last time we restarted.
     live_config = local_state.setdefault('live_config', {})
 
     # Pull in a list of PostgreSQL settings.
     cur = db_cursor()
     cur.execute("SELECT name, context FROM pg_settings")
-    requires_restart = False
+    restart = False
     for name, context in cur.fetchall():
         live_value = live_config.get(name, None)
         new_value = saved_config.get(name, None)
@@ -296,23 +300,27 @@
             if context == 'postmaster':
                 # A setting has changed that requires PostgreSQL to be
                 # restarted before it will take effect.
-                requires_restart = True
-
-    if requires_restart:
-        # A change has been requested that requires a restart.
-        log(
-            "Configuration change requires PostgreSQL restart. Restarting.",
+                restart = True
+    return restart
+
+
+def postgresql_reload_or_restart():
+    """Reload PostgreSQL configuration, restarting if necessary."""
+    if requires_restart():
+        log("Configuration change requires PostgreSQL restart. Restarting.",
             WARNING)
-        rc = postgresql_restart()
+        success = postgresql_restart()
+        if not success or requires_restart():
+            log("Configuration changes failed to apply", WARNING)
+            success = False
     else:
-        log("PostgreSQL reload, config changes taking effect.", DEBUG)
-        rc = postgresql_reload()  # No pending need to bounce, just reload.
+        success = host.service_reload('postgresql')
 
-    if rc == 0 and 'saved_config' in local_state:
-        local_state['live_config'] = local_state['saved_config']
+    if success:
+        local_state['saved_config'] = local_state['live_config']
         local_state.save()
 
-    return rc
+    return success
 
 
 def get_service_port(postgresql_config):
@@ -344,8 +352,6 @@
             config_data["shared_buffers"] = \
                 "%sMB" % (int(int(total_ram) * 0.15),)
         # XXX: This is very messy - should probably be a subordinate charm
-        # file overlaps with __builtin__.file ... renaming to conf_file
-        # negronjl
         conf_file = open("/etc/sysctl.d/50-postgresql.conf", "w")
         conf_file.write("kernel.sem = 250 32000 100 1024\n")
         conf_file.write("kernel.shmall = %s\n" %
@@ -579,7 +585,7 @@
 
 
 def db_cursor(autocommit=False, db='template1', user='postgres',
-              host=None, timeout=120):
+              host=None, timeout=30):
     import psycopg2
     if host:
         conn_str = "dbname={} host={} user={}".format(db, host, user)
@@ -855,14 +861,16 @@
 
 @hooks.hook()
 def start():
-    if not postgresql_restart():
+    if not postgresql_reload_or_restart():
         raise SystemExit(1)
 
 
 @hooks.hook()
 def stop():
-    if not postgresql_stop():
-        raise SystemExit(1)
+    if postgresql_is_running():
+        with restart_lock(hookenv.local_unit(), True):
+            if not postgresql_stop():
+                raise SystemExit(1)
 
 
 def quote_identifier(identifier):
@@ -1163,7 +1171,7 @@
 def db_relation_broken():
     from psycopg2.extensions import AsIs
 
-    relid = os.environ['JUJU_RELATION_ID']
+    relid = hookenv.relation_id()
     if relid not in local_state['relations']['db']:
         # This was to be a hot standby, but it had not yet got as far as
         # receiving and handling credentials from the master.
@@ -1174,7 +1182,7 @@
     # we used from there. Instead, we have to persist this information
     # ourselves.
     relation = local_state['relations']['db'][relid]
-    unit_relation_data = relation[os.environ['JUJU_UNIT_NAME']]
+    unit_relation_data = relation[hookenv.local_unit()]
 
     if local_state['state'] in ('master', 'standalone'):
         user = unit_relation_data.get('user', None)
@@ -1303,27 +1311,75 @@
         log("I am already the master", DEBUG)
         return hookenv.local_unit()
 
+    if local_state['state'] == 'hot standby':
+        log("I am already following {}".format(
+            local_state['following']), DEBUG)
+        return local_state['following']
+
+    replication_relid = hookenv.relation_ids('replication')[0]
+    replication_units = hookenv.related_units(replication_relid)
+
+    if local_state['state'] == 'standalone':
+        log("I'm a standalone unit wanting to participate in replication")
+        existing_replication = False
+        for unit in replication_units:
+            # If another peer thinks it is the master, believe it.
+            remote_state = hookenv.relation_get(
+                'state', unit, replication_relid)
+            if remote_state == 'master':
+                log("{} thinks it is the master, believing it".format(
+                    unit), DEBUG)
+                return unit
+
+            # If we find a peer that isn't standalone, we know
+            # replication has already been setup at some point.
+            if remote_state != 'standalone':
+                existing_replication = True
+
+        # If we are joining a peer relation where replication has
+        # already been setup, but there is currently no master, wait
+        # until one of the remaining participating units has been
+        # promoted to master. Only they have the data we need to
+        # preserve.
+        if existing_replication:
+            log("Peers participating in replication need to elect a master",
+                DEBUG)
+            return None
+
+        # There are no peers claiming to be master, and there is no
+        # election in progress, so lowest numbered unit wins.
+        units = replication_units + [hookenv.local_unit()]
+        master = unit_sorted(units)[0]
+        if master == hookenv.local_unit():
+            log("I'm Master - lowest numbered unit in new peer group")
+            return master
+        else:
+            log("Waiting on {} to declare itself Master".format(master), DEBUG)
+            return None
+
     if local_state['state'] == 'failover':
         former_master = local_state['following']
         log("Failover from {}".format(former_master))
 
         units_not_in_failover = set()
-        for relid in hookenv.relation_ids('replication'):
-            for unit in hookenv.related_units(relid):
-                if unit == former_master:
-                    log("Found dying master {}".format(unit), DEBUG)
-                    continue
-
-                relation = hookenv.relation_get(unit=unit, rid=relid)
-
-                if relation['state'] == 'master':
-                    log(
-                        "{} says it already won the election".format(unit),
-                        INFO)
-                    return unit
-
-                if relation['state'] != 'failover':
-                    units_not_in_failover.add(unit)
+        candidates = set()
+        for unit in replication_units:
+            if unit == former_master:
+                log("Found dying master {}".format(unit), DEBUG)
+                continue
+
+            relation = hookenv.relation_get(unit=unit, rid=replication_relid)
+
+            if relation['state'] == 'master':
+                log("{} says it already won the election".format(unit),
+                    INFO)
+                return unit
+
+            if relation['state'] == 'failover':
+                candidates.add(unit)
+
+            elif relation['state'] != 'standalone':
+                units_not_in_failover.add(unit)
 
         if units_not_in_failover:
             log("{} unaware of impending election. Deferring result.".format(
@@ -1333,35 +1389,24 @@
         log("Election in progress")
         winner = None
         winning_offset = -1
-        for relid in hookenv.relation_ids('replication'):
-            candidates = set(hookenv.related_units(relid))
-            candidates.add(hookenv.local_unit())
-            candidates.discard(former_master)
-            # Sort the unit lists so we get consistent results in a tie
-            # and lowest unit number wins.
-            for unit in unit_sorted(candidates):
-                relation = hookenv.relation_get(unit=unit, rid=relid)
-                if int(relation['wal_received_offset']) > winning_offset:
-                    winner = unit
-                    winning_offset = int(relation['wal_received_offset'])
+        candidates.add(hookenv.local_unit())
+        # Sort the unit lists so we get consistent results in a tie
+        # and lowest unit number wins.
+        for unit in unit_sorted(candidates):
+            relation = hookenv.relation_get(unit=unit, rid=replication_relid)
+            if int(relation['wal_received_offset']) > winning_offset:
+                winner = unit
+                winning_offset = int(relation['wal_received_offset'])
 
         # All remaining hot standbys are in failover mode and have
         # reported their wal_received_offset. We can declare victory.
-        log("{} won the election as is the new master".format(winner))
-        return winner
-
-    # Maybe another peer thinks it is the master?
-    for relid in hookenv.relation_ids('replication'):
-        for unit in hookenv.related_units(relid):
-            if hookenv.relation_get('state', unit, relid) == 'master':
-                return unit
-
-    # New peer group. Lowest numbered unit will be the master.
-    for relid in hookenv.relation_ids('replication'):
-        units = hookenv.related_units(relid) + [hookenv.local_unit()]
-        master = unit_sorted(units)[0]
-        log("New peer group. {} is elected master".format(master))
-        return master
+        if winner == hookenv.local_unit():
+            log("I won the election, announcing myself winner")
+            return winner
+        else:
+            log("Waiting for {} to announce its victory".format(winner),
+                DEBUG)
+            return None
 
 
 @hooks.hook('replication-relation-joined', 'replication-relation-changed')
@@ -1419,10 +1464,7 @@
             log("Fresh unit. I will clone {} and become a hot standby".format(
                 master))
 
-            # Before we start destroying anything, ensure that the
-            # master is contactable.
             master_ip = hookenv.relation_get('private-address', master)
-            wait_for_db(db='postgres', user='juju_replication', host=master_ip)
 
             clone_database(master, master_ip)
 
@@ -1592,8 +1634,55 @@
         os.chdir(org_dir)
 
 
+@contextmanager
+def restart_lock(unit, exclusive):
+    '''Aquire the database restart lock on the given unit.
+
+    A database needing a restart should grab an exclusive lock before
+    doing so. To block a remote database from doing a restart, grab a shared
+    lock.
+    '''
+    import psycopg2
+    key = long(config_data['advisory_lock_restart_key'])
+    if exclusive:
+        lock_function = 'pg_advisory_lock'
+    else:
+        lock_function = 'pg_advisory_lock_shared'
+    q = 'SELECT {}({})'.format(lock_function, key)
+
+    # We will get an exception if the database is rebooted while waiting
+    # for a shared lock. If the connection is killed, we retry a few
+    # times to cope.
+    num_retries = 3
+
+    for count in range(0, num_retries):
+        try:
+            if unit == hookenv.local_unit():
+                cur = db_cursor(autocommit=True)
+            else:
+                host = hookenv.relation_get('private-address', unit)
+                cur = db_cursor(
+                    autocommit=True, db='postgres',
+                    user='juju_replication', host=host)
+            cur.execute(q)
+            break
+        except psycopg2.Error:
+            if count == num_retries - 1:
+                raise
+
+    try:
+        yield
+    finally:
+        # Close our connection, swallowing any exceptions as the database
+        # may be being rebooted now we have released our lock.
+        try:
+            del cur
+        except psycopg2.Error:
+            pass
+
+
 def clone_database(master_unit, master_host):
-    with pgpass():
+    with restart_lock(master_unit, False):
         postgresql_stop()
         log("Cloning master {}".format(master_unit))
 
@@ -1607,9 +1696,10 @@
             shutil.rmtree(postgresql_cluster_dir)
 
         try:
-            # Change directory the postgres user can read.
-            with switch_cwd('/tmp'):
-                # Run the sudo command.
+            # Change directory the postgres user can read, and need
+            # .pgpass too.
+            with switch_cwd('/tmp'), pgpass():
+                # Clone the master with pg_basebackup.
                 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
             log(output, DEBUG)
             # Debian by default expects SSL certificates in the datadir.
@@ -1626,8 +1716,8 @@
             # can retry hooks again. Even assuming the charm is
             # functioning correctly, the clone may still fail
             # due to eg. lack of disk space.
-            log("Clone failed, db cluster destroyed", ERROR)
             log(x.output, ERROR)
+            log("Clone failed, local db destroyed", ERROR)
             if os.path.exists(postgresql_cluster_dir):
                 shutil.rmtree(postgresql_cluster_dir)
             if os.path.exists(postgresql_config_dir):
@@ -1652,6 +1742,15 @@
         os.path.join(postgresql_cluster_dir, 'backup_label'))
 
 
+def pg_basebackup_is_running():
+    cur = db_cursor(autocommit=True)
+    cur.execute("""
+        SELECT count(*) FROM pg_stat_activity
+        WHERE usename='juju_replication' AND application_name='pg_basebackup'
+        """)
+    return cur.fetchone()[0] > 0
+
+
 def postgresql_wal_received_offset():
     """How much WAL we have.
 
@@ -1694,7 +1793,7 @@
     try:
         nagios_uid = getpwnam('nagios').pw_uid
         nagios_gid = getgrnam('nagios').gr_gid
-    except:
+    except Exception:
         hookenv.log("Nagios user not set up.", hookenv.DEBUG)
         return
 

=== modified file 'test.py'
--- test.py	2013-08-23 09:40:15 +0000
+++ test.py	2013-08-23 09:40:16 +0000
@@ -74,12 +74,12 @@
         return None
 
     def deploy(self, charm, name=None, num_units=1):
-        # The first time we deploy a charm in the test run, it needs to
-        # deploy with --update to ensure we are testing the desired
-        # revision of the charm. Subsequent deploys we do not use
-        # --update to avoid overhead and needless incrementing of the
+        # The first time we deploy a local: charm in the test run, it
+        # needs to deploy with --update to ensure we are testing the
+        # desired revision of the charm. Subsequent deploys we do not
+        # use --update to avoid overhead and needless incrementing of the
         # revision number.
-        if charm.startswith('cs:') or charm in self._deployed_charms:
+        if not charm.startswith('local:') or charm in self._deployed_charms:
             cmd = ['deploy']
         else:
             cmd = ['deploy', '-u']
@@ -102,7 +102,7 @@
         self.status = self.get_result(['status'])
         return self.status
 
-    def wait_until_ready(self):
+    def wait_until_ready(self, extra=45):
         ready = False
         while not ready:
             self.refresh_status()
@@ -128,7 +128,7 @@
         # enough that our system is probably stable. This means we have
         # extremely slow and flaky tests, but that is possibly better
         # than no tests.
-        time.sleep(45)
+        time.sleep(extra)
 
     def setUp(self):
         DEBUG("JujuFixture.setUp()")
@@ -156,7 +156,7 @@
         # Per Bug #1190250 (WONTFIX), we need to wait for dying services
         # to die before we can continue.
         if found_services:
-            self.wait_until_ready()
+            self.wait_until_ready(0)
 
         # We shouldn't reuse machines, as we have no guarantee they are
         # still in a usable state, so tear them down too. Per
@@ -305,15 +305,18 @@
         self.juju.do(['add-relation', 'postgresql:db', 'psql:db'])
         self.juju.wait_until_ready()
 
-        # On a freshly setup service, lowest numbered unit is always the
-        # master.
-        units = unit_sorted(
-            self.juju.status['services']['postgresql']['units'].keys())
-        master_unit, standby_unit_1, standby_unit_2 = units
-
-        self.assertIs(True, self.is_master(master_unit))
-        self.assertIs(False, self.is_master(standby_unit_1))
-        self.assertIs(False, self.is_master(standby_unit_2))
+        # Even on a freshly setup service, we have no idea which unit
+        # will become the master as we have no control over which two
+        # units join the peer relation first.
+        units = sorted((self.is_master(unit), unit)
+            for unit in
+                self.juju.status['services']['postgresql']['units'].keys())
+        self.assertFalse(units[0][0])
+        self.assertFalse(units[1][0])
+        self.assertTrue(units[2][0])
+        standby_unit_1 = units[0][1]
+        standby_unit_2 = units[1][1]
+        master_unit = units[2][1]
 
         self.sql('CREATE TABLE Token (x int)', master_unit)
 
@@ -390,11 +393,18 @@
         self.juju.do(['add-relation', 'postgresql:db-admin', 'psql:db-admin'])
         self.juju.wait_until_ready()
 
-        # On a freshly setup service, lowest numbered unit is always the
-        # master.
-        units = unit_sorted(
-            self.juju.status['services']['postgresql']['units'].keys())
-        master_unit, standby_unit_1, standby_unit_2 = units
+        # Even on a freshly setup service, we have no idea which unit
+        # will become the master as we have no control over which two
+        # units join the peer relation first.
+        units = sorted((self.is_master(unit, 'postgres'), unit)
+            for unit in
+                self.juju.status['services']['postgresql']['units'].keys())
+        self.assertFalse(units[0][0])
+        self.assertFalse(units[1][0])
+        self.assertTrue(units[2][0])
+        standby_unit_1 = units[0][1]
+        standby_unit_2 = units[1][1]
+        master_unit = units[2][1]
 
         # Shutdown PostgreSQL on standby_unit_1 and ensure
         # standby_unit_2 will have received more WAL information from