Fix the README in order to show BDR-related PostgreSQL options.

Currently, BDR and physical replication at the same time
is not supported. BDR requires additional postgresql
configuration and a patched version of PostgreSQL 9.4 + BDR plugin.
Included is the Docker image to try it locally.

Dockerfile

@@ -6,15 +6,22 @@ MAINTAINER Feike Steenbergen <[email protected]>
 # We need curl
 RUN apt-get update -y && apt-get install curl -y
 
-# Add PGDG repositories
+# Add PGDG and BDR repositories
 RUN echo "deb http://apt.postgresql.org/pub/repos/apt/ $(lsb_release -cs)-pgdg main" > /etc/apt/sources.list.d/pgdg.list
+RUN echo "deb http://packages.2ndquadrant.com/bdr/apt/ $(lsb_release -cs)-2ndquadrant main" >> /etc/apt/sources.list.d/pgdg.list
 RUN curl https://www.postgresql.org/media/keys/ACCC4CF8.asc | apt-key add -
+# import the BDR key
+RUN curl -s -o - http://packages.2ndquadrant.com/bdr/apt/AA7A6805.asc | sudo apt-key add -
+
 RUN apt-get update -y
 RUN apt-get upgrade -y
 
 ENV PGVERSION 9.4
-RUN apt-get install python python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo python-pip -y
-RUN apt-get install python-dev postgresql-server-dev-${PGVERSION} -y
+ENV PGTYPE postgresql-bdr
+ENV PGCOMPATIBLETYPE postgresql
+
+RUN apt-get install python python-yaml python-requests python-boto ${PGTYPE}-${PGVERSION} ${PGTYPE}-${PGVERSION}-bdr-plugin python-dnspython python-kazoo python-pip -y
+RUN apt-get install python-dev ${PGTYPE}-server-dev-${PGVERSION} -y
 RUN pip install python-etcd psycopg2
 
 ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH

README.rst

@@ -75,6 +75,10 @@ For an example file, see ``postgres0.yml``. Regarding settings:
         -  *port*: Exhibitor port.
         -  *hosts*: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.
 
+-  *bdr*:
+    - *enable*: on if you want to enable BDR
+    - *database*: database name to support BDR (only a single database is supported)
+
 -  *postgresql*:
     -  *name*: the name of the Postgres host. Must be unique for the cluster.
     -  *listen*: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication.
@@ -110,8 +114,16 @@ For an example file, see ``postgres0.yml``. Regarding settings:
         -  *username*: admin username; user is created during initialization. It will have CREATEDB and CREATEROLE privileges.
         -  *password*: admin password; user is created during initialization.
 
-    -  *recovery\_conf*: additional configuration settings written to recovery.conf when configuring the follower.
-        -  *parameters*: list of configuration settings for Postgres. Many of these are required for replication to work.
+    -  *recovery\_conf*: additional configuration settings written to recovery.conf when configuring follower.
+        -  *parameters*: list of configuration settings for Postgres.  Many of these are required for replication to work.
+
+    -  *create\_replica\_methods*: an ordered list of the create methods for turning a patroni node into a new replica.
+       "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured
+       as its own config item.
+
+    -  *replica\_method* for each create_replica_method other than basebackup, you would add a configuration section
+       of the same name.  At a minimum, this should include "command" with a full path to the actual script to be
+       executed.  Other configuration parameters will be passed along to the script in the form "parameter=value".
 
 Replication Choices
 -------------------
@@ -155,6 +167,31 @@ Choosing your replication schema is dependent on your business
 considerations. Investigate both async and sync replication, as well as other
 HA solutions, to determine which solution is best for you.
 
+You can also use BDR (bi-directional replication) if you have a compatible
+PostgreSQL version with the BDR plugin installed (see http://bdr-project.org/docs/next/installation.html).
+It will require adding a BDR shared_library to your configuration, as well as
+setting the following options for postgresql (see http://bdr-project.org/docs/next/settings-prerequisite.html):
+
+.. code:: YAML
+
+        max_worker_processes: 10
+        max_replication_slots: 10
+        max_wal_senders: 10
+        shared_preload_libraries: 'bdr'
+        track_commit_timestamp: 'on'
+        wal_level: 'logical'
+
+At the moment Patroni BDR is not compatible with a streaming replication,
+if BDR is enabled normal replica node won't be able to join. This is not
+a principal limitation of BDR, and we might resolve this in the future
+(although 'promotion' will only work between nodes running a physical
+replication, i.e. a replica won't be able to attach to a different
+multimaster node).
+
+Another limitation is that only one database is supported at the moment.
+BDR requires the replication user to be also a superuser, so you might
+want to excersie extra caution when choosing the password for this user.
+
 Applications Should Not Use Superusers
 --------------------------------------
 

docker/entrypoint.sh

@@ -92,8 +92,11 @@ etcd:
   scope: *scope
   ttl: *ttl
   host: ${ETCD_CLUSTER}
+bdr:
+   enable: 'on'
+   database: 'bdrtest'
 postgresql:
-  name: postgresql_${DOCKER_IP//./_} ## Replication slots do not allow dots in their name
+  name: ${HOSTNAME}
   scope: *scope
   listen: 0.0.0.0:5432
   connect_address: ${DOCKER_IP}:5432
@@ -119,20 +122,24 @@ postgresql:
     archive_command: 'true'
     max_wal_senders: 20
     listen_addresses: 0.0.0.0
-    wal_keep_segments: 8
+    checkpoint_segments: 64
+    wal_keep_segments: 64
     archive_timeout: 1800s
     max_replication_slots: 20
     hot_standby: "on"
+    max_worker_processes: 10
+    max_replication_slots: 10
+    max_wal_senders: 10
+    shared_preload_libraries: 'bdr'
+    track_commit_timestamp: true
+    wal_level: 'logical'
 __EOF__
 
 cat /patroni/postgres.yml
 
 if [ ! -z $CHEAT ]
 then
-    while :
-    do
-        sleep 60
-    done
+    exec bash
 else
     exec python /patroni.py /patroni/postgres.yml
 fi

patroni/__init__.py

@@ -19,7 +19,8 @@ class Patroni:
     def __init__(self, config):
         self.nap_time = config['loop_wait']
         self.tags = config.get('tags', dict())
-        self.postgresql = Postgresql(config['postgresql'])
+        self.bdr = config.get('bdr', dict())
+        self.postgresql = Postgresql(config['postgresql'], self.bdr)
         self.dcs = self.get_dcs(self.postgresql.name, config)
         self.api = RestApiServer(self, config['restapi'])
         self.ha = Ha(self)

patroni/api.py

@@ -242,14 +242,17 @@ class RestApiHandler(BaseHTTPRequestHandler):
             }
         except (psycopg2.Error, RetryFailedError, PostgresConnectionException):
             state = self.server.patroni.postgresql.state
-            if state in ['stopped', 'starting', 'stopping', 'restarting', 'running']:
+            if state == 'running':
                 logger.exception('get_postgresql_status')
-                state = 'unknown' if state == 'running' else state
+                state = 'unknown'
             return {'state': state}
 
     def get_tags(self):
         return {'tags': self.server.patroni.tags}
 
+    def log_message(self, format, *args):
+        logger.debug("API thread: " + format % args)
+
 
 class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
 
@@ -278,7 +281,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
     def query(self, sql, *params):
         cursor = None
         try:
-            with self.patroni.postgresql.connection().cursor() as cursor:
+            with self.patroni.postgresql.connection('postgres').cursor() as cursor:
                 cursor.execute(sql, params)
                 return [r for r in cursor]
         except psycopg2.Error as e:

patroni/ctl.py

@@ -111,7 +111,7 @@ def post_patroni(member, endpoint, content, headers={'Content-Type': 'applicatio
     url = urlparse(member.api_url)
     logging.debug(url)
     return requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers,
-                         data=json.dumps(content), timeout=5)
+                         data=json.dumps(content), timeout=60)
 
 
 def print_output(columns, rows=[], alignment=None, format='pretty', header=True, delimiter='\t'):
@@ -506,7 +506,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
     t_started = time.time()
     r = None
     try:
-        r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'candidate': candidate or ''})
+        r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'member': candidate or ''})
         if r.status_code == 200:
             logging.debug(r)
             logging.debug(r.text)
@@ -540,6 +540,8 @@ def output_members(cluster, name=None, format='pretty'):
     if cluster.leader:
         leader_name = cluster.leader.member.name
 
+    xlog_location_cluster = cluster.last_leader_operation or 0
+
     # Mainly for consistent pretty printing and watching we sort the output
     cluster.members.sort(key=lambda x: x.name)
     for m in cluster.members:
