Merge branch 'master' of github.com:zalando/governor

helpers/etcd.py

@@ -47,7 +47,7 @@ class Etcd:
     def put_client_path(self, path, **data):
         try:
             response = requests.put(self.client_url(path), data=data)
-            return response.status_code in [200, 201]
+            return response.status_code in [200, 201, 202, 204]
         except:
             logger.exception('PUT %s data=%s', path, data)
             return False
@@ -55,7 +55,7 @@ class Etcd:
     def delete_client_path(self, path):
         try:
             response = requests.delete(self.client_url(path))
-            return response.status_code == 204
+            return response.status_code in [200, 202, 204]
         except:
             logger.exception('DELETE %s', path)
             return False

helpers/postgresql.py

@@ -1,6 +1,8 @@
 import logging
 import os
 import psycopg2
+import shutil
+import subprocess
 import sys
 import time
 
@@ -42,7 +44,13 @@ class Postgresql:
         self.superuser = config['superuser']
         self.admin = config['admin']
         self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf')
+        self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'),
+                                      os.path.join(self.data_dir, 'postgresql.conf'))
         self._pg_ctl = 'pg_ctl -w -D ' + self.data_dir
+        self.wal_e = config.get('wal_e', None)
+        if self.wal_e:
+            self.wal_e_path = 'envdir {} wal-e --aws-instance-profile '.\
+                               format(self.wal_e.get('env_dir', '/home/postgres/etc/wal-e.d/env'))
 
         self.config = config
 
@@ -101,11 +109,98 @@ class Postgresql:
 
         try:
             os.environ['PGPASSFILE'] = pgpass
-            return os.system('pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format(
-                data_dir=self.data_dir, **r)) == 0
+            return self.create_replica(leader.address, r) == 0
         finally:
             os.environ.pop('PGPASSFILE')
 
+    def create_replica(self, master_connurl, master_connection):
+        """ creates a new replica using either pg_basebackup or WAL-E """
+        if self.should_use_s3_to_create_replica(master_connurl):
+            result = self.create_replica_with_s3()
+            # if restore from the backup on S3 failed - try with the pg_basebackup
+            if result == 0:
+                return result
+        return self.create_replica_with_pg_basebackup(master_connection)
+
+    def create_replica_with_pg_basebackup(self, master_connection):
+        return os.system('pg_basebackup -R -D {data_dir} --host={hostname} --port={port} -U {username}'.format(
+                data_dir=self.data_dir, **master_connection))
+
+    def create_replica_with_s3(self):
+        if not self.wal_e or not self.wal_e_path:
+            return 1
+
+        ret = os.system(self.wal_e_path + ' backup-fetch {} LATEST'.format(self.data_dir))
+        self.restore_configuration_files()
+        return ret
+
+    def should_use_s3_to_create_replica(self, master_connurl):
+        """ determine whether it makes sense to use S3 and not pg_basebackup """
+        if not self.wal_e or not self.wal_e_path:
+            return False
+
+        threshold_megabytes = self.wal_e.get('threshold_megabytes', 10240)
+        threshold_backup_size_percentage = self.wal_e.get('threshold_backup_size_percentage', 30)
+
+        try:
+            latest_backup = subprocess.check_output(self.wal_e_path.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 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(master_connurl)
+            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 is_leader(self):
         return not self.query('SELECT pg_is_in_recovery()').fetchone()[0]
 
@@ -125,6 +220,7 @@ class Postgresql:
 
         ret = os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0
         ret and self.load_replication_slots()
+        self.save_configuration_files()
         return ret
 
     def stop(self):
@@ -219,6 +315,22 @@ primary_conninfo = '{}'
         self.write_recovery_conf(leader)
         self.restart()
 
+    def save_configuration_files(self):
+        """
+            copy postgresql.conf to postgresql.conf.backup to preserve it in the WAL-e backup.
+            see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239
+        """
+        for f in self.configuration_to_save:
+            shutil.copy(f, f+'.backup')
+
+    def restore_configuration_files(self):
+        """ restore a previously saved postgresql.conf """
+        try:
+            for f in self.configuration_to_save:
+                shutil.copy(f+'.backup', f)
+        except Exception as e:
+            logger.error("unable to restore configuration from WAL-E backup: {}".format(e))
+
     def promote(self):
         return os.system(self._pg_ctl + ' promote') == 0
 

postgres0.yml

@@ -19,6 +19,10 @@ 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
   #recovery_conf:
     #restore_command: cp ../wal_archive/%f %p
   parameters:

postgres1.yml

@@ -21,6 +21,10 @@ postgresql:
     password: admin
   #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
   parameters:
     archive_mode: "on"
     wal_level: hot_standby

setup.py

@@ -12,8 +12,8 @@ import setuptools
 from setuptools.command.test import test as TestCommand
 from setuptools import setup
 
-if sys.version_info < (2, 6, 0):
-    sys.stderr.write('FATAL: governor needs to be run with Python 2.6+\n')
+if sys.version_info < (2, 7, 0):
+    sys.stderr.write('FATAL: governor needs to be run with Python 2.7+\n')
     sys.exit(1)
 
 __location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect.currentframe())))
