Bumped version to 0.7

Merge pull request #82 from zalando/feature/patroni_cli_or_ctl_tbd
Feature/patroni cli or ctl tbd
2026-08-26 07:30:14 +00:00 · 2015-11-18 16:45:05 +01:00 · 2015-11-18 16:17:02 +01:00 · 2015-11-18 15:08:59 +01:00 · 2015-11-18 15:01:35 +01:00 · 2015-11-18 12:55:11 +01:00
29 changed files with 1596 additions and 193 deletions
@@ -1,3 +1,11 @@
 data/*
 *.pyc
-helpers/*.pyc
+*.egg/
+*.egg-info/
+.cache/
+.coverage
+.eggs/
+build/
+coverage.xml
+junit.xml
+pgpass
@@ -13,8 +13,9 @@ RUN apt-get update -y
 RUN apt-get upgrade -y

 ENV PGVERSION 9.4
-RUN apt-get install python python-psycopg2 python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo python-pip -y
-RUN pip install python-etcd
+RUN apt-get install python python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo python-pip -y
+RUN apt-get install python-dev postgresql-server-dev-${PGVERSION} -y
+RUN pip install python-etcd psycopg2

 ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH

@@ -5,10 +5,10 @@ Patroni: A Template for PostgreSQL HA with ZooKeeper or etcd

 Patroni was previously known as Governor.

-*There are many ways to run high availability with PostgreSQL; here we
-present a template for you to create your own custom fit high
-availability solution using python and distributed configuration store
-(like ZooKeeper or etcd) for maximum accessibility.*
+*There are many ways to run high availability with PostgreSQL. Here, we
+present a template for you to create your own customized, high-availability 
+solution using Python and — for maximum accessibility — a distributed 
+configuration store like ZooKeeper or etcd.*

 Getting Started
 ---------------
@@ -22,8 +22,8 @@ To get started, do the following from different terminals:
    > ./patroni.py postgres1.yml

 From there, you will see a high-availability cluster start up. Test
-different settings in the YAML files to see how behavior changes. Kill
-some of the different components to see how the system behaves.
+different settings in the YAML files to see how its behavior changes. Kill
+some of the components to see how the system behaves.

 Add more ``postgres*.yml`` files to create an even larger cluster.

@@ -39,77 +39,81 @@ run:

    > psql --host 127.0.0.1 --port 5000 postgres

-How Patroni works
+How Patroni Works
 -----------------

-For a diagram of the high availability decision loop, see the included a
-PDF:
+For a diagram of the high availability decision loop, review this PDF:
 `postgres-ha.pdf <https://github.com/zalando/patroni/blob/master/postgres-ha.pdf>`__

 YAML Configuration
 ------------------

-For an example file, see ``postgres0.yml``. Below is an explanation of
-settings:
+For an example file, see ``postgres0.yml``. Regarding settings:

-  *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated.
+-  *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process.
 -  *loop\_wait*: the number of seconds the loop will sleep

 -  *restapi*:
-    -  *listen*: ip address + port that Patroni will listen to provide health-check information for haproxy.
-    -  *connect\_address*: ip address + port through which restapi is accessible.
-    -  *auth*: (optional) 'username:password' to protect some dangerous REST API endpoints.
-    -  *certfile*: (optional) Specifies a file with the certificate in the PEM format. If certfile is not specified or empty API server will work without SSL.
+    -  *listen*: IP address + port that Patroni will listen to, to provide health-check information for haproxy.
+    -  *connect\_address*: IP address + port through which restapi is accessible.
+    -  *auth*: (optional) 'username:password' to protect dangerous REST API endpoints.
+    -  *certfile*: (optional) Specifies a file with the certificate in the PEM format. If the certfile is not specified or is left empty, the API server will work without SSL.
    -  *keyfile*: (optional) Specifies a file with the secret key in the PEM format.

 -  *etcd*:
-    -  *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd
-    -  *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated.
-    -  *host*: the host:port for the etcd endpoint
+    -  *scope*: the relative path used on etcd's HTTP API for this deployment; makes it possible to run multiple HA deployments from a single etcd.
+    -  *ttl*: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process.
+    -  *host*: the host:port for the etcd endpoint.

 -  *zookeeper*:
-    -  *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd
-    -  *session\_timeout*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated.
-    -  *reconnect\_timeout*: how long we should try to reconnect to ZooKeeper after connection loss. After this timeout we assume that we don't have lock anymore and will restart in read-only mode.
+    -  *scope*: the relative path used on etcd's HTTP API for this deployment; makes it possible to run multiple HA deployments from a single etcd.
+    -  *session\_timeout*: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process.
+    -  *reconnect\_timeout*: how long we should try to reconnect to ZooKeeper after a connection loss. After this timeout, assume that you no longer have a lock and restart in read-only mode.
    -  *hosts*: list of ZooKeeper cluster members in format: ['host1:port1', 'host2:port2', 'etc...']
-    -  *exhibitor*: if you are running ZooKeeper cluster under Exhibitor supervisory the following section could be interesting for you
-        -  *poll\_interval*: how often list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor
-        -  *port*: Exhibitor port
-        -  *hosts*: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list would be updated automatically when Exhibitor (ZooKeeper) cluster topology changes.
+    -  *exhibitor*: if you are running a ZooKeeper cluster under the Exhibitor supervisory, the following section might interest you:
+        -  *poll\_interval*: how often the list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor
+        -  *port*: Exhibitor port.
+        -  *hosts*: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.

 -  *postgresql*:
-    -  *name*: the name of the Postgres host, must be unique for the cluster
-    -  *listen*: ip address + port that Postgres listening. Must be accessible from other nodes in the cluster if using streaming replication.
-    -  *connect\_address*: ip address + port through which Postgres is accessible from other nodes and applications.
-    -  *data\_dir*: file path to initialize and store Postgres data files
-    -  *maximum\_lag\_on\_failover*: the maximum bytes a follower may lag
-    -  *use\_slots*: whether or not to use replication_slots.  Must be False for PostgreSQL 9.3, and you should comment out max_replication_slots. before it is not eligible become leader
-    -  *pg\_hba*: list of lines which should be added to pg\_hba.conf
-        -  *- host all all 0.0.0.0/0 md5*
+    -  *name*: the name of the Postgres host. Must be unique for the cluster.
+    -  *listen*: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication.
+    -  *connect\_address*: IP address + port through which Postgres is accessible from other nodes and applications.
+    -  *data\_dir*: file path to initialize and store Postgres data files.
+    -  *maximum\_lag\_on\_failover*: the maximum bytes a follower may lag.
+    -  *use\_slots*: whether or not to use replication_slots. Must be False for PostgreSQL 9.3. You should comment out max_replication_slots before it becomes ineligible for leader status.
+
+    -  *initdb*:  List options to be passed on to initdb
+        -  *encoding*: default encoding for new databases
+        -  *locale*: default locale for new databases
+        -  *data-checksums*  # When pg_rewind is needed on 9.3, this needs to be enabled
+
+    -  *pg\_hba*: list of lines which should be added to pg\_hba.conf.
+        -  *- host all all 0.0.0.0/0 md5*.

    -  *replication*:
-        -  *username*: replication username, user will be created during initialization
-        -  *password*: replication password, user will be created during initialization
-        -  *network*: network setting for replication in pg\_hba.conf
+        -  *username*: replication username; user will be created during initialization.
+        -  *password*: replication password; user will be created during initialization.
+        -  *network*: network setting for replication in pg\_hba.conf.

-    -  *callbacks* callback scripts to run on certain actions. Patroni will pass current action, role and cluster name. See scripts/aws.py as an example on how to write them.
-        -  *on\_start*: a script to run when the cluster starts
-        -  *on\_stop*: a script to run when the cluster stops
-        -  *on\_restart*: a script to run when the cluster restarts
-        -  *on\_reload*: a script to run when configuration reload is triggered
-        -  *on\_role\_change*: a script to run when the cluster is being promoted or demoted
+    -  *callbacks* callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. See scripts/aws.py as an example on how to write them.
+        -  *on\_start*: a script to run when the cluster starts.
+        -  *on\_stop*: a script to run when the cluster stops.
+        -  *on\_restart*: a script to run when the cluster restarts.
+        -  *on\_reload*: a script to run when configuration reload is triggered.
+        -  *on\_role\_change*: a script to run when the cluster is being promoted or demoted.

    -  *superuser*:
-        -  *password*: password for postgres user. It would be set during initialization
+        -  *password*: password for the Postgres user, set during initialization.

    -  *admin*:
-        -  *username*: admin username, user will be created during initialization. It would have CREATEDB and CREATEROLE privileges
-        -  *password*: admin password, user will be created during initialization.
+        -  *username*: admin username; user is created during initialization. It will have CREATEDB and CREATEROLE privileges.
+        -  *password*: admin password; user is created during initialization.

-    -  *recovery\_conf*: additional configuration settings written to recovery.conf when configuring follower
-        -  *parameters*: list of configuration settings for Postgres.  Many of these are required for replication to work.
+    -  *recovery\_conf*: additional configuration settings written to recovery.conf when configuring the follower.
+        -  *parameters*: list of configuration settings for Postgres. Many of these are required for replication to work.

-Replication choices
+Replication Choices
 -------------------

 Patroni uses Postgres' streaming replication. By default, this
@@ -123,43 +127,43 @@ not occur if a follower is more than a certain number of bytes behind
 the follower. This setting should be increased or decreased based on
 business requirements.

-When asynchronous replication is not best for your use-case, investigate
+When asynchronous replication is not optimal for your use case, investigate
 how Postgres's `synchronous
 replication <http://www.postgresql.org/docs/current/static/warm-standby.html#SYNCHRONOUS-REPLICATION>`__
 works. Synchronous replication ensures consistency across a cluster by
 confirming that writes are written to a secondary before returning to
 the connecting client with a success. The cost of synchronous
-replication will be reduced throughput on writes. This throughput will
+replication: reduced throughput on writes. This throughput will
 be entirely based on network performance. In hosted datacenter
 environments (like AWS, Rackspace, or any network you do not control),
-synchrous replication increases the variability of write performance
-significantly. If followers become inaccessible from the leader, the
-leader will becomes effectively readonly.
+synchrous replication significantly increases the variability of write 
+performance. If followers become inaccessible from the leader, the
+leader effectively becomes readonly.

 To enable a simple synchronous replication test, add the follow lines to
-the ``parameters`` section of your YAML configuration files.
+the ``parameters`` section of your YAML configuration files:

 .. code:: YAML

        synchronous_commit: "on"
        synchronous_standby_names: "*"

-When using synchronous replication, use at least a 3-Postgres data nodes
+When using synchronous replication, use at least three Postgres data nodes
 to ensure write availability if one host fails.

-Choosing your replication schema is dependent on the many business
-decisions. Investigate both async and sync replication, as well as other
+Choosing your replication schema is dependent on your business
+considerations. Investigate both async and sync replication, as well as other
 HA solutions, to determine which solution is best for you.

-Applications should not use superusers
+Applications Should Not Use Superusers
 --------------------------------------

 When connecting from an application, always use a non-superuser. Patroni
 requires access to the database to function properly. By using a
-superuser from application, you can potentially use the entire
+superuser from an application, you can potentially use the entire
 connection pool, including the connections reserved for superusers with
 the ``superuser_reserved_connections`` setting. If Patroni cannot access
-the Primary, because the connection pool is full, behavior will be
+the Primary because the connection pool is full, behavior will be
 undesireable.

 Requirements on a Mac
@@ -175,14 +179,13 @@ Run the following on a Mac to install requirements:
 Notice
 ------

-There are many different ways to do HA with PostgreSQL, see `the
+There are many different ways to do HA with PostgreSQL: See `the
 PostgreSQL
 documentation <https://wiki.postgresql.org/wiki/Replication,_Clustering,_and_Connection_Pooling>`__
 for a complete list.

-We call this project a "template" because it is far from a one-size fits
-all, or a plug-and-play replication system. It will have it's own
-caveats. Use wisely.
+We call Patroni a "template" because it is far from being a one-size-fits-all
+or plug-and-play replication system. It will have its own caveats. Use wisely.

 .. |Build Status| image:: https://travis-ci.org/zalando/patroni.svg?branch=master
   :target: https://travis-ci.org/zalando/patroni
@@ -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


@@ -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)
@@ -83,10 +83,11 @@ cat > /patroni/postgres.yml <<__EOF__

 ttl: &ttl 30
 loop_wait: &loop_wait 10
-scope: &scope ${PATRONI_SCOPE}
+scope: &scope '${PATRONI_SCOPE}'
+namespace: 'patroni'
 restapi:
-  listen: 127.0.0.1:8008
-  connect_address: 127.0.0.1:8008
+  listen: 0.0.0.0:8008
+  connect_address: ${DOCKER_IP}:8008
 etcd:
  scope: *scope
  ttl: *ttl
@@ -115,7 +116,7 @@ postgresql:
  parameters:
    archive_mode: "on"
    wal_level: hot_standby
-    archive_command: mkdir -p ../wal_archive && cp %p ../wal_archive/%f
+    archive_command: 'true'
    max_wal_senders: 20
    listen_addresses: 0.0.0.0
    wal_keep_segments: 8
@@ -15,7 +15,7 @@ frontend ft_postgresql
 	default_backend bk_db

 backend bk_db
-	option httpchk GET
+	option httpchk

-  server postgresql_127.0.0.1_5432 127.0.0.1:5432 maxconn 100 check port 8008
-  server postgresql_127.0.0.1_5433 127.0.0.1:5433 maxconn 100 check port 8009
+    server postgresql_127.0.0.1_5432 127.0.0.1:5432 maxconn 100 check port 8008
+    server postgresql_127.0.0.1_5433 127.0.0.1:5433 maxconn 100 check port 8009
@@ -18,20 +18,24 @@ class Patroni:

    def __init__(self, config):
        self.nap_time = config['loop_wait']
+        self.tags = config.get('tags', dict())
        self.postgresql = Postgresql(config['postgresql'])
        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()

+    @property
+    def nofailover(self):
+        return self.tags.get('nofailover', False)
+
    @staticmethod
    def get_dcs(name, config):
        if 'etcd' in config:
            return Etcd(name, config['etcd'])
        if 'zookeeper' in config:
            return ZooKeeper(name, config['zookeeper'])
-        raise Exception('Can not find sutable configuration of distributed configuration store')
+        raise Exception('Can not find suitable configuration of distributed configuration store')

    def schedule_next_run(self):
        self.next_run += self.nap_time
@@ -3,6 +3,8 @@ import fcntl
 import json
 import logging
 import psycopg2
+import socket
+import time

 from patroni.exceptions import PostgresConnectionException
 from patroni.utils import Retry, RetryFailedError
@@ -36,16 +38,29 @@ class RestApiHandler(BaseHTTPRequestHandler):
        self.end_headers()
        self.wfile.write(body.encode('utf-8'))

+    def finish(self, *args, **kwargs):
+        try:
+            if not self.wfile.closed:
+                self.wfile.flush()
+                self.wfile.close()
+        except socket.error:
+            pass
+        self.rfile.close()
+
    def check_auth_header(self):
        auth_header = self.headers.get('Authorization')
        status = self.server.check_auth_header(auth_header)
        return not status or self.send_auth_request(status)

-    def do_GET(self):
+    def do_OPTIONS(self):
+        self.do_GET(options=True)
+
+    def do_GET(self, options=False):
        """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()
+        response.update(self.get_tags())

        patroni = self.server.patroni
        cluster = patroni.dcs.cluster
@@ -69,12 +84,14 @@ class RestApiHandler(BaseHTTPRequestHandler):
            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'))
+        if not options:
+            self.send_header('Content-Type', 'application/json')
+            self.end_headers()
+            self.wfile.write(json.dumps(response).encode('utf-8'))

    def do_GET_patroni(self):
        response = self.get_postgresql_status(True)
+        response.update(self.get_tags())

        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
@@ -121,6 +138,56 @@ class RestApiHandler(BaseHTTPRequestHandler):
        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, tags in self.server.patroni.ha.fetch_nodes_statuses(members):
+            if reachable and not tags.get('nofailover', False):
+                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

@@ -139,6 +206,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
                self.command = mname
        return ret

+    def handle_one_request(self):
+        try:
+            BaseHTTPRequestHandler.handle_one_request(self)
+        except socket.error:
+            pass
+
    def query(self, sql, *params, **kwargs):
        if not kwargs.get('retry', False):
            return self.server.query(sql, *params)
@@ -174,6 +247,9 @@ class RestApiHandler(BaseHTTPRequestHandler):
                state = 'unknown' if state == 'running' else state
            return {'state': state}

+    def get_tags(self):
+        return {'tags': self.server.patroni.tags}
+

 class RestApiServer(ThreadingMixIn, HTTPServer, Thread):

@@ -0,0 +1,613 @@
+'''
+Patroni Control
+'''
+
+import click
+import os
+import yaml
+import json
+import time
+import psycopg2
+import random
+import requests
+import datetime
+from prettytable import PrettyTable
+from six.moves.urllib_parse import urlparse
+import logging
+
+from .etcd import Etcd
+from .exceptions import PatroniCtlException
+from .postgresql import parseurl
+
+CONFIG_DIR_PATH = click.get_app_dir('patroni')
+CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronictl.yaml')
+LOGLEVEL = 'WARNING'
+
+
+def parse_dcs(dcs):
+    """
+    Break up the provided dcs string
+    >>> parse_dcs('localhost') == {'scheme': 'etcd', 'hostname': 'localhost', 'port': 4001}
+    True
+    >>> parse_dcs('localhost:8500') == {'scheme': 'consul', 'hostname': 'localhost', 'port': 8500}
+    True
+    >>> parse_dcs('zookeeper://localhost') == {'scheme': 'zookeeper', 'hostname': 'localhost', 'port': 2181}
+    True
+    """
+
+    if not dcs:
+        return {}
+
+    parsed = urlparse(dcs)
+    scheme = parsed.scheme
+    if scheme == '' and parsed.netloc == '':
+        parsed = urlparse('//' + dcs)
+
+    if scheme == '':
+        default_schemes = {'2181': 'zookeeper', '8500': 'consul'}
+        scheme = default_schemes.get(str(parsed.port), 'etcd')
+
+    port = parsed.port
+    if port is None:
+        default_ports = {'consul': 8500, 'zookeeper': 2181}
+        port = default_ports.get(str(scheme), 4001)
+
+    return {'scheme': str(scheme), 'hostname': str(parsed.hostname), 'port': int(port)}
+
+
+def load_config(path, dcs):
+    logging.debug('Loading configuration from file {}'.format(path))
+    config = dict()
+    try:
+        with open(path, 'rb') as fd:
+            config = yaml.safe_load(fd)
+    except:
+        logging.exception('Could not load configuration file')
+
+    if dcs:
+        config['dcs'] = parse_dcs(dcs)
+    else:
+        config['dcs'] = parse_dcs(config.get('dcs_api'))
+
+    return config
+
+
+def store_config(config, path):
+    dir_path = os.path.dirname(path)
+    if dir_path:
+        if not os.path.isdir(dir_path):
+            os.makedirs(dir_path)
+    with open(path, 'w') as fd:
+        yaml.dump(config, fd)
+
+
+option_config_file = click.option('--config-file', '-c', help='Configuration file', default=CONFIG_FILE_PATH)
+option_format = click.option('--format', '-f', help='Output format (pretty, json)', default='pretty')
+option_dcs = click.option('--dcs', '-d', help='Use this DCS', envvar='DCS')
+option_watchrefresh = click.option('-w', '--watch', type=float, help='Auto update the screen every X seconds')
+option_watch = click.option('-W', is_flag=True, help='Auto update the screen every 2 seconds')
+option_force = click.option('--force', is_flag=True, help='Do not ask for confirmation at any point')
+
+
+@click.group()
+@click.pass_context
+def ctl(ctx):
+    global LOGLEVEL
+    LOGLEVEL = os.environ.get('LOGLEVEL', LOGLEVEL)
+
+    logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=LOGLEVEL)
+
+
+def get_dcs(config, scope):
+    scheme, hostname, port = map(config.get('dcs', {}).get, ('scheme', 'hostname', 'port'))
+
+    if scheme == 'etcd':
+        return Etcd(name=scope, config={'scope': scope, 'host': '{}:{}'.format(hostname, port)})
+
+    raise PatroniCtlException('Can not find suitable configuration of distributed configuration store')
+
+
+def post_patroni(member, endpoint, content, headers={'Content-Type': 'application/json'}):
+    url = urlparse(member.api_url)
+    logging.debug(url)
+    return requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers,
+                         data=json.dumps(content), timeout=5)
+
+
+def print_output(columns, rows=[], alignment=None, format='pretty', header=True, delimiter='\t'):
+    if format == 'pretty':
+        t = PrettyTable(columns)
+        for k, v in (alignment or {}).items():
+            t.align[k] = v
+        for r in rows:
+            t.add_row(r)
+        click.echo(t)
+        return
+
+    if format == 'json':
+        elements = list()
+        for r in rows:
+            elements.append(dict(zip(columns, r)))
+
+        click.echo(json.dumps(elements))
+
+    if format == 'tsv':
+        if columns is not None and header:
+            click.echo(delimiter.join(columns) + '\n')
+
+        for r in rows or []:
+            c = [str(c) for c in r]
+            click.echo(delimiter.join(c))
+
+
+def watching(w, watch, max_count=None, clear=True):
+    """
+    >>> len(list(watching(True, 1, 0)))
+    1
+    >>> len(list(watching(True, 1, 1)))
+    2
+    >>> len(list(watching(True, None, 0)))
+    1
+    """
+
+    if w and not watch:
+        watch = 2
+    if watch and clear:
+        click.clear()
+    yield 0
+
+    if max_count is not None and max_count < 1:
+        return
+
+    counter = 1
+    while watch and counter <= (max_count or counter):
+        time.sleep(watch)
+        counter += 1
+        if clear:
+            click.clear()
+        yield 0
+
+
+def build_connect_parameters(conn_url, connect_parameters={}):
+    params = connect_parameters.copy()
+    parsed = parseurl(conn_url)
+    params['host'] = parsed['host']
+    params['port'] = parsed['port']
+    params['fallback_application_name'] = 'Patroni ctl'
+    params['connect_timeout'] = '5'
+
+    return params
+
+
+def get_all_members(cluster, role='master'):
+    if role == 'master':
+        if cluster.leader is not None:
+            yield cluster.leader
+        return
+
+    leader_name = (cluster.leader.member.name if cluster.leader else None)
+    for m in cluster.members:
+        if role == 'any' or role == 'replica' and m.name != leader_name:
+            yield m
+
+
+def get_any_member(cluster, role='master', member=None):
+    members = get_all_members(cluster=cluster, role=role)
+    for m in members:
+        if member is None or m.name == member:
+            return m
+
+    return None
+
+
+def get_cursor(cluster, role='master', member=None, connect_parameters={}):
+    member = get_any_member(cluster=cluster, role=role, member=member)
+    if member is None:
+        return None
+
+    params = build_connect_parameters(member.conn_url, connect_parameters=connect_parameters)
+
+    conn = psycopg2.connect(**params)
+    conn.autocommit = True
+    cursor = conn.cursor()
+    if role == 'any':
+        return cursor
+
+    cursor.execute('SELECT pg_is_in_recovery()')
+    in_recovery = cursor.fetchone()[0]
+
+    if in_recovery and role == 'replica' or not in_recovery and role == 'master':
+        return cursor
+
+    conn.close()
+
+    return None
+
+
+@ctl.command('dsn', help='Generate a dsn for the provided member, defaults to a dsn of the master')
+@click.option('--role', '-r', help='Give a dsn of any member with this role', type=click.Choice(['master', 'replica',
+              'any']), default=None)
+@click.option('--member', '-m', help='Generate a dsn for this member', type=str)
+@option_dcs
+@option_config_file
+@click.argument('cluster_name')
+def dsn(cluster_name, config_file, dcs, role, member):
+    if role is not None and member is not None:
+        raise PatroniCtlException('--role and --member are mutually exclusive options')
+    if member is None and role is None:
+        role = 'master'
+
+    config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
+    m = get_any_member(cluster=cluster, role=role, member=member)
+    if m is None:
+        raise PatroniCtlException('Can not find a suitable member')
+
+    params = build_connect_parameters(m.conn_url)
+    click.echo('host={} port={}'.format(params['host'], params['port']))
+
+
+@ctl.command('query', help='Query a Patroni PostgreSQL member')
+@click.argument('cluster_name')
+@option_config_file
+@option_format
+@click.option('--format', help='Output format (pretty, json)', default='tsv')
+@click.option('--file', '-f', help='Execute the SQL commands from this file', type=click.File('rb'))
+@option_dcs
+@option_watch
+@option_watchrefresh
+@click.option('--role', '-r', help='The role of the query', type=click.Choice(['master', 'replica', 'any']),
+              default=None)
+@click.option('--member', '-m', help='Query a specific member', type=str)
+@click.option('--delimiter', help='The column delimiter', default='\t')
+@click.option('--command', '-c', help='The SQL commands to execute')
+def query(
+    cluster_name,
+    config_file,
+    dcs,
+    role,
+    member,
+    w,
+    watch,
+    delimiter,
+    command,
+    file,
+    format='tsv',
+):
+    if role is not None and member is not None:
+        raise PatroniCtlException('--role and --member are mutually exclusive options')
+    if member is None and role is None:
+        role = 'master'
+
+    if file is not None and command is not None:
+        raise PatroniCtlException('--file and --command are mutually exclusive options')
+
+    if file is not None:
+        command = file.read()
+
+    config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
+
+    cursor = None
+    for _ in watching(w, watch, clear=False):
+
+        output, cursor = query_member(cluster=cluster, cursor=cursor, member=member, role=role, command=command)
+        print_output(None, output, format=format, delimiter=delimiter)
+
+        if cursor is None:
+            cluster = dcs.get_cluster()
+
+
+def query_member(cluster, cursor, member, role, command):
+    try:
+        if cursor is None:
+            cursor = get_cursor(cluster, role=role, member=member)
+
+        if cursor is None:
+            if role is None:
+                message = 'No connection to member {} is available'.format(member)
+            else:
+                message = 'No connection to role={} is available'.format(role)
+            logging.debug(message)
+            return [[timestamp(0), message]], None
+
+        cursor.execute('SELECT pg_is_in_recovery()')
+        in_recovery = cursor.fetchone()[0]
+
+        if in_recovery and role == 'master' or not in_recovery and role == 'replica':
+            cursor.connection.close()
+            return None, None
+
+        cursor.execute(command)
+        return cursor.fetchall(), cursor
+    except (psycopg2.OperationalError, psycopg2.DatabaseError) as oe:
+        logging.debug(oe)
+        if cursor is not None and not cursor.connection.closed:
+            cursor.connection.close()
+        message = oe.pgcode or oe.pgerror or str(oe)
+        message = message.replace('\n', ' ')
+        return [[timestamp(0), 'ERROR, SQLSTATE: {}'.format(message)]], None
+
+
+@ctl.command('remove', help='Remove cluster from DCS')
+@click.argument('cluster_name')
+@option_config_file
+@option_format
+@option_dcs
+def remove(config_file, cluster_name, format, dcs):
+    config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
+
+    if not isinstance(dcs, Etcd):
+        raise PatroniCtlException('We have not implemented this for DCS of type {}'.format(type(dcs)))
+
+    output_members(cluster, format=format)
+
+    confirm = click.prompt('Please confirm the cluster name to remove', type=str)
+    if confirm != cluster_name:
+        raise PatroniCtlException('Cluster names specified do not match')
+
+    message = 'Yes I am aware'
+    confirm = \
+        click.prompt('You are about to remove all information in DCS for {}, please type: "{}"'.format(cluster_name,
+                     message), type=str)
+    if message != confirm:
+        raise PatroniCtlException('You did not exactly type "{}"'.format(message))
+
+    if cluster.leader:
+        confirm = click.prompt('This cluster currently is healthy. Please specify the master name to continue')
+        if confirm != cluster.leader.name:
+            raise PatroniCtlException('You did not specify the current master of the cluster')
+
+    dcs.client.delete(dcs._base_path, recursive=True)
+
+
+def wait_for_leader(dcs, timeout=30):
+    t_stop = time.time() + timeout
+    timeout /= 2
+
+    while time.time() < t_stop:
+        dcs.watch(timeout)
+        cluster = dcs.get_cluster()
+
+        if cluster.leader:
+            return cluster
+
+    raise PatroniCtlException('Timeout occured')
+
+
+def empty_post_to_members(cluster, member_names, force, endpoint):
+    candidates = dict()
+    for m in cluster.members:
+        candidates[m.name] = m
+
+    if len(member_names) == 0:
+        member_names = [click.prompt('Which member do you want to {} [{}]?'.format(endpoint,
+                        ', '.join(candidates.keys())), type=str, default='')]
+
+    for mn in member_names:
+        if mn not in candidates.keys():
+            raise PatroniCtlException('{} is not a member of cluster'.format(mn))
+
+    if not force:
+        confirm = click.confirm('Are you sure you want to {} members {}?'.format(endpoint, ', '.join(member_names)))
+        if not confirm:
+            raise PatroniCtlException('Aborted {}'.format(endpoint))
+
+    for mn in member_names:
+        r = post_patroni(candidates[mn], endpoint, '')
+        if r.status_code != 200:
+            click.echo('{} failed for member {}, status code={}, ({})'.format(endpoint, mn, r.status_code, r.text))
+        else:
+            click.echo('Succesful {} on member {}'.format(endpoint, mn))
+
+
+def ctl_load_config(cluster_name, config_file, dcs):
+    config = load_config(config_file, dcs)
+    dcs = get_dcs(config, cluster_name)
+    cluster = dcs.get_cluster()
+
+    return config, dcs, cluster
+
+
+@ctl.command('restart', help='Restart cluster member')
+@click.argument('cluster_name')
+@click.argument('member_names', nargs=-1)
+@click.option('--role', '-r', help='Restart only members with this role', default='any',
+              type=click.Choice(['master', 'replica', 'any']))
+@click.option('--any', help='Restart a single member only', is_flag=True)
+@option_config_file
+@option_force
+@option_dcs
+def restart(cluster_name, member_names, config_file, dcs, force, role, any):
+    config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
+
+    role_names = [m.name for m in get_all_members(cluster=cluster, role=role)]
+
+    if len(member_names) > 0:
+        member_names = list(set(member_names) & set(role_names))
+    else:
+        member_names = role_names
+
+    if any:
+        random.shuffle(member_names)
+        member_names = member_names[:1]
+
+    output_members(cluster)
+    empty_post_to_members(cluster, member_names, force, 'restart')
+
+
+@ctl.command('reinit', help='Reinitialize cluster member')
+@click.argument('cluster_name')
+@click.argument('member_names', nargs=-1)
+@option_config_file
+@option_force
+@option_dcs
+def reinit(cluster_name, member_names, config_file, dcs, force):
+    config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
+    empty_post_to_members(cluster, member_names, force, 'reinitialize')
+
+
+@ctl.command('failover', help='Failover to a replica')
+@click.argument('cluster_name')
+@click.option('--master', help='The name of the current master', default=None)
+@click.option('--candidate', help='The name of the candidate', default=None)
+@click.option('--force', is_flag=True)
+@option_config_file
+@option_dcs
+def failover(config_file, cluster_name, master, candidate, force, dcs):
+    """
+        We want to trigger a failover for the specified cluster name.
+
+        We verify that the cluster name, master name and candidate name are correct.
+        If so, we trigger a failover and keep the client up to date.
+    """
+
+    config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
+
+    if cluster.leader is None:
+        raise PatroniCtlException('This cluster has no master')
+
+    if master is None:
+        if force:
+            master = cluster.leader.member.name
+        else:
+            master = click.prompt('Master', type=str, default=cluster.leader.member.name)
+
+    if cluster.leader.member.name != master:
+        raise PatroniCtlException('Member {} is not the leader of cluster {}'.format(master, cluster_name))
+
+    candidate_names = [str(m.name) for m in cluster.members if m.name != master]
+    # We sort the names for consistent output to the client
+    candidate_names.sort()
+
+    if len(candidate_names) == 0:
+        raise PatroniCtlException('No candidates found to failover to')
+
+    if candidate is None and not force:
+        candidate = click.prompt('Candidate ' + str(candidate_names), type=str, default='')
+
+    if candidate == master:
+        raise PatroniCtlException('Failover target and source are the same.')
+
+    if candidate and candidate not in candidate_names:
+        raise PatroniCtlException('Member {} does not exist in cluster {}'.format(candidate, cluster_name))
+
+    # By now we have established that the leader exists and the candidate exists
+    click.echo('Current cluster topology')
+    output_members(dcs.get_cluster(), name=cluster_name)
+
+    if not force:
+        a = \
+            click.confirm('Are you sure you want to failover cluster {}, demoting current master {}?'.format(
+                cluster_name, master))
+        if not a:
+            raise PatroniCtlException('Aborting failover')
+
+    failover_value = '{}:{}'.format(master, candidate or '')
+
+    t_started = time.time()
+    r = None
+    try:
+        r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'candidate': candidate or ''})
+        if r.status_code == 200:
+            logging.debug(r)
+            logging.debug(r.text)
+            cluster = dcs.get_cluster()
+            click.echo(timestamp() + ' Failing over to new leader: {}'.format(cluster.leader.member.name))
+        else:
+            click.echo('Failover failed, details: {}, {}'.format(r.status_code, r.text))
+            return
+    except:
+        logging.exception(r)
+        logging.warning('Failing over to DCS')
+        click.echo(timestamp() + ' Could not failover using Patroni api, falling back to DCS')
+        dcs.set_failover_value(failover_value)
+        click.echo(timestamp() + ' Initialized failover from master {}'.format(master))
+        # The failover process should within a minute update the failover key, we will keep watching it until it changes
+        # or we timeout
+        cluster = wait_for_leader(dcs, timeout=60)
+        if cluster.leader.member.name == master:
+            click.echo('Failover failed, master did not change after {:0.1f} seconds'.format(time.time() - t_started))
+            return
+
+    click.echo(timestamp() + ' Failover completed in {:0.1f} seconds, new leader is {}'.format(time.time() - t_started,
+               str(cluster.leader.member.name)))
+    output_members(cluster, name=cluster_name)
+
+
+def output_members(cluster, name=None, format='pretty'):
+    rows = []
+    logging.debug(cluster)
+    leader_name = None
+    if cluster.leader:
+        leader_name = cluster.leader.member.name
+
+    # Mainly for consistent pretty printing and watching we sort the output
+    cluster.members.sort(key=lambda x: x.name)
+    for m in cluster.members:
+        logging.debug(m)
+
+        leader = ''
+        if m.name == leader_name:
+            leader = '*'
+
+        host = build_connect_parameters(m.conn_url)['host']
+
+        xlog_location = m.data.get('xlog_location')
+        lag = ''
+        if xlog_location is not None:
+            lag = round(((cluster.last_leader_operation or 0) - m.data.get('xlog_location', 0)) / 1024 / 1024)
+
+        rows.append([
+            name,
+            m.name,
+            host,
+            leader,
+            m.data.get('state', ''),
+            lag,
+        ])
+
+    columns = [
+        'Cluster',
+        'Member',
+        'Host',
+        'Leader',
+        'State',
+        'Lag in MB',
+    ]
+    alignment = {'Cluster': 'l', 'Member': 'l', 'Host': 'l'}
+
+    print_output(columns, rows, alignment, format)
+
+
+@ctl.command('list', help='List the Patroni members for a given Patroni')
+@click.argument('cluster_names', nargs=-1)
+@option_config_file
+@option_format
+@option_watch
+@option_watchrefresh
+@option_dcs
+def members(config_file, cluster_names, format, watch, w, dcs):
+    if len(cluster_names) == 0:
+        logging.warning('Listing members: No cluster names were provided')
+        return
+
+    config = load_config(config_file, dcs)
+    for cn in cluster_names:
+        dcs = get_dcs(config, cn)
+
+        for _ in watching(w, watch):
+            output_members(dcs.get_cluster(), name=cn, format=format)
+
+
+def timestamp(precision=6):
+    return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:precision - 7]
+
+
+@ctl.command('configure', help='Create configuration file')
+@click.option('--config-file', '-c', help='Configuration file', prompt='Configuration file', default=CONFIG_FILE_PATH)
+@click.option('--dcs', '-d', help='The DCS connect url', prompt='DCS connect url', default='etcd://localhost:4001')
+@click.option('--namespace', '-n', help='The namespace', prompt='Namespace', default='/service/')
+def configure(config_file, dcs, namespace):
+    config = dict()
+    config['dcs_api'] = str(dcs)
+    config['namespace'] = str(namespace)
+    store_config(config, config_file)
@@ -63,6 +63,10 @@ class Member(namedtuple('Member', 'index,name,session,data')):
    def api_url(self):
        return self.data.get('api_url', None)

+    @property
+    def nofailover(self):
+        return self.data.get('tags', {}).get('nofailover', False)
+

 class Leader(namedtuple('Leader', 'index,session,member')):

@@ -122,8 +126,8 @@ class AbstractDCS:
            i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc...
        """
        self._name = name
-        self._scope = config['scope']
-        self._base_path = '/service/' + self._scope
+        self._namespace = '/{}'.format(config.get('namespace', '/service/').strip('/'))
+        self._base_path = '/'.join([self._namespace, config['scope']])

        self._cluster = None
        self._cluster_thread_lock = Lock()
@@ -240,8 +244,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`
@@ -52,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):
@@ -139,6 +143,10 @@ def catch_etcd_errors(func):
            return not func(*args, **kwargs) is None
        except (RetryFailedError, etcd.EtcdException):
            return False
+        except:
+            logger.exception("")
+            raise EtcdError("unexpected error")
+
    return wrapper


@@ -146,7 +154,7 @@ class Etcd(AbstractDCS):

    def __init__(self, name, config):
        super(Etcd, self).__init__(name, config)
-        self.ttl = config['ttl']
+        self.ttl = config.get('ttl', 30)
        self._retry = Retry(deadline=10, max_delay=1, max_tries=-1,
                            retry_exceptions=(etcd.EtcdConnectionFailed,
                                              etcd.EtcdLeaderElectionInProgress,
@@ -177,7 +185,8 @@ class Etcd(AbstractDCS):
            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)
@@ -235,8 +244,8 @@ class Etcd(AbstractDCS):
        return self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl)

    @catch_etcd_errors
-    def initialize(self):
-        return self.retry(self.client.write, self.initialize_path, self._name, prevExist=False)
+    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):
@@ -244,7 +253,7 @@ class Etcd(AbstractDCS):

    @catch_etcd_errors
    def cancel_initialization(self):
-        return self.retry(self.client.delete, self.initialize_path, prevValue=self._name)
+        return self.retry(self.client.delete, self.initialize_path)

    def watch(self, timeout):
        cluster = self.cluster
@@ -13,6 +13,10 @@ class PatroniException(Exception):
        return repr(self.value)


+class PatroniCtlException(Exception):
+    pass
+
+
 class PostgresException(PatroniException):
    pass

@@ -2,6 +2,7 @@ import json
 import logging
 import psycopg2
 import requests
+import sys

 from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
@@ -50,7 +51,8 @@ class Ha:
            'conn_url': self.state_handler.connection_string,
            'api_url': self.patroni.api.connection_string,
            'state': self.state_handler.state,
-            'role': self.state_handler.role
+            'role': self.state_handler.role,
+            'tags': self.patroni.tags
        }
        if data['state'] in ['running', 'restarting', 'starting']:
            try:
@@ -72,10 +74,11 @@ class Ha:
            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
+        elif not self.cluster.initialize and not self.patroni.nofailover:  # no initialize key
+            if self.dcs.initialize(create_new=True):  # race for initialization
                try:
                    self.state_handler.bootstrap()
+                    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")
@@ -130,6 +133,7 @@ class Ha:
            return message
        else:
            self.state_handler.promote()
+            self.touch_member()
            return promote_message

    @staticmethod
@@ -139,7 +143,9 @@ class Ha:

        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."""
+        xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role.
+        tags - dictionary with values of different tags (i.e. nofailover)
+        """

        try:
            response = requests.get(member.api_url, timeout=2, verify=False)
@@ -147,10 +153,11 @@ class Ha:
            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)
+            tags = json.get('tags', dict())
+            return (member, True, not is_master, xlog_location, tags)
        except:
            logging.exception('request failed: GET %s', member.api_url)
-        return (member, False, None, 0)
+        return (member, False, None, 0, {})

    def fetch_nodes_statuses(self, members):
        pool = ThreadPool(len(members))
@@ -165,16 +172,19 @@ class Ha:
        if self.state_handler.is_leader():
            return True

+        if self.patroni.nofailover is True:
+            return False
+
        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]
+        members = [m for m in members if m.name != self.state_handler.name and not m.nofailover 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
+            for member, reachable, in_recovery, xlog_location, tags in self.fetch_nodes_statuses(members):
+                if reachable and not tags.get('nofailover', False):  # If the node is unreachable it's not healhy
                    if not in_recovery:
                        logger.warning('Master (%s) is still alive', member.name)
                        return False
@@ -184,13 +194,15 @@ class Ha:

    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]
+        members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url]
        if members:
-            for member, reachable, in_recovery, xlog_location in self.fetch_nodes_statuses(members):
-                if reachable:
+            for member, reachable, in_recovery, xlog_location, tags in self.fetch_nodes_statuses(members):
+                if reachable and not tags.get('nofailover', False):
                    ret = True  # TODO: check xlog_location
-                else:
+                elif not reachable:
                    logger.info('Member %s is not reachable', member.name)
+                elif tags.get('nofailover', False):
+                    logger.info('Member %s is not allowed to promote', member.name)
        else:
            logger.warning('manual failover: members list is empty')
        return ret
@@ -204,12 +216,15 @@ class Ha:
            # 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
+                member, reachable, in_recovery, xlog_location, tags = self.fetch_node_status(members[0])
+                if reachable and not tags.get('nofailover', False):  # 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)
+                if not reachable:
+                    logger.warning('manual failover: member %s is unhealthy', member.name)
+                elif tags.get('nofailover', False):
+                    logger.warning('manual failover: member %s is not allowed to promote', member.name)

            # at this point we should consider all members as a candidates for failover
            # i.e. we assume that failover.member is None
@@ -218,7 +233,7 @@ class Ha:
        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]
+                members = [m for m in self.cluster.members if m.name not in (failover.member, failover.leader)]
                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
@@ -231,6 +246,13 @@ class Ha:
        return self._is_healthiest_node(members, check_replication_lag=False)

    def is_healthiest_node(self):
+
+        if self.state_handler.is_leader():  # leader is always the healthiest
+            return True
+
+        if self.patroni.nofailover:  # nofailover tag makes node always unhealthy
+            return False
+
        if self.cluster.failover:
            return self.manual_failover_process_no_leader()

@@ -242,6 +264,7 @@ class Ha:
        if delete_leader:
            self.state_handler.stop()
            self.dcs.delete_leader()
+            self.touch_member()
            self.dcs.reset_cluster()
        self.state_handler.follow_the_leader(None)

@@ -269,14 +292,18 @@ class Ha:
        if self.is_healthiest_node():
            if self.acquire_lock():
                if self.cluster.failover:
-                    logger.info('Cleanning up failover key after acquiring leader lock...')
+                    logger.info('Cleaning up failover key after acquiring leader lock...')
                    self.dcs.manual_failover('', '')
+                self.dcs.get_cluster()
                return self.enforce_master_role('acquired session lock as a leader',
                                                'promoted self to leader by acquiring session lock')
            else:
                return self.follow_the_leader('demoted self due after trying and failing to obtain lock',
                                              'following new leader after trying and failing to obtain lock')
        else:
+            if self.patroni.nofailover:
+                return self.follow_the_leader('demoting self because I am not allowed to become master',
+                                              'following a different leader because I am not allowed to promote')
            return self.follow_the_leader('demoting self because i am not the healthiest node',
                                          'following a different leader because i am not the healthiest node')

@@ -350,6 +377,11 @@ class Ha:
        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()
@@ -357,8 +389,8 @@ class Ha:
            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()
@@ -372,8 +404,14 @@ class Ha:
            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 not self.state_handler.is_healthy():
@@ -4,10 +4,12 @@ import psycopg2
 import shlex
 import shutil
 import subprocess
+import tempfile
 import time

 from patroni.exceptions import PostgresConnectionException, PostgresException
 from patroni.utils import Retry, RetryFailedError
+from six import string_types
 from six.moves.urllib_parse import urlparse
 from threading import Lock

@@ -42,12 +44,15 @@ class Postgresql:
    def __init__(self, config):
        self.config = config
        self.name = config['name']
+        self.server_parameters = config.get('parameters', {})
        self.scope = config['scope']
        self.listen_addresses, self.port = config['listen'].split(':')
        self.data_dir = config['data_dir']
        self.replication = config['replication']
        self.superuser = config['superuser']
        self.admin = config['admin']
+        self.initdb_options = config.get('initdb', [])
+        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)
@@ -69,6 +74,7 @@ class Postgresql:
        self._connection = None
        self._cursor_holder = None
        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)

@@ -100,10 +106,14 @@ class Postgresql:
            return False
        # check if the cluster's configuration permits pg_rewind
        data = self.controldata()
-        if data:
-            return data.get('wal_log_hints setting', 'off') == 'on' or\
-                data.get('Data page checksum version', '0') != '0'
-        return False
+        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
@@ -127,9 +137,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:
@@ -152,9 +168,39 @@ class Postgresql:
    def data_directory_empty(self):
        return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []

+    @staticmethod
+    def initdb_allowed_option(name):
+        if name in ['pgdata', 'nosync', 'pwfile', 'sync-only']:
+            raise Exception('{} option for initdb is not allowed'.format(name))
+        return True
+
+    def get_initdb_options(self):
+        options = []
+        for o in self.initdb_options:
+            if isinstance(o, string_types) and self.initdb_allowed_option(o):
+                options.append('--{}'.format(o))
+            elif isinstance(o, dict):
+                keys = list(o.keys())
+                if len(keys) != 1 or not isinstance(keys[0], string_types) or not self.initdb_allowed_option(keys[0]):
+                    raise Exception('Invalid option: {}'.format(o))
+                options.append('--{}={}'.format(keys[0], o[keys[0]]))
+            else:
+                raise Exception('Unknown type of initdb option: {}'.format(o))
+        return options
+
    def initialize(self):
        self.set_state('initalizing new cluster')
-        ret = subprocess.call(self._pg_ctl + ['initdb', '-o', '--encoding=UTF8']) == 0
+        options = self.get_initdb_options()
+        pwfile = None
+        if self.superuser and 'username' not in self.superuser and 'password' in self.superuser:
+            (fd, pwfile) = tempfile.mkstemp()
+            os.write(fd, self.superuser['password'].encode())
+            os.close(fd)
+            options.append('--pwfile={}'.format(pwfile))
+
+        ret = subprocess.call(self._pg_ctl + ['initdb'] + ['-o', ' '.join(options)] if options else []) == 0
+        if pwfile:
+            os.remove(pwfile)
        if ret:
            self.write_pg_hba()
        else:
@@ -165,12 +211,12 @@ class Postgresql:
        os.path.exists(self.trigger_file) and os.unlink(self.trigger_file)

    def write_pgpass(self, record):
-        pgpass = 'pgpass'
-        with open(pgpass, 'w') as f:
+        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'] = pgpass
+        env['PGPASSFILE'] = self.pgpass
        return env

    def sync_from_leader(self, leader):
@@ -269,6 +315,12 @@ class Postgresql:
            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')
@@ -305,7 +357,7 @@ class Postgresql:

    def server_options(self):
        options = "--listen_addresses='{}' --port={}".format(self.listen_addresses, self.port)
-        for setting, value in self.config['parameters'].items():
+        for setting, value in self.server_parameters.items():
            options += " --{}='{}'".format(setting, value)
        return options

@@ -316,13 +368,14 @@ class Postgresql:
        return True

    def check_replication_lag(self, last_leader_operation):
-        return last_leader_operation - self.xlog_position() <= self.config.get('maximum_lag_on_failover', 0)
+        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:
            f.write('\nhost replication {username} {network} md5\n'.format(**self.replication))
            for line in self.config.get('pg_hba', []):
-                if line.split()[0].strip() == 'hostssl' and self.config['parameters'].get('ssl', 'off').lower() != 'on':
+                if line.split()[0].strip() == 'hostssl' and self.server_parameters.get('ssl', 'off').lower() != 'on':
                    continue
                f.write(line + '\n')

@@ -378,8 +431,8 @@ recovery_target_timeline = 'latest'
        try:
            data = subprocess.check_output(['pg_controldata', self.data_dir])
            if data:
-                data = data.splitlines()
-                result = {l.split(':')[0]: l.split(':')[1].strip() for l in data if l}
+                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:
@@ -472,19 +525,23 @@ recovery_target_timeline = 'latest'

    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':
@@ -500,21 +557,27 @@ recovery_target_timeline = 'latest'
    def demote(self):
        self.follow_the_leader(None)

+    def create_or_update_role(self, name, password, options):
+        self.query("""DO $$
+BEGIN
+    SET local synchronous_commit = 'local';
+    PERFORM * FROM pg_authid WHERE rolname = %s;
+    IF FOUND THEN
+        ALTER ROLE "{0}" WITH LOGIN {1} PASSWORD %s;
+    ELSE
+        CREATE ROLE "{0}" WITH LOGIN {1} PASSWORD %s;
+    END IF;
+END;
+$$""".format(name, options), name, password, password)
+
    def create_replication_user(self):
-        self.query('CREATE USER "{}" WITH REPLICATION ENCRYPTED PASSWORD %s'.format(
-            self.replication['username']), self.replication['password'])
+        self.create_or_update_role(self.replication['username'], self.replication['password'], 'REPLICATION')

    def create_connection_users(self):
-        if self.superuser:
-            if 'username' in self.superuser:
-                self.query('CREATE ROLE "{0}" WITH LOGIN SUPERUSER PASSWORD %s'.format(
-                    self.superuser['username']), self.superuser['password'])
-            else:
-                rolsuper = self.query("""SELECT rolname FROM pg_authid WHERE rolsuper = 't'""").fetchone()[0]
-                self.query('ALTER ROLE "{0}" WITH PASSWORD %s'.format(rolsuper), self.superuser['password'])
+        if 'username' in self.superuser:
+            self.create_or_update_role(self.superuser['username'], self.superuser['password'], 'SUPERUSER')
        if self.admin:
-            self.query('CREATE ROLE "{0}" WITH LOGIN CREATEDB CREATEROLE PASSWORD %s'.format(
-                self.admin['username']), self.admin['password'])
+            self.create_or_update_role(self.admin['username'], self.admin['password'], 'CREATEDB CREATEROLE')

    def xlog_position(self):
        return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
@@ -572,6 +635,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
@@ -1 +1 @@
-__version__ = '0.2'
+__version__ = '0.7'
@@ -139,7 +139,7 @@ class ZooKeeper(AbstractDCS):
            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 []
@@ -198,13 +198,14 @@ class ZooKeeper(AbstractDCS):
            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.encode('utf-8')))
+            return value == '' or (not index and self._create(self.failover_path, value))
        except:
            logging.exception('set_failover_value')
            return False

-    def initialize(self):
-        return self._create(self.initialize_path, self._name, makepath=True)
+    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
@@ -270,7 +271,7 @@ class ZooKeeper(AbstractDCS):

    def _cancel_initialization(self):
        node = self.get_node(self.initialize_path)
-        if node and node[0] == self._name:
+        if node:
            self.client.delete(self.initialize_path, version=node[1].version)

    def cancel_initialization(self):
@@ -0,0 +1,5 @@
+#!/usr/bin/env python
+from patroni.ctl import ctl
+
+if __name__ == '__main__':
+    ctl()
@@ -34,6 +34,24 @@ postgresql:
  data_dir: data/postgresql0
  maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
  use_slots: True
+  pgpass: /tmp/pgpass0
+  initdb:  ## We allow the following options to be passed on to initdb
+  # - auth: authmethod
+  # - auth-host: authmethod
+  # - auth-local: authmethod
+  - encoding: UTF8
+  # - data-checksums  # When pg_rewind is needed on 9.3, this needs to be enabled
+  # - locale: locale
+  # - lc-collate: locale
+  # - lc-ctype: locale
+  # - lc-messages: locale
+  # - lc-monetary: locale
+  # - lc-numeric: locale
+  # - lc-time: locale
+  # - text-search-config: CFG
+  # - xlogdir: directory
+  # - debug
+  # - noclean
  pg_rewind:
    username: postgres
    password: zalando
@@ -67,3 +85,8 @@ postgresql:
    max_replication_slots: 5
    hot_standby: "on"
    wal_log_hints: "on"
+tags:
+    nofailover: False
+    noloadbalance: False
+    clonefrom: False
+    replicatefrom: 127.0.0.1
@@ -34,6 +34,24 @@ postgresql:
  data_dir: data/postgresql1
  maximum_lag_on_failover: 1048576 # 1 megabyte in bytes
  use_slots: True
+  pgpass: /tmp/pgpass1
+  initdb:  ## We allow the following options to be passed on to initdb
+  # - auth: authmethod
+  # - auth-host: authmethod
+  # - auth-local: authmethod
+  - encoding: UTF8
+  # - data-checksums  # When pg_rewind is needed on 9.3, this needs to be enabled
+  # - locale: locale
+  # - lc-collate: locale
+  # - lc-ctype: locale
+  # - lc-messages: locale
+  # - lc-monetary: locale
+  # - lc-numeric: locale
+  # - lc-time: locale
+  # - text-search-config: CFG
+  # - xlogdir: directory
+  # - debug
+  # - noclean
  pg_rewind:
    username: postgres
    password: zalando
@@ -67,3 +85,8 @@ postgresql:
    max_replication_slots: 5
    hot_standby: "on"
    wal_log_hints: "on"
+tags:
+    nofailover: False
+    noloadbalance: False
+    clonefrom: False
+    replicatefrom: 127.0.0.1
@@ -1,9 +1,11 @@
 boto
 dnspython
 mock
-psycopg2
+psycopg2>=2.6.1
 PyYAML
 requests
 six >= 1.7
 kazoo>=2.2.1
 python-etcd>=0.4.1
+click>=4.1
+prettytable>=0.7
@@ -1,9 +1,11 @@
 boto
 mock
 dnspython3
-psycopg2
+psycopg2>=2.6.1
 PyYAML
 requests
 six
 kazoo>=2.2.1
 python-etcd>=0.4.1
+click>=4.1
+prettytable>=0.7
@@ -56,7 +56,7 @@ CLASSIFIERS = [
    'Programming Language :: Python :: Implementation :: CPython',
 ]

-CONSOLE_SCRIPTS = ['patroni = patroni:main']
+CONSOLE_SCRIPTS = ['patroni = patroni:main', 'patronictl = patroni.ctl:ctl']


 class PyTest(TestCommand):
@@ -3,8 +3,11 @@ 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 six.moves.BaseHTTPServer import BaseHTTPRequestHandler
+import socket
 from test_postgresql import psycopg2_connect, MockCursor


@@ -38,12 +41,16 @@ class MockHa(Mock):
    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()
+    tags = {}


 class MockRequest:
@@ -86,6 +93,20 @@ class TestRestApiHandler(unittest.TestCase):
            MockRestApiServer(RestApiHandler, b'GET /master')
        MockRestApiServer(RestApiHandler, b'GET /master')

+    def test_do_OPTIONS(self):
+        MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
+
+        with patch.object(BaseHTTPRequestHandler, 'handle_one_request') as mock_handle_request:
+            mock_handle_request.side_effect = socket.error("foo")
+            MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
+
+        # make sure socket.error gets propagated via wfile object in finalize()
+        with patch.object(MockRequest, 'makefile') as makefile:
+            makefile.return_value.closed = False
+            makefile.return_value.readline.side_effect = lambda x: b"foo"
+            makefile.return_value.flush = Mock(side_effect=socket.error("foo"))
+            MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
+
    def test_do_GET_patroni(self):
        MockRestApiServer(RestApiHandler, b'GET /patroni')

@@ -117,3 +138,31 @@ class TestRestApiHandler(unittest.TestCase):
            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)
@@ -0,0 +1,376 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import os
+import pytest
+import unittest
+import psycopg2
+import requests
+import patroni.exceptions
+import etcd
+from mock import patch, Mock
+
+from click.testing import CliRunner
+from patroni.ctl import ctl, members, store_config, load_config, output_members, post_patroni, get_dcs, \
+    wait_for_leader, get_all_members, get_any_member, get_cursor, query_member, configure
+from patroni.ha import Ha
+from patroni.etcd import Etcd, Client
+from test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, \
+    get_cluster_initialized_with_only_leader, MockPostgresql, MockPatroni, run_async, \
+    get_cluster_not_initialized_without_leader
+from test_etcd import etcd_read, etcd_write, requests_get, MockResponse
+from test_postgresql import MockConnect, psycopg2_connect
+
+CONFIG_FILE_PATH = './test-ctl.yaml'
+
+def test_rw_config():
+    runner = CliRunner()
+    config = {'a':'b'}
+    with runner.isolated_filesystem():
+        store_config(config, CONFIG_FILE_PATH + '/dummy')
+        os.remove(CONFIG_FILE_PATH + '/dummy')
+        os.rmdir(CONFIG_FILE_PATH)
+
+        with pytest.raises(Exception):
+            result = load_config(CONFIG_FILE_PATH, None)
+            assert 'Could not load configuration file' in result.output
+
+        os.mkdir(CONFIG_FILE_PATH)
+        with pytest.raises(Exception):
+            store_config(config, CONFIG_FILE_PATH)
+
+        os.rmdir(CONFIG_FILE_PATH)
+
+    store_config(config, CONFIG_FILE_PATH)
+    load_config(CONFIG_FILE_PATH, None)
+    load_config(CONFIG_FILE_PATH, '0.0.0.0')
+
+@patch('patroni.ctl.load_config', Mock(return_value={'dcs': {'scheme': 'etcd', 'hostname': 'localhost', 'port': 4001}}))
+class TestCtl(unittest.TestCase):
+
+    @patch.object(Client, 'machines')
+    def setUp(self, mock_machines):
+        mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
+        self.p = MockPostgresql()
+        self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'})
+        self.e.client.read = etcd_read
+        self.e.client.write = etcd_write
+        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()
+
+    @patch('psycopg2.connect', psycopg2_connect)
+    def test_get_cursor(self):
+        c = get_cursor(get_cluster_initialized_without_leader(), role='master')
+        assert c is None
+
+        c = get_cursor(get_cluster_initialized_with_leader(), role='master')
+        assert c is not None
+
+        c = get_cursor(get_cluster_initialized_with_leader(), role='replica')
+        # # MockCursor returns pg_is_in_recovery as false
+        assert c is None
+
+        c = get_cursor(get_cluster_initialized_with_leader(), role='any')
+        assert c is not None
+
+    def test_output_members(self):
+        cluster = get_cluster_initialized_with_leader()
+        output_members(cluster, name='abc', format='pretty')
+        output_members(cluster, name='abc', format='json')
+        output_members(cluster, name='abc', format='tsv')
+
+    @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
+    @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None))
+    @patch('patroni.etcd.Etcd.set_failover_value', Mock(return_value=None))
+    @patch('patroni.ctl.wait_for_leader', Mock(return_value=get_cluster_initialized_with_leader()))
+    @patch('requests.get', requests_get)
+    @patch('requests.post', requests_get)
+    @patch('patroni.ctl.post_patroni', Mock(return_value=MockResponse()))
+    def test_failover(self):
+        runner = CliRunner()
+
+        with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())):
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+y''')
+            assert 'Failing over to new leader' in result.output
+
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+N''')
+            assert 'Aborting failover' in str(result.exception)
+
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+leader
+y''')
+            assert 'target and source are the same' in str(result.exception)
+
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+Reality
+y''')
+            assert 'Reality does not exist' in str(result.exception)
+
+            result = runner.invoke(ctl, ['failover', 'dummy', '--force'])
+            assert 'Failing over to new leader' in result.output
+
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='dummy')
+            assert 'is not the leader of cluster' in str(result.exception)
+
+        with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_only_leader())):
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+y''')
+            assert 'No candidates found to failover to' in str(result.exception)
+
+        with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())):
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+y''')
+            assert 'This cluster has no master' in str(result.exception)
+
+        with patch('patroni.ctl.post_patroni', Mock(side_effect=Exception())):
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+y''')
+            assert 'falling back to DCS' in result.output
+            assert 'Failover failed' in result.output
+
+        mocked = Mock()
+        mocked.return_value.status_code = 500
+        with patch('patroni.ctl.post_patroni', Mock(return_value=mocked)):
+            result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
+other
+y''')
+            assert 'Failover failed, details' in result.output
+
+#        with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())):
+#            result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='nonsense')
+#            assert 'is not the leader of cluster' in str(result.exception)
+
+            # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8', '--master', 'nonsense'])
+            # assert 'is not the leader of cluster' in str(result.exception)
+
+            # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nn')
+            # assert 'Aborting failover' in str(result.exception)
+
+            # with patch('patroni.ctl.wait_for_leader', Mock(return_value = get_cluster_initialized_with_leader())):
+                # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY')
+                # assert 'master did not change after' in result.output
+
+                # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY')
+                # assert 'Failover failed' in result.output
+
+    def test_(self):
+        self.assertRaises(patroni.exceptions.PatroniCtlException, get_dcs, {'scheme': 'dummy'}, 'dummy')
+
+    @patch('psycopg2.connect', psycopg2_connect)
+    @patch('patroni.ctl.query_member', Mock(return_value=([['mock column']], None)))
+    def test_query(self):
+        runner = CliRunner()
+
+        with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)):
+            result = runner.invoke(ctl, [
+                'query',
+                'alpha',
+                '--member',
+                'abc',
+                '--role',
+                'master',
+            ])
+            assert 'mutually exclusive' in str(result.exception)
+
+            with runner.isolated_filesystem():
+                dummy_file = open('dummy', 'w')
+                dummy_file.write('SELECT 1')
+                dummy_file.close()
+
+                result = runner.invoke(ctl, [
+                    'query',
+                    'alpha',
+                    '--file',
+                    'dummy',
+                    '--command',
+                    'dummy',
+                ])
+                assert 'mutually exclusive' in str(result.exception)
+
+                result = runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy'])
+
+                os.remove('dummy')
+
+            result = runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1'])
+            assert 'mock column' in result.output
+
+    @patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor()))
+    def test_query_member(self):
+        rows = query_member(None, None, None, 'master', 'SELECT pg_is_in_recovery()')
+        assert 'False' in str(rows)
+
+        rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()')
+        assert rows == (None, None)
+
+        with patch('patroni.ctl.get_cursor', Mock(return_value=None)):
+            rows = query_member(None, None, None, None, 'SELECT pg_is_in_recovery()')
+            assert 'No connection to' in str(rows)
+
+            rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()')
+            assert 'No connection to' in str(rows)
+
+        with patch('patroni.ctl.get_cursor', Mock(side_effect=psycopg2.OperationalError('bla'))):
+            rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()')
+
+        with patch('test_postgresql.MockCursor.execute', Mock(side_effect=psycopg2.OperationalError('bla'))):
+            rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()')
+
+    @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
+    def test_dsn(self):
+        runner = CliRunner()
+
+        with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)):
+            result = runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8'])
+            assert 'host=127.0.0.1 port=5435' in result.output
+
+            result = runner.invoke(ctl, [
+                'dsn',
+                'alpha',
+                '--role',
+                'master',
+                '--member',
+                'dummy',
+            ])
+            assert 'mutually exclusive' in str(result.exception)
+
+            result = runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy'])
+            assert 'Can not find' in str(result.exception)
+
+        # result = runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8', '--role', 'replica'])
+        # assert 'host=127.0.0.1 port=5436' in result.output
+
+    @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
+    @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None))
+    @patch('requests.get', requests_get)
+    @patch('requests.post', requests_get)
+    def test_restart_reinit(self):
+        runner = CliRunner()
+
+        result = runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
+        result = runner.invoke(ctl, ['reinit', 'alpha', '--dcs', '8.8.8.8'], input='y')
+
+        result = runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='N')
+        result = runner.invoke(ctl, [
+            'restart',
+            'alpha',
+            '--dcs',
+            '8.8.8.8',
+            'dummy',
+            '--any',
+        ], input='y')
+        assert 'not a member' in str(result.exception)
+
+        with patch('requests.post', Mock(return_value=MockResponse())):
+            result = runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
+
+    @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
+    @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None))
+    def test_remove(self):
+        runner = CliRunner()
+
+        result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nslave')
+        assert 'Please confirm' in result.output
+        assert 'You are about to remove all' in result.output
+        assert 'You did not exactly type' in str(result.exception)
+
+        result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
+Yes I am aware
+slave''')
+        assert 'You did not specify the current master of the cluster' in str(result.exception)
+
+        result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader')
+        assert 'Cluster names specified do not match' in str(result.exception)
+
+        with patch('patroni.etcd.Etcd.get_cluster', get_cluster_initialized_with_leader):
+            result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'],
+                                   input='''alpha
+Yes I am aware
+leader''')
+            assert 'object has no attribute' in str(result.exception)
+
+        with patch('patroni.ctl.get_dcs', Mock(return_value=Mock())):
+            result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'],
+                                   input='''alpha
+Yes I am aware
+leader''')
+            assert 'We have not implemented this for DCS of type' in str(result.exception)
+
+    @patch('patroni.etcd.Etcd.watch', Mock(return_value=None))
+    @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
+    def test_wait_for_leader(self):
+        dcs = self.e
+        self.assertRaises(patroni.exceptions.PatroniCtlException, wait_for_leader, dcs, 0)
+
+        cluster = wait_for_leader(dcs=dcs, timeout=2)
+        assert cluster.leader.member.name == 'leader'
+
+    def test_post_patroni(self):
+        member = get_cluster_initialized_with_leader().leader.member
+        self.assertRaises(requests.exceptions.ConnectionError, post_patroni, member, 'dummy', {})
+
+    def test_ctl(self):
+        runner = CliRunner()
+
+        runner.invoke(ctl, ['list'])
+
+        result = runner.invoke(ctl, ['--help'])
+        assert 'Usage:' in result.output
+
+    def test_get_any_member(self):
+        m = get_any_member(get_cluster_initialized_without_leader(), role='master')
+        assert m is None
+
+        m = get_any_member(get_cluster_initialized_with_leader(), role='master')
+        assert m.name == 'leader'
+
+    def test_get_all_members(self):
+        r = list(get_all_members(get_cluster_initialized_without_leader(), role='master'))
+        assert len(r) == 0
+
+        r = list(get_all_members(get_cluster_initialized_with_leader(), role='master'))
+        assert len(r) == 1
+        assert r[0].name == 'leader'
+
+        r = list(get_all_members(get_cluster_initialized_with_leader(), role='replica'))
+        assert len(r) == 1
+        assert r[0].name == 'other'
+
+        r = list(get_all_members(get_cluster_initialized_without_leader(), role='replica'))
+        assert len(r) == 2
+
+    @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
+    @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None))
+    @patch('requests.get', requests_get)
+    @patch('requests.post', requests_get)
+    def test_members(self):
+        runner = CliRunner()
+
+        result = runner.invoke(members, ['alpha'])
+        assert result.exit_code == 0
+
+    def test_configure(self):
+        runner = CliRunner()
+
+        result = runner.invoke(configure, [
+            '--dcs',
+            'abc',
+            '-c',
+            'dummy',
+            '-n',
+            'bla',
+        ])
+
+        assert result.exit_code == 0
+
+
@@ -8,7 +8,7 @@ import unittest
 from dns.exception import DNSException
 from mock import Mock, patch
 from patroni.dcs import Cluster, DCSError, Leader
-from patroni.etcd import Client, Etcd
+from patroni.etcd import Client, Etcd, EtcdError


 class MockResponse:
@@ -17,6 +17,7 @@ class MockResponse:
        self.status_code = 200
        self.content = '{}'
        self.ok = True
+        self.text = ''

    def json(self):
        return json.loads(self.content)
@@ -50,7 +51,7 @@ def requests_get(url, **kwargs):
    if url.startswith('http://local'):
        raise requests.exceptions.RequestException()
    elif ':8011/patroni' in url:
-        response.content = '{"role": "replica", "xlog": {"replayed_location": 0}}'
+        response.content = '{"role": "replica", "xlog": {"replayed_location": 0}, "tags": {}}'
    elif url.endswith('/members'):
        if url.startswith('http://error'):
            response.content = '[{}]'
@@ -80,7 +81,7 @@ def etcd_watch(key, index=None, timeout=None, recursive=None):
 def etcd_write(key, value, **kwargs):
    if key == '/service/exists/leader':
        raise etcd.EtcdAlreadyExist
-    if key == '/service/test/leader':
+    if key == '/service/test/leader' or key == '/patroni/test/leader':
        if kwargs.get('prevValue', None) == 'foo' or not kwargs.get('prevExist', True):
            return True
    raise etcd.EtcdException
@@ -106,13 +107,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:[email protected]:5434/postgres"
-                        + "?application_name=http://127.0.0.1:8009/patroni",
+                     "value": "postgres://replicator:[email protected]: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:[email protected]:5433/postgres"
-                        + "?application_name=http://127.0.0.1:8008/patroni",
+                     "value": "postgres://replicator:[email protected]: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}}
@@ -165,6 +166,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,11 +205,14 @@ class TestEtcd(unittest.TestCase):
    def setUp(self):
        with patch.object(Client, 'machines') as mock_machines:
            mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
-            self.etcd = Etcd('foo', {'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'})
+            self.etcd = Etcd('foo', {'namespace': '/patroni/', 'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'})
            self.etcd.client.write = etcd_write
            self.etcd.client.read = etcd_read
            self.etcd.client.delete = Mock(side_effect=etcd.EtcdException())

+    def test_base_path(self):
+        self.assertEquals(self.etcd._base_path, '/patroni/test')
+
    @patch('dns.resolver.query', dns_query)
    def test_get_etcd_client(self):
        with patch.object(etcd.Client, 'machines') as mock_machines:
@@ -258,3 +267,7 @@ class TestEtcd(unittest.TestCase):
        self.etcd.watch(4.5)
        self.etcd.watch(9.5)
        self.etcd.watch(100)
+
+    @patch('patroni.etcd.Etcd.retry', Mock(side_effect=AttributeError("foo")))
+    def test_other_exceptions(self):
+        self.assertRaises(EtcdError, self.etcd.cancel_initialization)
@@ -1,6 +1,7 @@
+import etcd
 import unittest

-from mock import Mock, patch
+from mock import Mock, MagicMock, patch
 from patroni.dcs import Cluster, Failover, Leader, Member
 from patroni.etcd import Client, Etcd
 from patroni.exceptions import DCSError, PostgresException
@@ -26,7 +27,7 @@ def get_cluster_not_initialized_without_leader():

 def get_cluster_initialized_without_leader(leader=False, failover=None):
    m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres',
-                                 'api_url': 'http://127.0.0.1:8008/patroni'})
+                                 'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location':4})
    l = Leader(0, 0, m) if leader else None
    o = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
                                'api_url': 'http://127.0.0.1:8011/patroni'})
@@ -36,6 +37,10 @@ def get_cluster_initialized_without_leader(leader=False, failover=None):
 def get_cluster_initialized_with_leader(failover=None):
    return get_cluster_initialized_without_leader(leader=True, failover=failover)

+def get_cluster_initialized_with_only_leader(failover=None):
+    l = get_cluster_initialized_without_leader(leader=True, failover=failover).leader
+    return get_cluster(True, l, [l], failover)
+

 class MockPostgresql(Mock):

@@ -81,6 +86,8 @@ class MockPatroni:
        self.postgresql = p
        self.dcs = d
        self.api = Mock()
+        self.tags = {}
+        self.nofailover = None
        self.api.connection_string = 'http://127.0.0.1:8008'


@@ -98,6 +105,7 @@ 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.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()
@@ -128,6 +136,12 @@ class TestHa(unittest.TestCase):
        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.p.is_leader = self.p.is_healthy = false
@@ -264,6 +278,11 @@ class TestHa(unittest.TestCase):
        f = Failover(0, MockPostgresql.name, '')
        self.ha.cluster = get_cluster_initialized_with_leader(f)
        self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself')
+        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'})
+        self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        # manual failover from the previous leader to us won't happen if we hold the nofailover flag
+        self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name))
+        self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')

    @patch('requests.get', requests_get)
    def test_manual_failover_process_no_leader(self):
@@ -272,24 +291,41 @@ class TestHa(unittest.TestCase):
        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.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.ha.fetch_node_status = lambda e: (e, False, True, 0, {})  # inaccessible, in_recovery
        self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+        # set failover flag to True for all members of the cluster
+        # this should elect the current member, as we are not going to call the API for it.
+        self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other'))
+        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'})  # accessible, in_recovery
+        self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+        # same as previous, but set the current member to nofailover. In no case it should be elected as a leader
+        self.ha.patroni.nofailover = True
+        self.assertEquals(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote')
+
+    def test_is_healthiest_node(self):
+        self.ha.state_handler.is_leader = false
+        self.ha.patroni.nofailover = False
+        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {})
+        self.assertTrue(self.ha.is_healthiest_node())

    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.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.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.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))
+        self.ha.patroni.nofailover = True
+        self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
+        self.ha.patroni.nofailover = False

    @patch('requests.get', requests_get)
    def test_fetch_node_status(self):
@@ -74,3 +74,9 @@ class TestPatroni(unittest.TestCase):
        self.p.schedule_next_run()
        self.p.next_run = time.time() - self.p.nap_time - 1
        self.p.schedule_next_run()
+
+    def test_nofailover(self):
+        self.p.tags['nofailover'] = True
+        self.assertTrue(self.p.nofailover)
+        self.p.tags['nofailover'] = None
+        self.assertFalse(self.p.nofailover)
@@ -4,12 +4,7 @@ import psycopg2
 import shutil
 import unittest

-from sys import version_info
-if version_info.major == 2:
-    import __builtin__ as builtins
-else:
-    import builtins
-
+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
@@ -19,6 +14,11 @@ 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:

    def __init__(self, connection):
@@ -56,6 +56,9 @@ class MockCursor:
    def fetchone(self):
        return self.results[0]

+    def fetchall(self):
+        return self.results
+
    def close(self):
        pass

@@ -86,7 +89,7 @@ class MockConnect(Mock):


 def pg_controldata_string(*args, **kwargs):
-    return """
+    return b"""
 pg_control version number:            942
 Catalog version number:               201509161
 Database system identifier:           6200971513092291716
@@ -118,12 +121,12 @@ Backup start location:                0/0
 Backup end location:                  0/0
 End-of-backup record required:        no
 wal_level setting:                    hot_standby
-wal_log_hints setting:                on
-max_connections setting:              100
-max_worker_processes setting:         8
-max_prepared_xacts setting:           0
-max_locks_per_xact setting:           64
-track_commit_timestamp setting:       off
+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
@@ -141,10 +144,10 @@ 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 \n'\
-        '&& cp %p ../wal_archive/%f" "--wal_log_hints=on" "--max_wal_senders=5" "--archive_timeout=1800s" "--archive_mode=on"'\
-        ' "--max_replication_slots=5"\n'
+    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):
@@ -162,7 +165,7 @@ class TestPostgresql(unittest.TestCase):
        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': ''},
+                             'superuser': {'password': 'test'},
                             'admin': {'username': 'admin', 'password': 'admin'},
                             'pg_rewind': {'username': 'admin', 'password': 'admin'},
                             'replication': {'username': 'replicator',
@@ -187,6 +190,16 @@ class TestPostgresql(unittest.TestCase):
    def test_data_directory_empty(self):
        self.assertTrue(self.p.data_directory_empty())

+    def test_get_initdb_options(self):
+        self.p.initdb_options = [{'encoding': 'UTF8'}, 'data-checksums']
+        self.assertEquals(self.p.get_initdb_options(), ['--encoding=UTF8', '--data-checksums'])
+        self.p.initdb_options = [{'pgdata': 'bar'}]
+        self.assertRaises(Exception, self.p.get_initdb_options)
+        self.p.initdb_options = [{'foo': 'bar', 1: 2}]
+        self.assertRaises(Exception, self.p.get_initdb_options)
+        self.p.initdb_options = [1]
+        self.assertRaises(Exception, self.p.get_initdb_options)
+
    def test_initialize(self):
        self.assertTrue(self.p.initialize())
        self.assertTrue(os.path.exists(os.path.join(self.p.data_dir, 'pg_hba.conf')))
@@ -210,10 +223,16 @@ class TestPostgresql(unittest.TestCase):
        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))

    @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
@@ -222,6 +241,7 @@ class TestPostgresql(unittest.TestCase):
    @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)
@@ -232,8 +252,8 @@ class TestPostgresql(unittest.TestCase):
        self.p.follow_the_leader(self.leader)
        self.p.require_rewind()
        with mock.patch('os.path.islink', MagicMock(return_value=True)):
-            with mock.patch('os.unlink', MagicMock(return_value=True)):
-                with mock.patch('patroni.postgresql.Postgresql.can_rewind', new_callable=PropertyMock(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)):
@@ -327,6 +347,7 @@ 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):
        with patch('subprocess.call', Mock(return_value=1)):
            self.assertRaises(PostgresException, self.p.bootstrap)
@@ -429,3 +450,19 @@ class TestPostgresql(unittest.TestCase):
        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()
@@ -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)
Author	SHA1	Message	Date
Feike Steenbergen	2a37e2c5d7	Bumped version to 0.7	2015-11-18 16:45:05 +01:00
Oleksii Kliukin	c003af294a	Merge pull request #82 from zalando/feature/patroni_cli_or_ctl_tbd Feature/patroni cli or ctl tbd	2015-11-18 16:17:02 +01:00
Oleksii Kliukin	0e7adb38fa	Merge pull request #100 from zalando/bugfix/etcd_exceptions Handle unexpected exceptions in etcd.	2015-11-18 15:08:59 +01:00
Feike Steenbergen	897024a297	PatroniCtl: Bugfixes for formatting and timeouts	2015-11-18 15:01:35 +01:00
Feike Steenbergen	f081f9d67e	Bugfix for Patronictl tests	2015-11-18 12:55:11 +01:00
Feike Steenbergen	cae025c3fb	Testing patronictl: Mock configuration and bugfix	2015-11-18 12:50:36 +01:00
Feike Steenbergen	4bb1e060c0	Bugfix for patronictl tests	2015-11-18 12:00:51 +01:00
Feike Steenbergen	ca4d9eaaf9	Patronictl: Expand tests to increase coverage	2015-11-18 11:51:24 +01:00
Oleksii Kliukin	fef7d45208	Handle unexpected exceptions in etcd. Previously, patroni would die after receiving an exception other than RetryFailedError, etcd.EtcdException from etcd. We have observed an AttributeError raised by etcd on some occasions. With this change, we demote ourselves, but not terminate on such exceptions.	2015-11-17 16:08:58 +01:00
Feike Steenbergen	2d9f5d9e4b	Refactoring and adding a dsn option to patronictl. Some refactoring to reuse some codepaths. A dsn option is now added, it is useful in scripts like so: psql -d "$(patronictl dsn alpha)" Restarting has been extended to allow restarting based on role.	2015-11-17 15:42:43 +01:00
Feike Steenbergen	e920416114	Patronictl: Extend test cases and create own Exception class.	2015-11-16 12:36:15 +01:00
Alexander Kukushkin	57f19fb149	Merge pull request #80 from zalando/feature/nofailover Feature/nofailover	2015-11-16 10:21:56 +01:00
Alexander Kukushkin	96dcb54429	Merge pull request #96 from zalando/bugfix/haproxy_errors Handle haproxy requests. Improve failover status code.	2015-11-16 10:19:19 +01:00
Alexander Kukushkin	490a54f922	Merge pull request #94 from zalando/feature/initdb Make it possible to specify custom options for initdb	2015-11-16 10:17:28 +01:00
Alexander Kukushkin	7d59a757fb	Merge pull request #93 from zalando/bugfix/parameters Do not fail when there is no postgresql/parameters section in a config file	2015-11-16 10:13:52 +01:00
Alexander Kukushkin	5290792dff	Merge pull request #92 from zalando/feature/api Call touch_member right after demote and promote.	2015-11-16 10:13:24 +01:00
Feike Steenbergen	990276c214	Install patronictl as a script	2015-11-13 13:02:40 +01:00
Feike Steenbergen	dcb5113f9d	Refactoring from patronicli to patronictl	2015-11-13 12:55:38 +01:00
Oleksii Kliukin	70cbbb4ef4	Ignore only socket.error at finalization stage of StreamRequestHandler, not other errors.	2015-11-12 17:55:02 +01:00
Oleksii Kliukin	28934350ef	Handle haproxy requests. Improve failover status code. By default, haproxy sens an OPTION request, which we didn't handle until now. In addition, all haproxy requests that doesn't examine the request body close the connection as soon as the status code is obtained. Such behavior breaks BaseHTTPRequestHandler, namely handle_one_request, which doesn't check for connection reset by peer and throw this error on a higher level, but since we don't call this function directly, there is no place in the code to catch it, therefore, we have to patch this function in the base class. In addition, patch the StreamRequestHandler finish() function in order to handle the connection reset error. Re-read the cluster from DCS right after the failover to supply the correct new values to the API thread. Fix a typo.	2015-11-12 17:38:22 +01:00
Alexander Kukushkin	092bf8defa	Merge branch 'master' of github.com:zalando/patroni into feature/initdb	2015-11-12 12:22:14 +01:00
Alexander Kukushkin	13cb8e7402	set synchronous_commit to local when creating roles	2015-11-12 12:20:05 +01:00
Oleksii Kliukin	9f0ee544cf	Merge pull request #95 from LappleApple/patch-1 Documentation improvements by @LappleApple	2015-11-12 12:18:13 +01:00
Alexander Kukushkin	d39d297d92	blacklist some options for initdb instead of having big whitelist	2015-11-12 12:07:36 +01:00
LappleApple	eb83d556e8	Update README.rst	2015-11-12 11:43:49 +01:00
Alexander Kukushkin	f0a6c86caa	Make it possible to specify custom options for initdb In the initial implementation we were using the only option --encoding=UTF8. In order to have pg_rewind working with postgresql-9.3 we have to enable data-checksums. The naive approach was to enable it globaly but taking into account some performance degradation it's better not to do it but make it possible to configure it. In addition to that fix all problems with setting up password of default postgres user: execute CREATE ROLE \| ALTER ROLE depending on content of pg_authid	2015-11-11 15:59:34 +01:00
Alexander Kukushkin	9c16830fc0	Do not fail when there is no postgresql/parameters section in a config file	2015-11-11 10:46:37 +01:00
Oleksii Kliukin	194aa92fd2	Merge pull request #89 from zalando/bugfix/cleanup Remove the useless code line and fix a typo in the error message. Per suggestion by @antban	2015-11-10 18:18:46 +01:00
Oleksii Kliukin	a8312dfefc	Fixed a typo in the error message.	2015-11-10 18:14:18 +01:00
Feike Steenbergen	92b23a2018	Merge branch 'feature/api' of github.com:zalando/patroni into feature/patroni_cli_or_ctl_tbd	2015-11-09 16:33:17 +01:00
Alexander Kukushkin	8a70cc4d7b	Call touch_member right after demote and promote. This is necessary to propagate actual information about node into DCS.	2015-11-09 16:22:51 +01:00
Feike Steenbergen	da23dd12f3	Do not wait for leader key to change after failover. Previously, the leader key was watched for changes after a failover. This resulted in a delay of up to 10 seconds to report a healthy failover back to the client. With this patch, we are not relying on the role of a member registered in the dcs anymore.	2015-11-06 15:02:13 +01:00
Oleksii Kliukin	e52a34de1c	Remove the useless codeline.	2015-11-03 16:04:01 +03:00
Oleksii Kliukin	87ead9ddaf	Fix haproxy configuration to work by default with Patroni.	2015-11-01 15:29:45 +03:00
Feike Steenbergen	3938359828	Command Line: Add reinit and restart commands.	2015-10-27 12:13:26 +01:00
Oleksii Kliukin	a1a7c9b4c5	Make sure nofailover tag is set to False by default.	2015-10-27 09:46:03 +01:00
Oleksii Kliukin	8d68130aac	Fix the test to account for an extra parameter returned by fetch_node_statuses.	2015-10-27 09:32:09 +01:00
Feike Steenbergen	7784f7db3d	Bumped version to 0.6	2015-10-27 09:20:24 +01:00
Oleksii Kliukin	da74aee2cd	Make sure that nofailover presents the node even from initializing the cluster.	2015-10-26 18:57:28 +01:00
Oleksii Kliukin	685363c5a8	Fix an invocation fo fetch_node_statuses in the API code to account for the tags	2015-10-26 18:57:05 +01:00
Oleksii Kliukin	1f9d8c039c	Add tests for nofailover.	2015-10-26 17:19:35 +01:00
Oleksii Kliukin	ed5e3e53e2	Use a shortcut to call patroni nofailover attribute.	2015-10-26 17:18:45 +01:00
Feike Steenbergen	98a0d8381b	Add prettytable to requirements	2015-10-26 15:07:08 +01:00
Feike Steenbergen	a2cb3f1889	Include Click as requirement for patroncli	2015-10-26 15:04:23 +01:00
Feike Steenbergen	e2aff13d3e	Patronictl: Create commandline tool that can manage clusters. For managing Patroni clusters, the Patroni api can be used. For many tasks, a command line interface for this api would be a useful addition. This commit adds patroncli (The name is still under debate). The command line interface needs access to the DCS; this is required for any operation. For some tasks it is required to have access to the Patroni api. A small summary of the additions to get the cli/ctl started: * Updated Docker image to use 'true' as the archive_command, to ensure disk not filling up during failover testing. * The cli currently can list members, failover a master and remove a given cluster from DCS. * The cli can be configured with a command, for repeated access to the same DCS * Added some simple tests for the cli, code coverage is very low	2015-10-26 14:57:59 +01:00
Oleksii Kliukin	2044d4db96	Move nofailover checks on the upper level. Instead of checking that nofailover node should not be marked as healthiest in one of the _failover functions, do make it unhealthy in the is_healthiest_node.	2015-10-26 14:52:34 +01:00
Oleksii Kliukin	84db64e0d5	Merge branch 'master' of https://github.com/zalando/patroni into feature/nofailover	2015-10-26 10:41:51 +01:00
Oleksii Kliukin	a2736cd2b7	Add nofailover property to Patroni and Cluster. Add checks for nofailover tag in is_failover_possible in order to save on API calls for the nodes with this tag set.	2015-10-26 10:40:34 +01:00
Alexander Kukushkin	2e5efc6cc9	Merge pull request #75 from zalando/feature/custom-namespace Make it possible to override default namespace (/service/) from a config file	2015-10-23 16:21:19 +02:00
Alexander Kukushkin	553129a981	Revert "Fix unit-test for Postgresql.controldata()" This reverts commit `deaaf8ad1a`.	2015-10-23 15:59:20 +02:00
Alexander Kukushkin	ad719297c2	Merge branch 'master' of github.com:zalando/patroni into feature/custom-namespace	2015-10-23 15:58:42 +02:00
Oleksii Kliukin	5f9276f5dc	Merge pull request #77 from zalando/bugfix/remove_initialize_key Fix removal of keys on failed initialization. Fixes the recent problem reported at #17	2015-10-23 15:42:48 +02:00
Oleksii Kliukin	0c5a21e57d	Fix removal of keys on failed initialization. The initialize key was checked against the value of the node name before removal, but it was changed recently to contain either an empty string, or cluster sysid. To fix this, the check for the previous value was simply removed: we can guarantee that the code path that removes the key is the one that created it.	2015-10-23 10:46:55 +02:00
Oleksii Kliukin	b7b47ffd79	Add support for the nofailover tag.	2015-10-23 10:11:38 +02:00
Alexander Kukushkin	8e41d14283	Merge branch 'master' of github.com:zalando/patroni into feature/custom-namespace	2015-10-23 09:13:52 +02:00
Feike Steenbergen	5ae6f3a56c	Change Docker registry	2015-10-22 09:30:12 +02:00
Feike Steenbergen	eaf63db886	Use a different namespace in the Docker container. Also bugfix: Patroni should advertise Docker ip as connect address	2015-10-22 09:28:00 +02:00
Feike Steenbergen	857caa1397	Revert casting to string	2015-10-22 09:24:31 +02:00
Feike Steenbergen	c751dfdebf	Typo in joining namespace to scope	2015-10-22 08:50:53 +02:00
Oleksii Kliukin	2200a4ce8c	Add support for per-member tags. Tags are labels assigned to individual members in order to alter its default behavior, i.e. exclude from the leader election or indicate a possibility to create base backups from the member. This commit only adds support for setting tags in the configuration file, exposes the tags to DCS /member subkey and returns the tags in a response of the API request. At the moment the tag names are not validated, nor they are interpreted in any way. Support for setting tags via the API is also in the scope of further work.	2015-10-21 15:51:21 +02:00
Feike Steenbergen	e0e4789b8a	Explicitly cast scope to string. Fixes issue #74	2015-10-21 15:49:20 +02:00
Alexander Kukushkin	deaaf8ad1a	Fix unit-test for Postgresql.controldata()	2015-10-21 15:38:51 +02:00
Alexander Kukushkin	2c7e3f60cc	Make possible to override default namespace (/service/) from a config file If the namespace is not specified in a config file /service/ would be used. Also it's possible to use just '/' as a namespace. It means we would have following structure: /scope1 /scope2 ...	2015-10-21 15:34:55 +02:00
Oleksii Kliukin	c61ac39e8f	Merge pull request #71 from zalando/feature/configurable_pgpass Feature/configurable pgpass	2015-10-21 14:45:02 +02:00
Oleksii Kliukin	c426b8c246	Merge branch 'master' of https://github.com/zalando/patroni into feature/configurable_pgpass	2015-10-21 13:07:42 +02:00
Oleksii Kliukin	9130891029	Move calculation of pgpass to the class constructor: better to fail fast in case of issues.	2015-10-21 13:06:54 +02:00
Oleksii Kliukin	3c3694dc7c	Merge pull request #73 from zalando/bugfix/update-machines-cache Bugfix/update machines cache	2015-10-21 12:11:11 +02:00
Oleksii Kliukin	e2261f531b	Merge pull request #72 from drnic/override-patroni-scope allow $PATRONI_SCOPE to be set via 'docker run -e PATRONI_SCOPE=ironman'	2015-10-21 12:10:46 +02:00
Oleksii Kliukin	2d7909e357	Merge pull request #69 from zalando/feature/clusterid Add clusterid to patroni in order to avoid connections from already existing nodes belonging to another cluster (master - replica system).	2015-10-21 12:09:55 +02:00
Oleksii Kliukin	0b69ef93cf	Merge branch 'master' of https://github.com/zalando/patroni into feature/clusterid	2015-10-21 12:02:55 +02:00
Oleksii Kliukin	44a73982d4	Do not try to fetch the element from the get_node result if the node is not there.	2015-10-21 12:00:03 +02:00
Oleksii Kliukin	b922b8ad0e	Merge pull request #70 from zalando/bugfix/save_configuration_files Add a missing call to restore_configuration_files.	2015-10-21 11:45:18 +02:00
Alexander Kukushkin	c4a6dd48d3	remove debug print statement	2015-10-21 11:09:37 +02:00
Alexander Kukushkin	8bd28507a9	format tests according to the latest pep8 standards	2015-10-21 11:08:06 +02:00
Alexander Kukushkin	0096b6b06f	Schedule update of machines cache when api_execute call has failed Such situation could happen if we replaced all etcd nodes except one which was used by patroni. After replacing the last node patroni will try to execute request on all other nodes from machines_cache but non of them are available. Michines cache would became empty and patroni will stick to the latest node which was available in the machines_cache and will never try to refresh machines_cache from dns for example. Currently machines cache is refreshed only when one request to the etcd cluster has failed, but probably it should be done periodically, for example every minute...	2015-10-21 10:56:43 +02:00
Alexander Kukushkin	71275fc303	Merge pull request #67 from zalando/feature/api Manual failover via REST API	2015-10-21 10:55:28 +02:00
Dr Nic Williams	5d7e4fe90a	allow $PATRONI_SCOPE to be set via 'docker run -e PATRONI_SCOPE=ironman'	2015-10-20 14:32:59 -05:00
Alexander Kukushkin	f53c968d8b	Improve tests	2015-10-20 14:36:49 +02:00
Oleksii Kliukin	35641ac072	Use distinct paths for pgpass from test nodes.	2015-10-20 11:40:52 +02:00
Oleksii Kliukin	92fe6a1de9	Make pgpass location configurable. One can use pgpass configuration parameter in the postgres subsection of Patroni. By default pgpass is written in ~/. Mock actual writes to pgpass in the tests.	2015-10-20 11:28:26 +02:00
Oleksii Kliukin	40c5d5e351	Match default param in the abstract class definition with those from the implementation.	2015-10-19 16:08:52 +02:00
Oleksii Kliukin	90c738d83a	Address the code review by Alex.	2015-10-19 16:03:21 +02:00
Oleksii Kliukin	8f606e4ff9	Add a missing call to restore_configuration_files. I accidentially removed the call when moving the backup functions to the external script. It is intended to save the configuration, so that at the restore phase one can just copy backup files. Its primary intention was to save configuration files in the WAL-E case (WAL-E just omits everything with .conf), but it is also useful in the pg_basebackup case, which omits all symlinks, leaving the cluster with .conf files symlinked in the broken state.	2015-10-19 15:13:24 +02:00
Alexander Kukushkin	18eebdadaa	Watch for change of failover key. If the value is empty and leader didn't changed, this probably means that failover failed. After 15 seconds timeout we will consider failover status = unknown	2015-10-19 15:00:06 +02:00
Oleksii Kliukin	4e448015f3	Increase the test coverage.	2015-10-19 10:13:14 +02:00
Oleksii Kliukin	11073ac644	Merge branch 'master' into feature/clusterid	2015-10-19 09:37:51 +02:00
Alexander Kukushkin	2ebbebac7f	Merge branch 'master' of github.com:zalando/patroni into feature/api	2015-10-19 09:37:11 +02:00
Oleksii Kliukin	a10b7248a6	Fix a flake8 warning	2015-10-19 09:19:25 +02:00
Oleksii Kliukin	83662f71cb	Exit right away if the node sysid is different from the cluster's one	2015-10-16 16:38:05 +02:00
Oleksii Kliukin	a844920489	Store the cluster sysid in the initialize flag. Make sure that the new PostgreSQL node will only join the cluster if its sysid matches the one stored in DCS.	2015-10-16 16:14:45 +02:00
Alexander Kukushkin	57ace2009c	Merge pull request #66 from zalando/fix/tests Fix/tests	2015-10-16 10:30:32 +02:00
Alexander Kukushkin	921e4fc323	psycopg2 should be not older than 2.6.1	2015-10-16 10:28:15 +02:00
Oleksii Kliukin	75e5901360	Merge pull request #63 from zalando/bugfix/leader_crash_readonly Make sure patroni is not using stale connections.	2015-10-16 09:40:55 +02:00
Alexander Kukushkin	3ed82ae22c	Manual failover via rest api curl -XPOST --data '{"leader": "leader_name", "member": "member_name"}' http://127.0.0.1:8008/failover It will execute some preliminary checks and write failover key into DCS. Afterward it will wait until new leader key will appear in a DCS. It's better to execute this request on the master node. It will send a signal to the main HA loop which makes possible to release leader key immidiately even if you are working with etcd.	2015-10-15 16:18:28 +02:00
Alexander Kukushkin	f35d109810	Bugfix: do not try to double encode data	2015-10-15 16:17:11 +02:00
Alexander Kukushkin	16a0a3481d	fix pep8 formatting	2015-10-15 09:08:33 +02:00
Alexander Kukushkin	2f0cf1db06	Mock etcd client delete method	2015-10-15 09:08:16 +02:00
Oleksii Kliukin	5c86b60cd2	Fix an exception in the (rather unusual) case of attaching Patroni to an existing running replica.	2015-10-14 17:05:09 +02:00
Feike Steenbergen	98b59354a9	Exclude more files from git.	2015-10-14 14:37:05 +02:00
Oleksii Kliukin	5747f561c0	Merge pull request #64 from zalando/bugfix/pg_controldata_changes Work around the differences in pg_controldata names.	2015-10-14 12:14:07 +02:00
Oleksii Kliukin	c7246e48d9	Work around the differences in pg_controldata names.	2015-10-14 09:46:20 +02:00
Oleksii Kliukin	953ea749bf	Make sure patroni is not using stale connections. After the PostgreSQL crash (i.e. with kill -9), the backend patroni connects to may still exist. In this case, patroni will get stale postgres role from this backend, preventing a restarted node with a leader lock from being promoted. Easily reproducible and also observed in a staging environment after the postgres crash due to out of disk space.	2015-10-13 15:00:16 +02:00