@@ -552,9 +554,10 @@ def output_members(cluster, name=None, format='pretty'):
         host = build_connect_parameters(m.conn_url)['host']
 
         xlog_location = m.data.get('xlog_location')
-        lag = ''
-        if xlog_location is not None:
-            lag = round(((cluster.last_leader_operation or 0) - m.data.get('xlog_location', 0)) / 1024 / 1024)
+        if xlog_location is None or (xlog_location_cluster < xlog_location):
+            lag = ''
+        else:
+            lag = round((xlog_location_cluster - xlog_location)/1024/1024)
 
         rows.append([
             name,
@@ -562,7 +565,7 @@ def output_members(cluster, name=None, format='pretty'):
             host,
             leader,
             m.data.get('state', ''),
-            lag,
+            lag
         ])
 
     columns = [
@@ -573,7 +576,7 @@ def output_members(cluster, name=None, format='pretty'):
         'State',
         'Lag in MB',
     ]
-    alignment = {'Cluster': 'l', 'Member': 'l', 'Host': 'l'}
+    alignment = {'Cluster': 'l', 'Member': 'l', 'Host': 'l', 'Lag in MB': 'r'}
 
     print_output(columns, rows, alignment, format)
 

patroni/ha.py

@@ -20,6 +20,11 @@ class Ha:
         self.cluster = None
         self.old_cluster = None
         self._async_executor = AsyncExecutor()
+        self.bdr = patroni.bdr
+        self.use_bdr = self.bdr and\
+            self.bdr.get('enable') and\
+            (self.bdr.get('database') is not None)
+        self.bdr_may_need_reinitialize = False
 
     def load_cluster_from_dcs(self):
         cluster = self.dcs.get_cluster()
@@ -46,7 +51,7 @@ class Ha:
         logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name)
         return lock_owner == self.state_handler.name
 
-    def touch_member(self):
+    def touch_member(self, ttl=None):
         data = {
             'conn_url': self.state_handler.connection_string,
             'api_url': self.patroni.api.connection_string,
@@ -59,7 +64,7 @@ class Ha:
                 data['xlog_location'] = self.state_handler.xlog_position()
             except:
                 pass
-        self.dcs.touch_member(json.dumps(data, separators=(',', ':')))
+        self.dcs.touch_member(json.dumps(data, separators=(',', ':')), ttl=ttl)
 
     def copy_backup_from_leader(self, leader):
         if self.state_handler.bootstrap(leader):
@@ -69,15 +74,29 @@ class Ha:
             self.state_handler.remove_data_directory()
             logger.error('failed to bootstrap from leader')
 
-    def bootstrap(self):
+    def initialize_bdr(self, leader):
+        if self.state_handler.initialize_bdr_database(leader):
+            logger.info("bootstrapped from leader")
+        else:
+            self.state_handler.stop('immediate')
+            self.state_handler.remove_data_directory()
+            logger.error('failed to bootstrap from leader')
+
+    def bootstrap(self, bdr=False):
         if not self.cluster.is_unlocked():  # cluster already has leader
             self._async_executor.schedule('bootstrap from leader')
-            self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, ))
+            if not bdr:
+                self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, ))
+            else:
+                self._async_executor.run_async(self.initialize_bdr, args=(self.cluster.leader, ))
             return 'trying to bootstrap from leader'
         elif not self.cluster.initialize and not self.patroni.nofailover:  # no initialize key
             if self.dcs.initialize(create_new=True):  # race for initialization
                 try:
-                    self.state_handler.bootstrap()
+                    if not bdr:
+                        self.state_handler.bootstrap()
+                    else:
+                        self.state_handler.initialize_bdr_database()
                     self.dcs.initialize(create_new=False, sysid=self.state_handler.sysid)
                 except:  # initdb or start failed
                     # remove initialization key and give a chance to other members
@@ -93,6 +112,9 @@ class Ha:
         else:
             return 'waiting for leader to bootstrap'
 
+    def bootstrap_bdr(self):
+        return self.bootstrap(bdr=True)
+
     def recover(self):
         has_lock = self.has_lock()
 
@@ -118,6 +140,22 @@ class Ha:
         logger.info('started as readonly because i had the session lock')
         self.load_cluster_from_dcs()
 
+    def recover_bdr(self):
+        if not self.bdr_may_need_reinitialize:
+            self.state_handler.start_bdr()
+            return "started as a member of BDR group"
+        has_lock = self.has_lock()
+        if self.cluster.is_unlocked():  # no leader yet, and we need to reinitialize from the leader
+            return "waiting for another member of the BDR group"
+        if has_lock:
+            self.dcs.delete_leader()
+            return "released the lock to another member of the BDR group"
+        if self.state_handler.initialize_bdr_database(self.cluster.leader, empty=False):
+            # we reinitialized this node, make sure the cluster is aware of it.
+            self.touch_member()
+            self.bdr_may_need_reinitialize = False
+            return "reinitialized and started as a member of the BDR group"
+
     def follow_the_leader(self, demote_reason, follow_reason, refresh=True):
         refresh and self.load_cluster_from_dcs()
         ret = demote_reason if self.state_handler.is_leader() else follow_reason
@@ -268,14 +306,18 @@ class Ha:
             self.dcs.reset_cluster()
         self.state_handler.follow_the_leader(None)
 
-    def process_manual_failover_from_leader(self):
+    def process_manual_failover_from_leader(self, bdr=False):
         failover = self.cluster.failover
         if not failover.leader or failover.leader == self.state_handler.name:
             if not failover.member or failover.member != self.state_handler.name:
                 members = [m for m in self.cluster.members if not failover.member or m.name == failover.member]
                 if self.is_failover_possible(members):  # check that there are healthy members
-                    self._async_executor.schedule('manual failover: demote')
-                    self._async_executor.run_async(self.demote)
+                    if not bdr:
+                        self._async_executor.schedule('manual failover: demote')
+                        self._async_executor.run_async(self.demote)
+                    else:
+                        self.dcs.delete_leader()
+                        self.dcs.reset_cluster()
                     return 'manual failover: demoting myself'
                 else:
                     logger.warning('manual failover: no healthy members found, failover is not possible')
@@ -288,6 +330,9 @@ class Ha:
         logger.info('Trying to clean up failover key')
         self.dcs.manual_failover('', '', self.cluster.failover.index)
 
+    def process_manual_failover_from_leader_bdr(self):
+        self.process_manual_failover_from_leader(bdr=True)
+
     def process_unhealthy_cluster(self):
         if self.is_healthiest_node():
             if self.acquire_lock():
@@ -326,6 +371,31 @@ class Ha:
         return self.follow_the_leader('demoting self because i do not have the lock and i was a leader',
                                       'no action.  i am a secondary and i am following a leader', False)
 
+    def process_unhealthy_cluster_bdr(self):
+        if self.acquire_lock():
+            if self.cluster.failover:
+                logger.info('Cleaning up failover key after acquiring leader lock...')
+                self.dcs.manual_failover('', '')
+            self.dcs.get_cluster()
+            return "acquired the session lock as a leader"
+        else:
+            return "lost the race to acquire the session lock"
+
+    def process_healthy_cluster_bdr(self):
+        if self.has_lock():
+            if self.cluster.failover:
+                msg = self.process_manual_failover_from_leader_bdr()
+                if msg is not None:
+                    return msg
+            if self.update_lock():
+                return "no action. I am the leader with the lock"
+            else:
+                logger.error("failed to update leader lock")
+                self.load_cluster_from_dcs()
+        else:
+            logger.info("does not have lock")
+        return "no action. I do not have the lock"
+
     def schedule(self, action):
         with self._async_executor:
             return self._async_executor.schedule(action)
@@ -386,7 +456,14 @@ class Ha:
         try:
             self.load_cluster_from_dcs()
 
-            self.touch_member()
+            # if member key is missing - we may need to destory
+            # and re-create the BDR database unless we are the first node in the cluster
+            if self.use_bdr and len([m for m in self.cluster.members if m.name == self.state_handler.name]) == 0:
+                self.bdr_may_need_reinitialize = True
+
+            # Create a member key with a very short expiration time if we may need to reinitialize this member
+            # in order to avoid lossing the "may need to reinitialize status" in case Patroni is restarted
+            self.touch_member(ttl=None if not self.bdr_may_need_reinitialize else self.bdr.get('min_ttl', 5))
 
             # cluster has leader key but not initialize key
             if not self.cluster.is_unlocked() and not self.sysid_valid(self.cluster.initialize) and self.has_lock():
@@ -402,33 +479,46 @@ class Ha:
 
             # is data directory empty?
             if self.state_handler.data_directory_empty():