@@ -25,9 +25,6 @@ HELPERS = 'helpers'
 VERSION = '0.1'
 DESCRIPTION = 'A Template for PostgreSQL HA with etcd'
 LICENSE = 'The MIT License'
-URL = 'https://github.com/zalando/governor'
-AUTHOR = 'Alexander Kukushkin'
-EMAIL = '[email protected]'
 
 COVERAGE_XML = True
 COVERAGE_HTML = False
@@ -43,17 +40,12 @@ CLASSIFIERS = [
     'License :: OSI Approved :: The MIT License',
     'Operating System :: POSIX :: Linux',
     'Programming Language :: Python',
-    'Programming Language :: Python :: 2',
-    'Programming Language :: Python :: 2.6',
     'Programming Language :: Python :: 2.7',
-    'Programming Language :: Python :: 3',
     'Programming Language :: Python :: 3.3',
     'Programming Language :: Python :: 3.4',
     'Programming Language :: Python :: Implementation :: CPython',
 ]
 
-CONSOLE_SCRIPTS = ['governor = governor:main']
-
 
 class PyTest(TestCommand):
 
@@ -102,18 +94,6 @@ def read(fname):
     return open(os.path.join(__location__, fname)).read()
 
 
-def check_deps(deps):
-    '''check dependency licenses'''
-    from pkg_resources import Requirement
-    import requests
-    for dep in deps:
-        dep = Requirement.parse(dep)
-        url = 'https://pypi.python.org/pypi/{}/json'.format(dep.project_name)
-        r = requests.get(url)
-        data = r.json()
-        print(data['info'].get('name'), data['info'].get('license'))
-
-
 def setup_package():
     # Assemble additional setup commands
     cmdclass = {}
@@ -124,8 +104,6 @@ def setup_package():
 
     install_reqs = get_install_requirements('requirements.txt')
 
-    # check_deps(install_reqs)
-
     command_options = {'test': {'test_suite': ('setup.py', 'tests')}}
     if JUNIT_XML:
         command_options['test']['junitxml'] = 'setup.py', 'junit.xml'
@@ -137,12 +115,9 @@ def setup_package():
     setup(
         name=NAME,
         version=version,
-        url=URL,
         description=DESCRIPTION,
-        author=AUTHOR,
-        author_email=EMAIL,
         license=LICENSE,
-        keywords='aws docker ec2 elb lb boto deployment route53 stack traffic',
+        keywords='etcd governor postgresql postgres ha',
         long_description=read('README.md'),
         classifiers=CLASSIFIERS,
         test_suite='tests',
@@ -153,7 +128,6 @@ def setup_package():
         cmdclass=cmdclass,
         tests_require=['pytest-cov', 'pytest'],
         command_options=command_options,
-        entry_points={'console_scripts': CONSOLE_SCRIPTS},
     )