diff --git a/.gitignore b/.gitignore index df367db5..699794c2 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,11 @@ data/* *.pyc -helpers/*.pyc +*.egg/ +*.egg-info/ +.cache/ +.coverage +.eggs/ +build/ +coverage.xml +junit.xml +pgpass diff --git a/docker/README.md b/docker/README.md index 74288a98..afadc4b6 100644 --- a/docker/README.md +++ b/docker/README.md @@ -1,7 +1,7 @@ # Patroni Dockerfile You can run Patroni in a docker container using this Dockerfile, or by using one of the Docker image at - https://os-registry.stups.zalan.do/v1/repositories/acid/patroni/tags + https://registry.opensource.zalan.do/v1/repositories/acid/patroni/tags This Dockerfile is meant in aiding development of Patroni and quick testing of features. It is not a production-worthy Dockerfile @@ -10,7 +10,7 @@ Dockerfile ## Standalone Patroni - docker run -d os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT + docker run -d registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT ## Multiple Patroni's communicating with a standalone etcd inside Docker @@ -36,12 +36,12 @@ To automate this you can run the following script: Example session: - $ ./dev_patroni_cluster.sh --image os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT --members=2 --name=bravo + $ ./dev_patroni_cluster.sh --image registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT --members=2 --name=bravo The etcd container is 6be871a11cb373406ca5ea1c6b39e1.0-SNAPSHOTfdde9fb1d6177212d6ad0c0d1bd9b563, ip=172.17.1.24 Started Patroni container 67e611f2eca7c40f9e6e0e24a4a8f2cba7e3e56d22a420e15ab9240a37a9d7a4, ip=172.17.1.25 Started Patroni container 47dd12ae635ab83b039f5889e250048b606ed5e48e3650b69e365e7e1d4acbcf, ip=172.17.1.26 $ docker ps CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES - 47dd12ae635a os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT "/bin/bash /entrypoi 10 seconds ago Up 8 seconds 4001/tcp, 5432/tcp, 2380/tcp bravo_OR64g8bx - 67e611f2eca7 os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT "/bin/bash /entrypoi 11 seconds ago Up 10 seconds 2380/tcp, 4001/tcp, 5432/tcp bravo_si9no8iz - 6be871a11cb3 os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT "/bin/bash /entrypoi 12 seconds ago Up 10 seconds 4001/tcp, 5432/tcp, 2380/tcp bravo_etcd + 47dd12ae635a registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT "/bin/bash /entrypoi 10 seconds ago Up 8 seconds 4001/tcp, 5432/tcp, 2380/tcp bravo_OR64g8bx + 67e611f2eca7 registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT "/bin/bash /entrypoi 11 seconds ago Up 10 seconds 2380/tcp, 4001/tcp, 5432/tcp bravo_si9no8iz + 6be871a11cb3 registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT "/bin/bash /entrypoi 12 seconds ago Up 10 seconds 4001/tcp, 5432/tcp, 2380/tcp bravo_etcd diff --git a/docker/dev_patroni_cluster.sh b/docker/dev_patroni_cluster.sh index e9a253dc..dcc18f87 100755 --- a/docker/dev_patroni_cluster.sh +++ b/docker/dev_patroni_cluster.sh @@ -1,6 +1,6 @@ #!/bin/bash -DOCKER_IMAGE="os-registry.stups.zalan.do/acid/patroni:1.0-SNAPSHOT" +DOCKER_IMAGE="registry.opensource.zalan.do/acid/patroni:1.0-SNAPSHOT" MEMBERS=3 diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index f4851a36..d94757b3 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -3,25 +3,25 @@ function usage() { cat <<__EOF__ -Usage: $0 +Usage: $0 Options: --etcd ETCD Provide an external etcd to connect to - --name NAME Give the cluster a specific name + --name NAME Give the cluster a specific name --etcd-only Do not run Patroni, run a standalone etcd Examples: $0 --etcd=127.17.0.84:4001 $0 --etcd-only - $0 + $0 $0 --name=true_scotsman __EOF__ } DOCKER_IP=$(hostname --ip-address) -PATRONI_SCOPE=batman +PATRONI_SCOPE=${PATRONI_SCOPE:-batman} optspec=":vh-:" while getopts "$optspec" optchar; do @@ -32,7 +32,7 @@ while getopts "$optspec" optchar; do exec etcd --data-dir /tmp/etcd.data \ -advertise-client-urls=http://${DOCKER_IP}:4001 \ -listen-client-urls=http://0.0.0.0:4001 \ - -listen-peer-urls=http://0.0.0.0:2380 + -listen-peer-urls=http://0.0.0.0:2380 exit 0 ;; cheat) diff --git a/patroni/__init__.py b/patroni/__init__.py index 10e70956..150334ba 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -8,7 +8,7 @@ from patroni.api import RestApiServer from patroni.etcd import Etcd from patroni.ha import Ha from patroni.postgresql import Postgresql -from patroni.utils import setup_signal_handlers, sleep, reap_children +from patroni.utils import setup_signal_handlers, reap_children from patroni.zookeeper import ZooKeeper logger = logging.getLogger(__name__) @@ -19,11 +19,11 @@ class Patroni: def __init__(self, config): self.nap_time = config['loop_wait'] self.postgresql = Postgresql(config['postgresql']) - self.ha = Ha(self.postgresql, self.get_dcs(self.postgresql.name, config)) + self.dcs = self.get_dcs(self.postgresql.name, config) host, port = config['restapi']['listen'].split(':') self.api = RestApiServer(self, config['restapi']) + self.ha = Ha(self) self.next_run = time.time() - self.shutdown_member_ttl = 300 @staticmethod def get_dcs(name, config): @@ -33,30 +33,13 @@ class Patroni: return ZooKeeper(name, config['zookeeper']) raise Exception('Can not find sutable configuration of distributed configuration store') - def touch_member(self, ttl=None): - connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string - if self.ha.cluster: - for m in self.ha.cluster.members: - # Do not update member TTL when it is far from being expired - if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: - return True - return self.ha.dcs.touch_member(connection_string, ttl) - - def initialize(self): - # wait for etcd to be available - while not self.touch_member(): - logger.info('waiting on DCS') - sleep(5) - - self.postgresql.schedule_load_slots = self.postgresql.is_running() and self.postgresql.use_slots - def schedule_next_run(self): self.next_run += self.nap_time current_time = time.time() nap_time = self.next_run - current_time if nap_time <= 0: self.next_run = current_time - elif self.ha.dcs.watch(nap_time): + elif self.dcs.watch(nap_time): self.next_run = time.time() def run(self): @@ -64,12 +47,7 @@ class Patroni: self.next_run = time.time() while True: - self.touch_member() logger.info(self.ha.run_cycle()) - try: - self.ha.cluster and self.ha.state_handler.sync_replication_slots(self.ha.cluster) - except: - logger.exception('Exception when changing replication slots') reap_children() self.schedule_next_run() @@ -87,13 +65,11 @@ def main(): config = yaml.load(f) patroni = Patroni(config) - patroni.initialize() try: patroni.run() except KeyboardInterrupt: pass finally: patroni.api.shutdown() - patroni.touch_member(patroni.shutdown_member_ttl) # schedule member removal patroni.postgresql.stop() - patroni.ha.dcs.delete_leader() + patroni.dcs.delete_leader() diff --git a/patroni/api.py b/patroni/api.py index 30bc8914..1fa5ca62 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -3,6 +3,7 @@ import fcntl import json import logging import psycopg2 +import time from patroni.exceptions import PostgresConnectionException from patroni.utils import Retry, RetryFailedError @@ -44,23 +45,35 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_GET(self): """Default method for processing all GET requests which can not be routed to other methods""" + path = '/master' if self.path == '/' else self.path response = self.get_postgresql_status() - path = '/master' if self.path == '/' else self.path - status_code = 200 if response['running'] and 'role' in response and response['role'] in path else 503 + patroni = self.server.patroni + cluster = patroni.dcs.cluster + if cluster: # dcs available + if cluster.leader and cluster.leader.name == patroni.postgresql.name: # is_leader + status_code = 200 if 'master' in path else 503 + elif 'role' not in response: + status_code = 503 + elif response['role'] == 'master': # running as master but without leader lock!!!! + status_code = 503 + elif response['role'] in path: + status_code = 200 + else: + status_code = 503 + elif 'role' in response and response['role'] in path: + status_code = 200 + elif patroni.ha.restart_scheduled() and patroni.postgresql.role == 'master' and 'master' in path: + # exceptional case for master node when the postgres is being restarted via API + status_code = 200 + else: + status_code = 503 self.send_response(status_code) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(response).encode('utf-8')) - @check_auth - def do_GET_sampleauth(self): - self.send_response(200) - self.send_header('Content-Type', 'text/html') - self.end_headers() - self.wfile.write(b'Hello!') - def do_GET_patroni(self): response = self.get_postgresql_status(True) @@ -69,6 +82,96 @@ class RestApiHandler(BaseHTTPRequestHandler): self.end_headers() self.wfile.write(json.dumps(response).encode('utf-8')) + @check_auth + def do_POST_restart(self): + status_code = 503 + data = b'restart failed' + try: + status, msg = self.server.patroni.ha.restart() + status_code = 200 if status else 503 + data = msg.encode('utf-8') + except: + logger.exception('Exception during restart') + + self.send_response(status_code) + self.send_header('Content-Type', 'text/html') + self.end_headers() + self.wfile.write(data) + + @check_auth + def do_POST_reinitialize(self): + ha = self.server.patroni.ha + cluster = ha.dcs.get_cluster() + if cluster.is_unlocked(): + status_code = 503 + data = b'Cluster has no leader, can not reinitialize' + elif cluster.leader.name == ha.state_handler.name: + status_code = 503 + data = b'I am the leader, can not reinitialize' + else: + action = ha.schedule_reinitialize() + if action is not None: + status_code = 503 + data = (action + ' already in progress').encode('utf-8') + else: + status_code = 200 + data = b'reinitialize scheduled' + + self.send_response(status_code) + self.send_header('Content-Type', 'text/html') + self.end_headers() + self.wfile.write(data) + + def poll_failover_result(self, leader, member): + for a in range(0, 15): + time.sleep(1) + try: + cluster = self.server.patroni.dcs.get_cluster() + if cluster.leader and cluster.leader.name != leader: + return 200, ('Successfully failed over to ' + cluster.leader.name).encode('utf-8') + if not cluster.failover: + return 503, b'Failover failed' + except: + pass + return 503, b'Failover status unknown' + + def is_failover_possible(self, cluster, leader, member): + if leader and not cluster.leader or cluster.leader.name != leader: + return b'leader name does not match' + if member: + members = [m for m in cluster.members if m.name == member] + if not members: + return b'member does not exists' + else: + members = [m for m in cluster.members if m.name != cluster.leader.name and m.api_url] + if not members: + return b'failover is not possible: cluster does not have members except leader' + for member, reachable, in_recovery, xlog_location in self.server.patroni.ha.fetch_nodes_statuses(members): + if reachable: + return None + return b'failover is not possible: no good candidates have been found' + + @check_auth + def do_POST_failover(self): + content_length = int(self.headers.get('content-length', 0)) + request = json.loads(self.rfile.read(content_length).decode('utf-8')) + leader = request.get('leader', None) + member = request.get('member', None) + cluster = self.server.patroni.ha.dcs.get_cluster() + status_code = 503 + data = self.is_failover_possible(cluster, leader, member) + if not data: + if not self.server.patroni.dcs.manual_failover(leader, member): + data = b'failed to write failover key into DCS' + else: + self.server.patroni.dcs.event.set() + status_code, data = self.poll_failover_result(cluster.leader and cluster.leader.name, member) + + self.send_response(status_code) + self.send_header('Content-Type', 'text/html') + self.end_headers() + self.wfile.write(data) + def parse_request(self): """Override parse_request method to enrich basic functionality of `BaseHTTPRequestHandler` class @@ -90,7 +193,7 @@ class RestApiHandler(BaseHTTPRequestHandler): def query(self, sql, *params, **kwargs): if not kwargs.get('retry', False): return self.server.query(sql, *params) - retry = Retry(delay=2, retry_exceptions=PostgresConnectionException) + retry = Retry(delay=1, retry_exceptions=PostgresConnectionException) return retry(self.server.query, sql, *params) def get_postgresql_status(self, retry=False): @@ -98,15 +201,16 @@ class RestApiHandler(BaseHTTPRequestHandler): row = self.query("""SELECT to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'), pg_is_in_recovery(), CASE WHEN pg_is_in_recovery() - THEN null - ELSE pg_current_xlog_location() END, - pg_last_xlog_receive_location(), - pg_last_xlog_replay_location(), + THEN 0 + ELSE pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint + END, + pg_xlog_location_diff(pg_last_xlog_receive_location(), '0/0')::bigint, + pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint, pg_is_in_recovery() AND pg_is_xlog_replay_paused()""", retry=retry)[0] return { - 'running': True, + 'state': self.server.patroni.postgresql.state, 'postmaster_start_time': row[0], - 'role': 'slave' if row[1] else 'master', + 'role': 'replica' if row[1] else 'master', 'xlog': ({ 'received_location': row[3], 'replayed_location': row[4], @@ -115,8 +219,11 @@ class RestApiHandler(BaseHTTPRequestHandler): }) } except (psycopg2.Error, RetryFailedError, PostgresConnectionException): - logger.exception('get_postgresql_status') - return {'running': self.server.patroni.postgresql.is_running()} + state = self.server.patroni.postgresql.state + if state in ['stopped', 'starting', 'stopping', 'restarting', 'running']: + logger.exception('get_postgresql_status') + state = 'unknown' if state == 'running' else state + return {'state': state} class RestApiServer(ThreadingMixIn, HTTPServer, Thread): diff --git a/patroni/async_executor.py b/patroni/async_executor.py new file mode 100644 index 00000000..fc222202 --- /dev/null +++ b/patroni/async_executor.py @@ -0,0 +1,55 @@ +import logging +from threading import Lock, Thread + +logger = logging.getLogger(__name__) + + +class AsyncExecutor: + + def __init__(self): + Lock.__init__(self) + self._busy = False + self._thread_lock = Lock() + self._scheduled_action = None + self._scheduled_action_lock = Lock() + + @property + def busy(self): + return self._busy + + def schedule(self, action, immediately=False): + with self._scheduled_action_lock: + if self._scheduled_action is not None: + return self._scheduled_action + self._scheduled_action = action + self._busy = immediately + return None + + @property + def scheduled_action(self): + with self._scheduled_action_lock: + return self._scheduled_action + + def reset_scheduled_action(self): + with self._scheduled_action_lock: + self._scheduled_action = None + + def run(self, func, args=()): + try: + return func(*args) if args else func() + except: + logger.exception('Exception during execution of long running task %s', self.scheduled_action) + finally: + with self: + self._busy = False + self.reset_scheduled_action() + + def run_async(self, func, args=()): + self._busy = True + Thread(target=self.run, args=(func, args)).start() + + def __enter__(self): + self._thread_lock.acquire() + + def __exit__(self, type, value, traceback): + self._thread_lock.release() diff --git a/patroni/dcs.py b/patroni/dcs.py index 908386b5..a8afda06 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -1,9 +1,10 @@ import abc +import json from collections import namedtuple from patroni.exceptions import DCSError -from patroni.utils import calculate_ttl, sleep from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl +from threading import Event, Lock def parse_connection_string(value): @@ -23,28 +24,52 @@ def parse_connection_string(value): return conn_url, api_url -class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')): +class Member(namedtuple('Member', 'index,name,session,data')): """Immutable object (namedtuple) which represents single member of PostgreSQL cluster. Consists of the following fields: :param index: modification index of a given member key in a Configuration Store :param name: name of PostgreSQL cluster member - :param conn_url: connection string containing host, user and password which could be used to access this member. - :param api_url: REST API url of patroni instance - :param expiration: expiration time of given member key - :param ttl: ttl of given member key in seconds""" + :param session: either session id or just ttl in seconds + :param data: arbitrary data i.e. conn_url, api_url, xlog location, state, role, tags, etc... - def real_ttl(self): - return calculate_ttl(self.expiration) or -1 + There are two mandatory keys in a data: + conn_url: connection string containing host, user and password which could be used to access this member. + api_url: REST API url of patroni instance""" + + @staticmethod + def from_node(index, name, session, data): + """ + >>> Member.from_node(-1, '', '', '{"conn_url": "postgres://foo@bar/postgres"}') is not None + True + >>> Member.from_node(-1, '', '', '{') + Member(index=-1, name='', session='', data={}) + """ + if data.startswith('postgres'): + conn_url, api_url = parse_connection_string(data) + data = {'conn_url': conn_url, 'api_url': api_url} + else: + try: + data = json.loads(data) + except: + data = {} + return Member(index, name, session, data) + + @property + def conn_url(self): + return self.data.get('conn_url', None) + + @property + def api_url(self): + return self.data.get('api_url', None) -class Leader(namedtuple('Leader', 'index,expiration,ttl,member')): +class Leader(namedtuple('Leader', 'index,session,member')): """Immutable object (namedtuple) which represents leader key. Consists of the following fields: :param index: modification index of a leader key in a Configuration Store - :param expiration: expiration time of the leader key - :param ttl: ttl of the leader key + :param session: either session id or just ttl in seconds :param member: reference to a `Member` object which represents current leader (see `Cluster.members`)""" @property @@ -56,7 +81,15 @@ class Leader(namedtuple('Leader', 'index,expiration,ttl,member')): return self.member.conn_url -class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): +class Failover(namedtuple('Failover', 'index,leader,member')): + + @staticmethod + def from_node(index, value): + t = [a.strip() for a in value.split(':')] + [''] + return Failover(index, t[0], t[1]) if t[0] or t[1] else None + + +class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members,failover')): """Immutable object (namedtuple) which represents PostgreSQL cluster. Consists of the following fields: @@ -64,7 +97,8 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem :param leader: `Leader` object which represents current leader of the cluster :param last_leader_operation: int or long object containing position of last known leader operation. This value is stored in `/optime/leader` key - :param members: list of Member object, all PostgreSQL cluster members including leader""" + :param members: list of Member object, all PostgreSQL cluster members including leader + :param failover: reference to `Failover` object""" def is_unlocked(self): return not (self.leader and self.leader.name) @@ -76,6 +110,7 @@ class AbstractDCS: _INITIALIZE = 'initialize' _LEADER = 'leader' + _FAILOVER = 'failover' _MEMBERS = 'members/' _OPTIME = 'optime' _LEADER_OPTIME = _OPTIME + '/' + _LEADER @@ -90,6 +125,10 @@ class AbstractDCS: self._scope = config['scope'] self._base_path = '/service/' + self._scope + self._cluster = None + self._cluster_thread_lock = Lock() + self.event = Event() + def client_path(self, path): return '/'.join([self._base_path, path.lstrip('/')]) @@ -109,25 +148,54 @@ class AbstractDCS: def leader_path(self): return self.client_path(self._LEADER) + @property + def failover_path(self): + return self.client_path(self._FAILOVER) + @property def leader_optime_path(self): return self.client_path(self._LEADER_OPTIME) @abc.abstractmethod + def _load_cluster(self): + """Internally this method should build `Cluster` object which + represents current state and topology of the cluster in DCS. + this method supposed to be called only by `get_cluster` method. + + raise `~DCSError` in case of communication or other problems with DCS. + If the current node was running as a master and exception raised, + instance would be demoted.""" + def get_cluster(self): - """:returns: `Cluster` object which represent current state and topology of the cluster - raise `~DCSError` in case of communication or other problems with DCS. If current instance was - running as a master and exception raised instance would be demoted.""" + with self._cluster_thread_lock: + try: + self._load_cluster() + except: + self._cluster = None + raise + return self._cluster + + @property + def cluster(self): + with self._cluster_thread_lock: + return self._cluster + + def reset_cluster(self): + with self._cluster_thread_lock: + self._cluster = None @abc.abstractmethod - def update_leader(self, state_handler): - """Update leader key (or session) ttl and `/optime/leader` key in DCS. + def write_leader_optime(self, last_operation): + """write current xlog location into `/optime/leader` key in DCS + :param last_operation: absolute xlog location in bytes""" + + @abc.abstractmethod + def update_leader(self): + """Update leader key (or session) ttl - :param state_handler: reference to `Postgresql` object :returns: `!True` if leader key (or session) has been updated successfully. If not, `!False` must be returned and current instance would be demoted. - If you failed to update `/optime/leader` this error is not critical and you can return `!True` You have to use CAS (Compare And Swap) operation in order to update leader key, for example for etcd `prevValue` parameter must be used.""" @@ -140,6 +208,13 @@ class AbstractDCS: Key must be created atomically. In case if key already exists it should not be overwritten and `!False` must be returned""" + @abc.abstractmethod + def set_failover_value(self, value, index=None): + """Create or update `/failover` key""" + + def manual_failover(self, leader, member, index=None): + return self.set_failover_value(leader + (':' + member if member else ''), index) + def current_leader(self): try: cluster = self.get_cluster() @@ -165,8 +240,11 @@ class AbstractDCS: overwriting the key if necessary.""" @abc.abstractmethod - def initialize(self): + def initialize(self, create_new=True, sysid=""): """Race for cluster initialization. + + :param create_new: False if the key should already exist (in the case we are setting the system_id) + :param sysid: PostgreSQL cluster system identifier, if specified, is written to the key :returns: `!True` if key has been created successfully. this method should create atomically initialize key and return `!True` @@ -188,5 +266,5 @@ class AbstractDCS: :param timeout: timeout in seconds :returns: `!True` if you would like to reschedule the next run of ha cycle""" - sleep(timeout) - return False + self.event.wait(timeout) + return self.event.isSet() diff --git a/patroni/etcd.py b/patroni/etcd.py index fc6d5a97..d84f5a50 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -10,7 +10,8 @@ import urllib3 from dns.exception import DNSException from dns import resolver -from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string +from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member +from patroni.exceptions import DCSError from patroni.utils import Retry, RetryFailedError, sleep from requests.exceptions import RequestException @@ -51,7 +52,11 @@ class Client(etcd.Client): def api_execute(self, path, method, **kwargs): # Update machines_cache if previous attempt of update has failed self._update_machines_cache and self._load_machines_cache() - return super(Client, self).api_execute(path, method, **kwargs) + try: + return super(Client, self).api_execute(path, method, **kwargs) + except etcd.EtcdConnectionFailed: + self._update_machines_cache = True + raise @staticmethod def get_srv_record(host): @@ -80,7 +85,7 @@ class Client(etcd.Client): for host, port in self.get_srv_record(discovery_srv): url = '{}://{}:{}/members'.format(self._protocol, host, port) try: - response = requests.get(url) + response = requests.get(url, timeout=5) if response.ok: for member in response.json(): ret.extend(member['clientURLs']) @@ -146,14 +151,12 @@ class Etcd(AbstractDCS): def __init__(self, name, config): super(Etcd, self).__init__(name, config) self.ttl = config['ttl'] - self.member_ttl = config.get('member_ttl', 3600) self._retry = Retry(deadline=10, max_delay=1, max_tries=-1, retry_exceptions=(etcd.EtcdConnectionFailed, etcd.EtcdLeaderElectionInProgress, etcd.EtcdWatcherCleared, etcd.EtcdEventIndexCleared)) self.client = self.get_etcd_client(config) - self.cluster = None def retry(self, *args, **kwargs): return self._retry.copy()(*args, **kwargs) @@ -170,16 +173,16 @@ class Etcd(AbstractDCS): @staticmethod def member(node): - conn_url, api_url = parse_connection_string(node.value) - return Member(node.modifiedIndex, os.path.basename(node.key), conn_url, api_url, node.expiration, node.ttl) + return Member.from_node(node.modifiedIndex, os.path.basename(node.key), node.ttl, node.value) - def get_cluster(self): + def _load_cluster(self): try: result = self.retry(self.client.read, self.client_path(''), recursive=True) nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves} # get initialize flag - initialize = bool(nodes.get(self._INITIALIZE, False)) + initialize = nodes.get(self._INITIALIZE, None) + initialize = initialize and initialize.value # get last leader operation last_leader_operation = nodes.get(self._LEADER_OPTIME, None) @@ -191,22 +194,25 @@ class Etcd(AbstractDCS): # get leader leader = nodes.get(self._LEADER, None) if leader: - member = Member(-1, leader.value, None, None, None, None) + member = Member(-1, leader.value, None, {}) member = ([m for m in members if m.name == leader.value] or [member])[0] - leader = Leader(leader.modifiedIndex, leader.expiration, leader.ttl, member) + leader = Leader(leader.modifiedIndex, leader.ttl, member) - self.cluster = Cluster(initialize, leader, last_leader_operation, members) + # failover key + failover = nodes.get(self._FAILOVER, None) + if failover: + failover = Failover.from_node(failover.modifiedIndex, failover.value) + + self._cluster = Cluster(initialize, leader, last_leader_operation, members, failover) except etcd.EtcdKeyNotFound: - self.cluster = Cluster(False, None, None, []) + self._cluster = Cluster(False, None, None, [], None) except: - self.cluster = None logger.exception('get_cluster') raise EtcdError('Etcd is not responding properly') - return self.cluster @catch_etcd_errors def touch_member(self, connection_string, ttl=None): - return self.retry(self.client.set, self.member_path, connection_string, ttl or self.member_ttl) + return self.retry(self.client.set, self.member_path, connection_string, ttl or self.ttl) @catch_etcd_errors def take_leader(self): @@ -222,18 +228,20 @@ class Etcd(AbstractDCS): return False @catch_etcd_errors - def write_leader_optime(self, state_handler): - return self.client.set(self.leader_optime_path, state_handler.last_operation()) + def set_failover_value(self, value, index=None): + return self.client.write(self.failover_path, value, prevIndex=index or 0) @catch_etcd_errors - def update_leader(self, state_handler): - ret = self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl) - ret and self.write_leader_optime(state_handler) - return ret + def write_leader_optime(self, last_operation): + return self.client.set(self.leader_optime_path, last_operation) @catch_etcd_errors - def initialize(self): - return self.client.write(self.initialize_path, self._name, prevExist=False) + def update_leader(self): + return self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl) + + @catch_etcd_errors + def initialize(self, create_new=True, sysid=""): + return self.retry(self.client.write, self.initialize_path, sysid, prevExist=(not create_new)) @catch_etcd_errors def delete_leader(self): @@ -241,13 +249,14 @@ class Etcd(AbstractDCS): @catch_etcd_errors def cancel_initialization(self): - return self.client.delete(self.initialize_path, prevValue=self._name) + return self.retry(self.client.delete, self.initialize_path, prevValue=self._name) def watch(self, timeout): + cluster = self.cluster # watch on leader key changes if it is defined and current node is not lock owner - if self.cluster and self.cluster.leader and self.cluster.leader.name != self._name: + if cluster and cluster.leader and cluster.leader.name != self._name: end_time = time.time() + timeout - index = self.cluster.leader.index + index = cluster.leader.index while index and timeout >= 1: # when timeout is too small urllib3 doesn't have enough time to connect try: @@ -263,4 +272,7 @@ class Etcd(AbstractDCS): timeout = end_time - time.time() - return timeout > 0 and super(Etcd, self).watch(timeout) + try: + return super(Etcd, self).watch(timeout) + finally: + self.event.clear() diff --git a/patroni/ha.py b/patroni/ha.py index 0d95d372..8ed18b16 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1,26 +1,31 @@ +import json import logging import psycopg2 +import requests +import sys +from patroni.async_executor import AsyncExecutor from patroni.exceptions import DCSError, PostgresConnectionException +from multiprocessing.pool import ThreadPool logger = logging.getLogger(__name__) class Ha: - def __init__(self, state_handler, etcd): - self.state_handler = state_handler - self.dcs = etcd + def __init__(self, patroni): + self.patroni = patroni + self.state_handler = patroni.postgresql + self.dcs = patroni.dcs self.cluster = None self.old_cluster = None + self._async_executor = AsyncExecutor() def load_cluster_from_dcs(self): cluster = self.dcs.get_cluster() # We want to keep the state of cluster when it was healhy - if cluster.is_unlocked() and self.cluster and not self.cluster.is_unlocked(): - self.old_cluster = self.cluster - if not self.old_cluster: + if not cluster.is_unlocked() or not self.old_cluster: self.old_cluster = cluster self.cluster = cluster @@ -28,26 +33,51 @@ class Ha: return self.dcs.attempt_to_acquire_leader() def update_lock(self): - return self.dcs.update_leader(self.state_handler) + ret = self.dcs.update_leader() + if ret: + try: + self.dcs.write_leader_optime(self.state_handler.last_operation()) + except: + pass + return ret def has_lock(self): lock_owner = self.cluster.leader and self.cluster.leader.name 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): + data = { + 'conn_url': self.state_handler.connection_string, + 'api_url': self.patroni.api.connection_string, + 'state': self.state_handler.state, + 'role': self.state_handler.role + } + if data['state'] in ['running', 'restarting', 'starting']: + try: + data['xlog_location'] = self.state_handler.xlog_position() + except: + pass + self.dcs.touch_member(json.dumps(data, separators=(',', ':'))) + + def copy_backup_from_leader(self, leader): + if self.state_handler.bootstrap(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): if not self.cluster.is_unlocked(): # cluster already has leader - logger.info('trying to bootstrap from leader', ) - if self.state_handler.bootstrap(self.cluster.leader): - return 'bootstrapped from leader' - else: - self.state_handler.stop('immediate') - self.state_handler.remove_data_directory() - return 'failed to bootstrap from leader' + self._async_executor.schedule('bootstrap from leader') + self._async_executor.run_async(self.copy_backup_from_leader, args=(self.cluster.leader, )) + return 'trying to bootstrap from leader' elif not self.cluster.initialize: # no initialize key - if self.dcs.initialize(): # race for initialization + if self.dcs.initialize(create_new=True): # race for initialization try: self.state_handler.bootstrap() + 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 logger.info("removing initialize key after failed attempt to initialize the cluster") @@ -63,20 +93,38 @@ class Ha: return 'waiting for leader to bootstrap' def recover(self): - if self.state_handler.is_healthy(): - return False has_lock = self.has_lock() - self.state_handler.write_recovery_conf(None if has_lock else self.cluster.leader) - self.state_handler.start() - if has_lock: - logger.info('started as readonly because i had the session lock') - self.load_cluster_from_dcs() - return True + + # try to see if we are the former master that crashed. If so - we likely need to run pg_rewind + # in order to join the former standby being promoted. + pg_controldata = self.state_handler.controldata() + if not has_lock and pg_controldata and\ + pg_controldata.get('Database cluster state', '') == 'in production': # crashed master + self.state_handler.require_rewind() + + # XXX: follow the leader calls stop, which might take quite some time. + # perhaps we should run sync asynchronously + # (we still need the exit code from follow_the_leader) + ret = self.state_handler.follow_the_leader(None if has_lock else self.cluster.leader, recovery=True) + if not ret: + if not has_lock: + return 'failed to start postgres' + self.dcs.delete_leader() + self.dcs.reset_cluster() + return 'removed leader key after trying and failing to start postgres' + if not has_lock: + return 'started as a secondary' + logger.info('started as readonly because i had the session lock') + self.load_cluster_from_dcs() 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 - self.state_handler.follow_the_leader(self.cluster.leader) + leader = self.cluster.leader + leader = None if (leader and leader.name) == self.state_handler.name else leader + if not self.state_handler.check_recovery_conf(leader): + self._async_executor.schedule('changing primary_conninfo and restarting') + self._async_executor.run_async(self.state_handler.follow_the_leader, (leader, )) return ret def enforce_master_role(self, message, promote_message): @@ -86,9 +134,145 @@ class Ha: self.state_handler.promote() return promote_message + @staticmethod + def fetch_node_status(member): + """This function perform http get request on member.api_url and fetches its status + :returns: tuple(`member`, reachable, in_recovery, xlog_location) + + reachable - `!False` if the node is not reachable or is not responding with correct JSON + in_recovery - `!True` if pg_is_in_recovery() == true + xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role.""" + + try: + response = requests.get(member.api_url, timeout=2, verify=False) + logger.info('Got response from %s %s: %s', member.name, member.api_url, response.content) + json = response.json() + is_master = json['role'] == 'master' + xlog_location = json['xlog']['location' if is_master else 'replayed_location'] + return (member, True, not is_master, xlog_location) + except: + logging.exception('request failed: GET %s', member.api_url) + return (member, False, None, 0) + + def fetch_nodes_statuses(self, members): + pool = ThreadPool(len(members)) + results = pool.map(self.fetch_node_status, members) # Run API calls on members in parallel + pool.close() + pool.join() + return results + + def _is_healthiest_node(self, members, check_replication_lag=True): + """This method tries to determine whether I am healthy enough to became a new leader candidate or not.""" + + if self.state_handler.is_leader(): + return True + + if check_replication_lag and not self.state_handler.check_replication_lag(self.cluster.last_leader_operation): + return False # Too far behind last reported xlog location on master + + # Prepare list of nodes to run check against + members = [m for m in members if m.name != self.state_handler.name and m.api_url] + + if members: + my_xlog_location = self.state_handler.xlog_position() + for member, reachable, in_recovery, xlog_location in self.fetch_nodes_statuses(members): + if reachable: # If the node is unreachable it's not healhy + if not in_recovery: + logger.warning('Master (%s) is still alive', member.name) + return False + if my_xlog_location < xlog_location: + return False + return True + + def is_failover_possible(self, members): + ret = False + members = [m for m in members if m.name != self.state_handler.name and m.api_url] + if members: + for member, reachable, in_recovery, xlog_location in self.fetch_nodes_statuses(members): + if reachable: + ret = True # TODO: check xlog_location + else: + logger.info('Member %s is not reachable', member.name) + else: + logger.warning('manual failover: members list is empty') + return ret + + def manual_failover_process_no_leader(self): + failover = self.cluster.failover + if failover.member: # manual failover to specific member + if failover.member == self.state_handler.name: # manual failover to me + return True + + # find specific node and check that it is healthy + members = [m for m in self.cluster.members if m.name == failover.member] + if members: + member, reachable, in_recovery, xlog_location = self.fetch_node_status(members[0]) + if reachable: # node is healthy + logger.info('manual failover: to %s, i am %s', member.name, self.state_handler.name) + return False + # we wanted to failover to specific member but it is not healthy + logger.warning('manual failover: member %s is unhealthy', member.name) + + # at this point we should consider all members as a candidates for failover + # i.e. we assume that failover.member is None + + # try to pick some other members to failover and check that they are healthy + if failover.leader: + if self.state_handler.name == failover.leader: # I was the leader + # exclude me and desired member which is unhealthy (failover.member can be None) + members = [m for m in self.cluster.members if m.name != failover.member] + if self.is_failover_possible(members): # check that there are healthy members + return False + else: # I was the leader and it looks like currently I am the only healthy member + return True + + # at this point we assume that our node is a candidate for a failover among all nodes except former leader + + # exclude former leader from the list (failover.leader can be None) + members = [m for m in self.cluster.members if m.name != failover.leader] + return self._is_healthiest_node(members, check_replication_lag=False) + + def is_healthiest_node(self): + if self.cluster.failover: + return self.manual_failover_process_no_leader() + + # run usual health check + members = {m.name: m for m in self.cluster.members + self.old_cluster.members} + return self._is_healthiest_node(members.values()) + + def demote(self, delete_leader=True): + if delete_leader: + self.state_handler.stop() + self.dcs.delete_leader() + self.dcs.reset_cluster() + self.state_handler.follow_the_leader(None) + + def process_manual_failover_from_leader(self): + 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) + return 'manual failover: demoting myself' + else: + logger.warning('manual failover: no healthy members found, failover is not possible') + else: + logger.warning('manual failover: I am already the leader, no need to failover') + else: + logger.warning('manual failover: leader name does not match: %s != %s', + self.cluster.failover.leader, self.state_handler.name) + + logger.info('Trying to clean up failover key') + self.dcs.manual_failover('', '', self.cluster.failover.index) + def process_unhealthy_cluster(self): - if self.state_handler.is_healthiest_node(self.old_cluster): + if self.is_healthiest_node(): if self.acquire_lock(): + if self.cluster.failover: + logger.info('Cleanning up failover key after acquiring leader lock...') + self.dcs.manual_failover('', '') return self.enforce_master_role('acquired session lock as a leader', 'promoted self to leader by acquiring session lock') else: @@ -100,6 +284,11 @@ class Ha: def process_healthy_cluster(self): if self.has_lock(): + if self.cluster.failover: + msg = self.process_manual_failover_from_leader() + if msg is not None: + return msg + if self.update_lock(): return self.enforce_master_role('no action. i am the leader with the lock', 'promoted self to leader because i had the session lock') @@ -112,34 +301,114 @@ 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 run_cycle(self): + def schedule(self, action): + with self._async_executor: + return self._async_executor.schedule(action) + + def restart_scheduled(self): + return self._async_executor.scheduled_action == 'restart' + + def schedule_reinitialize(self): + return self.schedule('reinitialize') + + def reinitialize_scheduled(self): + return self._async_executor.scheduled_action == 'reinitialize' + + def restart(self): + with self._async_executor: + prev = self._async_executor.schedule('restart', True) + if prev is not None: + return (False, prev + ' already in progress') + if self._async_executor.run(self.state_handler.restart): + return (True, 'restarted successfully') + else: + return (False, 'restart failed') + + def reinitialize(self, cluster): + self.state_handler.stop('immediate') + self.state_handler.remove_data_directory() + self.copy_backup_from_leader(cluster.leader) + + def process_scheduled_action(self): + if self.reinitialize_scheduled(): + if self.cluster.is_unlocked(): + logger.error('Cluster has no leader, can not reinitialize') + self._async_executor.reset_scheduled_action() + elif self.has_lock(): + logger.error('I am the leader, can not reinitialize') + self._async_executor.reset_scheduled_action() + else: + self._async_executor.run_async(self.reinitialize, args=(self.cluster, )) + return 'reinitialize started' + + def handle_long_action_in_progress(self): + if self.has_lock(): + if self.update_lock(): + return 'updated leader lock during ' + self._async_executor.scheduled_action + else: + return 'failed to update leader lock during ' + self._async_executor.scheduled_action + elif self.cluster.is_unlocked(): + return 'not healthy enough for leader race' + else: + return self._async_executor.scheduled_action + ' in progress' + + def sysid_valid(self, sysid): + # sysid does tv_sec << 32, where tv_sec is the number of seconds sine 1970, + # so even 1 << 32 would have 10 digits. + return str(sysid) and len(str(sysid)) >= 10 and str(sysid).isdigit() + + def _run_cycle(self): try: self.load_cluster_from_dcs() + self.touch_member() + # cluster has leader key but not initialize key - if not self.cluster.is_unlocked() and not self.cluster.initialize: - self.dcs.initialize() # fix it + if not self.cluster.is_unlocked() and not self.sysid_valid(self.cluster.initialize) and self.has_lock(): + self.dcs.initialize(create_new=(self.cluster.initialize is None), sysid=self.state_handler.sysid) + + if self._async_executor.busy: + return self.handle_long_action_in_progress() + + # currently it can trigger only reinitialize + msg = self.process_scheduled_action() + if msg is not None: + return msg # is data directory empty? if self.state_handler.data_directory_empty(): return self.bootstrap() # new node # "bootstrap", but data directory is not empty - elif not self.cluster.initialize and self.cluster.is_unlocked(): - self.dcs.initialize() + elif not self.sysid_valid(self.cluster.initialize) and self.cluster.is_unlocked(): + 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: + 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 self.recover() and not self.has_lock(): - # no lock, do not try to promote immediately - return 'started as a secondary' + if not self.state_handler.is_healthy(): + msg = self.recover() + if msg is not None: + return msg - if self.cluster.is_unlocked(): - return self.process_unhealthy_cluster() - else: - return self.process_healthy_cluster() + try: + if self.cluster.is_unlocked(): + return self.process_unhealthy_cluster() + else: + return self.process_healthy_cluster() + finally: + self.state_handler.sync_replication_slots(self.cluster) except DCSError: logger.error('Error communicating with DCS') - if self.state_handler.is_leader(): - self.state_handler.demote(None) + if self.state_handler.is_running() and self.state_handler.is_leader(): + self.demote(delete_leader=False) return 'demoted self because DCS is not accessible and i was a leader' except (psycopg2.Error, PostgresConnectionException): - logger.exception('Error communicating with Postgresql. Will try again') + logger.exception('Error communicating with Postgresql. Will try again later') + + def run_cycle(self): + with self._async_executor: + return self._run_cycle() diff --git a/patroni/postgresql.py b/patroni/postgresql.py index a68b8dfd..09571b80 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -9,6 +9,7 @@ import time from patroni.exceptions import PostgresConnectionException, PostgresException from patroni.utils import Retry, RetryFailedError from six.moves.urllib_parse import urlparse +from threading import Lock logger = logging.getLogger(__name__) @@ -47,6 +48,8 @@ class Postgresql: self.replication = config['replication'] self.superuser = config['superuser'] self.admin = config['admin'] + self.pgpass = config.get('pgpass', None) or os.path.join(os.path.expanduser('~'), 'pgpass') + self.pg_rewind = config.get('pg_rewind', {}) self.callback = config.get('callbacks', {}) self.use_slots = config.get('use_slots', True) self.schedule_load_slots = self.use_slots @@ -56,7 +59,6 @@ class Postgresql: self.postmaster_pid = os.path.join(self.data_dir, 'postmaster.pid') self.trigger_file = config.get('recovery_conf', {}).get('trigger_file', None) or 'promote' self.trigger_file = os.path.abspath(os.path.join(self.data_dir, self.trigger_file)) - self._role = 'replica' self._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir] @@ -67,8 +69,50 @@ class Postgresql: self._connection = None self._cursor_holder = None - self.members = [] # list of already existing replication slots - self.retry = Retry(max_tries=-1, deadline=10, max_delay=1, retry_exceptions=PostgresConnectionException) + self._need_rewind = False + self._sysid = None + self.replication_slots = [] # list of already existing replication slots + self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException) + + self._state = 'stopped' + self._state_lock = Lock() + self._role = 'replica' + self._role_lock = Lock() + + if self.is_running(): + self._state = 'running' + self._role = 'master' if self.is_leader() else 'replica' + + @property + def can_rewind(self): + """ check if pg_rewind executable is there and that pg_controldata indicates + we have either wal_log_hints or checksums turned on + """ + # low-hanging fruit: check if pg_rewind configuration is there + if not self.pg_rewind or\ + not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')): + return False + + cmd = ['pg_rewind', '--help'] + try: + ret = subprocess.call(cmd, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) + if ret != 0: # pg_rewind is not there, close up the shop and go home + return False + except OSError: + return False + # check if the cluster's configuration permits pg_rewind + data = self.controldata() + return data.get('wal_log_hints setting', 'off') == 'on' or data.get('Data page checksum version', '0') != '0' + + @property + def sysid(self): + if not self._sysid: + data = self.controldata() + self._sysid = data.get('Database system identifier', "") + return self._sysid + + def require_rewind(self): + self._need_rewind = True def get_local_address(self): listen_addresses = self.listen_addresses.split(',') @@ -89,9 +133,15 @@ class Postgresql: def _cursor(self): 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() 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") + def _query(self, sql, *params): cursor = None try: @@ -101,6 +151,8 @@ class Postgresql: except psycopg2.Error as e: if cursor and cursor.connection.closed == 0: raise e + if self.state == 'restarting': + raise RetryFailedError('cluster is being restarted') raise PostgresConnectionException('connection problems') def query(self, sql, *params): @@ -113,23 +165,30 @@ class Postgresql: return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == [] def initialize(self): + self.set_state('initalizing new cluster') ret = subprocess.call(self._pg_ctl + ['initdb', '-o', '--encoding=UTF8']) == 0 - ret and self.write_pg_hba() + if ret: + self.write_pg_hba() + else: + self.set_state('initdb failed') return ret def delete_trigger_file(self): os.path.exists(self.trigger_file) and os.unlink(self.trigger_file) + def write_pgpass(self, record): + with open(self.pgpass, 'w') as f: + os.fchmod(f.fileno(), 0o600) + f.write('{host}:{port}:*:{user}:{password}\n'.format(**record)) + + env = os.environ.copy() + env['PGPASSFILE'] = self.pgpass + return env + def sync_from_leader(self, leader): r = parseurl(leader.conn_url) - pgpass = 'pgpass' - with open(pgpass, 'w') as f: - os.fchmod(f.fileno(), 0o600) - f.write('{host}:{port}:*:{user}:{password}\n'.format(**r)) - - env = os.environ.copy() - env['PGPASSFILE'] = pgpass + env = self.write_pgpass(r) return self.create_replica(r, env) == 0 @staticmethod @@ -213,40 +272,82 @@ class Postgresql: @property def role(self): - return self._role + with self._role_lock: + return self._role + + def set_role(self, value): + with self._role_lock: + self._role = value + + @property + def state(self): + with self._state_lock: + return self._state + + def set_state(self, value): + with self._state_lock: + self._state = value def start(self, block_callbacks=False): if self.is_running(): - self._role = 'master' if self.is_leader() else 'replica' - self.schedule_load_slots = self.use_slots logger.error('Cannot start PostgreSQL because one is already running.') - return False + return True - self._role = 'replica' if os.path.exists(self.recovery_conf) else 'master' + self.set_role('replica' if os.path.exists(self.recovery_conf) else 'master') if os.path.exists(self.postmaster_pid): os.remove(self.postmaster_pid) logger.info('Removed %s', self.postmaster_pid) + if not block_callbacks: + self.set_state('starting') + ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0 + + self.set_state('running' if ret else 'start failed') + 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 ret and self.call_nowait(ACTION_ON_START) + ret and not block_callbacks and self.call_nowait(ACTION_ON_START) return ret + def checkpoint(self): + try: + r = parseurl('postgres://{}/postgres'.format(self.local_address)) + r['options'] = '-c statement_timeout=0' + with psycopg2.connect(**r) as conn: + conn.autocommit = True + with conn.cursor() as cur: + cur.execute('CHECKPOINT') + except: + logging.exception('Exception during CHECKPOINT') + def stop(self, mode='fast', block_callbacks=False): + # make sure we close all connections established against + # the former node, otherwise, we might get a stalled one + # after kill -9, which would report incorrect data to + # patroni. + + self.close_connection() + if not self.is_running(): + if not block_callbacks: + self.set_state('stopped') + return True + if block_callbacks: - try: - self.query('SET statement_timeout TO 0') - self.query('CHECKPOINT') - except: - logging.exception('Exception diring CHECKPOINT') + self.checkpoint() + else: + self.set_state('stopping') ret = subprocess.call(self._pg_ctl + ['stop', '-m', mode]) == 0 # 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_STOP) + if not ret: + self.set_state('stop failed') + elif not block_callbacks: + self.set_state('stopped') + self.call_nowait(ACTION_ON_STOP) return ret def reload(self): @@ -255,8 +356,12 @@ class Postgresql: return ret def restart(self): + self.set_state('restarting') ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True) - ret and self.call_nowait(ACTION_ON_RESTART) + if ret: + self.call_nowait(ACTION_ON_RESTART) + else: + self.set_state('restart failed ({})'.format(self.state)) return ret def server_options(self): @@ -271,36 +376,9 @@ class Postgresql: return False return True - def is_healthiest_node(self, cluster): - if self.is_leader(): - return True - - if cluster.last_leader_operation - self.xlog_position() > self.config.get('maximum_lag_on_failover', 0): - return False - - for member in cluster.members: - if member.name == self.name: - continue - try: - r = parseurl(member.conn_url) - member_conn = psycopg2.connect(**r) - member_conn.autocommit = True - member_cursor = member_conn.cursor() - member_cursor.execute( - "SELECT pg_is_in_recovery(), %s - pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')", - (self.xlog_position(),)) - row = member_cursor.fetchone() - member_cursor.close() - member_conn.close() - logger.error([self.name, member.name, row]) - if not row[0]: - logger.warning('Master (%s) is still alive', member.name) - return False - if row[1] < 0: - return False - except psycopg2.Error: - continue - return True + def check_replication_lag(self, last_leader_operation): + return (last_leader_operation if last_leader_operation else 0) - self.xlog_position() <=\ + self.config.get('maximum_lag_on_failover', 0) def write_pg_hba(self): with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f: @@ -324,10 +402,7 @@ class Postgresql: with open(self.recovery_conf, 'r') as f: for line in f: if line.startswith('primary_conninfo'): - if not pattern: - return False - return pattern in line - + return pattern and (pattern in line) return not pattern def write_recovery_conf(self, leader): @@ -342,40 +417,154 @@ recovery_target_timeline = 'latest' for name, value in self.config.get('recovery_conf', {}).items(): f.write("{} = '{}'\n".format(name, value)) - def follow_the_leader(self, leader): - if not self.check_recovery_conf(leader): + def rewind(self, leader): + # prepare pg_rewind connection + r = parseurl(leader.conn_url) + r.update(self.pg_rewind) + r['user'] = r['username'] + env = self.write_pgpass(r) + pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r) + logger.info("running pg_rewind from {}".format(pc)) + pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc] + try: + ret = (subprocess.call(pg_rewind, env=env) == 0) + except: + ret = False + if ret: self.write_recovery_conf(leader) - run_callback = self.role == 'master' - self.restart() - run_callback and self.call_nowait(ACTION_ON_ROLE_CHANGE) + return ret + + def controldata(self): + """ return the contents of pg_controldata, or non-True value if pg_controldata call failed """ + result = {} + try: + data = subprocess.check_output(['pg_controldata', self.data_dir]) + if data: + data = data.decode().splitlines() + result = {l.split(':')[0].replace('Current ', '', 1): l.split(':')[1].strip() for l in data if l} + except subprocess.CalledProcessError: + logger.exception("Error when calling pg_controldata") + finally: + return result + + def read_postmaster_opts(self): + """ returns the list of option names/values from postgres.opts, Empty dict if read failed or no file """ + result = {} + try: + with open(os.path.join(self.data_dir, "postmaster.opts")) as f: + data = f.read() + opts = [opt.strip('"\n') for opt in data.split(' "')] + for opt in opts: + if '=' in opt and opt.startswith('--'): + name, val = opt.split('=', 1) + name = name.strip('-') + result[name] = val + except IOError: + logger.exception('Error when reading postmaster.opts') + finally: + return result + + def single_user_mode(self, command=None, options={}): + """ run a given command in a single-user mode. If the command is empty - then just start and stop """ + cmd = ['postgres', '--single', '-D', self.data_dir] + for opt in sorted(options): + cmd.extend(['-c', '{0}={1}'.format(opt, options[opt])]) + # need a database name to connect + cmd.append('postgres') + p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT) + if p: + command and p.communicate('{}\n'.format(command)) + p.stdin.close() + return p.wait() + return 1 + + def cleanup_archive_status(self): + status_dir = os.path.join(self.data_dir, 'pg_xlog', 'archive_status') + if os.path.isdir(status_dir): + for f in os.listdir(status_dir): + path = os.path.join(status_dir, f) + try: + if os.path.islink(path): + os.unlink(path) + elif os.path.isfile(path): + os.remove(path) + except: + logger.exception("Unable to remove {}".format(path)) + + def follow_the_leader(self, leader, recovery=False): + if not self.check_recovery_conf(leader) or recovery: + change_role = (self.role == 'master') + + self._need_rewind = (self._need_rewind or change_role) and self.can_rewind + if self._need_rewind: + logger.info("set the rewind flag after demote") + self.write_recovery_conf(leader) + if not leader or not self._need_rewind: # do not rewind until the leader becomes available + ret = self.restart() + else: # we have a leader and need to rewind + if self.is_running(): + self.stop() + # at present, pg_rewind only runs when the cluster is shut down cleanly + # and not shutdown in recovery. We have to remove the recovery.conf if present + # and start/shutdown in a single user mode to emulate this. + # XXX: if recovery.conf is linked, it will be written anew as a normal file. + if os.path.islink(self.recovery_conf): + os.unlink(self.recovery_conf) + else: + os.remove(self.recovery_conf) + # Archived segments might be useful to pg_rewind, + # clean the flags that tell we should remove them. + self.cleanup_archive_status() + # Start in a single user mode and stop to produce a clean shutdown + opts = self.read_postmaster_opts() + opts['archive_mode'] = 'on' + opts['archive_command'] = 'false' + self.single_user_mode(options=opts) + if self.rewind(leader): + ret = self.start() + else: + logger.error("unable to rewind the former master") + self.remove_data_directory() + ret = True + self._need_rewind = False + change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE) + return ret + else: + return True 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 + copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files + - originally stored as symlinks, those are normally skipped by pg_basebackup + - in case of WAL-E basebackup (see http://comments.gmane.org/gmane.comp.db.postgresql.wal-e/239) """ - for f in self.configuration_to_save: - shutil.copy(f, f + '.backup') + try: + for f in self.configuration_to_save: + os.path.isfile(f) and shutil.copy(f, f + '.backup') + except: + logger.exception('unable to create backup copies of configuration files') def restore_configuration_files(self): """ restore a previously saved postgresql.conf """ try: for f in self.configuration_to_save: - 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 from WAL-E backup') + logger.exception('unable to restore configuration files from backup') def promote(self): if self.role == 'master': return True ret = subprocess.call(self._pg_ctl + ['promote']) == 0 if ret: - self._role = 'master' + self.set_role('master') + logger.info("cleared rewind flag after becoming the leader") + self._need_rewind = False self.call_nowait(ACTION_ON_ROLE_CHANGE) return ret - def demote(self, leader): - self.follow_the_leader(leader) + def demote(self): + self.follow_the_leader(None) def create_replication_user(self): self.query('CREATE USER "{}" WITH REPLICATION ENCRYPTED PASSWORD %s'.format( @@ -397,31 +586,34 @@ recovery_target_timeline = 'latest' return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery() THEN pg_last_xlog_replay_location() ELSE pg_current_xlog_location() - END, '0/0')""").fetchone()[0] + END, '0/0')::bigint""").fetchone()[0] def load_replication_slots(self): if self.use_slots and self.schedule_load_slots: cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'") - self.members = [r[0] for r in cursor] + self.replication_slots = [r[0] for r in cursor] self.schedule_load_slots = False def sync_replication_slots(self, cluster): if self.use_slots: - self.load_replication_slots() - members = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else [] - # drop unused slots - for slot in set(self.members) - set(members): - self.query("""SELECT pg_drop_replication_slot(%s) - WHERE EXISTS(SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + try: + self.load_replication_slots() + slots = [m.name for m in cluster.members if m.name != self.name] if self.role == 'master' else [] + # drop unused slots + for slot in set(self.replication_slots) - set(slots): + self.query("""SELECT pg_drop_replication_slot(%s) + WHERE EXISTS(SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) - # create new slots - for slot in set(members) - set(self.members): - self.query("""SELECT pg_create_physical_replication_slot(%s) - WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots - WHERE slot_name = %s)""", slot, slot) + # create new slots + for slot in set(slots) - set(self.replication_slots): + self.query("""SELECT pg_create_physical_replication_slot(%s) + WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots + WHERE slot_name = %s)""", slot, slot) - self.members = members + self.replication_slots = slots + except: + logger.exception('Exception when changing replication slots') def last_operation(self): return str(self.xlog_position()) @@ -446,6 +638,7 @@ recovery_target_timeline = 'latest' raise PostgresException("Could not bootstrap master PostgreSQL") else: if self.sync_from_leader(current_leader): + self.restore_configuration_files() self.write_recovery_conf(current_leader) ret = self.start() return ret diff --git a/patroni/scripts/aws.py b/patroni/scripts/aws.py index b172a8c3..a0622f15 100755 --- a/patroni/scripts/aws.py +++ b/patroni/scripts/aws.py @@ -1,4 +1,4 @@ -#!/usr/bin/python +#!/usr/bin/env python import logging import requests diff --git a/patroni/scripts/restore.py b/patroni/scripts/restore.py index 4ac091f3..6b20e3e8 100755 --- a/patroni/scripts/restore.py +++ b/patroni/scripts/restore.py @@ -1,4 +1,4 @@ -#!/usr/bin/python +#!/usr/bin/env python # arguments are: # - cluster scope # - cluster role diff --git a/patroni/utils.py b/patroni/utils.py index 681ca39c..9b040294 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -36,6 +36,8 @@ def calculate_ttl(expiration): """ >>> calculate_ttl(None) >>> calculate_ttl('2015-06-10 12:56:30.552539016Z') + >>> calculate_ttl('2015-06-10T12:56:30.552539016Z') < 0 + True """ if not expiration: return None diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 9cd15fbc..1fa2cca3 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -5,7 +5,8 @@ import time from kazoo.client import KazooClient, KazooState from kazoo.exceptions import NoNodeError, NodeExistsError -from patroni.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string +from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member +from patroni.exceptions import DCSError from patroni.utils import sleep from requests.exceptions import RequestException @@ -90,9 +91,8 @@ class ZooKeeper(AbstractDCS): 'max_tries': -1}, connection_retry={'max_delay': 1, 'max_tries': -1}) self.client.add_listener(self.session_listener) - self.cluster_event = self.client.handler.event_object() - self.cluster = None + self._my_member_data = None self.fetch_cluster = True self.last_leader_operation = 0 @@ -104,7 +104,7 @@ class ZooKeeper(AbstractDCS): def cluster_watcher(self, event): self.fetch_cluster = True - self.cluster_event.set() + self.event.set() def get_node(self, key, watch=None): try: @@ -115,8 +115,7 @@ class ZooKeeper(AbstractDCS): @staticmethod def member(name, value, znode): - conn_url, api_url = parse_connection_string(value) - return Member(znode.version, name, conn_url, api_url, None, None) + return Member.from_node(znode.version, name, znode.ephemeralOwner, value) def get_children(self, key, watch=None): try: @@ -133,17 +132,20 @@ class ZooKeeper(AbstractDCS): return members def _inner_load_cluster(self): - self.cluster_event.clear() - nodes = set(self.get_children(self.client_path(''))) + self.fetch_cluster = False + self.event.clear() + nodes = set(self.get_children(self.client_path(''), self.cluster_watcher)) + if not nodes: + self.fetch_cluster = True # get initialize flag - initialize = self._INITIALIZE in nodes + initialize = (self.get_node(self.initialize_path) or [None])[0] if self._INITIALIZE in nodes else None # get list of members members = self.load_members() if self._MEMBERS[:-1] in nodes else [] # get leader - leader = self.get_node(self.leader_path, self.cluster_watcher) if self._LEADER in nodes else None + leader = self.get_node(self.leader_path) if self._LEADER in nodes else None if leader: client_id = self.client.client_id if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner: @@ -152,17 +154,22 @@ class ZooKeeper(AbstractDCS): leader = None if leader: - member = Member(-1, leader[0], None, None, None, None) + member = Member(-1, leader[0], None, {}) member = ([m for m in members if m.name == leader[0]] or [member])[0] - leader = Leader(leader[1].version, None, None, member) + leader = Leader(leader[1].version, leader[1].ephemeralOwner, member) self.fetch_cluster = member.index == -1 - # get last leader operation - self.last_leader_operation = self.get_node(self.leader_optime_path) if self.fetch_cluster else None - self.last_leader_operation = 0 if self.last_leader_operation is None else int(self.last_leader_operation[0]) - self.cluster = Cluster(initialize, leader, self.last_leader_operation, members) + # failover key + failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None + if failover: + failover = Failover.from_node(failover[1].version, failover[0]) - def get_cluster(self): + # get last leader operation + optime = self.get_node(self.leader_optime_path) if self._OPTIME in nodes and self.fetch_cluster else None + self.last_leader_operation = 0 if optime is None else int(optime[0]) + self._cluster = Cluster(initialize, leader, self.last_leader_operation, members, failover) + + def _load_cluster(self): if self.exhibitor and self.exhibitor.poll(): self.client.set_hosts(self.exhibitor.zookeeper_hosts) @@ -170,11 +177,9 @@ class ZooKeeper(AbstractDCS): try: self.client.retry(self._inner_load_cluster) except: - self.cluster = None logger.exception('get_cluster') self.session_listener(KazooState.LOST) raise ZooKeeperError('ZooKeeper in not responding properly') - return self.cluster def _create(self, path, value, **kwargs): try: @@ -188,31 +193,61 @@ class ZooKeeper(AbstractDCS): ret or logger.info('Could not take out TTL lock') return ret - def initialize(self): - return self._create(self.initialize_path, self._name, makepath=True) - - def touch_member(self, connection_string, ttl=None): - if self.cluster and any(m.name == self._name for m in self.cluster.members): - return True - path = self.member_path - connection_string = connection_string.encode('utf-8') + def set_failover_value(self, value, index=None): try: - self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True) + self.client.retry(self.client.set, self.failover_path, value.encode('utf-8'), version=index or -1) + return True + except NoNodeError: + return value == '' or (not index and self._create(self.failover_path, value)) + except: + logging.exception('set_failover_value') + return False + + def initialize(self, create_new=True, sysid=""): + return self._create(self.initialize_path, sysid, makepath=True) if create_new \ + else self.client.retry(self.client.set, self.initialize_path, sysid.encode("utf-8")) + + def touch_member(self, data, ttl=None): + cluster = self.cluster + me = cluster and ([m for m in cluster.members if m.name == self._name] or [None])[0] + path = self.member_path + data = data.encode('utf-8') + create = not me + if me and self.client.client_id is not None and me.session != self.client.client_id[0]: + try: + self.client.retry(self.client.delete, path) + except NoNodeError: + pass + except: + return False + create = True + + if not create and data == self._my_member_data: + return True + + try: + if create: + self.client.retry(self.client.create, path, data, makepath=True, ephemeral=True) + else: + self.client.retry(self.client.set, path, data) + self._my_member_data = data return True except NodeExistsError: try: - self.client.retry(self.client.delete, path) - self.client.retry(self.client.create, path, connection_string, makepath=True, ephemeral=True) + self.client.retry(self.client.set, path, data) + self._my_member_data = data return True except: logger.exception('touch_member') + except: + logger.exception('touch_member') return False def take_leader(self): return self.attempt_to_acquire_leader() - def update_leader(self, state_handler): - last_operation = state_handler.last_operation().encode('utf-8') + def write_leader_optime(self, last_operation): + last_operation = last_operation.encode('utf-8') if last_operation != self.last_leader_operation: self.last_leader_operation = last_operation path = self.leader_optime_path @@ -225,11 +260,14 @@ class ZooKeeper(AbstractDCS): logger.exception('Failed to create %s', path) except: logger.exception('Failed to update %s', path) + + def update_leader(self): return True def delete_leader(self): - if isinstance(self.cluster, Cluster) and self.cluster.leader.name == self._name: - self.client.delete(self.leader_path, version=self.cluster.leader.index) + self.client.restart() + self._my_member_data = None + return True def _cancel_initialization(self): node = self.get_node(self.initialize_path) @@ -243,8 +281,6 @@ class ZooKeeper(AbstractDCS): logger.exception("Unable to delete initialize key") def watch(self, timeout): - self.cluster_event.wait(timeout) - if self.cluster_event.isSet(): + if super(ZooKeeper, self).watch(timeout): self.fetch_cluster = True - return not self.cluster or not self.cluster.leader or self.cluster.leader.name != self._name - return False + return self.fetch_cluster diff --git a/postgres0.yml b/postgres0.yml index 50f4065f..41fca705 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -34,6 +34,10 @@ postgresql: data_dir: data/postgresql0 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True + pgpass: /tmp/pgpass0 + pg_rewind: + username: postgres + password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -42,6 +46,7 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: + username: postgres password: zalando admin: username: admin @@ -62,3 +67,4 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" + wal_log_hints: "on" diff --git a/postgres1.yml b/postgres1.yml index dd2e2369..68597f2a 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -34,6 +34,10 @@ postgresql: data_dir: data/postgresql1 maximum_lag_on_failover: 1048576 # 1 megabyte in bytes use_slots: True + pgpass: /tmp/pgpass1 + pg_rewind: + username: postgres + password: zalando pg_hba: - host all all 0.0.0.0/0 md5 - hostssl all all 0.0.0.0/0 md5 @@ -42,6 +46,7 @@ postgresql: password: rep-pass network: 127.0.0.1/32 superuser: + user: postgres password: zalando admin: username: admin @@ -62,3 +67,4 @@ postgresql: archive_timeout: 1800s max_replication_slots: 5 hot_standby: "on" + wal_log_hints: "on" diff --git a/requirements-py2.txt b/requirements-py2.txt index d57df720..fde9c79a 100644 --- a/requirements-py2.txt +++ b/requirements-py2.txt @@ -1,7 +1,7 @@ boto dnspython mock -psycopg2 +psycopg2>=2.6.1 PyYAML requests six >= 1.7 diff --git a/requirements-py3.txt b/requirements-py3.txt index 0fd9dfb3..cc00965b 100644 --- a/requirements-py3.txt +++ b/requirements-py3.txt @@ -1,7 +1,7 @@ boto mock dnspython3 -psycopg2 +psycopg2>=2.6.1 PyYAML requests six diff --git a/tests/test_api.py b/tests/test_api.py index b8f9e038..72d5c2b0 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -3,6 +3,7 @@ import unittest from mock import Mock, patch from patroni.api import RestApiHandler, RestApiServer +from patroni.dcs import Member from six import BytesIO as IO from six.moves import BaseHTTPServer from test_postgresql import psycopg2_connect, MockCursor @@ -10,6 +11,10 @@ from test_postgresql import psycopg2_connect, MockCursor class MockPostgresql(Mock): + name = 'test' + state = 'running' + role = 'master' + def connection(self): return psycopg2_connect() @@ -17,9 +22,32 @@ class MockPostgresql(Mock): return True +class MockHa(Mock): + + dcs = Mock() + state_handler = MockPostgresql() + + def schedule_restart(self): + return 'restart' + + def schedule_reinitialize(self): + return 'reinitialize' + + def restart(self): + return (True, '') + + def restart_scheduled(self): + return False + + def fetch_nodes_statuses(self, members): + return [[None, True, None, None]] + + class MockPatroni: postgresql = MockPostgresql() + ha = MockHa() + dcs = Mock() class MockRequest: @@ -47,21 +75,77 @@ class MockRestApiServer(RestApiServer): class TestRestApiHandler(unittest.TestCase): def test_do_GET(self): - MockRestApiServer(RestApiHandler, b'GET /') - with patch.object(RestApiServer, 'query', Mock(side_effect=psycopg2.OperationalError())): - MockRestApiServer(RestApiHandler, b'GET /') - - def test_do_GET_sampleauth(self): - MockRestApiServer(RestApiHandler, b'GET /sampleauth') - MockRestApiServer(RestApiHandler, b'GET /sampleauth\nAuthorization:') - MockRestApiServer(RestApiHandler, b'GET /sampleauth\nAuthorization: Basic dGVzdDp0ZXN0') + MockRestApiServer(RestApiHandler, b'GET /replica') + with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})): + MockRestApiServer(RestApiHandler, b'GET /replica') + with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})): + MockRestApiServer(RestApiHandler, b'GET /replica') + MockRestApiServer(RestApiHandler, b'GET /master') + MockPatroni.dcs.cluster.leader.name = MockPostgresql.name + MockRestApiServer(RestApiHandler, b'GET /replica') + MockPatroni.dcs.cluster = None + with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'master'})): + MockRestApiServer(RestApiHandler, b'GET /master') + with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)): + MockRestApiServer(RestApiHandler, b'GET /master') + MockRestApiServer(RestApiHandler, b'GET /master') def test_do_GET_patroni(self): MockRestApiServer(RestApiHandler, b'GET /patroni') + def test_basicauth(self): + MockRestApiServer(RestApiHandler, b'POST /restart HTTP/1.0') + MockRestApiServer(RestApiHandler, b'POST /restart HTTP/1.0\nAuthorization:') + + def test_do_POST_restart(self): + request = b'POST /restart HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0' + MockRestApiServer(RestApiHandler, request) + with patch.object(MockHa, 'restart', Mock(side_effect=Exception)): + MockRestApiServer(RestApiHandler, request) + + @patch.object(MockHa, 'dcs') + def test_do_POST_reinitialize(self, dcs): + cluster = dcs.get_cluster.return_value + request = b'POST /reinitialize HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0' + MockRestApiServer(RestApiHandler, request) + cluster.is_unlocked.return_value = False + MockRestApiServer(RestApiHandler, request) + with patch.object(MockHa, 'schedule_reinitialize', Mock(return_value=None)): + MockRestApiServer(RestApiHandler, request) + cluster.leader.name = 'test' + MockRestApiServer(RestApiHandler, request) + @patch('time.sleep', Mock()) def test_RestApiServer_query(self): with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)): MockRestApiServer(RestApiHandler, b'GET /patroni') with patch.object(MockPostgresql, 'connection', Mock(side_effect=psycopg2.OperationalError)): MockRestApiServer(RestApiHandler, b'GET /patroni') + + @patch('time.sleep', Mock()) + @patch.object(MockHa, 'dcs') + def test_do_POST_failover(self, dcs): + cluster = dcs.get_cluster.return_value + request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\ + b'Content-Length: 25\n\n{"leader": "postgresql1"}' + MockRestApiServer(RestApiHandler, request) + cluster.leader.name = 'postgresql1' + MockRestApiServer(RestApiHandler, request) + cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'})] + MockRestApiServer(RestApiHandler, request) + with patch.object(MockPatroni, 'dcs') as d: + cluster = d.get_cluster.return_value + cluster.leader.name = 'postgresql0' + MockRestApiServer(RestApiHandler, request) + cluster.leader.name = 'postgresql1' + cluster.failover = None + MockRestApiServer(RestApiHandler, request) + d.get_cluster = Mock(side_effect=Exception()) + MockRestApiServer(RestApiHandler, request) + d.manual_failover.return_value = False + MockRestApiServer(RestApiHandler, request) + with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])): + MockRestApiServer(RestApiHandler, request) + request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\ + b'Content-Length: 50\n\n{"leader": "postgresql1", "member": "postgresql2"}' + MockRestApiServer(RestApiHandler, request) diff --git a/tests/test_async_executor.py b/tests/test_async_executor.py new file mode 100644 index 00000000..17d7f94a --- /dev/null +++ b/tests/test_async_executor.py @@ -0,0 +1,18 @@ +import unittest + +from mock import Mock, patch +from patroni.async_executor import AsyncExecutor +from threading import Thread + + +class TestAsyncExecutor(unittest.TestCase): + + def setUp(self): + self.a = AsyncExecutor() + + @patch.object(Thread, 'start', Mock()) + def test_run_async(self): + self.a.run_async(Mock(return_value=True)) + + def test_run(self): + self.a.run(Mock(side_effect=Exception())) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index d2824699..d0d01d71 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -1,4 +1,3 @@ -import datetime import etcd import json import requests @@ -8,7 +7,7 @@ import unittest from dns.exception import DNSException from mock import Mock, patch -from patroni.dcs import Cluster, DCSError, Leader, Member +from patroni.dcs import Cluster, DCSError, Leader from patroni.etcd import Client, Etcd @@ -50,6 +49,8 @@ def requests_get(url, **kwargs): response = MockResponse() if url.startswith('http://local'): raise requests.exceptions.RequestException() + elif ':8011/patroni' in url: + response.content = '{"role": "replica", "xlog": {"replayed_location": 0}}' elif url.endswith('/members'): if url.startswith('http://error'): response.content = '[{}]' @@ -92,6 +93,8 @@ def etcd_read(key, **kwargs): raise etcd.EtcdKeyNotFound response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [ + {"key": "/service/batman5/failover", "value": "", + "modifiedIndex": 1582, "createdIndex": 1582}, {"key": "/service/batman5/initialize", "value": "postgresql0", "modifiedIndex": 1582, "createdIndex": 1582}, {"key": "/service/batman5/leader", "value": "postgresql1", @@ -103,13 +106,13 @@ def etcd_read(key, **kwargs): "modifiedIndex": 20437, "createdIndex": 20437}, {"key": "/service/batman5/members", "dir": True, "nodes": [ {"key": "/service/batman5/members/postgresql1", - "value": "postgres://replicator:rep-pass@127.0.0.1:5434/postgres" - + "?application_name=http://127.0.0.1:8009/patroni", + "value": "postgres://replicator:rep-pass@127.0.0.1:5434/postgres" + + "?application_name=http://127.0.0.1:8009/patroni", "expiration": "2015-05-15T09:10:59.949384522Z", "ttl": 21, "modifiedIndex": 20727, "createdIndex": 20727}, {"key": "/service/batman5/members/postgresql0", - "value": "postgres://replicator:rep-pass@127.0.0.1:5433/postgres" - + "?application_name=http://127.0.0.1:8008/patroni", + "value": "postgres://replicator:rep-pass@127.0.0.1:5433/postgres" + + "?application_name=http://127.0.0.1:8008/patroni", "expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30, "modifiedIndex": 20730, "createdIndex": 20730}], "modifiedIndex": 1581, "createdIndex": 1581}], "modifiedIndex": 1581, "createdIndex": 1581}} @@ -145,15 +148,6 @@ def http_request(method, url, **kwargs): raise socket.error -class TestMember(unittest.TestCase): - - def test_real_ttl(self): - now = datetime.datetime.utcnow() - member = Member(0, 'a', 'b', 'c', (now + datetime.timedelta(seconds=2)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None) - self.assertLess(member.real_ttl(), 2) - self.assertEquals(Member(0, 'a', 'b', 'c', '', None).real_ttl(), -1) - - @patch('dns.resolver.query', dns_query) @patch('socket.getaddrinfo', socket_getaddrinfo) @patch('requests.get', requests_get) @@ -171,6 +165,11 @@ class TestClient(unittest.TestCase): self.client._base_uri = 'http://localhost:4001' self.client._machines_cache = ['http://localhost:2379'] self.client.api_execute('/', 'GET') + self.client._update_machines_cache = False + self.client._base_uri = 'http://localhost:4001' + self.client._machines_cache = [] + self.assertRaises(etcd.EtcdConnectionFailed, self.client.api_execute, '/', 'GET') + self.assertTrue(self.client._update_machines_cache) def test_get_srv_record(self): self.assertEquals(self.client.get_srv_record('blabla'), []) @@ -199,7 +198,6 @@ class TestClient(unittest.TestCase): self.assertRaises(etcd.EtcdException, self.client._load_machines_cache) -@patch('time.sleep', Mock()) @patch('requests.get', requests_get) class TestEtcd(unittest.TestCase): @@ -242,8 +240,11 @@ class TestEtcd(unittest.TestCase): self.etcd._base_path = '/service/failed' self.assertFalse(self.etcd.attempt_to_acquire_leader()) + def test_write_leader_optime(self): + self.etcd.write_leader_optime('0') + def test_update_leader(self): - self.assertTrue(self.etcd.update_leader(MockPostgresql())) + self.assertTrue(self.etcd.update_leader()) def test_initialize(self): self.assertFalse(self.etcd.initialize()) @@ -256,7 +257,7 @@ class TestEtcd(unittest.TestCase): def test_watch(self): self.etcd.client.watch = etcd_watch - self.etcd.watch(100) + self.etcd.watch(0) self.etcd.get_cluster() self.etcd.watch(1.5) self.etcd.watch(4.5) diff --git a/tests/test_ha.py b/tests/test_ha.py index 925beb33..e34f9b8e 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1,11 +1,12 @@ +import etcd import unittest -from mock import Mock, patch -from patroni.dcs import Cluster, DCSError, Leader, Member +from mock import Mock, MagicMock, patch +from patroni.dcs import Cluster, Failover, Leader, Member from patroni.etcd import Client, Etcd -from patroni.exceptions import PostgresException +from patroni.exceptions import DCSError, PostgresException from patroni.ha import Ha -from test_etcd import socket_getaddrinfo, etcd_read, etcd_write +from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get def true(*args, **kwargs): @@ -16,28 +17,33 @@ def false(*args, **kwargs): return False -def get_cluster(initialize, leader): - return Cluster(initialize, leader, None, None) +def get_cluster(initialize, leader, members, failover): + return Cluster(initialize, leader, None, members, failover) def get_cluster_not_initialized_without_leader(): - return get_cluster(None, None) + return get_cluster(None, None, [], None) -def get_cluster_initialized_without_leader(): - return get_cluster(True, None) +def get_cluster_initialized_without_leader(leader=False, failover=None): + m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', + 'api_url': 'http://127.0.0.1:8008/patroni'}) + l = Leader(0, 0, m) if leader else None + o = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', + 'api_url': 'http://127.0.0.1:8011/patroni'}) + return get_cluster(True, l, [m, o], failover) -def get_cluster_initialized_with_leader(): - return get_cluster(True, Leader(0, 0, 0, - Member(0, 'leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', - None, None, 28))) +def get_cluster_initialized_with_leader(failover=None): + return get_cluster_initialized_without_leader(leader=True, failover=failover) class MockPostgresql(Mock): name = 'postgresql0' role = 'replica' + state = 'running' + connection_string = 'postgres://foo@bar/postgres' def is_healthy(self): return True @@ -51,6 +57,9 @@ class MockPostgresql(Mock): def is_leader(self): return True + def xlog_position(self): + return 0 + def last_operation(self): return 0 @@ -60,6 +69,25 @@ class MockPostgresql(Mock): def bootstrap(self, *args, **kwargs): return True + def check_replication_lag(self, last_leader_operation): + return True + + def check_recovery_conf(self, leader): + return False + + +class MockPatroni: + + def __init__(self, p, d): + self.postgresql = p + self.dcs = d + self.api = Mock() + self.api.connection_string = 'http://127.0.0.1:8008' + + +def run_async(func, args=()): + func(*args) if args else func() + class TestHa(unittest.TestCase): @@ -71,23 +99,45 @@ class TestHa(unittest.TestCase): self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'}) self.e.client.read = etcd_read self.e.client.write = etcd_write - self.ha = Ha(self.p, self.e) - self.ha.load_cluster_from_dcs() + self.e.client.delete = Mock(side_effect=etcd.EtcdException()) + self.ha = Ha(MockPatroni(self.p, self.e)) + self.ha._async_executor.run_async = run_async + self.ha.old_cluster = self.e.get_cluster() self.ha.cluster = get_cluster_not_initialized_without_leader() self.ha.load_cluster_from_dcs = Mock() - def test_load_cluster_from_dcs(self): - ha = Ha(self.p, self.e) - ha.load_cluster_from_dcs() - self.e.get_cluster = get_cluster_not_initialized_without_leader - ha.load_cluster_from_dcs() + def test_update_lock(self): + self.p.last_operation = Mock(side_effect=PostgresException('')) + self.assertTrue(self.ha.update_lock()) - def test_start_as_slave(self): + def test_touch_member(self): + self.p.xlog_position = Mock(side_effect=Exception) + self.ha.touch_member() + + def test_start_as_replica(self): self.p.is_healthy = false self.assertEquals(self.ha.run_cycle(), 'started as a secondary') + def test_recover_replica_failed(self): + self.p.controldata = lambda: {'Database cluster state': 'in production'} + self.p.is_healthy = false + self.p.follow_the_leader = false + self.assertEquals(self.ha.run_cycle(), 'failed to start postgres') + + def test_recover_master_failed(self): + self.p.follow_the_leader = false + self.p.is_healthy = false + self.ha.has_lock = true + self.assertEquals(self.ha.run_cycle(), 'removed leader key after trying and failing to start postgres') + + @patch('sys.exit', return_value=1) + @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True)) + def test_sysid_no_match(self, exit_mock): + self.ha.run_cycle() + exit_mock.assert_called_once_with(1) + + @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_start_as_readonly(self): - self.ha.cluster.is_unlocked = false self.p.is_leader = self.p.is_healthy = false self.ha.has_lock = true self.assertEquals(self.ha.run_cycle(), 'promoted self to leader because i had the session lock') @@ -96,6 +146,7 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'acquired session lock as a leader') def test_promoted_by_acquiring_lock(self): + self.ha.is_healthiest_node = true self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') @@ -104,16 +155,17 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'demoted self due after trying and failing to obtain lock') def test_follow_new_leader_after_failing_to_obtain_lock(self): + self.ha.is_healthiest_node = true self.ha.acquire_lock = false self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'following new leader after trying and failing to obtain lock') def test_demote_because_not_healthiest(self): - self.p.is_healthiest_node = false + self.ha.is_healthiest_node = false self.assertEquals(self.ha.run_cycle(), 'demoting self because i am not the healthiest node') def test_follow_new_leader_because_not_healthiest(self): - self.p.is_healthiest_node = false + self.ha.is_healthiest_node = false self.p.is_leader = false self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') @@ -148,13 +200,9 @@ class TestHa(unittest.TestCase): self.assertEquals(self.ha.run_cycle(), 'demoted self because DCS is not accessible and i was a leader') def test_bootstrap_from_leader(self): - self.ha.cluster = get_cluster_initialized_with_leader() - self.assertEquals(self.ha.bootstrap(), 'bootstrapped from leader') - - def test_bootstrap_from_leader_failed(self): self.ha.cluster = get_cluster_initialized_with_leader() self.p.bootstrap = false - self.assertEquals(self.ha.bootstrap(), 'failed to bootstrap from leader') + self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from leader') def test_bootstrap_waiting_for_leader(self): self.ha.cluster = get_cluster_initialized_without_leader() @@ -174,3 +222,86 @@ class TestHa(unittest.TestCase): self.e.initialize = true self.p.bootstrap = Mock(side_effect=PostgresException("Could not bootstrap master PostgreSQL")) self.assertRaises(PostgresException, self.ha.bootstrap) + + def test_reinitialize(self): + self.ha.schedule_reinitialize() + self.ha.schedule_reinitialize() + self.ha.run_cycle() + self.assertIsNone(self.ha._async_executor.scheduled_action) + + self.ha.cluster = get_cluster_initialized_with_leader() + self.ha.has_lock = true + self.ha.schedule_reinitialize() + self.ha.run_cycle() + self.assertIsNone(self.ha._async_executor.scheduled_action) + + self.ha.has_lock = false + self.ha.schedule_reinitialize() + self.ha.run_cycle() + + def test_restart(self): + self.assertEquals(self.ha.restart(), (True, 'restarted successfully')) + self.p.restart = false + self.assertEquals(self.ha.restart(), (False, 'restart failed')) + self.ha.schedule_reinitialize() + self.assertEquals(self.ha.restart(), (False, 'reinitialize already in progress')) + + def test_restart_in_progress(self): + self.ha._async_executor.schedule('restart', True) + self.assertTrue(self.ha.restart_scheduled()) + self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race') + + self.ha.cluster = get_cluster_initialized_with_leader() + self.assertEquals(self.ha.run_cycle(), 'restart in progress') + + self.ha.has_lock = true + self.assertEquals(self.ha.run_cycle(), 'updated leader lock during restart') + + self.ha.update_lock = false + self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart') + + @patch('requests.get', requests_get) + def test_manual_failover_from_leader(self): + self.ha.has_lock = true + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '')) + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', MockPostgresql.name)) + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') + self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla')) + self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock') + f = Failover(0, MockPostgresql.name, '') + self.ha.cluster = get_cluster_initialized_with_leader(f) + self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself') + + @patch('requests.get', requests_get) + def test_manual_failover_process_no_leader(self): + self.p.is_leader = false + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', MockPostgresql.name)) + self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader')) + self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + self.ha.fetch_node_status = lambda e: (e, True, True, 0) # accessible, in_recovery + self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') + self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, MockPostgresql.name, '')) + self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') + self.ha.fetch_node_status = lambda e: (e, False, True, 0) # accessible, in_recovery + self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + + def test__is_healthiest_node(self): + self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + self.p.is_leader = false + self.ha.fetch_node_status = lambda e: (e, True, True, 0) # accessible, in_recovery + self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + self.ha.fetch_node_status = lambda e: (e, True, False, 0) # accessible, not in_recovery + self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + self.ha.fetch_node_status = lambda e: (e, True, True, 1) # accessible, in_recovery, xlog location ahead + self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + self.p.check_replication_lag = false + self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + + @patch('requests.get', requests_get) + def test_fetch_node_status(self): + member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'}) + self.ha.fetch_node_status(member) + member = Member(0, 'test', 1, {'api_url': 'http://localhost:8011/patroni'}) + self.ha.fetch_node_status(member) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 7cac2eba..52b10d7a 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -1,4 +1,3 @@ -import datetime import sys import time import unittest @@ -6,7 +5,7 @@ import yaml from mock import Mock, patch from patroni.api import RestApiServer -from patroni.dcs import Cluster, Member +from patroni.async_executor import AsyncExecutor from patroni.etcd import Etcd from patroni import Patroni, main from patroni.zookeeper import ZooKeeper @@ -26,6 +25,7 @@ def time_sleep(*args): @patch.object(Postgresql, 'write_pg_hba', Mock()) @patch.object(Postgresql, 'write_recovery_conf', Mock()) @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock()) +@patch.object(AsyncExecutor, 'run', Mock()) class TestPatroni(unittest.TestCase): @patch.object(Client, 'machines') @@ -48,7 +48,6 @@ class TestPatroni(unittest.TestCase): self.assertRaises(Exception, self.p.get_dcs, '', {}) @patch('time.sleep', Mock(side_effect=SleepException())) - @patch.object(Patroni, 'initialize', Mock()) @patch.object(Etcd, 'delete_leader', Mock()) @patch.object(Client, 'machines') def test_patroni_main(self, mock_machines): @@ -56,16 +55,13 @@ class TestPatroni(unittest.TestCase): sys.argv = ['patroni.py', 'postgres0.yml'] mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) - with patch.object(Patroni, 'touch_member', self.touch_member): - with patch.object(Patroni, 'run', Mock(side_effect=SleepException())): - self.assertRaises(SleepException, main) - with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())): - main() + with patch.object(Patroni, 'run', Mock(side_effect=SleepException())): + self.assertRaises(SleepException, main) + with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())): + main() @patch('time.sleep', Mock(side_effect=SleepException())) def test_run(self): - self.p.touch_member = self.touch_member - self.p.ha.state_handler.sync_replication_slots = time_sleep self.p.ha.dcs.watch = time_sleep self.assertRaises(SleepException, self.p.run) @@ -73,24 +69,6 @@ class TestPatroni(unittest.TestCase): self.p.api.start = Mock() self.assertRaises(SleepException, self.p.run) - def touch_member(self, ttl=None): - if not self.touched: - self.touched = True - return False - return True - - def test_touch_member(self): - self.p.touch_member() - now = datetime.datetime.utcnow() - member = Member(0, self.p.postgresql.name, 'b', 'c', (now + datetime.timedelta( - seconds=self.p.shutdown_member_ttl + 10)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None) - self.p.ha.cluster = Cluster(True, member, 0, [member]) - self.p.touch_member() - - def test_patroni_initialize(self): - self.p.touch_member = self.touch_member - self.p.initialize() - def test_schedule_next_run(self): self.p.ha.dcs.watch = Mock(return_value=True) self.p.schedule_next_run() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index b4988c25..0ed04a15 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -1,14 +1,22 @@ +import mock # for the mock.call method, importing it without a namespace breaks python3 import os import psycopg2 import shutil import unittest -from mock import Mock, patch +from six.moves import builtins +from mock import Mock, MagicMock, PropertyMock, patch, mock_open from patroni.dcs import Cluster, Leader, Member from patroni.exceptions import PostgresException, PostgresConnectionException from patroni.postgresql import Postgresql from patroni.utils import RetryFailedError from test_ha import false +import subprocess + + +def is_file_raise_on_backup(*args, **kwargs): + if args[0].endswith('.backup'): + raise Exception("foo") class MockCursor: @@ -25,18 +33,9 @@ class MockCursor: raise RetryFailedError('retry') elif sql.startswith('SELECT slot_name'): self.results = [('blabla',), ('foobar',)] - elif sql.startswith('SELECT pg_current_xlog_location()'): - self.results = [(0,)] - elif sql.startswith('SELECT pg_is_in_recovery(), %s'): - if params[0][0] == 1: - raise psycopg2.OperationalError() - elif params[0][0] == 2: - self.results = [(True, -1)] - else: - self.results = [(False, 0)] elif sql.startswith('SELECT pg_xlog_location_diff'): self.results = [(0,)] - elif sql.startswith('SELECT pg_is_in_recovery()'): + elif sql == 'SELECT pg_is_in_recovery()': self.results = [(False, )] elif sql.startswith('SELECT to_char(pg_postmaster_start_time'): self.results = [('', True, '', '', '', False)] @@ -79,22 +78,93 @@ class MockConnect(Mock): def cursor(self): return MockCursor(self) + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + +def pg_controldata_string(*args, **kwargs): + return b""" +pg_control version number: 942 +Catalog version number: 201509161 +Database system identifier: 6200971513092291716 +Database cluster state: shut down in recovery +pg_control last modified: Fri Oct 2 10:57:06 2015 +Latest checkpoint location: 0/30000C8 +Prior checkpoint location: 0/2000060 +Latest checkpoint's REDO location: 0/3000090 +Latest checkpoint's REDO WAL file: 000000020000000000000003 +Latest checkpoint's TimeLineID: 2 +Latest checkpoint's PrevTimeLineID: 2 +Latest checkpoint's full_page_writes: on +Latest checkpoint's NextXID: 0/943 +Latest checkpoint's NextOID: 24576 +Latest checkpoint's NextMultiXactId: 1 +Latest checkpoint's NextMultiOffset: 0 +Latest checkpoint's oldestXID: 931 +Latest checkpoint's oldestXID's DB: 1 +Latest checkpoint's oldestActiveXID: 943 +Latest checkpoint's oldestMultiXid: 1 +Latest checkpoint's oldestMulti's DB: 1 +Latest checkpoint's oldestCommitTs: 0 +Latest checkpoint's newestCommitTs: 0 +Time of latest checkpoint: Fri Oct 2 10:56:54 2015 +Fake LSN counter for unlogged rels: 0/1 +Minimum recovery ending location: 0/30241F8 +Min recovery ending loc's timeline: 2 +Backup start location: 0/0 +Backup end location: 0/0 +End-of-backup record required: no +wal_level setting: hot_standby +Current wal_log_hints setting: on +Current max_connections setting: 100 +Current max_worker_processes setting: 8 +Current max_prepared_xacts setting: 0 +Current max_locks_per_xact setting: 64 +Current track_commit_timestamp setting: off +Maximum data alignment: 8 +Database block size: 8192 +Blocks per segment of large relation: 131072 +WAL block size: 8192 +Bytes per WAL segment: 16777216 +Maximum length of identifiers: 64 +Maximum columns in an index: 32 +Maximum size of a TOAST chunk: 1996 +Size of a large-object chunk: 2048 +Date/time type storage: 64-bit integers +Float4 argument passing: by value +Float8 argument passing: by value +Data page checksum version: 0 +""" + + +def postmaster_opts_string(*args, **kwargs): + return '/usr/local/pgsql/bin/postgres "-D" "data/postgresql0" "--listen_addresses=127.0.0.1" \ +"--port=5432" "--hot_standby=on" "--wal_keep_segments=8" "--wal_level=hot_standby" \ +"--archive_command=mkdir -p ../wal_archive && cp %p ../wal_archive/%f" "--wal_log_hints=on" \ +"--max_wal_senders=5" "--archive_timeout=1800s" "--archive_mode=on" "--max_replication_slots=5"\n' + def psycopg2_connect(*args, **kwargs): return MockConnect() @patch('subprocess.call', Mock(return_value=0)) -@patch('shutil.copy', Mock()) @patch('psycopg2.connect', psycopg2_connect) +@patch('shutil.copy', Mock()) class TestPostgresql(unittest.TestCase): + @patch('subprocess.call', Mock(return_value=0)) + @patch('psycopg2.connect', psycopg2_connect) def setUp(self): 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'}, + 'pg_rewind': {'username': 'admin', 'password': 'admin'}, 'replication': {'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, @@ -106,10 +176,10 @@ class TestPostgresql(unittest.TestCase): 'restore': 'true'}) if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) - self.leadermem = Member(0, 'leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, None, 28) - self.leader = Leader(-1, None, 28, self.leadermem) - self.other = Member(0, 'test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, None, 28) - self.me = Member(0, 'test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, None, 28) + self.leadermem = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres'}) + self.leader = Leader(-1, 28, self.leadermem) + self.other = Member(0, 'test1', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres'}) + self.me = Member(0, 'test0', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres'}) def tearDown(self): shutil.rmtree('data') @@ -121,23 +191,77 @@ 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_stop(self): - self.assertFalse(self.p.start()) + def test_start(self): + self.assertTrue(self.p.start()) self.p.is_running = false - with open(os.path.join(self.p.data_dir, 'postmaster.pid'), 'w'): - pass + open(os.path.join(self.p.data_dir, 'postmaster.pid'), 'w').close() self.assertTrue(self.p.start()) - self.assertTrue(self.p.stop()) + def test_stop(self): + self.assertTrue(self.p.stop()) + with patch('subprocess.call', Mock(return_value=1)): + self.assertTrue(self.p.stop()) + self.p.is_running = Mock(return_value=True) + self.assertFalse(self.p.stop()) + + def test_restart(self): + self.p.start = false + self.p.is_running = false + self.assertFalse(self.p.restart()) + self.assertEquals(self.p.state, 'restart failed (restarting)') + + @patch.object(builtins, 'open', MagicMock()) + def test_write_pgpass(self): + self.p.write_pgpass({'host': 'localhost', 'port': '5432', 'user': 'foo', 'password': 'bar'}) + + @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) def test_sync_from_leader(self): self.assertTrue(self.p.sync_from_leader(self.leader)) - def test_follow_the_leader(self): - self.p.demote(self.leader) + @patch('subprocess.call', side_effect=Exception("Test")) + @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) + def test_pg_rewind(self, mock_call): + self.assertTrue(self.p.rewind(self.leader)) + subprocess.call = mock_call + self.assertFalse(self.p.rewind(self.leader)) + + @patch('patroni.postgresql.Postgresql.rewind', return_value=False) + @patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True)) + @patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1)) + @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) + def test_follow_the_leader(self, mock_pg_rewind): + self.p.demote() self.p.follow_the_leader(None) - self.p.demote(self.leader) + self.p.demote() self.p.follow_the_leader(self.leader) - self.p.follow_the_leader(Leader(-1, None, 28, self.other)) + self.p.follow_the_leader(Leader(-1, 28, self.other)) + self.p.rewind = mock_pg_rewind + self.p.follow_the_leader(self.leader) + self.p.require_rewind() + with mock.patch('os.path.islink', MagicMock(return_value=True)): + with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): + with mock.patch('os.unlink', MagicMock(return_value=True)): + self.p.follow_the_leader(self.leader, recovery=True) + self.p.require_rewind() + with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(return_value=True)): + self.p.rewind.return_value = True + self.p.follow_the_leader(self.leader, recovery=True) + self.p.rewind.return_value = False + self.p.follow_the_leader(self.leader, recovery=True) + + def test_can_rewind(self): + tmp = self.p.pg_rewind + self.p.pg_rewind = None + self.assertFalse(self.p.can_rewind) + self.p.pg_rewind = tmp + with mock.patch('subprocess.call', MagicMock(return_value=1)): + self.assertFalse(self.p.can_rewind) + with mock.patch('subprocess.call', side_effect=OSError("foo")): + self.assertFalse(self.p.can_rewind) + tmp = self.p.controldata() + self.p.controldata = lambda: {'wal_log_hints setting': 'on'} + self.assertTrue(self.p.can_rewind) + self.p.controldata = tmp def test_create_replica(self): self.p.delete_trigger_file = Mock(side_effect=OSError()) @@ -151,29 +275,25 @@ class TestPostgresql(unittest.TestCase): def test_sync_replication_slots(self): self.p.start() - cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem]) + cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem], None) + self.p.sync_replication_slots(cluster) + self.p.query = Mock(side_effect=psycopg2.OperationalError) + self.p.schedule_load_slots = True self.p.sync_replication_slots(cluster) @patch.object(MockConnect, 'closed', 2) def test__query(self): self.assertRaises(PostgresConnectionException, self.p._query, 'blabla') + self.p._state = 'restarting' + self.assertRaises(RetryFailedError, self.p._query, 'blabla') def test_query(self): self.p.query('select 1') self.assertRaises(PostgresConnectionException, self.p.query, 'RetryFailedError') self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla') - def test_is_healthiest_node(self): - cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem]) - self.assertTrue(self.p.is_healthiest_node(cluster)) - self.p.is_leader = false - self.assertFalse(self.p.is_healthiest_node(cluster)) - self.p.xlog_position = lambda: 1 - self.assertTrue(self.p.is_healthiest_node(cluster)) - self.p.xlog_position = lambda: 2 - self.assertFalse(self.p.is_healthiest_node(cluster)) - self.p.config['maximum_lag_on_failover'] = -3 - self.assertFalse(self.p.is_healthiest_node(cluster)) + def test_is_leader(self): + self.assertTrue(self.p.is_leader()) def test_reload(self): self.assertTrue(self.p.reload()) @@ -184,6 +304,7 @@ class TestPostgresql(unittest.TestCase): self.assertFalse(self.p.is_healthy()) def test_promote(self): + self.p._role = 'replica' self.assertTrue(self.p.promote()) self.assertTrue(self.p.promote()) @@ -202,6 +323,9 @@ class TestPostgresql(unittest.TestCase): self.p.query = Mock(side_effect=psycopg2.OperationalError("not supported")) self.assertTrue(self.p.stop()) + def test_check_replication_lag(self): + self.assertTrue(self.p.check_replication_lag(0)) + @patch('os.rename', Mock()) @patch('os.path.isdir', Mock(return_value=True)) def test_move_data_directory(self): @@ -210,10 +334,12 @@ class TestPostgresql(unittest.TestCase): with patch('os.rename', Mock(side_effect=OSError())): self.p.move_data_directory() + @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict())) def test_bootstrap(self): - self.assertRaises(PostgresException, self.p.bootstrap) - self.p.start = Mock(return_value=True) + with patch('subprocess.call', Mock(return_value=1)): + self.assertRaises(PostgresException, self.p.bootstrap) self.p.bootstrap() + self.p.bootstrap(self.leader) def test_remove_data_directory(self): self.p.data_dir = 'data_dir' @@ -226,3 +352,104 @@ class TestPostgresql(unittest.TestCase): with patch('os.unlink', Mock(side_effect=Exception)): self.p.remove_data_directory() self.p.remove_data_directory() + + @patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string)) + @patch('subprocess.check_output', side_effect=subprocess.CalledProcessError) + @patch('subprocess.check_output', side_effect=Exception('Failed')) + def test_controldata(self, check_output_call_error, check_output_generic_exception): + data = self.p.controldata() + self.assertEquals(len(data), 50) + self.assertEquals(data['Database cluster state'], 'shut down in recovery') + self.assertEquals(data['wal_log_hints setting'], 'on') + self.assertEquals(int(data['Database block size']), 8192) + + subprocess.check_output = check_output_call_error + data = self.p.controldata() + self.assertEquals(data, dict()) + + subprocess.check_output = check_output_generic_exception + self.assertRaises(Exception, self.p.controldata()) + + def test_read_postmaster_opts(self): + m = mock_open(read_data=postmaster_opts_string()) + with patch.object(builtins, 'open', m): + data = self.p.read_postmaster_opts() + self.assertEquals(data['wal_level'], 'hot_standby') + self.assertEquals(int(data['max_replication_slots']), 5) + self.assertEqual(data.get('D'), None) + + m.side_effect = IOError("foo") + data = self.p.read_postmaster_opts() + self.assertEqual(data, dict()) + + m.side_effect = Exception("foo") + self.assertRaises(Exception, self.p.read_postmaster_opts()) + + @patch('subprocess.Popen') + @patch.object(builtins, 'open', MagicMock(return_value=42)) + def test_single_user_mode(self, subprocess_popen_mock): + subprocess_popen_mock.return_value.wait.return_value = 0 + self.assertEquals(self.p.single_user_mode(options=dict(archive_mode='on', archive_command='false')), 0) + subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir, + '-c', 'archive_command=false', '-c', 'archive_mode=on', + 'postgres'], stdin=subprocess.PIPE, + stdout=42, + stderr=subprocess.STDOUT) + subprocess_popen_mock.reset_mock() + self.assertEquals(self.p.single_user_mode(command="CHECKPOINT"), 0) + subprocess_popen_mock.assert_called_once_with(['postgres', '--single', '-D', self.p.data_dir, + 'postgres'], stdin=subprocess.PIPE, + stdout=42, + stderr=subprocess.STDOUT) + subprocess_popen_mock.return_value = None + self.assertEquals(self.p.single_user_mode(), 1) + + def fake_listdir(path): + if path.endswith(os.path.join('pg_xlog', 'archive_status')): + return ["a", "b", "c"] + return [] + + @patch('os.listdir', MagicMock(side_effect=fake_listdir)) + @patch('os.path.isdir', MagicMock(return_value=True)) + @patch('os.unlink', return_value=True) + @patch('os.remove', return_value=True) + @patch('os.path.islink', return_value=False) + @patch('os.path.isfile', return_value=True) + def test_cleanup_archive_status(self, mock_file, mock_link, mock_remove, mock_unlink): + ap = os.path.join(self.p.data_dir, 'pg_xlog', 'archive_status/') + self.p.cleanup_archive_status() + mock_remove.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')]) + mock_unlink.assert_not_called() + + mock_remove.reset_mock() + + mock_file.return_value = False + mock_link.return_value = True + self.p.cleanup_archive_status() + mock_unlink.assert_has_calls([mock.call(ap+'a'), mock.call(ap+'b'), mock.call(ap+'c')]) + mock_remove.assert_not_called() + + mock_unlink.reset_mock() + mock_remove.reset_mock() + + mock_file.side_effect = Exception("foo") + mock_link.side_effect = Exception("foo") + self.p.cleanup_archive_status() + mock_unlink.assert_not_called() + mock_remove.assert_not_called() + + @patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string)) + def test_sysid(self): + self.assertEqual(self.p.sysid, "6200971513092291716") + + @patch('os.path.isfile', MagicMock(return_value=True)) + @patch('shutil.copy', side_effect=Exception) + def test_save_configuration_files(self, mock_copy): + shutil.copy = mock_copy + self.p.save_configuration_files() + + @patch('os.path.isfile', MagicMock(side_effect=is_file_raise_on_backup)) + @patch('shutil.copy', side_effect=Exception) + def test_restore_configuration_files(self, mock_copy): + shutil.copy = mock_copy + self.p.restore_configuration_files() diff --git a/tests/test_utils.py b/tests/test_utils.py index 45b194dc..265f98ab 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -67,7 +67,9 @@ class TestRetrySleeper(unittest.TestCase): self.assertRaises(RetryFailedError, retry, self._fail(times=100)) def test_copy(self): - _sleep = lambda t: None + def _sleep(t): + None + retry = self._makeOne(sleep_func=_sleep) rcopy = retry.copy() self.assertTrue(rcopy.sleep_func is _sleep) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index afbdad5f..84807270 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -7,7 +7,7 @@ from patroni.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperErr from kazoo.client import KazooState from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.protocol.states import ZnodeStat -from test_etcd import MockPostgresql, SleepException, requests_get +from test_etcd import SleepException, requests_get class MockKazooClient(Mock): @@ -31,7 +31,7 @@ class MockKazooClient(Mock): elif '/members/' in path: return ( b'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni', - ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) + ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0 if self.exists else -1, 0, 0, 0) ) elif path.endswith('/optime/leader'): return (b'1', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) @@ -41,14 +41,15 @@ class MockKazooClient(Mock): return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) elif path.endswith('/initialize'): return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) + return (b'', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) def get_children(self, path, watch=None, include_data=False): if not isinstance(path, six.string_types): raise TypeError("Invalid type for 'path' (string expected)") - if path == '/no_node': + if path.startswith('/no_node'): raise NoNodeError elif path in ['/service/bla/', '/service/test/']: - return ['initialize', 'leader', 'members', 'optime'] + return ['initialize', 'leader', 'members', 'optime', 'failover'] return ['foo', 'bar', 'buzz'] def create(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False): @@ -68,6 +69,14 @@ class MockKazooClient(Mock): raise TypeError("Invalid type for 'value' (must be a byte string)") if path == '/service/bla/optime/leader': raise Exception + if path == '/service/test/members/bar': + if value == b'retry': + return + if path == '/service/test/failover': + if value == b'Exception': + raise Exception + elif value == b'ok': + return raise NoNodeError def delete(self, path, version=-1, recursive=False): @@ -79,7 +88,9 @@ class MockKazooClient(Mock): return self.leader = True raise Exception - elif path.endswith('/initialize'): + elif path == '/service/test/members/buzz': + raise Exception + elif path.endswith('/initialize') or path == '/service/test/members/bar': raise NoNodeError @@ -110,6 +121,8 @@ class TestZooKeeper(unittest.TestCase): def test__inner_load_cluster(self): self.zk._base_path = self.zk._base_path.replace('test', 'bla') self.zk._inner_load_cluster() + self.zk._base_path = self.zk._base_path = '/no_node' + self.zk._inner_load_cluster() def test_get_cluster(self): self.assertRaises(ZooKeeperError, self.zk.get_cluster) @@ -119,6 +132,11 @@ class TestZooKeeper(unittest.TestCase): self.zk.touch_member('foo') self.zk.delete_leader() + def test_set_failover_value(self): + self.zk.set_failover_value('') + self.zk.set_failover_value('ok') + self.zk.set_failover_value('Exception') + def test_initialize(self): self.assertFalse(self.zk.initialize()) @@ -126,21 +144,33 @@ class TestZooKeeper(unittest.TestCase): self.zk.cancel_initialization() def test_touch_member(self): + self.zk._name = 'buzz' + self.zk.get_cluster() self.zk.touch_member('new') + self.zk._name = 'bar' + self.zk.touch_member('new') + self.zk._name = 'na' + self.zk.client.exists = 1 self.zk.touch_member('exists') + self.zk._name = 'bar' + self.zk.touch_member('retry') + self.zk.fetch_cluster = True + self.zk.get_cluster() self.zk.touch_member('retry') def test_take_leader(self): self.zk.take_leader() def test_update_leader(self): - self.zk.last_leader_operation = -1 - self.assertTrue(self.zk.update_leader(MockPostgresql())) + self.assertTrue(self.zk.update_leader()) + + def test_write_leader_optime(self): + self.zk.last_leader_operation = '0' + self.zk.write_leader_optime('1') self.zk._base_path = self.zk._base_path.replace('test', 'bla') - self.zk.last_leader_operation = -1 - self.assertTrue(self.zk.update_leader(MockPostgresql())) + self.zk.write_leader_optime('2') def test_watch(self): self.zk.watch(0) - self.zk.cluster_event.isSet = lambda: False + self.zk.event.isSet = lambda: True self.zk.watch(0)