-                return self.bootstrap()  # new node
+                if self.use_bdr:
+                    # member key was missing, but it's a new database - no need to reinitialize
+                    self.bdr_may_need_reinitialize = False
+                    self.touch_member()
+                return self.bootstrap(self.use_bdr)  # new node
             # "bootstrap", but data directory is not empty
-            elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked():
+            elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked() and \
+                    not self.bdr_may_need_reinitialize:
                 self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid)
             else:
                 # check if we are allowed to join
-                if self.sysid_valid(self.cluster.initialize) and self.cluster.initialize != self.state_handler.sysid:
+                if not self.use_bdr and \
+                   self.sysid_valid(self.cluster.initialize) and self.cluster.initialize != self.state_handler.sysid:
                     logger.fatal("system ID mismatch, node {0} belongs to a different cluster".
                                  format(self.state_handler.name))
                     sys.exit(1)
 
             # try to start dead postgres
-            if not self.state_handler.is_healthy():
-                msg = self.recover()
-                if msg is not None:
-                    return msg
+            if self.use_bdr:
+                if self.bdr_may_need_reinitialize or not self.state_handler.is_healthy():
+                    msg = self.recover_bdr()
+                    if msg is not None:
+                        return msg
+            else:
+                if not self.state_handler.is_healthy():
+                    msg = self.recover()
+                    if msg is not None:
+                        return msg
 
             try:
                 if self.cluster.is_unlocked():
-                    return self.process_unhealthy_cluster()
+                    return self.process_unhealthy_cluster() if not self.use_bdr else self.process_unhealthy_cluster_bdr()
                 else:
-                    return self.process_healthy_cluster()
+                    return self.process_healthy_cluster() if not self.use_bdr else self.process_healthy_cluster_bdr()
             finally:
-                self.state_handler.sync_replication_slots(self.cluster)
+                if not self.use_bdr:
+                    self.state_handler.sync_replication_slots(self.cluster)
         except DCSError:
             logger.error('Error communicating with DCS')
-            if self.state_handler.is_running() and self.state_handler.is_leader():
+            if self.state_handler.is_running() and self.state_handler.is_leader() and not self.use_bdr:
                 self.demote(delete_leader=False)
                 return 'demoted self because DCS is not accessible and i was a leader'
         except (psycopg2.Error, PostgresConnectionException):

patroni/postgresql.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import psycopg2
+import re
 import shlex
 import shutil
 import subprocess
@@ -41,8 +42,9 @@ def parseurl(url):
 
 class Postgresql:
 
-    def __init__(self, config):
+    def __init__(self, config, config_bdr={}):
         self.config = config
+        self.bdr = config_bdr
         self.name = config['name']
         self.server_parameters = config.get('parameters', {})
         self.scope = config['scope']
@@ -72,6 +74,7 @@ class Postgresql:
             connect_address=connect_address, **self.replication)
 
         self._connection = None
+        self._connection_db = None
         self._cursor_holder = None
         self._need_rewind = False
         self._sysid = None
@@ -128,28 +131,32 @@ class Postgresql:
                 break
         return local_address + ':' + self.port
 
-    def connection(self):
+    def connection(self, dbname):
         if not self._connection or self._connection.closed != 0:
-            r = parseurl('postgres://{}/postgres'.format(self.local_address))
+            r = parseurl('postgres://{0}/{1}'.format(self.local_address, dbname))
             self._connection = psycopg2.connect(**r)
+            self._connection_db = dbname
             self._connection.autocommit = True
         return self._connection
 
-    def _cursor(self):
+    def _cursor(self, dbname='postgres'):
+        if self._connection_db != dbname:
+            self.close_connection()
         if not self._cursor_holder or self._cursor_holder.closed or self._cursor_holder.connection.closed != 0:
-            logger.info("established a new patroni connection to the postgres cluster")
-            self._cursor_holder = self.connection().cursor()
+            logger.info("established a new patroni connection to the {0} cluster".format(dbname))
+            self._cursor_holder = self.connection(dbname).cursor()
         return self._cursor_holder
 
     def close_connection(self):
         if self._cursor_holder and self._cursor_holder.connection and self._cursor_holder.connection.closed == 0:
             self._cursor_holder.connection.close()
-            logger.info("closed patroni connection to the postgresql cluster")
+            logger.info("closed patroni connection to the {0} cluster".format(self._connection_db))
+            self._connection_db = None
 
-    def _query(self, sql, *params):
+    def _query(self, sql, dbname, *params):
         cursor = None
         try:
-            cursor = self._cursor()
+            cursor = self._cursor(dbname)
             cursor.execute(sql, params)
             return cursor
         except psycopg2.Error as e:
@@ -159,9 +166,10 @@ class Postgresql:
                 raise RetryFailedError('cluster is being restarted')
             raise PostgresConnectionException('connection problems')
 
-    def query(self, sql, *params):
+    def query(self, sql, *params, **kwargs):
+        dbname = kwargs.get('dbname', 'postgres')
         try:
-            return self.retry(self._query, sql, *params)
+            return self.retry(self._query, sql, dbname, *params)
         except RetryFailedError as e:
             raise PostgresConnectionException(str(e))
 
@@ -198,7 +206,7 @@ class Postgresql:
             os.close(fd)
             options.append('--pwfile={}'.format(pwfile))
 
-        ret = subprocess.call(self._pg_ctl + ['initdb'] + ['-o', ' '.join(options)] if options else []) == 0
+        ret = subprocess.call(self._pg_ctl + ['initdb'] + (['-o', ' '.join(options)] if options else [])) == 0
         if pwfile:
             os.remove(pwfile)
         if ret:
@@ -207,6 +215,81 @@ class Postgresql:
             self.set_state('initdb failed')
         return ret
 
+    def initialize_bdr_database(self, leader=None, empty=True):
+        """ initialize BDR database. If empty = False, reinitialize an existing one.
+            If leader is None, create a new group, otherwise, join an exising one.
+            non-empty database with an empty leader is not supported
+        """
+        if not empty and not leader:
+            ret = False
+        else:
+            try:
+                ret = (self.initialize() and self.start_bdr()) if empty else\
+                    (self.drop_bdr_database() and self.start_bdr())
+            except:
+                logger.exception("exception")
+                ret = False
+        if ret:
+            # convert all connection URLs to name=value strings
+            leader_dsn = self.primary_conninfo(leader.conn_url, include_db=self.bdr['database']) if leader else None
+            self_dsn = self.primary_conninfo(self.connection_string, include_db=self.bdr['database'])
+            try:
+                self.create_replication_user(superuser=True)
+                ret = self.create_bdr_database() and (self.join_bdr_group(self_dsn, leader_dsn) if leader
+                                                      else self.create_bdr_group(self_dsn))
+            except:
+                logger.exception("exception")
+                ret = False
+        if not ret and not leader:
+            raise Exception("Could not bootstrap initial PostgreSQL BDR node")
+        return ret
+
+    def start_bdr(self):
+        logger.info("starting new BDR instance {0}".format(self.name))
+        return self.start(bdr=True)
+
+    def drop_bdr_database(self):
+        logger.info("Dropping BDR database for instance {0}".format(self.name))
+        # stop the database if it's running in order to turn off bdr
+        if self.is_running():
+            self.stop()
+        ret = self.start(bdr=False)
+        if ret:
+            self.query("DROP DATABASE IF EXISTS {0}".format(self.bdr['database']))
+        return ret and self.restart(bdr=True)
+
+    def create_bdr_database(self):
+        logger.info("Creating BDR database and extensions for instance {0}".format(self.name))
+        self.query("CREATE DATABASE {0}".format(self.bdr['database']))
+        self.query("CREATE EXTENSION IF NOT EXISTS btree_gist", dbname=self.bdr['database'])
+        self.query("CREATE EXTENSION IF NOT EXISTS bdr", dbname=self.bdr['database'])
+        return True
+
+    def join_bdr_group(self, self_dsn, join_dsn):
+        logger.info("Joining BDR group {0}, host {1}, dsn {2}".format(join_dsn, self.name, self_dsn))
+        self.query("SELECT bdr.bdr_group_join(local_node_name:=%s, join_using_dsn:=%s, node_external_dsn:=%s)",
+                   self.name, join_dsn, self_dsn, dbname=self.bdr['database'])
+        return True
+
+    def create_bdr_group(self, self_dsn):
+        logger.info("Creating new BDR group {0}, host {1}".format(self_dsn, self.name))
+        self.query("SELECT bdr.bdr_group_create(local_node_name:=%s, node_external_dsn:=%s)",
+                   self.name, self_dsn, dbname=self.bdr['database'])
+        return True
+
+    def remove_bdr_nodes(self, nodes):
+        logger.info("BDR node {0}, removing nodes {1}".format(self.name, nodes))
+        self.query("SELECT bdr.part_by_node_names(%s)", nodes, dbname=self.bdr['database'])
+
+    def remove_bdr_disconnected_members(self, cluster):
+        # get all members
+        result = self.query("SELECT node_name FROM bdr.bdr_nodes WHERE node_status = 'r'")
+        if result:
+            nodes_db = set([node[0] for node in result.fetchall()])
+            nodes_etcd = set([member.name for member in self.members])
+            nodes_remove = nodes_db.difference(nodes_etcd)
+            self.remove_bdr_nodes(list(nodes_remove))
+
     def delete_trigger_file(self):
         os.path.exists(self.trigger_file) and os.unlink(self.trigger_file)
 
