From aa5a72fb3748aa1ecd8cf6e488661dd4b3108ed0 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 9 Jul 2015 15:45:10 +0200 Subject: [PATCH 01/27] Add on_start, on_stop, on_role_change, on_restart and on_reload callbacks. --- helpers/postgresql.py | 33 ++++++++++++++++++++++++++++++--- 1 file changed, 30 insertions(+), 3 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 560d7a98..a7c2f742 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -1,6 +1,7 @@ import logging import os import psycopg2 +import shlex import shutil import subprocess import sys @@ -42,6 +43,7 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] + self.callback = config.get(['callback'], {}) 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')) @@ -238,6 +240,16 @@ class Postgresql: def is_running(self): return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0 + def call_nowait(self, cb_name): + """ pick a callback command and call it without waiting for it to finish """ + if not self.callback or cb_name not in self.callback: + return False + cmd = self.callback[cb_name] + is_master = self.is_leader() + name = self.name + subprocess.Popen(shlex.split(cmd)+[is_master, name]) + return True + def start(self): if self.is_running(): self.load_replication_slots() @@ -251,18 +263,29 @@ class Postgresql: ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0 ret and self.load_replication_slots() self.save_configuration_files() + if ret and 'on_start' in self.callback: + self.call_nowait('on_start') if self.on_change_callback: self.on_change_callback('replica' if os.path.exists(self.recovery_conf) else 'master') return ret def stop(self): - return subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) != 0 + ret = subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) + if ret == 0 and 'on_stop' in self.callback: + self.call_nowait('on_stop') + return ret != 0 def reload(self): - return subprocess.call(self._pg_ctl + ['reload']) == 0 + ret = subprocess.call(self._pg_ctl + ['reload']) + if ret == 0 and 'on_reload' in self.callback: + self.call_nowait('on_reload') + return ret == 0 def restart(self): - return subprocess.call(self._pg_ctl + ['restart', '-m', 'fast']) == 0 + ret = subprocess.call(self._pg_ctl + ['restart', '-m', 'fast']) + if ret == 0 and 'on_restart' in self.callback: + self.call_nowait('on_restart') + return ret == 0 def server_options(self): options = "--listen_addresses='{}' --port={}".format(self.listen_addresses, self.port) @@ -349,6 +372,8 @@ primary_conninfo = '{}' if not self.check_recovery_conf(leader): self.write_recovery_conf(leader) self.restart() + if self.on_change_callback['on_role_change']: + self.call_nowait('on_role_change') if self.on_change_callback: self.on_change_callback('replica') @@ -370,6 +395,8 @@ primary_conninfo = '{}' def promote(self): self.is_promoted = subprocess.call(self._pg_ctl + ['promote']) == 0 + if self.is_promoted and self.on_change_callback['on_role_change']: + self.call_nowait('on_role_change') if self.on_change_callback: self.on_change_callback('master') return self.is_promoted From 2c1627c2f3aec4befc535c0239097ffccaf89671 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 10 Jul 2015 10:33:07 +0200 Subject: [PATCH 02/27] get rid of yaml module, since AWS instance-identity natively sends JSON. --- helpers/aws.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/helpers/aws.py b/helpers/aws.py index b47e4e0f..26c65e89 100644 --- a/helpers/aws.py +++ b/helpers/aws.py @@ -1,8 +1,6 @@ import logging -import re import requests from requests.exceptions import RequestException -import yaml import boto.ec2 logger = logging.getLogger(__name__) @@ -27,7 +25,7 @@ class AWSConnection: return if r.ok: try: - content = yaml.load(r.content) + content = r.json() self.instance_id = content['instanceId'] self.region = content['region'] except Exception as e: From 78f0b16543541ce20b42c0ab97878b8b57b47d17 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 10 Jul 2015 10:57:24 +0200 Subject: [PATCH 03/27] Enable aws.py to run as a stand-alone program. --- helpers/aws.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/helpers/aws.py b/helpers/aws.py index 26c65e89..02ddd7fe 100644 --- a/helpers/aws.py +++ b/helpers/aws.py @@ -1,6 +1,7 @@ import logging import requests from requests.exceptions import RequestException +import sys import boto.ec2 logger = logging.getLogger(__name__) @@ -67,3 +68,15 @@ class AWSConnection: def on_role_change(self, new_role): ret = self._tag_ec2(new_role) return self._tag_ebs(new_role) and ret + + +if __name__ == '__main__': + if len(sys.argv) != 3: + print ("Usage: {0} action role name".format(sys.argv[0])) + return 1 + action, role, name = sys.argv[1:] + if action in ('on_start', 'on_stop', 'on_role_change'): + aws = gAWSConnection({'cluster_name': name}) + aws.on_role_change(role) + return 0 + return 2 From c0cc59093820bd134935e508281e67e76647df6b Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 10 Jul 2015 12:58:37 +0200 Subject: [PATCH 04/27] Remove usage of AWS.py from Patroni Functionality provided by that module will be achieved via callbacks. - fix a typo in aws.py - fix some mixups of old code and new callbacks in postgresql.py --- helpers/aws.py | 2 +- helpers/postgresql.py | 13 +++---------- patroni.py | 2 -- 3 files changed, 4 insertions(+), 13 deletions(-) diff --git a/helpers/aws.py b/helpers/aws.py index 02ddd7fe..6a04da92 100644 --- a/helpers/aws.py +++ b/helpers/aws.py @@ -76,7 +76,7 @@ if __name__ == '__main__': return 1 action, role, name = sys.argv[1:] if action in ('on_start', 'on_stop', 'on_role_change'): - aws = gAWSConnection({'cluster_name': name}) + aws = AWSConnection({'cluster_name': name}) aws.on_role_change(role) return 0 return 2 diff --git a/helpers/postgresql.py b/helpers/postgresql.py index a7c2f742..5dff6578 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -35,7 +35,7 @@ def parseurl(url): class Postgresql: - def __init__(self, config, on_change_callback=None): + def __init__(self, config): self.config = config self.name = config['name'] self.listen_addresses, self.port = config['listen'].split(':') @@ -66,7 +66,6 @@ class Postgresql: self._connection = None self._cursor_holder = None self.members = [] # list of already existing replication slots - self.on_change_callback = on_change_callback def get_local_address(self): listen_addresses = self.listen_addresses.split(',') @@ -265,8 +264,6 @@ class Postgresql: self.save_configuration_files() if ret and 'on_start' in self.callback: self.call_nowait('on_start') - if self.on_change_callback: - self.on_change_callback('replica' if os.path.exists(self.recovery_conf) else 'master') return ret def stop(self): @@ -372,10 +369,8 @@ primary_conninfo = '{}' if not self.check_recovery_conf(leader): self.write_recovery_conf(leader) self.restart() - if self.on_change_callback['on_role_change']: + if 'on_role_change' in self.callback: self.call_nowait('on_role_change') - if self.on_change_callback: - self.on_change_callback('replica') def save_configuration_files(self): """ @@ -395,10 +390,8 @@ primary_conninfo = '{}' def promote(self): self.is_promoted = subprocess.call(self._pg_ctl + ['promote']) == 0 - if self.is_promoted and self.on_change_callback['on_role_change']: + if self.is_promoted and 'on_role_change' in self.callback: self.call_nowait('on_role_change') - if self.on_change_callback: - self.on_change_callback('master') return self.is_promoted def demote(self, leader): diff --git a/patroni.py b/patroni.py index d90e5c64..6f06a056 100755 --- a/patroni.py +++ b/patroni.py @@ -6,7 +6,6 @@ import time import yaml from helpers.api import RestApiServer -from helpers.aws import AWSConnection from helpers.etcd import Etcd from helpers.ha import Ha from helpers.postgresql import Postgresql @@ -18,7 +17,6 @@ class Patroni: def __init__(self, config): self.nap_time = config['loop_wait'] - self.aws = AWSConnection(config) self.postgresql = Postgresql(config['postgresql']) self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config)) host, port = config['restapi']['listen'].split(':') From 6d0a9bbdc0030734f415e61801bc40928735590f Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 13 Jul 2015 10:25:42 +0200 Subject: [PATCH 05/27] create a new directory for callback scripts and move aws volume tags there. --- {helpers => scripts}/aws.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename {helpers => scripts}/aws.py (100%) diff --git a/helpers/aws.py b/scripts/aws.py similarity index 100% rename from helpers/aws.py rename to scripts/aws.py From f734272f3f437bf0c38a92ff0183e4b2340bdd5a Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 13 Jul 2015 11:45:25 +0200 Subject: [PATCH 06/27] Process the command paths relative to the patroni dir correctly, fix the error of not supplying a callback action to the actual command. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 5dff6578..de73ee93 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -246,7 +246,7 @@ class Postgresql: cmd = self.callback[cb_name] is_master = self.is_leader() name = self.name - subprocess.Popen(shlex.split(cmd)+[is_master, name]) + subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, is_master, name]) return True def start(self): From 15f94f42241d46b54cca38376fd1f927c28506e8 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 13 Jul 2015 11:47:55 +0200 Subject: [PATCH 07/27] Use actual role name and not boolean flag if it is a master to tag AWS objects (EBS volumes and EC2 instances). --- scripts/aws.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/aws.py b/scripts/aws.py index 6a04da92..7930323e 100644 --- a/scripts/aws.py +++ b/scripts/aws.py @@ -74,9 +74,9 @@ if __name__ == '__main__': if len(sys.argv) != 3: print ("Usage: {0} action role name".format(sys.argv[0])) return 1 - action, role, name = sys.argv[1:] + action, is_master, name = sys.argv[1:] if action in ('on_start', 'on_stop', 'on_role_change'): aws = AWSConnection({'cluster_name': name}) - aws.on_role_change(role) + aws.on_role_change('master' if is_master else 'replica') return 0 return 2 From d8aa1ff849989979e8bcfae3ae3b0e1f58237513 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Tue, 14 Jul 2015 09:22:49 +0200 Subject: [PATCH 08/27] fix a typo. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index de73ee93..41bc13c8 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -43,7 +43,7 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] - self.callback = config.get(['callback'], {}) + self.callback = config.get('callback', {}) 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')) From 11004f8c48062ebd3dea0bccda8985975e0217b7 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 17 Jul 2015 11:19:05 +0200 Subject: [PATCH 09/27] Be more robust on callbacks when cluster is down. For 'on_stop' or 'on_restart' callbacks we cannot call is_leader, because the cluster would be stopped. Instead, call is_leader before running a callback. In addition, make the call_nowait more robust by handling the exceptions related to the DB cluster unavailability. --- helpers/postgresql.py | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 41bc13c8..4d72fcec 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -239,14 +239,19 @@ class Postgresql: def is_running(self): return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0 - def call_nowait(self, cb_name): + def call_nowait(self, cb_name, is_leader=None): """ pick a callback command and call it without waiting for it to finish """ if not self.callback or cb_name not in self.callback: return False cmd = self.callback[cb_name] - is_master = self.is_leader() + if not is_leader: + try: + is_leader = self.is_leader() + except psycopg2.OperationalError as e: + logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e)) + return False name = self.name - subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, is_master, name]) + subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, is_leader, name]) return True def start(self): @@ -267,9 +272,14 @@ class Postgresql: return ret def stop(self): + try: + is_leader = self.is_leader() + except: + is_leader = None + pass ret = subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) if ret == 0 and 'on_stop' in self.callback: - self.call_nowait('on_stop') + self.call_nowait('on_stop', is_leader=is_leader) return ret != 0 def reload(self): @@ -279,9 +289,14 @@ class Postgresql: return ret == 0 def restart(self): + try: + is_leader = self.is_leader() + except: + is_leader = None + pass ret = subprocess.call(self._pg_ctl + ['restart', '-m', 'fast']) if ret == 0 and 'on_restart' in self.callback: - self.call_nowait('on_restart') + self.call_nowait('on_restart', is_leader=is_leader) return ret == 0 def server_options(self): From 33b5e1b22e95a66f55819defc443ab4fa14f502f Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 17 Jul 2015 15:22:21 +0200 Subject: [PATCH 10/27] Make sure is_leader is False does not trigger the leader re-check. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 4d72fcec..3fcb44da 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -244,7 +244,7 @@ class Postgresql: if not self.callback or cb_name not in self.callback: return False cmd = self.callback[cb_name] - if not is_leader: + if is_leader is None: try: is_leader = self.is_leader() except psycopg2.OperationalError as e: From a11d94aff53d00595e0e8c6c598a5af84a48d0e5 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 17 Jul 2015 15:39:34 +0200 Subject: [PATCH 11/27] Pass text names of cluster roles to the script. Seems better than passing a bool flag and converting it from/to string. --- helpers/postgresql.py | 2 +- scripts/aws.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 3fcb44da..371c187a 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -251,7 +251,7 @@ class Postgresql: logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e)) return False name = self.name - subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, is_leader, name]) + subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, "master" if is_leader else "replica", name]) return True def start(self): diff --git a/scripts/aws.py b/scripts/aws.py index 7930323e..6a04da92 100644 --- a/scripts/aws.py +++ b/scripts/aws.py @@ -74,9 +74,9 @@ if __name__ == '__main__': if len(sys.argv) != 3: print ("Usage: {0} action role name".format(sys.argv[0])) return 1 - action, is_master, name = sys.argv[1:] + action, role, name = sys.argv[1:] if action in ('on_start', 'on_stop', 'on_role_change'): aws = AWSConnection({'cluster_name': name}) - aws.on_role_change('master' if is_master else 'replica') + aws.on_role_change(role) return 0 return 2 From e9934be9aa80f04990eb07eb6708c50422cff4aa Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 17 Jul 2015 17:26:28 +0200 Subject: [PATCH 12/27] Do not terminate on callback failure. Show the error if a callback command fails and move on. Set the executable bit for the aws callback. --- helpers/postgresql.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 371c187a..0c31ddac 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -251,7 +251,12 @@ class Postgresql: logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e)) return False name = self.name - subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, "master" if is_leader else "replica", name]) + try: + role = "master" if is_leader else "replica" + subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, role, name]) + except Exception as e: + logger.warning("callback {0} {1} {2} {3} failed: {4}".format(os.path.abspath(cmd), cb_name, role, name, e)) + return False return True def start(self): From f1d76611b6b7c2f1b1a15c72976e5c1029f3b4a8 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 17 Jul 2015 17:26:52 +0200 Subject: [PATCH 13/27] Use the executable bit on AWS callback. --- scripts/aws.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) mode change 100644 => 100755 scripts/aws.py diff --git a/scripts/aws.py b/scripts/aws.py old mode 100644 new mode 100755 From c9c678fb5b8cbd80b806302d5d63159d1c31f922 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 23 Jul 2015 16:16:22 +0200 Subject: [PATCH 14/27] Pass scope, the cluster name, instead of the name, which is the instance name, to the callback. --- helpers/postgresql.py | 3 ++- scripts/aws.py | 10 ++++++---- 2 files changed, 8 insertions(+), 5 deletions(-) mode change 100755 => 100644 scripts/aws.py diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 0c31ddac..6348c3ff 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -38,6 +38,7 @@ class Postgresql: def __init__(self, config): self.config = config self.name = config['name'] + self.scope = config['scope'] self.listen_addresses, self.port = config['listen'].split(':') self.data_dir = config['data_dir'] self.replication = config['replication'] @@ -250,7 +251,7 @@ class Postgresql: except psycopg2.OperationalError as e: logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e)) return False - name = self.name + name = self.scope try: role = "master" if is_leader else "replica" subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, role, name]) diff --git a/scripts/aws.py b/scripts/aws.py old mode 100755 new mode 100644 index 6a04da92..0025e44e --- a/scripts/aws.py +++ b/scripts/aws.py @@ -1,3 +1,5 @@ +#!/usr/bin/python + import logging import requests from requests.exceptions import RequestException @@ -71,12 +73,12 @@ class AWSConnection: if __name__ == '__main__': - if len(sys.argv) != 3: + if len(sys.argv) != 4: print ("Usage: {0} action role name".format(sys.argv[0])) - return 1 + sys.exit(1) action, role, name = sys.argv[1:] if action in ('on_start', 'on_stop', 'on_role_change'): aws = AWSConnection({'cluster_name': name}) aws.on_role_change(role) - return 0 - return 2 + sys.exit(0) + sys.exit(2) From 35fe536e8b6ff2a073682592243744c7d1b36f4a Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 23 Jul 2015 17:09:39 +0200 Subject: [PATCH 15/27] Set +x bit on aws.py --- scripts/aws.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) mode change 100644 => 100755 scripts/aws.py diff --git a/scripts/aws.py b/scripts/aws.py old mode 100644 new mode 100755 From dac3d7997fdb2eb9994a0dfe2de543cb9de1c17b Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Fri, 24 Jul 2015 10:47:26 +0200 Subject: [PATCH 16/27] Change a typo. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 6348c3ff..a5ae8b36 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -44,7 +44,7 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] - self.callback = config.get('callback', {}) + self.callback = config.get('callbacks', {}) 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')) From 859da0db0adb98505ad840afd0cd4f7527399554 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 10:10:29 +0200 Subject: [PATCH 17/27] Make unit tets run again. --- postgres0.yml | 6 ++++-- postgres1.yml | 6 ++++-- tests/test_aws.py | 32 +++++++++++++++++++++++++------- tests/test_postgresql.py | 10 +++++----- 4 files changed, 38 insertions(+), 16 deletions(-) diff --git a/postgres0.yml b/postgres0.yml index 0cb694a1..6c9fb82d 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -1,20 +1,22 @@ ttl: &ttl 30 loop_wait: &loop_wait 10 +scope: &scope batman restapi: listen: 127.0.0.1:8008 connect_address: 127.0.0.1:8008 etcd: - scope: batman + scope: *scope ttl: *ttl host: 127.0.0.1:4001 #discovery_srv: my-etcd.domain #zookeeper: -# scope: batman +# scope: *scope # session_timeout: *ttl # reconnect_timeout: *loop_wait # hosts: 127.0.0.1:2181 postgresql: name: postgresql0 + scope: *scope listen: 127.0.0.1:5432 connect_address: 127.0.0.1:5432 data_dir: data/postgresql0 diff --git a/postgres1.yml b/postgres1.yml index 680091f1..1928cee7 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -1,20 +1,22 @@ ttl: &ttl 30 loop_wait: &loop_wait 10 +scope: &scope batman restapi: listen: 127.0.0.1:8009 connect_address: 127.0.0.1:8009 etcd: - scope: batman + scope: *scope ttl: *ttl host: 127.0.0.1:4001 #discovery_srv: my-etcd.domain #zookeeper: -# scope: batman +# scope: *scope # session_timeout: *ttl # reconnect_timeout: *loop_wait # hosts: 127.0.0.1:2181 postgresql: name: postgresql1 + scope: *scope listen: 127.0.0.1:5433 connect_address: 127.0.0.1:5433 data_dir: data/postgresql1 diff --git a/tests/test_aws.py b/tests/test_aws.py index 63b50734..4331badb 100644 --- a/tests/test_aws.py +++ b/tests/test_aws.py @@ -2,7 +2,7 @@ import unittest import requests import boto.ec2 from collections import namedtuple -from helpers.aws import AWSConnection +from scripts.aws import AWSConnection from requests.exceptions import RequestException import yaml @@ -24,6 +24,16 @@ class MockEc2Connection: return True +class MockResponse: + + def __init__(self, content): + self.content = content + self.ok = True + + def json(self): + return self.content + + class TestAWSConnection(unittest.TestCase): def __init__(self, method_name='runTest'): @@ -44,25 +54,28 @@ class TestAWSConnection(unittest.TestCase): result = namedtuple('Request', 'ok content') result.ok = True if url.split('/')[-1] == 'document': - result.content = '{\n "instanceId" : "012345",\n "region" : "eu-west-1"\n}' + result = {"instanceId": "012345", "region": "eu-west-1"} else: - result.content = 'foo' - return result + result = 'foo' + return MockResponse(result) def setUp(self): self.error = False requests.get = self.requests_get boto.ec2.connect_to_region = self.boto_ec2_connect_to_region self.config_string = """ -loop_wait: 10 +scope: &scope test +ttl: &ttl 30 +loop_wait: &loop_wait 10 restapi: listen: 0.0.0.0:8008 connect_address: 127.0.0.1:5432 etcd: - scope: test - ttl: 30 + scope: *scope + ttl: *ttl host: 127.0.0.1:8080 postgresql: + scope: *scope name: postgresql_foo listen: 0.0.0.0:5432 connect_address: 127.0.0.1:5432 @@ -76,6 +89,11 @@ postgresql: admin: username: admin password: admin + callbacks: + on_start: patroni/scripts/aws.py + on_stop: patroni/scripts/aws.py + on_restart: patroni/scripts/aws.py + on_role_change: patroni/scripts/aws.py parameters: archive_mode: "on" wal_level: hot_standby diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index d6c01f5b..069c4622 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -110,15 +110,15 @@ class TestPostgresql(unittest.TestCase): def set_up(self): subprocess.call = subprocess_call shutil.copy = nop - self.p = Postgresql({'name': 'test0', 'data_dir': 'data/test0', 'listen': '127.0.0.1, *:5432', - 'connect_address': '127.0.0.2:5432', + self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': 'data/test0', + 'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432', 'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'], - 'superuser': {'password': ''}, 'admin': {'username': 'admin', 'password': 'admin'}, + 'superuser': {'password': ''}, + 'admin': {'username': 'admin', 'password': 'admin'}, 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, - 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}}, - on_change_callback=lambda state: True) + 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}}) psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) From 201f9bf4b29e3bd6cb9c9ccb4a25e62663d04b7a Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 11:08:19 +0200 Subject: [PATCH 18/27] Make sure role check during callbacks does not accidentially modify the cluster state (is_promoted flag or trigger file). --- helpers/postgresql.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index a5ae8b36..71eb7d9c 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -230,9 +230,9 @@ class Postgresql: 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): + def is_leader(self, check_only=False): ret = not self.query('SELECT pg_is_in_recovery()').fetchone()[0] - if ret and self.is_promoted: + if ret and self.is_promoted and not check_only: self.delete_trigger_file() self.is_promoted = False return ret @@ -247,7 +247,7 @@ class Postgresql: cmd = self.callback[cb_name] if is_leader is None: try: - is_leader = self.is_leader() + is_leader = self.is_leader(check_only=True) except psycopg2.OperationalError as e: logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e)) return False @@ -279,7 +279,7 @@ class Postgresql: def stop(self): try: - is_leader = self.is_leader() + is_leader = self.is_leader(check_only=True) except: is_leader = None pass @@ -296,7 +296,7 @@ class Postgresql: def restart(self): try: - is_leader = self.is_leader() + is_leader = self.is_leader(check_only=True) except: is_leader = None pass From 115ba1d9ae2cf4c87526d5fcb2ccc92b58414fde Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 11:23:08 +0200 Subject: [PATCH 19/27] Make sure postgresql.stop also return True if pg_ctl stop returned 0. --- helpers/postgresql.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 71eb7d9c..da2d1fdf 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -286,7 +286,7 @@ class Postgresql: ret = subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) if ret == 0 and 'on_stop' in self.callback: self.call_nowait('on_stop', is_leader=is_leader) - return ret != 0 + return ret == 0 def reload(self): ret = subprocess.call(self._pg_ctl + ['reload']) From 56983d867db449c218af97766043b8d9966fb079 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 13:03:12 +0200 Subject: [PATCH 20/27] Run coverage test for the data in scripts as well as helpers. --- setup.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/setup.py b/setup.py index 5ac4a89f..db327c37 100644 --- a/setup.py +++ b/setup.py @@ -22,6 +22,7 @@ __location__ = os.path.join(os.getcwd(), os.path.dirname(inspect.getfile(inspect NAME = 'patroni' MAIN_PACKAGE = 'patroni.py' HELPERS = 'helpers' +SCRIPTS = 'scripts' VERSION = '0.1' DESCRIPTION = 'A Template for PostgreSQL HA with etcd' LICENSE = 'The MIT License' @@ -61,7 +62,8 @@ class PyTest(TestCommand): def finalize_options(self): TestCommand.finalize_options(self) if self.cov_xml or self.cov_html: - self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov-report', 'term-missing'] + self.cov = ['--cov', MAIN_PACKAGE, '--cov', HELPERS, '--cov', SCRIPTS, '--cov-report', + 'term-missing'] if self.cov_xml: self.cov.extend(['--cov-report', 'xml']) if self.cov_html: From b92bfb34a59555523ca66172e908da697b3c961f Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 13:04:01 +0200 Subject: [PATCH 21/27] Futher test for postgresql module (includes callbacks) --- tests/test_postgresql.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 069c4622..c93a7a3c 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -118,7 +118,11 @@ class TestPostgresql(unittest.TestCase): 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, - 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}}) + 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}, + 'callbacks': {'on_start': '/usr/bin/true', 'on_stop': '/usr/bin/true', + 'on_restart': '/usr/bin/true', 'on_role_change': '/bin/true', + 'on_reload': '/usr/bin/true' + }}) psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) @@ -129,6 +133,9 @@ class TestPostgresql(unittest.TestCase): def tear_down(self): shutil.rmtree('data') + def mock_query(self, p): + raise psycopg2.OperationalError("not supported") + def test_data_directory_empty(self): self.assertTrue(self.p.data_directory_empty()) @@ -136,12 +143,13 @@ class TestPostgresql(unittest.TestCase): self.assertTrue(self.p.initialize()) self.assertTrue(os.path.exists(os.path.join(self.p.data_dir, 'pg_hba.conf'))) - def test_start(self): + def test_start_stop(self): self.assertFalse(self.p.start()) self.p.is_running = is_running with open(os.path.join(self.p.data_dir, 'postmaster.pid'), 'w'): pass self.assertTrue(self.p.start()) + self.assertTrue(self.p.stop()) def test_sync_from_leader(self): self.assertTrue(self.p.sync_from_leader(self.leader)) @@ -196,3 +204,11 @@ class TestPostgresql(unittest.TestCase): def test_last_operation(self): self.assertEquals(self.p.last_operation(), '0') + + def test_non_existing_callback(self): + self.assertFalse(self.p.call_nowait('foobar')) + + def test_is_leader_exception(self): + self.p.start() + self.p.query = self.mock_query + self.assertTrue(self.p.stop()) From d85451efae29d4d7a4e10ff36aa557d2f22556b5 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 13:04:45 +0200 Subject: [PATCH 22/27] Increase test coverage for aws script. --- scripts/aws.py | 24 +++++--------------- tests/test_aws.py | 58 +++++++++-------------------------------------- 2 files changed, 17 insertions(+), 65 deletions(-) diff --git a/scripts/aws.py b/scripts/aws.py index 0025e44e..0d10701c 100755 --- a/scripts/aws.py +++ b/scripts/aws.py @@ -10,16 +10,9 @@ logger = logging.getLogger(__name__) class AWSConnection: - def __init__(self, config): + def __init__(self, cluster_name): self.available = False - self.config = config - - if 'cluster_name' in config: - self.cluster_name = config.get('cluster_name') - elif 'etcd' in config and isinstance(config['etcd'], dict): - self.cluster_name = config['etcd'].get('scope', 'unknown') - else: - self.cluster_name = 'unknown' + self.cluster_name = cluster_name if cluster_name is not None else 'unknown' try: # get the instance id r = requests.get('http://169.254.169.254/latest/dynamic/instance-identity/document', timeout=0.1) @@ -73,12 +66,7 @@ class AWSConnection: if __name__ == '__main__': - if len(sys.argv) != 4: - print ("Usage: {0} action role name".format(sys.argv[0])) - sys.exit(1) - action, role, name = sys.argv[1:] - if action in ('on_start', 'on_stop', 'on_role_change'): - aws = AWSConnection({'cluster_name': name}) - aws.on_role_change(role) - sys.exit(0) - sys.exit(2) + 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])) diff --git a/tests/test_aws.py b/tests/test_aws.py index 4331badb..84d495fe 100644 --- a/tests/test_aws.py +++ b/tests/test_aws.py @@ -4,7 +4,6 @@ import boto.ec2 from collections import namedtuple from scripts.aws import AWSConnection from requests.exceptions import RequestException -import yaml class MockEc2Connection: @@ -42,8 +41,8 @@ class TestAWSConnection(unittest.TestCase): def set_error(self): self.error = True - def set_ok(self): - self.error = False + def set_json_error(self): + self.json_error = True def boto_ec2_connect_to_region(self, region): return MockEc2Connection(self.error) @@ -53,7 +52,7 @@ class TestAWSConnection(unittest.TestCase): raise RequestException("foo") result = namedtuple('Request', 'ok content') result.ok = True - if url.split('/')[-1] == 'document': + if url.split('/')[-1] == 'document' and not self.json_error: result = {"instanceId": "012345", "region": "eu-west-1"} else: result = 'foo' @@ -61,50 +60,10 @@ class TestAWSConnection(unittest.TestCase): def setUp(self): self.error = False + self.json_error = False requests.get = self.requests_get boto.ec2.connect_to_region = self.boto_ec2_connect_to_region - self.config_string = """ -scope: &scope test -ttl: &ttl 30 -loop_wait: &loop_wait 10 -restapi: - listen: 0.0.0.0:8008 - connect_address: 127.0.0.1:5432 -etcd: - scope: *scope - ttl: *ttl - host: 127.0.0.1:8080 -postgresql: - scope: *scope - name: postgresql_foo - listen: 0.0.0.0:5432 - connect_address: 127.0.0.1:5432 - data_dir: /home/postgres/pgdata/data - replication: - username: standby - password: standby - network: 0.0.0.0/0 - superuser: - password: zalando - admin: - username: admin - password: admin - callbacks: - on_start: patroni/scripts/aws.py - on_stop: patroni/scripts/aws.py - on_restart: patroni/scripts/aws.py - on_role_change: patroni/scripts/aws.py - parameters: - archive_mode: "on" - wal_level: hot_standby - max_wal_senders: 5 - wal_keep_segments: 8 - archive_timeout: 1800s - max_replication_slots: 5 - hot_standby: "on" - ssl: "on" -""" - self.conn = AWSConnection(yaml.load(self.config_string)) + self.conn = AWSConnection('test') def test_aws_available(self): self.assertTrue(self.conn.aws_available()) @@ -116,11 +75,16 @@ postgresql: def test_non_aws(self): self.set_error() - conn = AWSConnection(yaml.load(self.config_string)) + conn = AWSConnection('test') self.assertFalse(conn.aws_available()) self.assertFalse(conn._tag_ebs('master')) self.assertFalse(conn._tag_ec2('master')) + def test_aws_bizare_response(self): + self.set_json_error() + conn = AWSConnection('test') + self.assertFalse(conn.aws_available()) + def test_aws_tag_ebs_error(self): self.set_error() self.assertFalse(self.conn._tag_ebs("master")) From 7d1bc32ad93d70b1d4814fc0b5bc2badb2aa1130 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 13:06:44 +0200 Subject: [PATCH 23/27] potentially run the doctests for the scripts. --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index db327c37..62d05a69 100644 --- a/setup.py +++ b/setup.py @@ -82,7 +82,7 @@ class PyTest(TestCommand): params['plugins'] = ['cov'] if self.junitxml: params['args'] += self.junitxml - params['args'] += ['--doctest-modules', HELPERS, '-s'] + params['args'] += ['--doctest-modules', HELPERS, '--doctest-modules', SCRIPTS, '-s'] errno = pytest.main(**params) sys.exit(errno) From 27bda5b4e46237f800bddb373a29b3366fb3eeeb Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 13:12:08 +0200 Subject: [PATCH 24/27] document the new callbacks option. --- README.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/README.md b/README.md index 3ecbcebb..be12dc71 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,12 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *username*: replication username, user will be created during initialization * *password*: replication password, user will be created during initialization * *network*: network setting for replication in pg_hba.conf + * *callbacks* callback scripts to run on certain actions. Patroni will pass current action, role and cluster name. See scripts/aws.py as an example on how to write them. + * *on_start*: a script to run when the cluster starts + * *on_stop*: a script to run when the cluster stops + * *on_restart*: a script to run when the cluster restarts + * *on_reload*: a script to run when configuration reload is triggered + * *on_role_change*: a script to run when the cluster is being promoted or demoted * *recovery_conf*: configuration settings written to recovery.conf when configuring follower * *parameters*: list of configuration settings for Postgres From 5c5bae69eaa8b03f893375038424b4db32fac397 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Mon, 27 Jul 2015 16:11:58 +0200 Subject: [PATCH 25/27] make sure we can import from scripts. --- scripts/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 scripts/__init__.py diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 00000000..e69de29b From 71a1200f64cbe60fe2123e19ea1a9035f97615e0 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Wed, 29 Jul 2015 12:09:17 +0200 Subject: [PATCH 26/27] Fix a mistake in checking for the number of parameters. --- scripts/aws.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/aws.py b/scripts/aws.py index b5e87fd7..b172a8c3 100755 --- a/scripts/aws.py +++ b/scripts/aws.py @@ -66,7 +66,7 @@ class AWSConnection: if __name__ == '__main__': - if len(sys.argv) != 4 and sys.argv[1] in ('on_start', 'on_stop', 'on_role_change'): + 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])) From a0fdd6398f0afcf5d29abd89101e9ee3da949b87 Mon Sep 17 00:00:00 2001 From: Oleksii Kliukin Date: Thu, 30 Jul 2015 10:17:17 +0200 Subject: [PATCH 27/27] Cosmetic changes: avoid magic strings in action names, name the parameter passed to the callback 'scope' for consistency. --- helpers/postgresql.py | 36 +++++++++++++++++++++--------------- 1 file changed, 21 insertions(+), 15 deletions(-) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 325d7080..84a94346 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -16,6 +16,12 @@ else: logger = logging.getLogger(__name__) +ACTION_ON_START = "on_start" +ACTION_ON_STOP = "on_stop" +ACTION_ON_RESTART = "on_restart" +ACTION_ON_RELOAD = "on_reload" +ACTION_ON_ROLE_CHANGE = "on_role_change" + def parseurl(url): r = urlparse(url) @@ -253,12 +259,12 @@ class Postgresql: except psycopg2.OperationalError as e: logger.warning("unable to perform {0} action, cannot obtain the cluster role: {1}".format(cb_name, e)) return False - name = self.scope + scope = self.scope try: role = "master" if is_leader else "replica" - subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, role, name]) + subprocess.Popen(shlex.split(os.path.abspath(cmd))+[cb_name, role, scope]) except Exception as e: - logger.warning("callback {0} {1} {2} {3} failed: {4}".format(os.path.abspath(cmd), cb_name, role, name, e)) + logger.warning("callback {0} {1} {2} {3} failed: {4}".format(os.path.abspath(cmd), cb_name, role, scope, e)) return False return True @@ -275,8 +281,8 @@ class Postgresql: ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0 ret and self.load_replication_slots() self.save_configuration_files() - if ret and 'on_start' in self.callback: - self.call_nowait('on_start') + if ret and ACTION_ON_START in self.callback: + self.call_nowait(ACTION_ON_START) return ret def stop(self): @@ -286,14 +292,14 @@ class Postgresql: is_leader = None pass ret = subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) - if ret == 0 and 'on_stop' in self.callback: - self.call_nowait('on_stop', is_leader=is_leader) + if ret == 0 and ACTION_ON_STOP in self.callback: + self.call_nowait(ACTION_ON_STOP, is_leader=is_leader) return ret == 0 def reload(self): ret = subprocess.call(self._pg_ctl + ['reload']) - if ret == 0 and 'on_reload' in self.callback: - self.call_nowait('on_reload') + if ret == 0 and ACTION_ON_RELOAD in self.callback: + self.call_nowait(ACTION_ON_RELOAD) return ret == 0 def restart(self): @@ -303,8 +309,8 @@ class Postgresql: is_leader = None pass ret = subprocess.call(self._pg_ctl + ['restart', '-m', 'fast']) - if ret == 0 and 'on_restart' in self.callback: - self.call_nowait('on_restart', is_leader=is_leader) + if ret == 0 and ACTION_ON_RESTART in self.callback: + self.call_nowait(ACTION_ON_RESTART, is_leader=is_leader) return ret == 0 def server_options(self): @@ -392,8 +398,8 @@ primary_conninfo = '{}' if not self.check_recovery_conf(leader): self.write_recovery_conf(leader) self.restart() - if 'on_role_change' in self.callback: - self.call_nowait('on_role_change') + if ACTION_ON_ROLE_CHANGE in self.callback: + self.call_nowait(ACTION_ON_ROLE_CHANGE) def save_configuration_files(self): """ @@ -413,8 +419,8 @@ primary_conninfo = '{}' def promote(self): self.is_promoted = subprocess.call(self._pg_ctl + ['promote']) == 0 - if self.is_promoted and 'on_role_change' in self.callback: - self.call_nowait('on_role_change') + if self.is_promoted and ACTION_ON_ROLE_CHANGE in self.callback: + self.call_nowait(ACTION_ON_ROLE_CHANGE) return self.is_promoted def demote(self, leader):