@@ -223,24 +306,63 @@ class Postgresql:
         r = parseurl(leader.conn_url)
 
         env = self.write_pgpass(r)
-        return self.create_replica(r, env) == 0
+        ret = self.create_replica(leader, env) == 0
+        ret and self.delete_trigger_file()
+        return ret
 
     @staticmethod
     def build_connstring(conn):
-        return "host={host} port={port} user={user}".format(**conn)
+        """
+        >>> Postgresql.build_connstring({'host': '127.0.0.1', 'port': '5432'}) == 'host=127.0.0.1 port=5432'
+        True
+        """
+        return ' '.join('{}={}'.format(param, val) for param, val in sorted(conn.items()))
+
+    def create_replica(self, leader, env):
+        # create the replica according to the replica_method
+        # defined by the user.  this is a list, so we need to
+        # loop through all methods the user supplies
+        connstring = leader.conn_url
+        # get list of replica methods from config.
+        # If there is no configuration key, or no value is specified, use basebackup
+        replica_methods = self.config.get('create_replica_method') or ['basebackup']
+        # go through them in priority order
+        ret = 1
+        for replica_method in replica_methods:
+            # if the method is basebackup, then use the built-in
+            if replica_method == "basebackup":
+                ret = self.basebackup(leader, env)
+                if ret == 0:
+                    logger.info("replica has been created using basebackup")
+                    # if basebackup succeeds, exit with success
+                    break
+            else:
+                cmd = replica_method
+                method_config = {}
+                # user-defined method; check for configuration
+                # not required, actually
+                if replica_method in self.config:
+                    method_config = self.config[replica_method].copy()
+                    # look to see if the user has supplied a full command path
+                    # if not, use the method name as the command
+                    cmd = method_config.pop('command', cmd)
+                    # add the default parameters
+                try:
+                    method_config.update({"scope": self.scope,
+                                          "role": "replica",
+                                          "datadir": self.data_dir,
+                                          "connstring": connstring})
+                    params = ["--{0}={1}".format(arg, val) for arg, val in method_config.items()]
+                    # call script with the full set of parameters
+                    ret = subprocess.call(shlex.split(cmd) + params, env=env)
+                    # if we succeeded, stop
+                    if ret == 0:
+                        logger.info("replica has been created using {0}".format(replica_method))
+                        break
+                except Exception as e:
+                    logger.exception('Error creating replica using method {0}: {1}'.format(replica_method, str(e)))
+                    ret = 1
 
-    def create_replica(self, master_connection, env):
-        self.set_state('building replica from {host}:{port}'.format(**master_connection))
-        connstring = self.build_connstring(master_connection)
-        cmd = self.config['restore']
-        try:
-            ret = subprocess.call(shlex.split(cmd) + [self.scope, "replica", self.data_dir, connstring], env=env)
-            self.delete_trigger_file()
-        except:
-            logger.exception('Error when creating replica')
-            ret = 1
-        if ret != 0:
-            self.set_state('failed to build replica from {host}:{port}'.format(**master_connection))
         return ret
 
     def is_leader(self):
@@ -279,12 +401,15 @@ class Postgresql:
         with self._state_lock:
             self._state = value
 
-    def start(self, block_callbacks=False):
+    def start(self, block_callbacks=False, bdr=False):
         if self.is_running():
             logger.error('Cannot start PostgreSQL because one is already running.')
             return True
 
-        self.set_role('replica' if os.path.exists(self.recovery_conf) else 'master')
+        if not bdr:
+            self.set_role('replica' if os.path.exists(self.recovery_conf) else 'master')
+        else:
+            self.set_role('master')
         if os.path.exists(self.postmaster_pid):
             os.remove(self.postmaster_pid)
             logger.info('Removed %s', self.postmaster_pid)
@@ -292,24 +417,25 @@ class Postgresql:
         if not block_callbacks:
             self.set_state('starting')
 
-        ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0
+        ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options(bdr)]) == 0
 
         self.set_state('running' if ret else 'start failed')
 
-        self.schedule_load_slots = ret and self.use_slots
-        self.save_configuration_files()
+        if not bdr:
+            self.schedule_load_slots = ret and self.use_slots
+            self.save_configuration_files()
         # block_callbacks is used during restart to avoid
         # running start/stop callbacks in addition to restart ones
         ret and not block_callbacks and self.call_nowait(ACTION_ON_START)
         return ret
 
-    def checkpoint(self):
+    def checkpoint(self, connstring=None):
         try:
-            r = parseurl('postgres://{}/postgres'.format(self.local_address))
-            r['options'] = '-c statement_timeout=0'
-            with psycopg2.connect(**r) as conn:
+            connstring = connstring or 'postgres://{}/postgres'.format(self.local_address)
+            with psycopg2.connect(connstring) as conn:
                 conn.autocommit = True
                 with conn.cursor() as cur:
+                    cur.execute("SET statement_timeout = 0")
                     cur.execute('CHECKPOINT')
         except:
             logging.exception('Exception during CHECKPOINT')
@@ -346,19 +472,28 @@ class Postgresql:
         ret and self.call_nowait(ACTION_ON_RELOAD)
         return ret
 
-    def restart(self):
+    def restart(self, bdr=False):
         self.set_state('restarting')
-        ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True)
+        ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True, bdr=bdr)
         if ret:
             self.call_nowait(ACTION_ON_RESTART)
         else:
             self.set_state('restart failed ({})'.format(self.state))
         return ret
 
-    def server_options(self):
+    def server_options(self, bdr=False):
+        bdr_set = False
         options = "--listen_addresses='{}' --port={}".format(self.listen_addresses, self.port)
         for setting, value in self.server_parameters.items():
+            if setting == 'shared_preload_libraries':
+                if not bdr and 'bdr' in value:
+                    value = ','.join([x for x in value.split(',') if x != 'bdr'])
+                elif bdr and 'bdr' not in value:
+                    bdr_set = True
+                    value = ','.join([x for x in value.split(',')]+['bdr'])
             options += " --{}='{}'".format(setting, value)
+        if bdr and not bdr_set:
+            options += " --shared_preload_libraries='bdr'"
         return options
 
     def is_healthy(self):
@@ -380,9 +515,10 @@ class Postgresql:
                 f.write(line + '\n')
 
     @staticmethod
-    def primary_conninfo(leader_url):
+    def primary_conninfo(leader_url, include_db=None):
         r = parseurl(leader_url)
-        return 'user={user} password={password} host={host} port={port} sslmode=prefer sslcompression=1'.format(**r)
+        ret = 'user={user} password={password} host={host} port={port} sslmode=prefer sslcompression=1'.format(**r)
+        return ret if not include_db else ret + ' dbname={0}'.format(include_db)
 
     def check_recovery_conf(self, leader):
         if not os.path.isfile(self.recovery_conf):
@@ -415,6 +551,9 @@ recovery_target_timeline = 'latest'
         r['user'] = r['username']
         env = self.write_pgpass(r)
         pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
+        # first run a checkpoint on a promoted master in order
+        # to make it store the new timeline ([email protected])
+        self.checkpoint(pc)
         logger.info("running pg_rewind from {}".format(pc))
         pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
         try:
@@ -539,7 +678,7 @@ recovery_target_timeline = 'latest'
         """ restore a previously saved postgresql.conf """
         try:
             for f in self.configuration_to_save:
-                not os.path.isfile(f) and os.path.isfile(f+'.backup') and shutil.copy(f + '.backup', f)
+                not os.path.isfile(f) and os.path.isfile(f + '.backup') and shutil.copy(f + '.backup', f)
         except:
             logger.exception('unable to restore configuration files from backup')
 
@@ -570,8 +709,9 @@ BEGIN
 END;
 $$""".format(name, options), name, password, password)
 
-    def create_replication_user(self):
-        self.create_or_update_role(self.replication['username'], self.replication['password'], 'REPLICATION')
+    def create_replication_user(self, superuser=False):
+        options = 'REPLICATION SUPERUSER' if superuser else 'REPLICATION'
+        self.create_or_update_role(self.replication['username'], self.replication['password'], '{0}'.format(options))
 
     def create_connection_users(self):
         if 'username' in self.superuser:
@@ -663,3 +803,27 @@ $$""".format(name, options), name, password, password)
         except:
             logger.exception('Could not remove data directory %s', self.data_dir)
             self.move_data_directory()
+
+    def basebackup(self, leader, env):
+        # creates a replica data dir using pg_basebackup.
+        # this is the default, built-in create_replica_method
+        # tries twice, then returns failure (as 1)
+        # uses "stream" as the xlog-method to avoid sync issues
+        master_connection = leader.conn_url
+        maxfailures = 2
+        ret = 1
+        for bbfailures in range(0, maxfailures):
+            try:
+                ret = subprocess.call(['pg_basebackup', '--pgdata=' + self.data_dir,
+                                       '--xlog-method=stream', "--dbname=" + master_connection], env=env)
+                if ret == 0:
+                    break
+
+            except Exception as e:
+                logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e))
+
+            if bbfailures < maxfailures - 1:
+                logger.error('Trying again in 5 seconds')
+                time.sleep(5)
+
+        return ret

patroni/scripts/aws.py

@@ -65,8 +65,11 @@ class AWSConnection:
         return self._tag_ebs(new_role) and ret
 
 
-if __name__ == '__main__':
+def main():
     if len(sys.argv) == 4 and sys.argv[1] in ('on_start', 'on_stop', 'on_role_change'):
         AWSConnection(cluster_name=sys.argv[3]).on_role_change(sys.argv[2])
     else:
         sys.exit("Usage: {0} action role name".format(sys.argv[0]))
+
+if __name__ == '__main__':
+    main()

patroni/scripts/restore.py

@@ -1,216 +0,0 @@
-#!/usr/bin/env python
-# arguments are:
-#   - cluster scope
-#   - cluster role
-#   - master connection string
-
-# for the AWS, the folliowing environment variables should be defined:
-# - WALE_ENV_DIR: directory where WAL-E environment is kept
-# - WAL_S3_BUCKET: a name of the S3 bucket for WAL-E
-# - WALE_BACKUP_THRESHOLD_MEGABYTES if WAL amount is above that - use pg_basebackup
-# - WALE_BACKUP_THRESHOLD_PERCENTAGE if WAL size exceeds a certain percentage of the
-#   latest backup size
-from collections import namedtuple
-import logging
-import os
-import psycopg2
-import subprocess
-import sys
-
-
-if sys.hexversion >= 0x03000000:
-    long = int
-
-logger = logging.getLogger(__name__)
-
-
-class Restore(object):
-
-    def __init__(self, scope, role, datadir, connstring, env=None):
-        self.scope = scope
-        self.role = role
-        self.master_connection = Restore.parse_connstring(connstring)
-        self.data_dir = datadir
-        self.env = os.environ.copy() if not env else env
-
-    @staticmethod
-    def parse_connstring(connstring):
-        # the connection string is in the form host= port= user=
-        # return the dictionary with all components as separare keys
-        result = {}
-        if connstring:
-            for x in connstring.split():
-                if x and '=' in x:
-                    key, val = x.split('=')
-                result[key.strip()] = val.strip()
-        return result
-
-    def setup(self):
-        pass
-
-    def replica_method(self):
-        return self.create_replica_with_pg_basebackup
-
-    def replica_fallback_method(self):
-        return None
-
-    def run(self):
-        """ creates a new replica using either pg_basebackup or WAL-E """
-        method_fn = self.replica_method()
-        ret = method_fn() if method_fn else 1
-        if ret != 0 and self.replica_fallback_method() is not None:
-            ret = (self.replica_fallback_method())()
-        return ret
-
-    def create_replica_with_pg_basebackup(self):
-        try:
-            ret = subprocess.call(['pg_basebackup', '-R', '-D',
-                                   self.data_dir, '--host=' + self.master_connection['host'],
-                                   '--port=' + str(self.master_connection['port']),
-                                   '-U', self.master_connection['user']],
-                                  env=self.env)
-        except Exception as e:
-            logger.error('Error when fetching backup with pg_basebackup: {0}'.format(e))
-            return 1
-        return ret
-
-
-class WALERestore(Restore):
-
-    def __init__(self, scope, role, datadir, connstring, env=None):
-        super(WALERestore, self).__init__(scope, role, datadir, connstring, env)
-        # check the environment variables
-        self.init_error = False
-
-    def setup(self):
-        if (self.env.get('WAL_S3_BUCKET') and
-            self.env.get('WALE_BACKUP_THRESHOLD_PERCENTAGE') and
-           self.env.get('WALE_BACKUP_THRESHOLD_MEGABYTES')) is None:
-                self.init_error = True
-        else:
-            self.wal_e = namedtuple('WALE',
-                                    'threshold_megabytes threshold_backup_size_percentage s3_bucket cmd dir env_file')
-
-            self.wal_e.dir = self.env.get('WALE_ENV_DIR', '/home/postgres/etc/wal-e.d/env')
-            self.wal_e.env_file = os.path.join(self.wal_e.dir, 'WALE_S3_PREFIX')
-
-            self.wal_e.cmd = 'envdir {} wal-e --aws-instance-profile '.\
-                format(self.wal_e.dir)
-            self.wal_e.s3_bucket = self.env['WAL_S3_BUCKET']
-            self.wal_e.threshold_megabytes = self.env['WALE_BACKUP_THRESHOLD_MEGABYTES']
-            self.wal_e.threshold_backup_size_percentage = self.env['WALE_BACKUP_THRESHOLD_PERCENTAGE']
-
-            # check that the env file exists, create it otherwise
-            try:
-                if not os.path.exists(self.wal_e.dir):
-                    os.makedirs(self.wal_e.dir)
-                # if this is a directory - make sure we have full access there
-                elif not (os.path.isdir(self.wal_e.dir) and os.access(self.wal_e.dir, os.R_OK | os.W_OK | os.X_OK)):
-                    logger.error("Unable to access {} or not a directory".format(self.wal_e.dir))
-                    self.init_error = True
-                # if WAL_S3_PREFIX is not there - create it and write the full path to bucket
-                if not self.init_error and not os.path.exists(self.wal_e.env_file):
-                    with open(self.wal_e.env_file, 'w') as f:
-                        f.write("s3://{0}/spilo/{1}/wal/\n".format(self.wal_e.s3_bucket, self.scope))
-
-            except (os.error, IOError) as e:
-                logger.error("{0}: WAL-e archiving is disabled".format(e))
-                self.init_error = True
-
-    def replica_method(self):
-        if self.should_use_s3_to_create_replica():
-            return self.create_replica_with_s3
-        return None
-
-    def replica_fallback_method(self):
-        return self.create_replica_with_pg_basebackup
-
-    def should_use_s3_to_create_replica(self):
-        """ determine whether it makes sense to use S3 and not pg_basebackup """
-        if self.init_error:
-            return False
-
-        threshold_megabytes = self.wal_e.threshold_megabytes
-        threshold_backup_size_percentage = self.wal_e.threshold_backup_size_percentage
-
-        try:
-            latest_backup = subprocess.check_output(self.wal_e.cmd.split() + ['backup-list', '--detail', 'LATEST'],
-                                                    env=self.env)
-            # name    last_modified   expanded_size_bytes wal_segment_backup_start    wal_segment_offset_backup_start
-            #                                                   wal_segment_backup_stop wal_segment_offset_backup_stop
-            # base_00000001000000000000007F_00000040  2015-05-18T10:13:25.000Z
-            # 20310671    00000001000000000000007F    00000040
-            # 00000001000000000000007F    00000240
-            backup_strings = latest_backup.splitlines() if latest_backup else ()
-            if len(backup_strings) != 2:
-                return False
-
-            names = backup_strings[0].split()
-            vals = backup_strings[1].split()
-            if (len(names) != len(vals)) or (len(names) != 7):
-                return False
-
-            backup_info = dict(zip(names, vals))
-        except subprocess.CalledProcessError as e:
-            logger.error("could not query wal-e latest backup: {}".format(e))
-            return False
-
-        try:
-            backup_size = backup_info['expanded_size_bytes']
-            backup_start_segment = backup_info['wal_segment_backup_start']
-            backup_start_offset = backup_info['wal_segment_offset_backup_start']
-        except Exception as e:
-            logger.error("unable to get some of S3 backup parameters: {}".format(e))
-            return False
-
-        # WAL filename is XXXXXXXXYYYYYYYY000000ZZ, where X - timeline, Y - LSN logical log file,
-        # ZZ - 2 high digits of LSN offset. The rest of the offset is the provided decimal offset,
-        # that we have to convert to hex and 'prepend' to the high offset digits.
-
-        lsn_segment = backup_start_segment[8:16]
-        # first 2 characters of the result are 0x and the last one is L
-        lsn_offset = hex((long(backup_start_segment[16:32], 16) << 24) + long(backup_start_offset))[2:-1]
-
-        # construct the LSN from the segment and offset
-        backup_start_lsn = '{}/{}'.format(lsn_segment, lsn_offset)
-
-        conn = None
-        cursor = None
-        diff_in_bytes = long(backup_size)
-        try:
-            # get the difference in bytes between the current WAL location and the backup start offset
-            conn = psycopg2.connect(**(self.master_connection))
-            conn.autocommit = True
-            cursor = conn.cursor()
-            cursor.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,))
-            diff_in_bytes = long(cursor.fetchone()[0])
-        except psycopg2.Error as e:
-            logger.error('could not determine difference with the master location: {}'.format(e))
-            return False
-        finally:
-            cursor and cursor.close()
-            conn and conn.close()
-
-        # if the size of the accumulated WAL segments is more than a certan percentage of the backup size
-        # or exceeds the pre-determined size - pg_basebackup is chosen instead.
-        return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\
-               (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100)
-
-    def create_replica_with_s3(self):
-        if self.init_error:
-            return 1
-        try:
-            ret = subprocess.call(self.wal_e.cmd + ' backup-fetch {} LATEST'.format(self.data_dir), env=self.env)
-        except Exception as e:
-            logger.error('Error when fetching backup with WAL-E: {0}'.format(e))
-            return 1
-        return ret
-
-
-if __name__ == '__main__':
-    if len(sys.argv) == 5:
-        # scope, role, datadir, connstring
-        restore = WALERestore(*(sys.argv[1:]))
-        restore.setup()
-        sys.exit(restore.run())
-    sys.exit("Usage: {0} scope role datadir connstring".format(sys.argv[0]))

patroni/scripts/wale_restore.py

@@ -0,0 +1,167 @@
+#!/usr/bin/python
+
+# sample script to clone new replicas using WAL-E restore
+# falls back to pg_basebackup if WAL-E restore fails, or if
+# WAL-E backup is too far behind
+# note that pg_basebackup still expects to use restore from
+# WAL-E for transaction logs
+
+# theoretically should work with SWIFT, but not tested on it
+
+# arguments are:
+#   - cluster scope
+#   - cluster role
+#   - master connection string
+#   - number of retries
+#   - envdir for the WALE env
+# - WALE_BACKUP_THRESHOLD_MEGABYTES if WAL amount is above that - use pg_basebackup
+# - WALE_BACKUP_THRESHOLD_PERCENTAGE if WAL size exceeds a certain percentage of the
+
+# this script depends on an envdir defining the S3 bucket (or SWIFT dir),and login
+# credentials per WALE Documentation.
+
+# currently also requires that you configure the restore_command to use wal_e, example:
+#       recovery_conf:
+#               restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1
+
+from collections import namedtuple
+import logging
+import os
+import psycopg2
+import subprocess
+import sys
+import argparse
+
+
+if sys.hexversion >= 0x03000000:
+    long = int
+
+logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class WALERestore(object):
+
+    def __init__(self, scope, datadir, connstring, env_dir, threshold_mb, threshold_pct, use_iam):
+        self.scope = scope
+        self.master_connection = connstring
+        self.data_dir = datadir
+        self.wal_e = namedtuple('wale', 'dir,threshold_mb,threshold_pct,iam_string,cmd')
+        self.wal_e.dir = env_dir
+        self.wal_e.threshold_mb = threshold_mb
+        self.wal_e.threshold_pct = threshold_pct
+        self.wal_e.iam_string = ' --aws-instance-profile ' if use_iam == 1 else ''
+        self.wal_e.cmd = 'envdir {0} wal-e {1} '.format(self.wal_e.dir, self.wal_e.iam_string)
+        self.init_error = (not os.path.exists(self.wal_e.dir))
+
+    def run(self):
+        """ creates a new replica using WAL-E """
+        if not self.init_error and self.should_use_s3_to_create_replica():
+            return self.create_replica_with_s3()
+        return 2
+
+    def should_use_s3_to_create_replica(self):
+        """ determine whether it makes sense to use S3 and not pg_basebackup """
+
+        threshold_megabytes = self.wal_e.threshold_mb
+        threshold_backup_size_percentage = self.wal_e.threshold_pct
+
+        try:
+            latest_backup = subprocess.check_output(self.wal_e.cmd.split() + ['backup-list', '--detail', 'LATEST'])
+            # name    last_modified   expanded_size_bytes wal_segment_backup_start    wal_segment_offset_backup_start
+            #                                                   wal_segment_backup_stop wal_segment_offset_backup_stop
+            # base_00000001000000000000007F_00000040  2015-05-18T10:13:25.000Z
+            # 20310671    00000001000000000000007F    00000040
+            # 00000001000000000000007F    00000240
+            backup_strings = latest_backup.splitlines() if latest_backup else ()
+            if len(backup_strings) != 2:
+                return False
+
+            names = backup_strings[0].split()
+            vals = backup_strings[1].split()
+            if (len(names) != len(vals)) or (len(names) != 7):
+                return False
+
+            backup_info = dict(zip(names, vals))
+        except subprocess.CalledProcessError as e:
+            logger.error("could not query wal-e latest backup: {}".format(e))
+            return False
+
+        try:
+            backup_size = backup_info['expanded_size_bytes']
+            backup_start_segment = backup_info['wal_segment_backup_start']
+            backup_start_offset = backup_info['wal_segment_offset_backup_start']
+        except Exception as e:
+            logger.error("unable to get some of WALE backup parameters: {}".format(e))
+            return False
+
+        # WAL filename is XXXXXXXXYYYYYYYY000000ZZ, where X - timeline, Y - LSN logical log file,
+        # ZZ - 2 high digits of LSN offset. The rest of the offset is the provided decimal offset,
+        # that we have to convert to hex and 'prepend' to the high offset digits.
+
+        lsn_segment = backup_start_segment[8:16]
+        # first 2 characters of the result are 0x and the last one is L
+        lsn_offset = hex((long(backup_start_segment[16:32], 16) << 24) + long(backup_start_offset))[2:-1]
+
+        # construct the LSN from the segment and offset
+        backup_start_lsn = '{}/{}'.format(lsn_segment, lsn_offset)
+
+        conn = None
+        cursor = None
+        diff_in_bytes = long(backup_size)
+        try:
+            # get the difference in bytes between the current WAL location and the backup start offset
+            conn = psycopg2.connect(self.master_connection)
+            conn.autocommit = True
+            cursor = conn.cursor()
+            cursor.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,))
+            diff_in_bytes = long(cursor.fetchone()[0])
+        except psycopg2.Error as e:
+            logger.error('could not determine difference with the master location: {}'.format(e))
+            return False
+        finally:
+            cursor and cursor.close()
+            conn and conn.close()
+
+        # if the size of the accumulated WAL segments is more than a certan percentage of the backup size
+        # or exceeds the pre-determined size - pg_basebackup is chosen instead.
+        return (diff_in_bytes < long(threshold_megabytes) * 1048576) and\
+            (diff_in_bytes < long(backup_size) * float(threshold_backup_size_percentage) / 100)
+
+    def create_replica_with_s3(self):
+        # if we're set up, restore the replica using fetch latest
+        try:
+            ret = subprocess.call(self.wal_e.cmd.split() + ['backup-fetch', '{}'.format(self.data_dir), 'LATEST'])
+        except Exception as e:
+            logger.error('Error when fetching backup with WAL-E: {0}'.format(e))
+            return 1
+
+        return ret
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Script to image replicas using WAL-E')
+    parser.add_argument('--scope', required=True)
+    parser.add_argument('--role', required=False)
+    parser.add_argument('--datadir', required=True)
+    parser.add_argument('--connstring', required=True)
+    parser.add_argument('--retries', type=int, default=1)
+    parser.add_argument('--envdir', required=True)
+    parser.add_argument('--threshold_megabytes', type=int, default=10240)
+    parser.add_argument('--threshold_backup_size_percentage', type=int, default=30)
+    parser.add_argument('--use_iam', type=int, default=0)
+    args = parser.parse_args()
+
+    # retry cloning in a loop
+    for retry in range(0, args.retries + 1):
+        restore = WALERestore(scope=args.scope, datadir=args.datadir, connstring=args.connstring,
+                              env_dir=args.envdir, threshold_mb=args.threshold_megabytes,
+                              threshold_pct=args.threshold_backup_size_percentage, use_iam=args.use_iam)
+        ret = restore.run()
+        if ret == 0:
+            break
+
+    sys.exit(ret)
+
+if __name__ == '__main__':
+    main()

patroni/version.py

@@ -1 +1 @@
-__version__ = '0.7'
+__version__ = '0.76'

postgres0.yml

@@ -4,7 +4,7 @@ scope: &scope batman
 restapi:
   listen: 127.0.0.1:8008
   connect_address: 127.0.0.1:8008
-  auth: 'username:password'
+  #auth: 'username:password'
 #  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
 #  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
 etcd:
@@ -26,6 +26,9 @@ etcd:
 #      - host1
 #      - host2
 #      - host3
+bdr:
+  enable: "on"
+  database: "bdrtest"
 postgresql:
   name: postgresql0
   scope: *scope
@@ -68,21 +71,33 @@ postgresql:
   admin:
     username: admin
     password: admin
-  wal_e:
-    env_dir: /home/postgres/etc/wal-e.d/env
-    threshold_megabytes: 10240
-    threshold_backup_size_percentage: 30
-  restore: patroni/scripts/restore.py
+  create_replica_method:
+    - basebackup
+#   - wal_e
+# commented-out example for wal-e provisioning
+  #create_replica_method: wal_e, basebackup
+  #wal_e:
+    #command: /patroni/scripts/wale_restore.py
+    #env_dir: /etc/wal-e.d/env
+    #threshold_megabytes: 10240
+    #threshold_backup_size_percentage: 30
+    #retries: 2
+    #use_iam: 1
   #recovery_conf:
-    #restore_command: cp ../wal_archive/%f %p
+    #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1
+  recovery_conf:
+    restore_command: cp ../wal_archive/%f %p
   parameters:
     archive_mode: "on"
-    wal_level: hot_standby
-    archive_command: mkdir -p ../wal_archive && cp %p ../wal_archive/%f
-    max_wal_senders: 5
+    wal_level: logical
+    archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
+    max_wal_senders: 10
     wal_keep_segments: 8
     archive_timeout: 1800s
-    max_replication_slots: 5
+    max_replication_slots: 10
+    max_worker_processes: 10
+    track_commit_timestamp: 'on'
+    shared_preload_libraries: 'bdr'
     hot_standby: "on"
     wal_log_hints: "on"
 tags:

postgres1.yml

@@ -4,7 +4,7 @@ scope: &scope batman
 restapi:
   listen: 127.0.0.1:8009
   connect_address: 127.0.0.1:8009
-  auth: 'username:password'
+  #auth: 'username:password'
 #  certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
 #  keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
 etcd:
@@ -26,6 +26,9 @@ etcd:
 #      - host1
 #      - host2
 #      - host3
+bdr:
+    enable: "on"
+    database: "bdrtest"
 postgresql:
   name: postgresql1
   scope: *scope
@@ -68,21 +71,33 @@ postgresql:
   admin:
     username: admin
     password: admin
+# commented-out example for wal-e provisioning
+  create_replica_method:
+    - basebackup
+#   - wal_e
+# commented-out example for wal-e provisioning
+  #wal_e:
+    #command: /patroni/scripts/wale_restore.py
+    #env_dir: /home/postgres/etc/wal-e.d/env
+    #threshold_megabytes: 10240
+    #threshold_backup_size_percentage: 30
+    #retries: 2
+    #use_iam: 1
   #recovery_conf:
-    #restore_command: cp ../wal_archive/%f %p
-  wal_e:
-    env_dir: /home/postgres/etc/wal-e.d/env
-    threshold_megabytes: 10240
-    threshold_backup_size_percentage: 30
-  restore: patroni/scripts/restore.py
+    #restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1
+  recovery_conf:
+    restore_command: cp ../wal_archive/%f %p
   parameters:
     archive_mode: "on"
-    wal_level: hot_standby
-    archive_command: mkdir -p ../wal_archive && cp %p ../wal_archive/%f
-    max_wal_senders: 5
+    wal_level: logical
+    archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
+    max_wal_senders: 10
+    max_worker_processes: 10
+    track_commit_timestamp: 'on'
+    shared_preload_libraries: 'bdr'
     wal_keep_segments: 8
     archive_timeout: 1800s
-    max_replication_slots: 5
+    max_replication_slots: 10
     hot_standby: "on"
     wal_log_hints: "on"
 tags:

setup.py

@@ -32,7 +32,7 @@ VERSION = read_version(MAIN_PACKAGE)
 DESCRIPTION = 'PostgreSQL High-Available orchestrator and CLI'
 LICENSE = 'The MIT License'
 URL = 'https://github.com/zalando/patroni'
-AUTHOR = 'Alexander Kukushkin, Alexey Klyukin, Feike Steenbergen'
+AUTHOR = 'Alexander Kukushkin, Oleksii Kliukin, Feike Steenbergen'
 AUTHOR_EMAIL = '[email protected], [email protected], [email protected]'
 KEYWORDS = 'etcd governor patroni postgresql postgres ha zookeeper streaming replication'
 
@@ -56,7 +56,10 @@ CLASSIFIERS = [
     'Programming Language :: Python :: Implementation :: CPython',
 ]
 
-CONSOLE_SCRIPTS = ['patroni = patroni:main', 'patronictl = patroni.ctl:ctl']
+CONSOLE_SCRIPTS = ['patroni = patroni:main',
+                   'patronictl = patroni.ctl:ctl',
+                   "patroni_wale_restore = patroni.scripts.wale_restore:main",
+                   "patroni_aws = patroni.scripts.aws:main"]
 
 
 class PyTest(TestCommand):

tests/test_ha.py

@@ -88,6 +88,7 @@ class MockPatroni:
         self.api = Mock()
         self.tags = {}
         self.nofailover = None
+        self.bdr = {}
         self.api.connection_string = 'http://127.0.0.1:8008'
 
 

tests/test_postgresql.py

@@ -276,9 +276,23 @@ class TestPostgresql(unittest.TestCase):
         self.assertTrue(self.p.can_rewind)
         self.p.controldata = tmp
 
+    @patch('time.sleep', Mock())
     def test_create_replica(self):
         self.p.delete_trigger_file = Mock(side_effect=OSError())
-        self.assertEquals(self.p.create_replica({'host': '', 'port': '', 'user': ''}, ''), 1)
+        with patch('subprocess.call', Mock(side_effect=[1, 0])):
+            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+        with patch('subprocess.call', Mock(side_effect=[Exception(), 0])):
+            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+
+        self.p.config['create_replica_method'] = ['wale', 'basebackup']
+        self.p.config['wale'] = {'command': 'foo'}
+        with patch('subprocess.call', Mock(return_value=0)):
+            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+            del self.p.config['wale']
+            self.assertEquals(self.p.create_replica(self.leader, ''), 0)
+
+        with patch('subprocess.call', Mock(side_effect=Exception("foo"))):
+            self.assertEquals(self.p.create_replica(self.leader, ''), 1)
 
     def test_create_connection_users(self):
         cfg = self.p.config
@@ -296,9 +310,9 @@ class TestPostgresql(unittest.TestCase):
 
     @patch.object(MockConnect, 'closed', 2)
     def test__query(self):
-        self.assertRaises(PostgresConnectionException, self.p._query, 'blabla')
+        self.assertRaises(PostgresConnectionException, self.p._query, 'blabla', 'postgres')
         self.p._state = 'restarting'
-        self.assertRaises(RetryFailedError, self.p._query, 'blabla')
+        self.assertRaises(RetryFailedError, self.p._query, 'blabla', 'postgres')
 
     def test_query(self):
         self.p.query('select 1')

tests/test_restore.py

@@ -1,111 +0,0 @@
-import unittest
-from mock import MagicMock, patch
-import os
-from patroni.scripts.restore import Restore, WALERestore
-
-
-def fake_cursor_fetchone(*args, **kwargs):
-    return ('16777216',)
-
-
-def fake_call_fail_for_wal_e(*args, **kwargs):
-    if len(args) > 0 and 'backup-fetch' in args[0]:
-        return 1
-    return 0
-
-
-def fake_call_fail_for_base_backup(*args, **kwargs):
-    if len(args) > 0 and 'backup-fetch' in args[0]:
-        return 0
-    return 1
-
-
-def fake_backup_data(self, *args, **kwargs):
-    """ return the fake result of WAL-E backup-list"""
-    return """name    last_modified   expanded_size_bytes wal_segment_backup_start    wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop
-base_00000001000000000000007F_00000040  2015-05-18T10:13:25.000Z 167772160   00000001000000000000007F    00000040 00000001000000000000007F    00000240
-"""
-
-
-class TestRestore(unittest.TestCase):
-
-    def setUp(self):
-        self.restore = Restore("batman", "master", "/data", "host=batman port=5432 user=batman")
-        pass
-
-    def tearDown(self):
-        pass
-
-    def test_parse_connstring(self):
-        self.assertDictEqual(self.restore.master_connection, {'host': 'batman', 'port': '5432', 'user': 'batman'})
-
-    @patch('subprocess.call', MagicMock(return_value=0))
-    def test_run(self):
-        ret = self.restore.run()
-        self.assertEqual(ret, 0)
-
-    @patch('subprocess.call', MagicMock(return_value=1))
-    def test_run_fail(self):
-        ret = self.restore.run()
-        self.assertEqual(ret, 1)
-
-
-@patch('os.access', MagicMock(return_value=True))
-@patch('os.makedirs', MagicMock(return_value=True))
-@patch('os.path.exists', MagicMock(return_value=True))
-@patch('os.path.isdir', MagicMock(return_value=True))
-@patch('psycopg2.extensions.cursor.fetchone', MagicMock(side_effect=fake_cursor_fetchone))
-@patch('psycopg2.extensions.cursor', MagicMock(autospec=True))
-@patch('psycopg2.extensions.connection', MagicMock(autospec=True))
-@patch('psycopg2.connect', MagicMock(autospec=True))
-@patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data))
-class TestWALERestore(unittest.TestCase):
-
-    def setUp(self):
-        env = {}
-        env['WAL_S3_BUCKET'] = 'batman'
-        env['WALE_BACKUP_THRESHOLD_PERCENTAGE'] = 100
-        env['WALE_BACKUP_THRESHOLD_MEGABYTES'] = 100
-        self.wale_restore = WALERestore("batman", "master", "/data", "host=batman port=5432 user=batman", env=env)
-
-    def tearDown(self):
-        pass
-
-    def test_setup(self):
-        self.wale_restore.setup()
-        self.assertFalse(self.wale_restore.init_error)
-
-    # have to redefine the class-level os.access mock inside the function
-    # since the class-level mock will be applied after the function level one.
-    @patch('os.access', return_value=False)
-    def test_setup_fail(self, mock_no_access):
-        os.access = mock_no_access
-        self.wale_restore.setup()
-        self.assertTrue(self.wale_restore.init_error)
-
-    # The 3 tests above only differ with the mock function instead of a subprocess call
-    # in the first one, subprocess call should return success only for wal-e command,
-    # checking the primary use-case of restoring from WAL-E backup.
-    # In the second one, we test fallbacks by failing at WAL-E, but succeeding at
-    # pg_basebackup.
-    # Finally, the last use case is when all subprocess.call fails. resulting in a
-    # failure to restore from replica
-    @patch('subprocess.call',
-           MagicMock(side_effect=lambda *args, **kwargs: 0 if 'wal-e' in args[0] else 1))
-    def test_run(self):
-        self.wale_restore.setup()
-        ret = self.wale_restore.run()
-        self.assertEqual(ret, 0)
-
-    @patch('subprocess.call',
-           MagicMock(side_effect=lambda *args, **kwargs: 0 if 'pg_basebackup' in args[0] else 1))
-    def test_run_fallback(self):
-        self.wale_restore.setup()
-        ret = self.wale_restore.run()
-        self.assertEqual(ret, 0)
-
-    @patch('subprocess.call', MagicMock(return_value=1))
-    def test_run_all_fail(self):
-        self.wale_restore.setup()
-        ret = self.wale_restore.run()
-        self.assertEqual(ret, 1)

tests/test_wale_restore.py

@@ -0,0 +1,91 @@
+import unittest
+from mock import MagicMock, patch, PropertyMock
+import os
+import psycopg2
+import subprocess
+from patroni.scripts.wale_restore import WALERestore
+
+
+def fake_cursor_fetchone(*args, **kwargs):
+    return ('16777216',)
+
+
+def fake_call_fail_for_wal_e(*args, **kwargs):
+    if len(args) > 0 and 'backup-fetch' in args[0]:
+        return 1
+    return 0
+
+
+def fake_call_fail_for_base_backup(*args, **kwargs):
+    if len(args) > 0 and 'backup-fetch' in args[0]:
+        return 0
+    return 1
+
+
+def fake_backup_data(self, *args, **kwargs):
+    """ return the fake result of WAL-E backup-list"""
+    return """name    last_modified   expanded_size_bytes wal_segment_backup_start    wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop
+base_00000001000000000000007F_00000040  2015-05-18T10:13:25.000Z 167772160   00000001000000000000007F    00000040 00000001000000000000007F    00000240
+"""
+
+def fake_backup_data_2(self, *args, **kwargs):
+    """ return the fake result of WAL-E backup-list"""
+    return """name    last_modified   expanded_size_bytes wal_segment_backup_start    wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop """
+
+def fake_backup_data_3(self, *args, **kwargs):
+    """ return the fake result of WAL-E backup-list"""
+    return """name    last_modified   expanded_size_bytes wal_segment_backup_start    wal_segment_offset_backup_start wal_segment_backup_stop
+base_00000001000000000000007F_00000040  2015-05-18T10:13:25.000Z 167772160   00000001000000000000007F    00000040 00000001000000000000007F    00000240
+"""
+
+def fake_backup_data_4(self, *args, **kwargs):
+    """ return the fake result of WAL-E backup-list"""
+    return """name    last_modified   expanded_size_foo wal_segment_backup_start    wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop
+base_00000001000000000000007F_00000040  2015-05-18T10:13:25.000Z 167772160   00000001000000000000007F    00000040 00000001000000000000007F    00000240
+"""
+
+
+@patch('os.access', MagicMock(return_value=True))
+@patch('os.makedirs', MagicMock(return_value=True))
+@patch('os.path.exists', MagicMock(return_value=True))
+@patch('os.path.isdir', MagicMock(return_value=True))
+@patch('psycopg2.extensions.cursor.fetchone', MagicMock(side_effect=fake_cursor_fetchone))
+@patch('psycopg2.extensions.cursor', MagicMock(autospec=True))
+@patch('psycopg2.extensions.connection', MagicMock(autospec=True))
+@patch('psycopg2.connect', MagicMock(autospec=True))
+@patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data))
+class TestWALERestore(unittest.TestCase):
+
+    def setUp(self):
+        self.wale_restore = WALERestore("batman", "/data",
+                                        "host=batman port=5432 user=batman", "/etc", 100, 100, 1)
+
+    def tearDown(self):
+        pass
+
+    def test_should_use_s3_to_create_replica(self):
+        with patch('psycopg2.connect', MagicMock(side_effect=psycopg2.Error("foo"))):
+            self.assertFalse(self.wale_restore.should_use_s3_to_create_replica())
+        with patch('subprocess.check_output', MagicMock(side_effect=subprocess.CalledProcessError(1, "cmd", "foo"))):
+            self.assertFalse(self.wale_restore.should_use_s3_to_create_replica())
+        with patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data_2)):
+            self.assertFalse(self.wale_restore.should_use_s3_to_create_replica())
+        with patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data_3)):
+            self.assertFalse(self.wale_restore.should_use_s3_to_create_replica())
+        with patch('subprocess.check_output', MagicMock(side_effect=fake_backup_data_4)):
+            self.assertFalse(self.wale_restore.should_use_s3_to_create_replica())
+
+        self.wale_restore.should_use_s3_to_create_replica()
+
+    def test_create_replica_with_s3(self):
+        with patch('subprocess.call', MagicMock(return_value=0)):
+            self.assertEqual(self.wale_restore.create_replica_with_s3(), 0)
+        with patch('subprocess.call', MagicMock(side_effect=Exception("foo"))):
+            self.assertEqual(self.wale_restore.create_replica_with_s3(), 1)
+
+    def test_run(self):
+        with patch.object(self.wale_restore, 'init_error', PropertyMock(return_value=True)):
+            self.assertEqual(self.wale_restore.run(), 2)
+        with patch.object(self.wale_restore, 'should_use_s3_to_create_replica', MagicMock(return_value=True)):
+            with patch.object(self.wale_restore, 'create_replica_with_s3', MagicMock(return_value=0)):
+                self.assertEqual(self.wale_restore.run(), 0)