gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #202 from zalando/feature/dynamic-configuration

Browse Source 
Feature/dynamic configuration
This commit is contained in:
Alexander Kukushkin
2016-06-14 10:10:26 +02:00
committed by GitHub
parent 1893f769a8 f2980b13fb
commit 0a32005cc8
31 changed files with 1757 additions and 573 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.rst
+7 -1
View File
@@ -75,7 +75,13 @@ run:
YAML Configuration
===============
Go `here <https://github.com/zalando/patroni/blob/master/SETTINGS.rst>`__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml <https://github.com/zalando/patroni/blob/master/postgres0.yml>`__.
Go `here <https://github.com/zalando/patroni/blob/master/docs/SETTINGS.rst>`__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml <https://github.com/zalando/patroni/blob/master/postgres0.yml>`__.
=========================
Environment Configuration
=========================
Go `here <https://github.com/zalando/patroni/blob/master/docs/ENVIRONMENT.rst>`__ for comprehensive information about configuring(overriding) settings via environment variables.
===============
Replication Choices
SETTINGS.rst
-77
View File
@@ -1,77 +0,0 @@
===========================
YAML Configuration Settings
===========================
Global/Universal
----------------
- **loop\_wait**: the number of seconds the loop will sleep.
- **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process.
Consul
------
- **host**: the host:port for the Consul endpoint.
- **scope**: the relative path used on Consul's HTTP API for this deployment; makes it possible to run multiple HA deployments from a single Consul cluster.
- **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process.
Etcd
----
- **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 cluster.
- **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process.
Exhibitor
---------
- **hosts**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.
- **poll\_interval**: how often the list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor
- **port**: Exhibitor port.
PostgreSQL
----------
- **admin**:
- **password**: admin password; user is created during initialization.
- **username**: admin username; user is created during initialization. It will have CREATEDB and CREATEROLE privileges.
- **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.)
- **on\_reload**: run this script when configuration reload is triggered.
- **on\_restart**: run this script when the cluster restarts.
- **on\_role\_change**: run this script when the cluster is being promoted or demoted.
- **on\_start**: run this script when the cluster starts.
- **on\_stop**: run this script when the cluster stops.
- **connect\_address**: IP address + port through which Postgres is accessible from other nodes and applications.
- **create\_replica\_methods**: an ordered list of the create methods for turning a Patroni node into a new replica. "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured as its own config item.
- **data\_dir**: file path to initialize and store Postgres data files.
- **initdb**: List options to be passed on to initdb.
- **data-checksums**: Must be enabled when pg_rewind is needed on 9.3.
- **encoding**: default encoding for new databases.
- **locale**: default locale for new databases.
- **listen**: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node.
- **maximum\_lag\_on\_failover**: the maximum bytes a follower may lag.
- **name**: the name of the Postgres host. Must be unique for the cluster.
- **pg\_hba**: list of lines that you should add to pg\_hba.conf.
- **- host all all 0.0.0.0/0 md5**.
- **- host replication replicator 127.0.0.1/32 md5**: A line like this is required for replication.
- **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.
- **replica\_method** for each create_replica_method other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed. Other configuration parameters will be passed along to the script in the form "parameter=value".
- **replication**:
- **username**: replication username; user will be created during initialization.
- **password**: replication password; user will be created during initialization.
- **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.
- **superuser**:
- **password**: password for the Postgres user, set during initialization.
REST API
--------
- **connect\_address**: IP address and port through which restapi is accessible.
- **listen**: IP address and port that Patroni will listen to, to provide health-check information for HAProxy.
- **Optional**:
- **auth**: 'username:password' to protect dangerous REST API endpoints.
- **certfile**: 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**: Specifies a file with the secret key in the PEM format.
ZooKeeper
----------
- **hosts**: list of ZooKeeper cluster members in format: ['host1:port1', 'host2:port2', 'etc...'].
- **reconnect\_timeout**: how long you 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.
- **scope**: the relative path used on ZooKeeper for this deployment. Makes it possible to run multiple HA deployments from a single ZooKeeper cluster.
- **session\_timeout**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process.
docs/ENVIRONMENT.rst
+58
View File
@@ -0,0 +1,58 @@
==================================
Environment Configuration Settings
==================================
It is possible to override some of the configuration parameters defined in the Patroni configuration file using the system environment variables. This document lists all environment variables handled by Patroni. The values set via those variables always take precedence over the ones set in the Patroni configuration file.
Global/Universal
----------------
- **PATRONI\_CONFIGURATION**: it is possible to set the entire configuration for the Patroni via ``PATRONI_CONFIGURATION`` environment variable. In this case any other environment variables will not be considered!
- **PATRONI\_NAME**: name of the node where the current instance of Patroni is running. Must be unique for the cluster.
- **PATRONI\_NAMESPACE**: path within the configuration store where Patroni will keep information about the cluster. Default value: "/service"
- **PATRONI\_SCOPE**: cluster name
Bootstrap configuration
-----------------------
It is possible to create new database users right after the successful initialization of a new cluster. This process is defined by the following variables:
- **PATRONI\_<username>\_PASSWORD='<password>'**
- **PATRONI\_<username>\_OPTIONS='list,of,options'**
Example: defining ``PATRONI_admin_PASSWORD=strongpasswd`` and ``PATRONI_admin_OPTIONS='createrole,createdb'`` will cause creation of the user **admin** with the password **strongpasswd** that is allowed to create other users and databases.
Consul
------
- **PATRONI\_CONSUL\_HOST**: the host:port for the Consul endpoint.
Etcd
----
- **PATRONI\_ETCD\_HOST**: the host:port for the etcd endpoint.
Exhibitor
---------
- **PATRONI\_EXHIBITOR\_HOSTS**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.
- **PATRONI\_EXHIBITOR\_PORT**: Exhibitor port.
PostgreSQL
----------
- **PATRONI\_POSTGRESQL\_LISTEN**: IP address + port that Postgres listens to. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node.
- **PATRONI\_POSTGRESQL\_CONNECT\_ADDRESS**: IP address + port through which Postgres is accessible from other nodes and applications.
- **PATRONI\_POSTGRESQL\_DATA\_DIR**: The location of the Postgres data directory, either existing or to be initialized by Patroni.
- **PATRONI\_POSTGRESQL\_PGPASS**: path to the `.pgpass <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ password file. Patroni creates this file before executing pg\_basebackup and under some other circumstances. The location must be writable by Patroni.
- **PATRONI\_REPLICATION\_USERNAME**: replication username; the user will be created during initialization. Replicas will use this user to access master via streaming replication
- **PATRONI\_REPLICATION\_PASSWORD**: replication password; the user will be created during initialization.
- **PATRONI\_SUPERUSER\_USERNAME**: name for the superuser, set during initialization (initdb) and later used by Patroni to connect to the postgres. Also this user is used by pg_rewind.
- **PATRONI\_SUPERUSER\_PASSWORD**: password for the superuser, set during initialization (initdb).
REST API
--------
- **PATRONI\_RESTAPI\_CONNECT\_ADDRESS**: IP address and port to access the REST API.
- **PATRONI\_RESTAPI\_LISTEN**: IP address and port that Patroni will listen to, to provide health-check information for HAProxy.
- **PATRONI\_RESTAPI\_USERNAME**: Basic-auth username to protect unsafe REST API endpoints.
- **PATRONI\_RESTAPI\_PASSWORD**: Basic-auth password to protect unsafe REST API endpoints.
- **PATRONI\_RESTAPI\_CERTFILE**: Specifies the 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.
- **PATRONI\_RESTAPI\_KEYFILE**: Specifies the file with the secret key in the PEM format.
ZooKeeper
---------
- **PATRONI\_ZOOKEEPER\_HOSTS**: comma separated list of ZooKeeper cluster members: 'host1:port1,host2:port2,etc...'
docs/SETTINGS.rst
+88
View File
@@ -0,0 +1,88 @@
===========================
YAML Configuration Settings
===========================
Global/Universal
----------------
- **name**: the name of the host. Must be unique for the cluster.
- **namespace**: path within the configuration store where Patroni will keep information about the cluster. Default value: "/service"
- **scope**: cluster name
Bootstrap configuration
-----------------------
- **dcs**: This section will be written into `/<namespace>/<scope>/config` of a given configuration store after initializing of new cluster. This is the global configuration for the cluster. If you want to change some parameters for all cluster nodes - just do it in DCS (or via Patroni API) and all nodes will apply this configuration.
- **loop\_wait**: the number of seconds the loop will sleep. Default value: 10
- **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process. Default value: 30
- **maximum\_lag\_on\_failover**: the maximum bytes a follower may lag to be able to participate in leader election.
- **postgresql**:
- **use\_pg\_rewind**:whether or not to use pg_rewind
- **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.
- **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.
- **initdb**: List options to be passed on to initdb.
- **- data-checksums**: Must be enabled when pg_rewind is needed on 9.3.
- **- encoding: UTF8**: default encoding for new databases.
- **- locale: UTF8**: default locale for new databases.
- **pg\_hba**: list of lines that you should add to pg\_hba.conf.
- **- host all all 0.0.0.0/0 md5**.
- **- host replication replicator 127.0.0.1/32 md5**: A line like this is required for replication.
- **users**: Some additional users users which needs to be created after initializing new cluster
- **admin**: the name of user
- **password: zalando**:
- **options**: list of options for CREATE USER statement
- **- createrole**
- **- createdb**
Consul
------
- **host**: the host:port for the Consul endpoint.
Etcd
----
- **host**: the host:port for the etcd endpoint.
Exhibitor
---------
- **hosts**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.
- **poll\_interval**: how often the list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor
- **port**: Exhibitor port.
PostgreSQL
----------
- **authentication**:
- **superuser**:
- **username**: name for the superuser, set during initialization (initdb) and later used by Patroni to connect to the postgres.
- **password**: password for the superuser, set during initialization (initdb).
- **replication**:
- **username**: replication username; the user will be created during initialization. Replicas will use this user to access master via streaming replication
- **password**: replication password; the user will be created during initialization.
- **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.)
- **on\_reload**: run this script when configuration reload is triggered.
- **on\_restart**: run this script when the cluster restarts.
- **on\_role\_change**: run this script when the cluster is being promoted or demoted.
- **on\_start**: run this script when the cluster starts.
- **on\_stop**: run this script when the cluster stops.
- **connect\_address**: IP address + port through which Postgres is accessible from other nodes and applications.
- **create\_replica\_methods**: an ordered list of the create methods for turning a Patroni node into a new replica. "basebackup" is the default method; other methods are assumed to refer to scripts, each of which is configured as its own config item.
- **data\_dir**: The location of the Postgres data directory, either existing or to be initialized by Patroni.
- **listen**: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node.
- **pgpass**: path to the `.pgpass <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ password file. Patroni creates this file before executing pg\_basebackup and under some other circumstances. The location must be writable by Patroni.
- **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.
- **replica\_method** for each create_replica_method other than basebackup, you would add a configuration section of the same name. At a minimum, this should include "command" with a full path to the actual script to be executed. Other configuration parameters will be passed along to the script in the form "parameter=value".
REST API
--------
- **connect\_address**: IP address and port to access the REST API.
- **listen**: IP address and port that Patroni will listen to, to provide health-check information for HAProxy.
- **Optional**:
- **authentication**:
- **username**: Basic-auth username to protect unsafe REST API endpoints.
- **password**: Basic-auth password to protect unsafe REST API endpoints.
- **certfile**: Specifies the 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**: Specifies the file with the secret key in the PEM format.
ZooKeeper
----------
- **hosts**: list of ZooKeeper cluster members in format: ['host1:port1', 'host2:port2', 'etc...'].
features/environment.py
+46 -18
View File
@@ -98,6 +98,13 @@ class PatroniController(AbstractController):
except IOError:
return None
def add_tag_to_config(self, tag, value):
with open(self._config) as r:
config = yaml.safe_load(r)
config['tags']['tag'] = value
with open(self._config, 'w') as w:
yaml.safe_dump(config, w, default_flow_style=False)
def _start(self):
return subprocess.Popen(['coverage', 'run', '--source=patroni', '-p', 'patroni.py', self._config],
stdout=self._log, stderr=subprocess.STDOUT, cwd=self._work_directory)
@@ -110,21 +117,25 @@ class PatroniController(AbstractController):
patroni_config_path = os.path.join(self._output_dir, patroni_config_name)
with open(patroni_config_name) as f:
config = yaml.load(f)
config = yaml.safe_load(f)
host = config['postgresql']['listen'].split(':')[0]
config['postgresql']['listen'] = config['postgresql']['connect_address'] = '{0}:{1}'.format(host, self.__PORT)
user = config['postgresql'].get('superuser', {})
user = config['postgresql'].get('authentication', config['postgresql']).get('superuser', {})
self._connkwargs = {k: user[n] for n, k in [('username', 'user'), ('password', 'password')] if n in user}
self._connkwargs.update({'host': host, 'port': self.__PORT, 'database': 'postgres'})
config['postgresql'].update({'name': name, 'data_dir': self._data_dir})
config['name'] = name
config['postgresql']['data_dir'] = self._data_dir
config['postgresql']['parameters'].update({
'logging_collector': 'on', 'log_destination': 'csvlog', 'log_directory': self._output_dir,
'log_filename': name + '.log', 'log_statement': 'all', 'log_min_messages': 'debug1'})
if 'bootstrap' in config and 'initdb' in config['bootstrap']:
config['bootstrap']['initdb'].extend([{'auth': 'md5'}, {'auth-host': 'md5'}])
if tags:
config['tags'] = tags
@@ -132,16 +143,14 @@ class PatroniController(AbstractController):
dcs_config = config.pop('etcd')
dcs_config.pop('host')
if dcs != 'consul':
dcs_config.update({'session_timeout': dcs_config.pop('ttl'), 'reconnect_timeout': config['loop_wait']})
if dcs == 'exhibitor':
dcs_config.update({'hosts': ['127.0.0.1'], 'port': 8181})
else:
dcs_config['hosts'] = ['127.0.0.1:2181']
if dcs == 'exhibitor':
dcs_config.update({'hosts': ['127.0.0.1'], 'port': 8181})
elif dcs == 'zookeeper':
dcs_config['hosts'] = ['127.0.0.1:2181']
config[dcs] = dcs_config
with open(patroni_config_path, 'w') as f:
yaml.dump(config, f, default_flow_style=False)
yaml.safe_dump(config, f, default_flow_style=False)
return patroni_config_path
@@ -180,7 +189,7 @@ class PatroniController(AbstractController):
class AbstractDcsController(AbstractController):
_CLUSTER_NODE = 'service/batman'
_CLUSTER_NODE = '/service/batman'
def _is_accessible(self):
return self._is_running()
@@ -191,10 +200,17 @@ class AbstractDcsController(AbstractController):
if self._work_directory:
shutil.rmtree(self._work_directory)
def path(self, key=None):
return self._CLUSTER_NODE + (key and '/' + key or '')
@abc.abstractmethod
def query(self, key):
""" query for a value of a given key """
@abc.abstractmethod
def set(self, key, value):
""" set a value to a given key """
@abc.abstractmethod
def cleanup_service_tree(self):
""" clean all contents stored in the tree used for the tests """
@@ -216,12 +232,18 @@ class ConsulController(AbstractDcsController):
except Exception:
return False
def path(self, key=None):
return super(ConsulController, self).path(key)[1:]
def query(self, key):
_, value = self._client.kv.get('{0}/{1}'.format(self._CLUSTER_NODE, key))
_, value = self._client.kv.get(self.path(key))
return value and value['Value'].decode('utf-8')
def set(self, key, value):
self._client.kv.put(self.path(key), value)
def cleanup_service_tree(self):
self._client.kv.delete(self._CLUSTER_NODE, recurse=True)
self._client.kv.delete(self.path(), recurse=True)
class EtcdController(AbstractDcsController):
@@ -238,13 +260,16 @@ class EtcdController(AbstractDcsController):
def query(self, key):
try:
return self._client.get('/{0}/{1}'.format(self._CLUSTER_NODE, key)).value
return self._client.get(self.path(key)).value
except etcd.EtcdKeyNotFound:
return None
def set(self, key, value):
self._client.set(self.path(key), value)
def cleanup_service_tree(self):
try:
self._client.delete('/' + self._CLUSTER_NODE, recursive=True)
self._client.delete(self.path(), recursive=True)
except (etcd.EtcdKeyNotFound, etcd.EtcdConnectionFailed):
return
except Exception as e:
@@ -271,13 +296,16 @@ class ZooKeeperController(AbstractDcsController):
def query(self, key):
try:
return self._client.get('/{0}/{1}'.format(self._CLUSTER_NODE, key))[0].decode('utf-8')
return self._client.get(self.path(key))[0].decode('utf-8')
except kazoo.exceptions.NoNodeError:
return None
def set(self, key, value):
self._client.set(self.path(key), value.encode('utf-8'))
def cleanup_service_tree(self):
try:
self._client.delete('/' + self._CLUSTER_NODE, recursive=True)
self._client.delete(self.path(), recursive=True)
except (kazoo.exceptions.NoNodeError):
return
except Exception as e:
@@ -326,7 +354,7 @@ class PatroniPoolController(object):
self._processes[pg_name].start(max_wait_limit)
def __getattr__(self, func):
if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to']:
if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to', 'add_tag_to_config']:
raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func))
def wrapper(pg_name, *args, **kwargs):
features/patroni_api.feature
+26 -5
View File
@@ -13,13 +13,35 @@ Scenario: check API requests on a stand-alone server
When I issue an empty POST request to http://127.0.0.1:8008/reinitialize
Then I receive a response code 503
And I receive a response text "I am the leader, can not reinitialize"
When I issue a POST request to http://127.0.0.1:8008/failover with leader=postgres0
When I issue a POST request to http://127.0.0.1:8008/failover with {"leader": "postgres0"}
Then I receive a response code 500
And I receive a response text "failover is not possible: cluster does not have members except leader"
And I receive a response text failover is not possible: cluster does not have members except leader
When I issue an empty POST request to http://127.0.0.1:8008/failover
Then I receive a response code 400
When I issue a POST request to http://127.0.0.1:8008/failover with {"foo": "bar"}
Then I receive a response code 400
And I receive a response text "No values given for required parameters leader and candidate"
Scenario: check local configuration reload
Given I issue an empty POST request to http://127.0.0.1:8008/reload
Then I receive a response code 200
And I receive a response text nothing changed
When I add tag new_tag new_value to postgres0 config
And I issue an empty POST request to http://127.0.0.1:8008/reload
Then I receive a response code 202
Scenario: check dynamic configuration change via DCS
Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "loop_wait": 1, "postgresql": {"parameters": {"max_connections": 101}}}
Then I receive a response code 200
And I receive a response loop_wait 1
And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds
When I issue a GET request to http://127.0.0.1:8008/config
Then I receive a response code 200
And I receive a response loop_wait 1
When I issue a GET request to http://127.0.0.1:8008/patroni
Then I receive a response code 200
And I receive a response tags {'tag': 'new_value'}
Scenario: check API requests for the primary-replica pair
Given I start postgres1
And replication works from postgres0 to postgres1 after 20 seconds
@@ -36,7 +58,7 @@ Scenario: check API requests for the primary-replica pair
Then postgres1 role is the secondary after 15 seconds
Scenario: check the failover via the API
Given I issue a POST request to http://127.0.0.1:8008/failover with leader=postgres0,candidate=postgres1
Given I issue a POST request to http://127.0.0.1:8008/failover with {"leader": "postgres0", "candidate": "postgres1"}
Then I receive a response code 200
And postgres1 is a leader after 5 seconds
And postgres1 role is the primary after 5 seconds
@@ -45,9 +67,8 @@ Scenario: check the failover via the API
Scenario: check the scheduled failover
Given I issue a scheduled failover at http://127.0.0.1:8009 from postgres1 to postgres0 in 1 seconds
Then I receive a response code 200
Then I receive a response code 202
And postgres0 is a leader after 20 seconds
And postgres0 role is the primary after 5 seconds
And postgres1 role is the secondary after 10 seconds
And replication works from postgres0 to postgres1 after 25 seconds
features/steps/patroni_api.py
+49 -29
View File
@@ -1,7 +1,9 @@
import json
import parse
import pytz
import requests
import time
import yaml
from behave import register_type, step, then
from datetime import datetime, timedelta
@@ -12,12 +14,7 @@ def parse_url(text):
return text
@parse.with_pattern(r'(?:\w+=(?:\w|\.|:|-|\+|\s)+,?)+')
def parse_data(text):
return text
register_type(url=parse_url, data=parse_data)
register_type(url=parse_url)
# there is no way we can find out if the node has already
@@ -39,6 +36,23 @@ def sleep_for_n_seconds(context, value):
time.sleep(int(value))
def _set_response(context, response):
context.status_code = response.status_code
data = response.content.decode('utf-8')
ct = response.headers.get('content-type', '')
if ct.startswith('application/json') or\
ct.startswith('text/yaml') or\
ct.startswith('text/x-yaml') or\
ct.startswith('application/yaml') or\
ct.startswith('application/x-yaml'):
try:
context.response = yaml.safe_load(data)
except ValueError:
context.response = data
else:
context.response = data
@step('I issue a GET request to {url:url}')
def do_get(context, url):
try:
@@ -47,38 +61,27 @@ def do_get(context, url):
context.status_code = None
context.response = None
else:
context.status_code = r.status_code
try:
context.response = r.json()
except ValueError:
context.response = r.content.decode('utf-8')
_set_response(context, r)
@step('I issue an empty POST request to {url:url}')
def do_post_empty(context, url):
do_post(context, url, None)
do_request(context, 'POST', url, None)
@step('I issue a POST request to {url:url} with {data:data}')
def do_post(context, url, data):
post_data = {}
if data:
post_components = data.split(',')
for pc in post_components:
if '=' in pc:
k, v = pc.split('=', 2)
post_data[k.strip()] = v.strip()
@step('I issue a {request_method:w} request to {url:url} with {data}')
def do_request(context, request_method, url, data):
data = data and json.loads(data) or {}
try:
r = requests.post(url, json=post_data)
if request_method == 'PATCH':
r = requests.patch(url, json=data)
else:
r = requests.post(url, json=data)
except requests.exceptions.RequestException:
context.status_code = None
context.response = None
else:
context.status_code = r.status_code
try:
context.response = r.json()
except ValueError:
context.response = r.content.decode('utf-8')
_set_response(context, r)
@then('I receive a response {component:w} {data}')
@@ -90,11 +93,28 @@ def check_response(context, component, data):
assert context.response == data.strip('"'), "response {0} does not contain {1}".format(context.response, data)
else:
assert component in context.response, "{0} is not part of the response".format(component)
assert context.response[component] == data, "{0} does not contain {1}".format(component, data)
assert str(context.response[component]) == str(data), "{0} does not contain {1}".format(component, data)
@step('I issue a scheduled failover at {at_url:url} from {from_host:w} to {to_host:w} in {in_seconds:d} seconds')
def scheduled_failover(context, at_url, from_host, to_host, in_seconds):
context.execute_steps(u"""
Given I issue a POST request to {0}/failover with leader={1},candidate={2},scheduled_at={3}
Given I issue a POST request to {0}/failover with {{"leader": "{1}", "candidate": "{2}", "scheduled_at": "{3}"}}
""".format(at_url, from_host, to_host, datetime.now(pytz.utc) + timedelta(seconds=int(in_seconds))))
@step('I add tag {tag:w} {value:w} to {pg_name:w} config')
def add_tag_to_config(context, tag, value, pg_name):
context.pctl.add_tag_to_config(pg_name, tag, value)
@then('Response on GET {url} contains {value} after {timeout:d} seconds')
def check_http_response(context, url, value, timeout):
for _ in range(int(timeout)):
r = requests.get(url)
if value in r.content.decode('utf-8'):
break
time.sleep(1)
else:
assert False,\
"Value {0} is not present in response after {1} seconds".format(value, timeout)
patroni/__init__.py
+61 -30
View File
@@ -1,11 +1,11 @@
import logging
import os
import sys
import signal
import time
import yaml
from patroni.api import RestApiServer
from patroni.config import Config
from patroni.dcs import get_dcs
from patroni.exceptions import DCSError
from patroni.ha import Ha
from patroni.postgresql import Postgresql
from patroni.utils import reap_children, set_ignore_sigterm, setup_signal_handlers
@@ -15,19 +15,54 @@ logger = logging.getLogger(__name__)
class Patroni(object):
PATRONI_CONFIG_VARIABLE = 'PATRONI_CONFIGURATION'
def __init__(self, config):
self.nap_time = config['loop_wait']
self.tags = {tag: value for tag, value in config.get('tags', {}).items()
if tag not in ('clonefrom', 'nofailover', 'noloadbalance') or value}
self.postgresql = Postgresql(config['postgresql'])
self.dcs = get_dcs(self.postgresql.name, config)
def __init__(self):
self.version = __version__
self.api = RestApiServer(self, config['restapi'])
self.config = Config()
self.dcs = get_dcs(self.config)
self.load_dynamic_configuration()
self.postgresql = Postgresql(self.config['postgresql'])
self.api = RestApiServer(self, self.config['restapi'])
self.ha = Ha(self)
self.tags = self.get_tags()
self.nap_time = self.config['loop_wait']
self.next_run = time.time()
self._reload_config_scheduled = False
self._received_sighup = False
def load_dynamic_configuration(self):
while True:
try:
cluster = self.dcs.get_cluster()
if cluster and cluster.config:
self.config.set_dynamic_configuration(cluster.config)
elif not self.config.dynamic_configuration and 'bootstrap' in self.config:
self.config.set_dynamic_configuration(self.config['bootstrap']['dcs'])
break
except DCSError:
logger.warning('Can not get cluster from dcs')
def get_tags(self):
return {tag: value for tag, value in self.config.get('tags', {}).items()
if tag not in ('clonefrom', 'nofailover', 'noloadbalance') or value}
def reload_config(self):
try:
self.tags = self.get_tags()
self.nap_time = self.config['loop_wait']
self.dcs.set_ttl(self.config.get('ttl') or 30)
self.dcs.set_retry_timeout(self.config.get('retry_timeout') or self.nap_time)
self.api.reload_config(self.config['restapi'])
self.postgresql.reload_config(self.config['postgresql'])
except Exception:
logger.exception('Failed to reload config_file=%s', self.config.config_file)
def sighup_handler(self, *args):
self._received_sighup = True
@property
def noloadbalance(self):
return self.tags.get('noloadbalance', False)
@@ -51,10 +86,24 @@ class Patroni(object):
def run(self):
self.api.start()
signal.signal(signal.SIGHUP, self.sighup_handler)
self.next_run = time.time()
while True:
if self._received_sighup:
self._received_sighup = False
if self.config.reload_local_configuration():
self.reload_config()
logger.info(self.ha.run_cycle())
cluster = self.dcs.cluster
if cluster and cluster.config and self.config.set_dynamic_configuration(cluster.config):
self.reload_config()
if not self.postgresql.data_directory_empty():
self.config.save_cache()
reap_children()
self.schedule_next_run()
@@ -64,25 +113,7 @@ def main():
logging.getLogger('requests').setLevel(logging.WARNING)
setup_signal_handlers()
# Patroni reads the configuration from the command-line argument if it exists, and from the environment otherwise.
use_env = False
use_file = (len(sys.argv) >= 2 and os.path.isfile(sys.argv[1]))
if not use_file:
config_env = os.environ.get(Patroni.PATRONI_CONFIG_VARIABLE)
use_env = config_env is not None
if not use_env:
print('Usage: {0} config.yml'.format(sys.argv[0]))
print('\tPatroni may also read the configuration from the {} environment variable'.
format(Patroni.PATRONI_CONFIG_VARIABLE))
return
if use_file:
with open(sys.argv[1], 'r') as f:
config = yaml.load(f)
elif use_env:
config = yaml.load(config_env)
patroni = Patroni(config)
patroni = Patroni()
try:
patroni.run()
except KeyboardInterrupt:
patroni/api.py
+134 -68
View File
@@ -3,14 +3,13 @@ import fcntl
import json
import logging
import psycopg2
import socket
import time
import dateutil
import datetime
import pytz
from patroni.exceptions import PostgresConnectionException
from patroni.utils import Retry, RetryFailedError
from patroni.utils import deep_compare, patch_config, Retry, RetryFailedError
from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from six.moves.socketserver import ThreadingMixIn
from threading import Thread
@@ -34,48 +33,39 @@ def check_auth(func):
class RestApiHandler(BaseHTTPRequestHandler):
def _write_response(self, status_code, body, headers=None):
def _write_response(self, status_code, body, content_type='text/html', headers=None):
self.send_response(status_code)
if body is not None:
headers = headers or {}
if 'Content-Type' not in headers:
headers['Content-Type'] = 'text/html'
for name, value in (headers or {}).items():
self.send_header(name, value)
self.end_headers()
self.wfile.write(body.encode('utf-8'))
headers = headers or {}
if content_type:
headers['Content-Type'] = content_type
for name, value in headers.items():
self.send_header(name, value)
self.end_headers()
self.wfile.write(body.encode('utf-8'))
def _write_json_response(self, status_code, response):
self._write_response(status_code, json.dumps(response), content_type='application/json')
def send_auth_request(self, body):
headers = {'WWW-Authenticate': 'Basic realm="' + self.server.patroni.__class__.__name__ + '"'}
self._write_response(401, body, headers)
def finish(self):
try:
if not self.wfile.closed:
self.wfile.flush()
self.wfile.close()
except socket.error:
pass
self.rfile.close()
self._write_response(401, body, headers=headers)
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 _write_status_response(self, status_code, response, options=False):
if options:
body = None
else:
patroni = self.server.patroni
response.update({'tags': patroni.tags} if patroni.tags else {})
if patroni.postgresql.sysid:
response['database_system_identifier'] = patroni.postgresql.sysid
response['patroni'] = {'version': patroni.version, 'scope': patroni.postgresql.scope}
body = json.dumps(response)
self._write_response(status_code, body, {'Content-Type': 'application/json'})
def _write_status_response(self, status_code, response):
patroni = self.server.patroni
response.update({'tags': patroni.tags} if patroni.tags else {})
if patroni.postgresql.sysid:
response['database_system_identifier'] = patroni.postgresql.sysid
if patroni.postgresql.pending_restart:
response['pending_restart'] = True
response['patroni'] = {'version': patroni.version, 'scope': patroni.postgresql.scope}
self._write_json_response(status_code, response)
def do_GET(self, options=False):
def do_GET(self, write_status_code_only=False):
"""Default method for processing all GET requests which can not be routed to other methods"""
path = '/master' if self.path == '/' else self.path
@@ -101,15 +91,77 @@ class RestApiHandler(BaseHTTPRequestHandler):
status_code = 200
else:
status_code = 503
self._write_status_response(status_code, response, options)
if write_status_code_only: # when haproxy sends OPTIONS request it reads only status code and nothing more
message = self.responses[status_code][0]
self.wfile.write('{0} {1} {2}\r\n'.format(self.protocol_version, status_code, message).encode('utf-8'))
else:
self._write_status_response(status_code, response)
def do_OPTIONS(self):
self.do_GET(options=True)
self.do_GET(write_status_code_only=True)
def do_GET_patroni(self):
response = self.get_postgresql_status(True)
self._write_status_response(200, response)
def do_GET_config(self):
cluster = self.server.patroni.ha.dcs.cluster or self.server.patroni.ha.dcs.get_cluster()
if cluster.config:
self._write_json_response(200, cluster.config.data)
else:
self.send_error(502)
def _read_json_content(self):
if 'content-length' not in self.headers:
return self.send_error(411)
try:
content_length = int(self.headers.get('content-length'))
request = json.loads(self.rfile.read(content_length).decode('utf-8'))
if isinstance(request, dict) and request:
return request
except Exception:
logger.exception('Bad request')
self.send_error(400)
@check_auth
def do_PATCH_config(self):
request = self._read_json_content()
if request:
cluster = self.server.patroni.ha.dcs.get_cluster()
data = cluster.config.data.copy()
if patch_config(data, request):
value = json.dumps(data, separators=(',', ':'))
if not self.server.patroni.ha.dcs.set_config_value(value, cluster.config.index):
return self.send_error(409)
self._write_json_response(200, data)
@check_auth
def do_PUT_config(self):
request = self._read_json_content()
if request:
cluster = self.server.patroni.ha.dcs.get_cluster()
if not deep_compare(request, cluster.config.data):
value = json.dumps(request, separators=(',', ':'))
if not self.server.patroni.ha.dcs.set_config_value(value):
return self.send_error(502)
self._write_json_response(200, request)
@check_auth
def do_POST_reload(self):
try:
if self.server.patroni.config.reload_local_configuration(True):
status_code = 202
response = 'reload scheduled'
self.server.patroni.sighup_handler()
else:
status_code = 200
response = 'nothing changed'
except Exception as e:
status_code = 500
response = str(e)
self._write_response(status_code, response)
@check_auth
def do_POST_restart(self):
status_code = 500
@@ -142,7 +194,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
self._write_response(status_code, data)
def poll_failover_result(self, leader, candidate):
for _ in range(0, 15):
timeout = 10 if self.server.patroni.nap_time < 10 else self.server.patroni.nap_time
for _ in range(0, timeout*2):
time.sleep(1)
try:
cluster = self.server.patroni.dcs.get_cluster()
@@ -175,11 +228,10 @@ class RestApiHandler(BaseHTTPRequestHandler):
@check_auth
def do_POST_failover(self):
content_length = int(self.headers.get('content-length', 0))
try:
request = json.loads(self.rfile.read(content_length).decode('utf-8'))
except ValueError:
request = {}
request = self._read_json_content()
if not request:
return
leader = request.get('leader')
candidate = request.get('candidate') or request.get('member')
scheduled_at = request.get('scheduled_at')
@@ -203,7 +255,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
elif self.server.patroni.dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at):
self.server.patroni.dcs.event.set()
data = 'Failover scheduled'
status_code = 200
status_code = 202
else:
data = 'failed to write failover key into DCS'
status_code = 503
@@ -243,12 +295,6 @@ 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)
@@ -294,25 +340,9 @@ class RestApiHandler(BaseHTTPRequestHandler):
class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
def __init__(self, patroni, config):
self._auth_key = base64.b64encode(config['auth'].encode('utf-8')).decode('utf-8') if 'auth' in config else None
host, port = config['listen'].split(':')
HTTPServer.__init__(self, (host, int(port)), RestApiHandler)
Thread.__init__(self, target=self.serve_forever)
self._set_fd_cloexec(self.socket)
protocol = 'http'
# wrap socket with ssl if 'certfile' is defined in a config.yaml
# Sometime it's also needed to pass reference to a 'keyfile'.
options = {option: config[option] for option in ['certfile', 'keyfile'] if option in config}
if options.get('certfile'):
import ssl
self.socket = ssl.wrap_socket(self.socket, server_side=True, **options)
protocol = 'https'
self.connection_string = '{0}://{1}/patroni'.format(protocol, config.get('connect_address', config['listen']))
self.patroni = patroni
self.__initialize(config)
self.__set_config_parameters(config)
self.daemon = True
def query(self, sql, *params):
@@ -332,11 +362,47 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
fcntl.fcntl(fd, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)
def check_basic_auth_key(self, key):
return self._auth_key == key
return self.__auth_key == key
def check_auth_header(self, auth_header):
if self._auth_key:
if self.__auth_key:
if auth_header is None:
return 'no auth header received'
if not auth_header.startswith('Basic ') or not self.check_basic_auth_key(auth_header[6:]):
return 'not authenticated'
@staticmethod
def __get_ssl_options(config):
return {option: config[option] for option in ['certfile', 'keyfile'] if option in config}
def __set_connection_string(self, connect_address):
self.connection_string = '{0}://{1}/patroni'.format(self.__protocol, connect_address or self.__listen)
def __set_config_parameters(self, config):
self.__auth_key = base64.b64encode(config['auth'].encode('utf-8')).decode('utf-8') if 'auth' in config else None
self.__set_connection_string(config.get('connect_address'))
def __initialize(self, config):
self.__ssl_options = self.__get_ssl_options(config)
self.__listen = config['listen']
host, port = config['listen'].split(':')
HTTPServer.__init__(self, (host, int(port)), RestApiHandler)
Thread.__init__(self, target=self.serve_forever)
self._set_fd_cloexec(self.socket)
self.__protocol = 'http'
# wrap socket with ssl if 'certfile' is defined in a config.yaml
# Sometime it's also needed to pass reference to a 'keyfile'.
if self.__ssl_options.get('certfile'):
import ssl
self.socket = ssl.wrap_socket(self.socket, server_side=True, **self.__ssl_options)
self.__protocol = 'https'
self.__set_connection_string(config.get('connect_address'))
def reload_config(self, config):
self.__set_config_parameters(config)
if self.__listen != config['listen'] or self.__ssl_options != self.__get_ssl_options(config):
self.shutdown()
self.__initialize(config)
self.start()
patroni/config.py
+308
View File
@@ -0,0 +1,308 @@
import json
import logging
import os
import sys
import tempfile
import yaml
from collections import defaultdict
from copy import deepcopy
from patroni.dcs import ClusterConfig
from patroni.postgresql import Postgresql
from patroni.utils import deep_compare, parse_int, patch_config
logger = logging.getLogger(__name__)
class Config(object):
"""
This class is responsible for:
1) Building and giving access to `effective_configuration` from:
* `Config.__DEFAULT_CONFIG` -- some sane default values
* `dynamic_configuration` -- configuration stored in DCS
* `local_configuration` -- configuration from `config.yml` or environment
2) Saving and loading `dynamic_configuration` into 'patroni.dynamic.json' file
located in local_configuration['postgresql']['data_dir'] directory.
This is necessary to be able to restore `dynamic_configuration`
if DCS was accidentally wiped
3) Loading of configuration file in the old format and converting it into new format
4) Mimicking some of the `dict` interfaces to make it possible
to work with it as with the old `config` object.
"""
PATRONI_ENV_PREFIX = 'PATRONI_'
PATRONI_CONFIG_VARIABLE = PATRONI_ENV_PREFIX + 'CONFIGURATION'
__CACHE_FILENAME = 'patroni.dynamic.json'
__DEFAULT_CONFIG = {
'ttl': 30, 'loop_wait': 10, 'retry_timeout': 10,
'maximum_lag_on_failover': 1048576,
'postgresql': {
'use_slots': True,
'parameters': {p: v[0] for p, v in Postgresql.CMDLINE_OPTIONS.items()}
}
}
def __init__(self):
self._modify_index = -1
self._dynamic_configuration = {}
self.__environment_configuration = self._build_environment_configuration()
# Patroni reads the configuration from the command-line argument if it exists, otherwise from the environment
self._config_file = len(sys.argv) >= 2 and os.path.isfile(sys.argv[1]) and sys.argv[1]
if self._config_file:
self._local_configuration = self._load_config_file()
else:
config_env = os.environ.pop(self.PATRONI_CONFIG_VARIABLE, None)
self._local_configuration = config_env and yaml.safe_load(config_env) or self.__environment_configuration
if not self._local_configuration:
print('Usage: {0} config.yml'.format(sys.argv[0]))
print('\tPatroni may also read the configuration from the {0} environment variable'.
format(self.PATRONI_CONFIG_VARIABLE))
exit(1)
self.__effective_configuration = self._build_effective_configuration(self._dynamic_configuration,
self._local_configuration)
self._data_dir = self.__effective_configuration['postgresql']['data_dir']
self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME)
self._load_cache()
self._cache_needs_saving = False
@property
def config_file(self):
return self._config_file
@property
def dynamic_configuration(self):
return deepcopy(self._dynamic_configuration)
def _load_config_file(self):
"""Loads config.yaml from filesystem and applies some values which were set via ENV"""
with open(self._config_file) as f:
config = yaml.safe_load(f)
patch_config(config, self.__environment_configuration)
return config
def _load_cache(self):
if os.path.isfile(self._cache_file):
try:
with open(self._cache_file) as f:
self.set_dynamic_configuration(json.load(f))
except Exception:
logger.exception('Exception when loading file: %s', self._cache_file)
def save_cache(self):
if self._cache_needs_saving:
tmpfile = fd = None
try:
(fd, tmpfile) = tempfile.mkstemp(prefix=self.__CACHE_FILENAME, dir=self._data_dir)
with os.fdopen(fd, 'w') as f:
fd = None
json.dump(self.dynamic_configuration, f)
tmpfile = os.rename(tmpfile, self._cache_file)
self._cache_needs_saving = False
except Exception:
logger.exception('Exception when saving file: %s', self._cache_file)
if fd:
try:
os.close(fd)
except Exception:
logger.error('Can not close temporary file %s', tmpfile)
if tmpfile and os.path.exists(tmpfile):
try:
os.remove(tmpfile)
except Exception:
logger.error('Can not remove temporary file %s', tmpfile)
# configuration could be either ClusterConfig or dict
def set_dynamic_configuration(self, configuration):
if isinstance(configuration, ClusterConfig):
if self._modify_index == configuration.modify_index:
return False # If the index didn't changed there is nothing to do
self._modify_index = configuration.modify_index
configuration = configuration.data
if not deep_compare(self._dynamic_configuration, configuration):
try:
self.__effective_configuration = self._build_effective_configuration(configuration,
self._local_configuration)
self._dynamic_configuration = configuration
self._cache_needs_saving = True
return True
except Exception:
logger.exception('Exception when setting dynamic_configuration')
def reload_local_configuration(self, dry_run=False):
if self.config_file:
try:
configuration = self._load_config_file()
if not deep_compare(self._local_configuration, configuration):
new_configuration = self._build_effective_configuration(self._dynamic_configuration, configuration)
if dry_run:
return not deep_compare(new_configuration, self.__effective_configuration)
self._local_configuration = configuration
self.__effective_configuration = new_configuration
return True
except Exception:
logger.exception('Exception when reloading local configuration from %s', self.config_file)
if dry_run:
raise
@staticmethod
def _process_postgresql_parameters(parameters, is_local=False):
ret = {}
for name, value in (parameters or {}).items():
if name not in Postgresql.CMDLINE_OPTIONS or not is_local and Postgresql.CMDLINE_OPTIONS[name][1](value):
ret[name] = value
return ret
def _safe_copy_dynamic_configuration(self, dynamic_configuration):
config = deepcopy(self.__DEFAULT_CONFIG)
for name, value in dynamic_configuration.items():
if name == 'postgresql':
for name, value in (value or {}).items():
if name == 'parameters':
config['postgresql'][name].update(self._process_postgresql_parameters(value))
elif name not in ('connect_address', 'listen', 'data_dir', 'pgpass', 'authentication'):
config['postgresql'][name] = deepcopy(value)
elif name in config: # only variables present in __DEFAULT_CONFIG allowed to be overriden from DCS
config[name] = int(value)
return config
@staticmethod
def _build_environment_configuration():
ret = defaultdict(dict)
def _popenv(name):
return os.environ.pop(Config.PATRONI_ENV_PREFIX + name.upper(), None)
for param in ('name', 'namespace', 'scope'):
value = _popenv(param)
if value:
ret[param] = value
def _set_section_values(section, params):
for param in params:
value = _popenv(section + '_' + param)
if value:
ret[section][param] = value
_set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile'])
_set_section_values('postgresql', ['listen', 'connect_address', 'data_dir', 'pgpass'])
def _get_auth(name):
ret = {}
for param in ('username', 'password'):
value = _popenv(name + '_' + param)
if value:
ret[param] = value
return len(ret) == 2 and ret or None
restapi_auth = _get_auth('restapi')
if restapi_auth:
ret['restapi']['authentication'] = restapi_auth
authentication = {}
for user_type in ('replication', 'superuser'):
entry = _get_auth(user_type)
if entry:
authentication[user_type] = entry
if authentication:
ret['postgresql']['authentication'] = authentication
users = {}
def _parse_list(value):
if not (value.strip().startswith('-') or '[' in value):
value = '[{0}]'.format(value)
try:
return yaml.safe_load(value)
except Exception:
return None
for param in list(os.environ.keys()):
if param.startswith(Config.PATRONI_ENV_PREFIX):
name, suffix = (param[8:].rsplit('_', 1) + [''])[:2]
if name and suffix:
# PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT)
if suffix in ('HOST', 'HOSTS', 'PORT') and '_' not in name:
value = os.environ.pop(param)
if suffix == 'PORT':
value = value and parse_int(value)
elif suffix == 'HOSTS':
value = value and _parse_list(value)
if value:
ret[name.lower()][suffix.lower()] = value
# PATRONI_<username>_PASSWORD=<password>, PATRONI_<username>_OPTIONS=<option1,option2,...>
# CREATE USER "<username>" WITH <OPTIONS> PASSWORD '<password>'
elif suffix == 'PASSWORD':
password = os.environ.pop(param)
if password:
users[name] = {'password': password}
options = os.environ.pop(param[:-9] + '_OPTIONS', None)
options = options and _parse_list(options)
if options:
users[name]['options'] = options
if users:
ret['bootstrap']['users'] = users
return ret
def _build_effective_configuration(self, dynamic_configuration, local_configuration):
config = self._safe_copy_dynamic_configuration(dynamic_configuration)
for name, value in local_configuration.items():
if name == 'postgresql':
for name, value in (value or {}).items():
if name == 'parameters':
config['postgresql'][name].update(self._process_postgresql_parameters(value, True))
elif name != 'use_slots': # replication slots must be enabled/disabled globally
config['postgresql'][name] = deepcopy(value)
elif name not in config:
config[name] = deepcopy(value) if value else {}
# restapi server expects to get restapi.auth = 'username:password'
if 'authentication' in config['restapi']:
restapi = config['restapi']
auth = restapi['authentication']
restapi['auth'] = '{0}:{1}'.format(auth['username'], auth['password'])
# special treatment for old config
# 'exhibitor' inside 'zookeeper':
if 'zookeeper' in config and 'exhibitor' in config['zookeeper']:
config['exhibitor'] = config['zookeeper'].pop('exhibitor')
config.pop('zookeeper')
pg_config = config['postgresql']
# no 'authentication' in 'postgresql', but 'replication' and 'superuser'
if 'authentication' not in pg_config:
pg_config['use_pg_rewind'] = 'pg_rewind' in pg_config
pg_config['authentication'] = {u: pg_config[u] for u in ('replication', 'superuser') if u in pg_config}
# no 'superuser' in 'postgresql'.'authentication'
if 'superuser' not in pg_config['authentication'] and 'pg_rewind' in pg_config:
pg_config['authentication']['superuser'] = pg_config['pg_rewind']
# no 'name' in config
if 'name' not in config and 'name' in pg_config:
config['name'] = pg_config['name']
pg_config.update({p: config[p] for p in ('name', 'scope', 'retry_timeout',
'maximum_lag_on_failover') if p in config})
return config
def get(self, key, default=None):
return self.__effective_configuration.get(key, default)
def __contains__(self, key):
return key in self.__effective_configuration
def __getitem__(self, key):
return self.__effective_configuration[key]
patroni/ctl.py
+3 -2
View File
@@ -95,8 +95,9 @@ def ctl(ctx):
def get_dcs(config, scope):
config.setdefault('scope', scope)
config.setdefault('name', scope)
try:
return _get_dcs(scope, config)
return _get_dcs(config)
except PatroniException as e:
raise PatroniCtlException(str(e))
@@ -533,7 +534,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled
r = None
try:
r = post_patroni(cluster.leader.member, 'failover', failover_value)
if r.status_code == 200:
if r.status_code in (200, 202):
logging.debug(r)
cluster = dcs.get_cluster()
logging.debug(cluster)
patroni/dcs/__init__.py
+45 -11
View File
@@ -30,7 +30,7 @@ def parse_connection_string(value):
return conn_url, api_url
def get_dcs(node_name, config):
def get_dcs(config):
available_implementations = []
for name in os.listdir(os.path.dirname(__file__)):
if name.endswith('.py') and not name.startswith('__'): # find module
@@ -44,8 +44,9 @@ def get_dcs(node_name, config):
available_implementations.append(name)
if name in config: # which has configuration section in the config file
# propagate some parameters
config[name].update({p: config[p] for p in ('namespace', 'scope', 'ttl') if p in config})
return value(node_name, config[name])
config[name].update({p: config[p] for p in ('namespace', 'name',
'scope', 'ttl', 'retry_timeout') if p in config})
return value(config[name])
raise PatroniException("""Can not find suitable configuration of distributed configuration store
Available implementations: """ + ', '.join(available_implementations))
@@ -163,11 +164,28 @@ class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')):
return Failover(index, data.get('leader'), data.get('member'), data.get('scheduled_at'))
class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members,failover')):
class ClusterConfig(namedtuple('ClusterConfig', 'index,data,modify_index')):
@staticmethod
def from_node(index, data, modify_index=None):
"""
>>> ClusterConfig.from_node(1, '{') is None
True
"""
try:
data = json.loads(data)
except (TypeError, ValueError):
return None
return ClusterConfig(index, data, modify_index or index)
class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_leader_operation,members,failover')):
"""Immutable object (namedtuple) which represents PostgreSQL cluster.
Consists of the following fields:
:param initialize: boolean, shows whether this cluster has initialization key stored in DC or not.
:param config: global dynamic configuration, reference to `ClusterConfig` object
:param leader: `Leader` object which represents current leader of the cluster
:param last_leader_operation: int or long object containing position of last known leader operation.
This value is stored in `/optime/leader` key
@@ -192,19 +210,19 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
class AbstractDCS(object):
_INITIALIZE = 'initialize'
_CONFIG = 'config'
_LEADER = 'leader'
_FAILOVER = 'failover'
_MEMBERS = 'members/'
_OPTIME = 'optime'
_LEADER_OPTIME = _OPTIME + '/' + _LEADER
def __init__(self, name, config):
def __init__(self, config):
"""
:param name: name of current instance (the same value as `~Postgresql.name`)
:param config: dict, reference to config section of selected DCS.
i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc...
"""
self._name = name
self._name = config['name']
self._namespace = '/{0}'.format(config.get('namespace', '/service/').strip('/'))
self._base_path = '/'.join([self._namespace, config['scope']])
@@ -219,6 +237,10 @@ class AbstractDCS(object):
def initialize_path(self):
return self.client_path(self._INITIALIZE)
@property
def config_path(self):
return self.client_path(self._CONFIG)
@property
def members_path(self):
return self.client_path(self._MEMBERS)
@@ -239,6 +261,14 @@ class AbstractDCS(object):
def leader_optime_path(self):
return self.client_path(self._LEADER_OPTIME)
@abc.abstractmethod
def set_ttl(self, ttl):
"""Set the new ttl value for leader key"""
@abc.abstractmethod
def set_retry_timeout(self, retry_timeout):
"""Set the new value for retry_timeout"""
@abc.abstractmethod
def _load_cluster(self):
"""Internally this method should build `Cluster` object which
@@ -306,15 +336,19 @@ class AbstractDCS(object):
if scheduled_at:
failover_value['scheduled_at'] = scheduled_at.isoformat()
return self.set_failover_value(json.dumps(failover_value), index)
return self.set_failover_value(json.dumps(failover_value, separators=(',', ':')), index)
@abc.abstractmethod
def touch_member(self, connection_string, ttl=None):
def set_config_value(self, value, index=None):
"""Create or update `/config` key"""
@abc.abstractmethod
def touch_member(self, data, ttl=None):
"""Update member key in DCS.
This method should create or update key with the name = '/members/' + `~self._name`
and value = connection_string in a given DCS.
and value = data in a given DCS.
:param connection_string: how this instance can be accessed by other instances
:param data: json serialized information about instance (including connection strings)
:param ttl: ttl for member key, optional parameter. If it is None `~self.member_ttl will be used`
:returns: `!True` on success otherwise `!False`
"""
patroni/dcs/consul.py
+44 -22
View File
@@ -5,7 +5,7 @@ import time
import six
from consul import ConsulException, NotFound, base, std
from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member
from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member
from patroni.exceptions import DCSError
from patroni.utils import sleep
from requests.exceptions import RequestException
@@ -21,9 +21,8 @@ class HTTPClient(std.HTTPClient):
def __init__(self, *args, **kwargs):
super(HTTPClient, self).__init__(*args, **kwargs)
self._patch_default_timeout()
def _patch_default_timeout(self):
def patch_default_timeout(self, timeout):
# Set a default timeout for the `request.session.request` method, that is used
# internally by the methods request.session.get, request.session.post and
# others. We monkey-patch here to avoid reimplementing each individual method from
@@ -35,7 +34,7 @@ class HTTPClient(std.HTTPClient):
defaults_attr_name = '__defaults__' if six.PY3 else 'func_defaults'
defaults = list(getattr(request_func, defaults_attr_name))
code = request_func.__code__ if six.PY3 else request_func.func_code
defaults[code.co_varnames[code.co_argcount - len(defaults):code.co_argcount].index('timeout')] = 5
defaults[code.co_varnames[code.co_argcount - len(defaults):code.co_argcount].index('timeout')] = timeout
setattr(request_func, defaults_attr_name, tuple(defaults)) # monkeypatching
def get(self, callback, path, params=None):
@@ -70,27 +69,37 @@ def catch_consul_errors(func):
class Consul(AbstractDCS):
def __init__(self, name, config):
super(Consul, self).__init__(name, config)
self.ttl = int((config.get('ttl') or 30)/2) # My experiments have shown that session expires after 2*ttl time
host, port = config.get('host', '127.0.0.1:8500').split(':')
self._client = ConsulClient(host=host, port=port)
self._scope = config['scope']
def __init__(self, config):
super(Consul, self).__init__(config)
self._ttl = None
self._session = None
self._my_member_data = None
self.create_or_restore_session()
self.set_ttl(config.get('ttl') or 30)
host, port = config.get('host', '127.0.0.1:8500').split(':')
self._client = ConsulClient(host=host, port=port)
self._client.http.patch_default_timeout(config['retry_timeout']/2.0)
self._scope = config['scope']
self.create_session()
self.__do_not_watch = False
def create_or_restore_session(self):
def create_session(self):
while not self._session:
try:
_, member = self._client.kv.get(self.member_path)
self._session = (member or {}).get('Session')
if self.refresh_session():
self._client.kv.delete(self.member_path)
except (ConsulException, RequestException):
self.refresh_session()
except ConsulError:
logger.info('waiting on consul')
sleep(5)
def set_ttl(self, ttl):
ttl = ttl/2.0 # My experiments have shown that session expires after 2*ttl time
if self._ttl != ttl:
self._session = None
self.__do_not_watch = True
self._ttl = ttl
def set_retry_timeout(self, retry_timeout):
self._client.http.patch_default_timeout(retry_timeout/2.0)
def refresh_session(self):
""":returns: `!True` if it had to create new session"""
if self._session:
@@ -101,7 +110,7 @@ class Consul(AbstractDCS):
if not self._session:
name = self._scope + '-' + self._name
try:
self._session = self._client.session.create(name=name, lock_delay=0, behavior='delete', ttl=self.ttl)
self._session = self._client.session.create(name=name, lock_delay=0, behavior='delete', ttl=self._ttl)
except (ConsulException, RequestException):
logger.exception('session.create')
if not self._session:
@@ -132,6 +141,10 @@ class Consul(AbstractDCS):
initialize = nodes.get(self._INITIALIZE)
initialize = initialize and initialize['Value']
# get global dynamic configuration
config = nodes.get(self._CONFIG)
config = config and ClusterConfig.from_node(config['ModifyIndex'], config['Value'])
# get last leader operation
last_leader_operation = nodes.get(self._LEADER_OPTIME)
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['Value'])
@@ -156,20 +169,21 @@ class Consul(AbstractDCS):
if failover:
failover = Failover.from_node(failover['ModifyIndex'], failover['Value'])
self._cluster = Cluster(initialize, leader, last_leader_operation, members, failover)
self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover)
except NotFound:
self._cluster = Cluster(False, None, None, [], None)
self._cluster = Cluster(False, None, None, None, [], None)
except:
logger.exception('get_cluster')
raise ConsulError('Consul is not responding properly')
def touch_member(self, data, **kwargs):
create_member = self.refresh_session()
cluster = self.cluster
member = cluster and ([m for m in cluster.members if m.name == self._name] or [None])[0]
if create_member and member:
create_member = self.refresh_session()
if member and (create_member or member.session != self._session):
try:
self._client.kv.delete(self.member_path)
create_member = True
except Exception:
return False
@@ -198,6 +212,10 @@ class Consul(AbstractDCS):
def set_failover_value(self, value, index=None):
return self._client.kv.put(self.failover_path, value, cas=index)
@catch_consul_errors
def set_config_value(self, value, index=None):
return self._client.kv.put(self.config_path, value, cas=index)
@catch_consul_errors
def write_leader_optime(self, last_operation):
return self._client.kv.put(self.leader_optime_path, last_operation)
@@ -226,6 +244,10 @@ class Consul(AbstractDCS):
return self._client.kv.delete(self.leader_path, cas=cluster.leader.index)
def watch(self, timeout):
if self.__do_not_watch:
self.__do_not_watch = False
return True
cluster = self.cluster
if cluster and cluster.leader and cluster.leader.name != self._name and cluster.leader.index:
end_time = time.time() + timeout
patroni/dcs/etcd.py
+33 -12
View File
@@ -9,7 +9,7 @@ import time
from dns.exception import DNSException
from dns import resolver
from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member
from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member
from patroni.exceptions import DCSError
from patroni.utils import Retry, RetryFailedError, sleep
from urllib3.exceptions import HTTPError, ReadTimeoutError
@@ -191,15 +191,16 @@ def catch_etcd_errors(func):
class Etcd(AbstractDCS):
def __init__(self, name, config):
super(Etcd, self).__init__(name, config)
self.ttl = config.get('ttl', 30)
self._retry = Retry(deadline=10, max_delay=1, max_tries=-1,
def __init__(self, config):
super(Etcd, self).__init__(config)
self._ttl = int(config.get('ttl') or 30)
self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
retry_exceptions=(etcd.EtcdConnectionFailed,
etcd.EtcdLeaderElectionInProgress,
etcd.EtcdWatcherCleared,
etcd.EtcdEventIndexCleared))
self._client = self.get_etcd_client(config)
self.__do_not_watch = False
def retry(self, *args, **kwargs):
return self._retry.copy()(*args, **kwargs)
@@ -215,6 +216,14 @@ class Etcd(AbstractDCS):
sleep(5)
return client
def set_ttl(self, ttl):
ttl = int(ttl)
self.__do_not_watch = self._ttl != ttl
self._ttl = ttl
def set_retry_timeout(self, retry_timeout):
self._retry.deadline = retry_timeout
@staticmethod
def member(node):
return Member.from_node(node.modifiedIndex, os.path.basename(node.key), node.ttl, node.value)
@@ -228,6 +237,10 @@ class Etcd(AbstractDCS):
initialize = nodes.get(self._INITIALIZE)
initialize = initialize and initialize.value
# get global dynamic configuration
config = nodes.get(self._CONFIG)
config = config and ClusterConfig.from_node(config.modifiedIndex, config.value)
# get last leader operation
last_leader_operation = nodes.get(self._LEADER_OPTIME)
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value)
@@ -247,24 +260,24 @@ class Etcd(AbstractDCS):
if failover:
failover = Failover.from_node(failover.modifiedIndex, failover.value)
self._cluster = Cluster(initialize, leader, last_leader_operation, members, failover)
self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover)
except etcd.EtcdKeyNotFound:
self._cluster = Cluster(False, None, None, [], None)
self._cluster = Cluster(False, None, None, None, [], None)
except:
logger.exception('get_cluster')
raise EtcdError('Etcd is not responding properly')
@catch_etcd_errors
def touch_member(self, connection_string, ttl=None):
return self.retry(self._client.set, self.member_path, connection_string, ttl or self.ttl)
def touch_member(self, data, ttl=None):
return self.retry(self._client.set, self.member_path, data, ttl or self._ttl)
@catch_etcd_errors
def take_leader(self):
return self.retry(self._client.set, self.leader_path, self._name, self.ttl)
return self.retry(self._client.set, self.leader_path, self._name, self._ttl)
def attempt_to_acquire_leader(self):
try:
return bool(self.retry(self._client.write, self.leader_path, self._name, ttl=self.ttl, prevExist=False))
return bool(self.retry(self._client.write, self.leader_path, self._name, ttl=self._ttl, prevExist=False))
except etcd.EtcdAlreadyExist:
logger.info('Could not take out TTL lock')
except (RetryFailedError, etcd.EtcdException):
@@ -275,13 +288,17 @@ class Etcd(AbstractDCS):
def set_failover_value(self, value, index=None):
return self._client.write(self.failover_path, value, prevIndex=index or 0)
@catch_etcd_errors
def set_config_value(self, value, index=None):
return self._client.write(self.config_path, value, prevIndex=index or 0)
@catch_etcd_errors
def write_leader_optime(self, last_operation):
return self._client.set(self.leader_optime_path, last_operation)
@catch_etcd_errors
def update_leader(self):
return self.retry(self._client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
return self.retry(self._client.test_and_set, self.leader_path, self._name, self._name, self._ttl)
@catch_etcd_errors
def initialize(self, create_new=True, sysid=""):
@@ -300,6 +317,10 @@ class Etcd(AbstractDCS):
return self.retry(self._client.delete, self.client_path(''), recursive=True)
def watch(self, timeout):
if self.__do_not_watch:
self.__do_not_watch = False
return True
cluster = self.cluster
# watch on leader key changes if it is defined and current node is not lock owner
if cluster and cluster.leader and cluster.leader.name != self._name and cluster.leader.index:
patroni/dcs/exhibitor.py
+2 -2
View File
@@ -62,12 +62,12 @@ class ExhibitorEnsembleProvider(object):
class Exhibitor(ZooKeeper):
def __init__(self, name, config):
def __init__(self, config):
interval = config.get('poll_interval', 300)
self._ensemble_provider = ExhibitorEnsembleProvider(config['hosts'], config['port'], poll_interval=interval)
config = config.copy()
config['hosts'] = self._ensemble_provider.zookeeper_hosts
super(Exhibitor, self).__init__(name, config)
super(Exhibitor, self).__init__(config)
def _load_cluster(self):
if self._ensemble_provider.poll():
patroni/dcs/zookeeper.py
+33 -12
View File
@@ -2,7 +2,7 @@ import logging
from kazoo.client import KazooClient, KazooState
from kazoo.exceptions import NoNodeError, NodeExistsError
from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member
from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member
from patroni.exceptions import DCSError
logger = logging.getLogger(__name__)
@@ -14,17 +14,15 @@ class ZooKeeperError(DCSError):
class ZooKeeper(AbstractDCS):
def __init__(self, name, config):
super(ZooKeeper, self).__init__(name, config)
def __init__(self, config):
super(ZooKeeper, self).__init__(config)
hosts = config.get('hosts', [])
if isinstance(hosts, list):
hosts = ','.join(hosts)
self._client = KazooClient(hosts=hosts, timeout=(config.get('session_timeout') or 30),
command_retry={'deadline': (config.get('reconnect_timeout') or 10),
'max_delay': 1, 'max_tries': -1},
connection_retry={'max_delay': 1, 'max_tries': -1})
self._client = KazooClient(hosts, timeout=config['ttl'], connection_retry={'max_delay': 1, 'max_tries': -1},
command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1})
self._client.add_listener(self.session_listener)
self._my_member_data = None
@@ -41,6 +39,16 @@ class ZooKeeper(AbstractDCS):
self._fetch_cluster = True
self.event.set()
def set_ttl(self, ttl):
ttl = int(ttl * 1000)
# I know, it's weird to access private attributes
if self._client._session_timeout != ttl:
self._client._session_timeout = ttl
self._client.restart()
def set_retry_timeout(self, retry_timeout):
self._client._retry.deadline = retry_timeout
def get_node(self, key, watch=None):
try:
ret = self._client.get(key, watch)
@@ -76,6 +84,10 @@ class ZooKeeper(AbstractDCS):
# get initialize flag
initialize = (self.get_node(self.initialize_path) or [None])[0] if self._INITIALIZE in nodes else None
# get global dynamic configuration
config = self.get_node(self.config_path, watch=self.cluster_watcher) if self._CONFIG in nodes else None
config = config and ClusterConfig.from_node(config[1].version, config[0], config[1].mzxid)
# get list of members
members = self.load_members() if self._MEMBERS[:-1] in nodes else []
@@ -96,13 +108,12 @@ class ZooKeeper(AbstractDCS):
# failover key
failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
if failover:
failover = Failover.from_node(failover[1].version, failover[0])
failover = failover and Failover.from_node(failover[1].version, failover[0])
# get last leader operation
optime = self.get_node(self.leader_optime_path) if self._OPTIME in nodes and self._fetch_cluster else None
self._last_leader_operation = 0 if optime is None else int(optime[0])
self._cluster = Cluster(initialize, leader, self._last_leader_operation, members, failover)
self._cluster = Cluster(initialize, config, leader, self._last_leader_operation, members, failover)
def _load_cluster(self):
if self._fetch_cluster or self._cluster is None:
@@ -110,7 +121,7 @@ class ZooKeeper(AbstractDCS):
self._client.retry(self._inner_load_cluster)
except:
logger.exception('get_cluster')
self.session_listener(KazooState.LOST)
self.cluster_watcher(None)
raise ZooKeeperError('ZooKeeper in not responding properly')
def _create(self, path, value, **kwargs):
@@ -131,11 +142,21 @@ 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))
return value == '' or (index is None and self._create(self.failover_path, value))
except:
logging.exception('set_failover_value')
return False
def set_config_value(self, value, index=None):
try:
self._client.retry(self._client.set, self.config_path, value.encode('utf-8'), version=index or -1)
return True
except NoNodeError:
return index is None and self._create(self.config_path, value)
except Exception:
logging.exception('set_config_value')
return False
def initialize(self, create_new=True, sysid=""):
return self._create(self.initialize_path, sysid, makepath=True) if create_new \
else self._client.retry(self._client.set, self.initialize_path, sysid.encode("utf-8"))
patroni/ha.py
+10 -3
View File
@@ -59,6 +59,8 @@ class Ha(object):
}
if self.patroni.tags:
data['tags'] = self.patroni.tags
if self.state_handler.pending_restart:
data['pending_restart'] = True
if not self._async_executor.busy and data['state'] in ['running', 'restarting', 'starting']:
try:
data['xlog_location'] = self.state_handler.xlog_position()
@@ -84,10 +86,11 @@ class Ha(object):
self._async_executor.schedule('bootstrap {0}'.format(msg))
self._async_executor.run_async(self.clone, args=(clone_member, msg))
return 'trying to bootstrap {0}'.format(msg)
elif not self.cluster.initialize and not self.patroni.nofailover: # no initialize key
# no initialize key and node is allowed to be master and has 'bootstrap' section in a configuration file
elif not (self.cluster.initialize or self.patroni.nofailover) and 'bootstrap' in self.patroni.config:
if self.dcs.initialize(create_new=True): # race for initialization
try:
self.state_handler.bootstrap()
self.state_handler.bootstrap(self.patroni.config['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
@@ -96,6 +99,7 @@ class Ha(object):
self.state_handler.stop('immediate')
self.state_handler.move_data_directory()
raise
self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
self.dcs.take_leader()
self.load_cluster_from_dcs()
return 'initialized a new cluster'
@@ -440,9 +444,12 @@ class Ha(object):
self.touch_member()
# cluster has leader key but not initialize key
if not self.cluster.is_unlocked() and not self.sysid_valid(self.cluster.initialize) and self.has_lock():
if not (self.cluster.is_unlocked() or 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 not (self.cluster.is_unlocked() or self.cluster.config and self.cluster.config.data) and self.has_lock():
self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
if self._async_executor.busy:
return self.handle_long_action_in_progress()
patroni/postgresql.py
+224 -72
View File
@@ -8,7 +8,7 @@ import tempfile
import time
from patroni.exceptions import PostgresConnectionException, PostgresException
from patroni.utils import Retry, RetryFailedError
from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError
from six import string_types
from six.moves.urllib_parse import urlparse
from threading import Lock
@@ -41,25 +41,59 @@ def parseurl(url):
class Postgresql(object):
# List of parameters which must be always passed to postmaster as command line options
# to make it not possible to change them with 'ALTER SYSTEM'.
# Some of these parameters have sane default value assigned and Patroni doesn't allow
# to decrease this value. E.g. 'wal_level' can't be lower then 'hot_standby' and so on.
# These parameters could be changed only globally, i.e. via DCS.
# P.S. 'listen_addresses' and 'port' are added here just for convenience, to mark them
# as a parameters which should always be passed through command line.
#
# Format:
# key - parameter name
# value - tuple(default_value, check_function, min_version)
# default_value -- some sane default value
# check_function -- if the new value is not correct must return `!False`
# min_version -- major version of PostgreSQL when parameter was introduced
CMDLINE_OPTIONS = {
'listen_addresses': (None, lambda _: False, 9.1),
'port': (None, lambda _: False, 9.1),
'cluster_name': (None, lambda _: False, 9.5),
'wal_level': ('hot_standby', lambda v: v.lower() in ('hot_standby', 'logical'), 9.1),
'hot_standby': ('on', lambda _: False, 9.1),
'max_connections': (100, lambda v: int(v) >= 100, 9.1),
'max_wal_senders': (5, lambda v: int(v) >= 5, 9.1),
'wal_keep_segments': (8, lambda v: int(v) >= 8, 9.1),
'max_prepared_transactions': (0, lambda v: int(v) >= 0, 9.1),
'max_locks_per_transaction': (64, lambda v: int(v) >= 64, 9.1),
'track_commit_timestamp': ('off', lambda v: parse_bool(v) is not None, 9.5),
'max_replication_slots': (5, lambda v: int(v) >= 5, 9.4),
'max_worker_processes': (8, lambda v: int(v) >= 8, 9.4),
'wal_log_hints': ('on', lambda _: False, 9.4)
}
def __init__(self, config):
self.config = config
self.name = config['name']
self.database = config.get('database', 'postgres')
self._server_parameters = self.get_server_parameters(config)
self._listen_addresses, self._port = (config['listen'] + ':5432').split(':')[:2]
self.scope = config['scope']
self._database = config.get('database', 'postgres')
self._data_dir = config['data_dir']
self.replication = config['replication']
self.superuser = config.get('superuser') or {}
self.admin = config.get('admin') or {}
self._pending_restart = False
self._server_parameters = self.get_server_parameters(config)
self.initdb_options = config.get('initdb') or []
self.pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
self.pg_rewind = config.get('pg_rewind') or {}
self.callback = config.get('callbacks') or {}
self.use_slots = config.get('use_slots', True)
self._connect_address = config.get('connect_address')
self._superuser = config['authentication'].get('superuser', {})
self._replication = config['authentication']['replication']
self.resolve_connection_addresses()
self._use_pg_rewind = config.get('use_pg_rewind', False)
self._use_slots = config.get('use_slots', True)
self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
self._major_version = self.get_major_version()
self._schedule_load_slots = self.use_slots
self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
self.callback = config.get('callbacks') or {}
config_base_name = config.get('config_base_name', 'postgresql')
self._postgresql_conf = os.path.join(self._data_dir, config_base_name + '.conf')
self._postgresql_base_conf_name = config_base_name + '.base.conf'
@@ -73,16 +107,12 @@ class Postgresql(object):
self._pg_ctl = ['pg_ctl', '-w', '-D', self._data_dir]
self.local_address = self.get_local_address()
connect_address = config.get('connect_address') or self.local_address
self.connection_string = 'postgres://{username}:{password}@{connect_address}/{database}'.format(
connect_address=connect_address, database=self.database, **self.replication)
self._connection = None
self._cursor_holder = None
self._sysid = None
self._replication_slots = [] # list of already existing replication slots
self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException)
self.retry = Retry(max_tries=-1, deadline=config['retry_timeout']/2.0, max_delay=1,
retry_exceptions=PostgresConnectionException)
self._state_lock = Lock()
self.set_state('stopped')
@@ -94,9 +124,97 @@ class Postgresql(object):
self.set_role('master' if self.is_leader() else 'replica')
self._write_postgresql_conf() # we are "joining" already running postgres
@staticmethod
def get_server_parameters(config):
return {p: v for p, v in (config.get('parameters') or {}).items() if p not in ('listen_addresses', 'port')}
@property
def use_slots(self):
return self._use_slots and self._major_version >= 9.4
def _version_file_exists(self):
return not self.data_directory_empty() and os.path.isfile(self._version_file)
def get_major_version(self):
if self._version_file_exists():
try:
with open(self._version_file) as f:
return float(f.read())
except Exception:
logger.exception('Failed to read PG_VERSION from %s', self._data_dir)
return 0.0
def get_server_parameters(self, config):
parameters = config['parameters'].copy()
listen_addresses, port = (config['listen'] + ':5432').split(':')[:2]
parameters.update({'cluster_name': self.scope, 'listen_addresses': listen_addresses, 'port': port})
return parameters
def resolve_connection_addresses(self):
self._local_address = self.get_local_address()
self.connection_string = 'postgres://{username}:{password}@{connect_address}/{database}'.format(
connect_address=self._connect_address or self._local_address, database=self._database, **self._replication)
def reload_config(self, config):
server_parameters = self.get_server_parameters(config)
listen_address_changed = pending_reload = pending_restart = False
if self.is_healthy():
changes = {p: v for p, v in server_parameters.items() if '.' not in p}
changes.update({p: None for p, v in self._server_parameters.items() if not ('.' in p or p in changes)})
if changes:
if 'wal_segment_size' not in changes:
changes['wal_segment_size'] = '16384kB'
# XXX: query can raise an exception
for r in self.query("""SELECT name, setting, unit, vartype, context
FROM pg_settings
WHERE name IN (""" + ', '.join(['%s'] * len(changes)) + """)
ORDER BY 1 DESC""", *(list(changes.keys()))):
if r[4] == 'internal':
if r[0] == 'wal_segment_size':
server_parameters.pop(r[0], None)
wal_segment_size = parse_int(r[2], 'kB')
if wal_segment_size is not None:
changes['wal_segment_size'] = '{0}kB'.format(int(r[1]) * wal_segment_size)
elif r[0] in changes:
unit = changes['wal_segment_size'] if r[0] in ('min_wal_size', 'max_wal_size') else r[2]
new_value = changes.pop(r[0])
if new_value is None or not compare_values(r[3], unit, r[1], new_value):
if r[4] == 'postmaster':
pending_restart = True
if r[0] in ('listen_addresses', 'port'):
listen_address_changed = True
else:
pending_reload = True
for param in changes:
if param in server_parameters:
logger.warning('Removing invalid parameter `%s` from postgresql.parameters', param)
server_parameters.pop(param)
# Check that user-defined-paramters have changed (parameters with period in name)
if not pending_reload:
for p, v in server_parameters.items():
if '.' in p and (p not in self._server_parameters or str(v) != str(self._server_parameters[p])):
pending_reload = True
break
if not pending_reload:
for p, v in self._server_parameters.items():
if '.' in p and (p not in server_parameters or str(v) != str(server_parameters[p])):
pending_reload = True
break
self.config = config
self._pending_restart = pending_restart
self._server_parameters = server_parameters
self._connect_address = config.get('connect_address')
if not listen_address_changed:
self.resolve_connection_addresses()
if pending_reload:
self._write_postgresql_conf()
self.reload()
self.retry.deadline = config['retry_timeout']/2.0
@property
def pending_restart(self):
return self._pending_restart
@property
def can_rewind(self):
@@ -104,8 +222,7 @@ class Postgresql(object):
we have either wal_log_hints or checksums turned on
"""
# low-hanging fruit: check if pg_rewind configuration is there
if not self.pg_rewind or\
not (self.pg_rewind.get('username', '') and self.pg_rewind.get('password', '')):
if not (self._use_pg_rewind and all(self._superuser.get(n) for n in ('username', 'password'))):
return False
cmd = ['pg_rewind', '--help']
@@ -127,14 +244,14 @@ class Postgresql(object):
return self._sysid
def get_local_address(self):
listen_addresses = self._listen_addresses.split(',')
listen_addresses = self._server_parameters['listen_addresses'].split(',')
local_address = listen_addresses[0].strip() # take first address from listen_addresses
for la in listen_addresses:
if la.strip() in ['*', '0.0.0.0']: # we are listening on *
if la.strip() in ('*', '0.0.0.0', '127.0.0.1', 'localhost'): # we are listening on '*' or localhost
local_address = 'localhost' # connection via localhost is preferred
break
return local_address + ':' + self._port
return local_address + ':' + self._server_parameters['port']
def get_postgres_role_from_data_directory(self):
if self.data_directory_empty():
@@ -146,11 +263,11 @@ class Postgresql(object):
@property
def _connect_kwargs(self):
r = parseurl('postgres://{0}/{1}'.format(self.local_address, self.database))
if 'username' in self.superuser:
r['user'] = self.superuser['username']
if 'password' in self.superuser:
r['password'] = self.superuser['password']
r = parseurl('postgres://{0}/{1}'.format(self._local_address, self._database))
if 'username' in self._superuser:
r['user'] = self._superuser['username']
if 'password' in self._superuser:
r['password'] = self._superuser['password']
return r
def connection(self):
@@ -199,9 +316,9 @@ class Postgresql(object):
raise Exception('{0} option for initdb is not allowed'.format(name))
return True
def get_initdb_options(self):
def get_initdb_options(self, config):
options = []
for o in self.initdb_options:
for o in config:
if isinstance(o, string_types) and self.initdb_allowed_option(o):
options.append('--{0}'.format(o))
elif isinstance(o, dict):
@@ -213,17 +330,17 @@ class Postgresql(object):
raise Exception('Unknown type of initdb option: {0}'.format(o))
return options
def initialize(self):
def _initialize(self, config):
self.set_state('initalizing new cluster')
options = self.get_initdb_options()
options = self.get_initdb_options(config.get('initdb') or [])
pwfile = None
if self.superuser:
if 'username' in self.superuser:
options.append('--username={0}'.format(self.superuser['username']))
if 'password' in self.superuser:
if self._superuser:
if 'username' in self._superuser:
options.append('--username={0}'.format(self._superuser['username']))
if 'password' in self._superuser:
(fd, pwfile) = tempfile.mkstemp()
os.write(fd, self.superuser['password'].encode('utf-8'))
os.write(fd, self._superuser['password'].encode('utf-8'))
os.close(fd)
options.append('--pwfile={0}'.format(pwfile))
@@ -231,7 +348,8 @@ class Postgresql(object):
if pwfile:
os.remove(pwfile)
if ret:
self.write_pg_hba()
self.write_pg_hba(config.get('pg_hba', []))
self._major_version = self.get_major_version()
else:
self.set_state('initdb failed')
return ret
@@ -241,12 +359,12 @@ class Postgresql(object):
os.unlink(self._trigger_file)
def write_pgpass(self, record):
with open(self.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'] = self.pgpass
env['PGPASSFILE'] = self._pgpass
return env
def replica_method_can_work_without_replication_connection(self, method):
@@ -329,7 +447,16 @@ class Postgresql(object):
return not self.query('SELECT pg_is_in_recovery()').fetchone()[0]
def is_running(self):
return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null 2>&1', shell=True) == 0
if not (self._version_file_exists() and os.path.isfile(self._postmaster_pid)):
return False
try:
with open(self._postmaster_pid) as f:
pid = int(f.readline())
if pid < 0:
pid = -pid
return pid > 0 and pid != os.getpid() and pid != os.getppid() and (os.kill(pid, 0) or True)
except Exception:
return False
def call_nowait(self, cb_name):
""" pick a callback command and call it without waiting for it to finish """
@@ -376,11 +503,17 @@ class Postgresql(object):
env = {'PATH': os.environ.get('PATH')}
# pg_ctl will write a FATAL if the username is incorrect. exporting PGUSER if necessary
if 'username' in self.superuser and self.superuser['username'] != os.environ.get('USER'):
env['PGUSER'] = self.superuser['username']
if 'username' in self._superuser and self._superuser['username'] != os.environ.get('USER'):
env['PGUSER'] = self._superuser['username']
self._write_postgresql_conf()
server_arguments = ['-o', "--listen_addresses='{0}' --port={1}".format(self._listen_addresses, self._port)]
ret = subprocess.call(self._pg_ctl + ['start'] + server_arguments, env=env, preexec_fn=os.setsid) == 0
self.resolve_connection_addresses()
options = ' '.join("--{0}='{1}'".format(p, self._server_parameters[p]) for p, v in self.CMDLINE_OPTIONS.items()
if self._major_version >= v[2])
ret = subprocess.call(self._pg_ctl + ['start', '-o', options], env=env, preexec_fn=os.setsid) == 0
self._pending_restart = False
self.set_state('running' if ret else 'start failed')
@@ -456,8 +589,9 @@ class Postgresql(object):
with open(self._postgresql_conf, 'w') as f:
f.write('# Do not edit this file manually!\n# It will be overwritten by Patroni!\n')
f.write("include '{0}'\n\n".format(self._postgresql_base_conf_name))
for setting, value in sorted(self._server_parameters.items()):
f.write("{0} = '{1}'\n".format(setting, value))
for name, value in sorted(self._server_parameters.items()):
if name not in self.CMDLINE_OPTIONS:
f.write("{0} = '{1}'\n".format(name, value))
def is_healthy(self):
if not self.is_running():
@@ -468,9 +602,9 @@ class Postgresql(object):
def check_replication_lag(self, last_leader_operation):
return (last_leader_operation or 0) - self.xlog_position() <= self.config.get('maximum_lag_on_failover', 0)
def write_pg_hba(self):
def write_pg_hba(self, config):
with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'a') as f:
f.write('\n{}\n'.format('\n'.join(self.config.get('pg_hba', []))))
f.write('\n{}\n'.format('\n'.join(config)))
def primary_conninfo(self, leader_url):
r = parseurl(leader_url)
@@ -504,9 +638,9 @@ class Postgresql(object):
def rewind(self, leader):
# prepare pg_rewind connection
r = parseurl(leader.conn_url)
r.update(self.pg_rewind)
r.update(self._superuser)
r['user'] = r.pop('username')
r['database'] = self.database
r['database'] = self._database
env = self.write_pgpass(r)
pc = "user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1".format(**r)
# first run a checkpoint on a promoted master in order
@@ -532,13 +666,29 @@ class Postgresql(object):
logger.exception("Error when calling pg_controldata")
return result
def read_postmaster_opts(self):
""" returns the list of option names/values from postgres.opts, Empty dict if read failed or no file """
result = {}
try:
with open(os.path.join(self._data_dir, "postmaster.opts")) as f:
data = f.read()
opts = [opt.strip('"\n') for opt in data.split(' "')]
for opt in opts:
if '=' in opt and opt.startswith('--'):
name, val = opt.split('=', 1)
name = name.strip('-')
result[name] = val
except IOError:
logger.exception('Error when reading postmaster.opts')
return result
def single_user_mode(self, command=None, options=None):
""" run a given command in a single-user mode. If the command is empty - then just start and stop """
cmd = ['postgres', '--single', '-D', self._data_dir]
for opt, val in sorted((options or {}).items()):
cmd.extend(['-c', '{0}={1}'.format(opt, val)])
# need a database name to connect
cmd.append(self.database)
cmd.append(self._database)
p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
if p:
if command:
@@ -569,7 +719,7 @@ class Postgresql(object):
need_rewind = change_role and self.can_rewind
if need_rewind:
logger.info("set the rewind flag after demote")
if leader and need_rewind: # we have a leader and need to rewind
if leader and leader.name != self.name and need_rewind: # we have a leader and need to rewind
if self.is_running():
self.stop()
# at present, pg_rewind only runs when the cluster is shut down cleanly
@@ -584,7 +734,9 @@ class Postgresql(object):
# clean the flags that tell we should remove them.
self.cleanup_archive_status()
# Start in a single user mode and stop to produce a clean shutdown
self.single_user_mode(options={'archive_mode': 'on', 'archive_command': 'false'})
opts = self.read_postmaster_opts()
opts.update({'archive_mode': 'on', 'archive_command': 'false'})
self.single_user_mode(options=opts)
if self.rewind(leader):
self.write_recovery_conf(member)
ret = self.start()
@@ -632,24 +784,21 @@ class Postgresql(object):
return ret
def create_or_update_role(self, name, password, options):
options = list(map(str.upper, options))
if 'NOLOGIN' not in options and 'LOGIN' not in options:
options.append('LOGIN')
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;
ALTER ROLE "{0}" WITH {1} PASSWORD %s;
ELSE
CREATE ROLE "{0}" WITH LOGIN {1} PASSWORD %s;
CREATE ROLE "{0}" WITH {1} PASSWORD %s;
END IF;
END;
$$""".format(name, options), name, password, password)
def create_replication_user(self):
self.create_or_update_role(self.replication['username'], self.replication['password'], 'REPLICATION')
def create_connection_user(self):
if self.admin:
self.create_or_update_role(self.admin['username'], self.admin['password'], 'CREATEDB CREATEROLE')
$$""".format(name, ' '.join(options)), name, password, password)
def xlog_position(self):
return self.query("""SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
@@ -708,15 +857,18 @@ $$""".format(name, options), name, password, password)
ret = self.create_replica(clone_member) == 0
if ret:
self._major_version = self.get_major_version()
self.delete_trigger_file()
self.restore_configuration_files()
return ret
def bootstrap(self):
def bootstrap(self, config):
""" Initialize a new node from scratch and start it. """
if self.initialize() and self.start():
self.create_replication_user()
self.create_connection_user()
if self._initialize(config) and self.start():
for name, value in config['users'].items():
if name not in (self._superuser.get('username'), self._replication['username']):
self.create_or_update_role(name, value['password'], value.get('options', []))
self.create_or_update_role(self._replication['username'], self._replication['password'], ['REPLICATION'])
else:
raise PostgresException("Could not bootstrap master PostgreSQL")
patroni/utils.py
+175
View File
@@ -2,6 +2,7 @@ import datetime
import os
import random
import signal
import six
import sys
import time
import pytz
@@ -33,6 +34,180 @@ def calculate_ttl(expiration):
return int((expiration - now).total_seconds())
def deep_compare(obj1, obj2):
"""
>>> deep_compare({'1': None}, {})
False
>>> deep_compare({'1': {}}, {'1': None})
False
>>> deep_compare({'1': [1]}, {'1': [2]})
False
>>> deep_compare({'1': 2}, {'1': '2'})
True
>>> deep_compare({'1': {'2': [3, 4]}}, {'1': {'2': [3, 4]}})
True
"""
if set(list(obj1.keys())) != set(list(obj2.keys())): # Objects have different sets of keys
return False
for key, value in obj1.items():
if isinstance(value, dict):
if not (isinstance(obj2[key], dict) and deep_compare(value, obj2[key])):
return False
elif str(value) != str(obj2[key]):
return False
return True
def patch_config(config, data):
"""recursively 'patch' `config` with `data`
:returns: `!True` if the `config` was changed"""
is_changed = False
for name, value in data.items():
if value is None:
if config.pop(name, None) is not None:
is_changed = True
elif name in config:
if isinstance(value, dict):
if isinstance(config[name], dict):
if patch_config(config[name], value):
is_changed = True
else:
config[name] = value
is_changed = True
elif str(config[name]) != str(value):
config[name] = value
is_changed = True
else:
config[name] = value
is_changed = True
return is_changed
def parse_bool(value):
"""
>>> parse_bool(1)
True
>>> parse_bool('off')
False
>>> parse_bool('foo')
"""
value = str(value).lower()
if value in ('on', 'true', 'yes', '1'):
return True
if value in ('off', 'false', 'no', '0'):
return False
def strtol(value, strict=True):
"""As most as possible close equivalent of strtol(3) function (with base=0),
used by postgres to parse parameter values.
>>> strtol(1) == (1, '')
True
>>> strtol(' +0x400MB') == (1024, 'MB')
True
>>> strtol(' -070d') == (-56, 'd')
True
>>> strtol(' d ') == (None, 'd')
True
>>> strtol(' s ', False) == (1, 's')
True
"""
value = str(value).strip()
l = len(value)
i = 0
# skip sign:
if i < l and value[i] in ('-', '+'):
i += 1
# we always expect to get digit in the beginning
if i < l and value[i].isdigit():
if value[i] == '0':
i += 1
if i < l and value[i] in ('x', 'X'): # '0' followed by 'x': HEX
base = 16
i += 1
else: # just starts with '0': OCT
base = 8
else: # any other digit: DEC
base = 10
ret = None
while i < l:
try: # try to find maximally long number
i += 1 # by giving to `int` longer and longer strings
ret = int(value[:i], base) if six.PY3 else long(value[:i], base)
except ValueError: # until we will not get an exception or end of the string
i -= 1
break
if ret is not None: # yay! there is a number in the beginning of the string
return ret, value[i:].strip() # return the number and the "rest"
return (None if strict else 1), value.strip()
def parse_int(value, base_unit=None):
"""
>>> parse_int('1') == 1
True
>>> parse_int(' 0x400 MB ', '16384kB') == 64
True
>>> parse_int('1MB', 'kB') == 1024
True
>>> parse_int('1000 ms', 's') == 1
True
>>> parse_int('1GB', 'MB') is None
True
"""
convert = {
'kB': {'kB': 1, 'MB': 1024, 'GB': 1024 * 1024, 'TB': 1024 * 1024 * 1024},
'ms': {'ms': 1, 's': 1000, 'min': 1000 * 60, 'h': 1000 * 60 * 60, 'd': 1000 * 60 * 60 * 24},
's': {'ms': -1000, 's': 1, 'min': 60, 'h': 60 * 60, 'd': 60 * 60 * 24},
'min': {'ms': -1000 * 60, 's': -60, 'min': 1, 'h': 60, 'd': 60 * 24}
}
value, unit = strtol(value)
if value is not None:
if not unit:
return value
if base_unit and base_unit not in convert:
base_value, base_unit = strtol(base_unit, False)
else:
base_value = 1
if base_unit in convert and unit in convert[base_unit]:
multiplier = convert[base_unit][unit]
if multiplier < 0:
value /= -multiplier
else:
value *= multiplier
return int(value/base_value)
def compare_values(vartype, unit, old_value, new_value):
"""
>>> compare_values('enum', None, 'remote_write', 'REMOTE_WRITE')
True
>>> compare_values('real', None, '1.23', 1.23)
True
"""
# if the integer or bool new_value is not correct this function will return False
if vartype == 'bool':
old_value = parse_bool(old_value)
new_value = parse_bool(new_value)
elif vartype == 'integer':
old_value = parse_int(old_value)
new_value = parse_int(new_value, unit)
elif vartype == 'enum':
return str(old_value).lower() == str(new_value).lower()
else: # ('string', 'real')
return str(old_value) == str(new_value)
return old_value is not None and new_value is not None and old_value == new_value
def set_ignore_sigterm(value=True):
global __ignore_sigterm
__ignore_sigterm = value
postgres0.yml
+61 -91
View File
@@ -1,104 +1,74 @@
ttl: &ttl 30
loop_wait: &loop_wait 10
scope: &scope batman
scope: batman
#namespace: /service/
name: postgresql0
restapi:
listen: 127.0.0.1:8008
# authentication:
# username: username
# password: password
connect_address: 127.0.0.1:8008
# auth: 'username:password'
# certfile: /etc/ssl/certs/ssl-cert-snakeoil.pem
# keyfile: /etc/ssl/private/ssl-cert-snakeoil.key
etcd:
scope: *scope
ttl: *ttl
host: 127.0.0.1:4001
#discovery_srv: my-etcd.domain
#consul:
# scope: *scope
# ttl: *ttl
# host: 127.0.0.1:8500
#zookeeper:
# scope: *scope
# session_timeout: *ttl
# reconnect_timeout: *loop_wait
# hosts:
# - 127.0.0.1:2181
# - 127.0.0.2:2181
#exhibitor:
# poll_interval: 300
# port: 8181
# hosts:
# - host1
# - host2
# - host3
bootstrap:
# this section will be written into Etcd:/<namespace>/<scope>/config after initializing new cluster
# and all other cluster members will use it as a `global configuration`
dcs:
ttl: 30
loop_wait: 10
retry_timeout: 10
maximum_lag_on_failover: 1048576
postgresql:
use_pg_rewind: true
# use_slots: true
parameters:
# wal_level: hot_standby
# hot_standby: "on"
# wal_keep_segments: 8
# max_wal_senders: 5
# max_replication_slots: 5
# wal_log_hints: "on"
archive_mode: "on"
archive_timeout: 1800s
archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
recovery_conf:
restore_command: cp ../wal_archive/%f %p
# some desired options for 'initdb'
initdb: # Note: It needs to be a list (some options need values, others are switches)
- encoding: UTF8
- data-checksums
pg_hba: # Add following lines to pg_hba.conf after running 'initdb'
- host replication replicator 127.0.0.1/32 md5
- host all all 0.0.0.0/0 md5
# - hostssl all all 0.0.0.0/0 md5
# Some additional users users which needs to be created after initializing new cluster
users:
admin:
password: admin
options:
- createrole
- createdb
postgresql:
name: postgresql0
scope: *scope
listen: 127.0.0.1:5432
connect_address: 127.0.0.1:5432
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
pg_hba:
- host replication replicator 127.0.0.1/32 md5
- host all all 0.0.0.0/0 md5
# - hostssl all all 0.0.0.0/0 md5
replication:
username: replicator
password: rep-pass
superuser:
username: postgres
password: zalando
admin:
username: admin
password: admin
create_replica_method:
- basebackup
# - wal_e
# commented-out example for wal-e provisioning
#wal_e:
#command: /patroni/scripts/wale_restore.py
#env_dir: /etc/wal-e.d/env
#threshold_megabytes: 10240
#threshold_backup_size_percentage: 30
#retries: 2
#use_iam: 1
#recovery_conf:
#restore_command: envdir /etc/wal-e.d/env wal-e wal-fetch "%f" "%p" -p 1
recovery_conf:
restore_command: cp ../wal_archive/%f %p
authentication:
replication:
username: replicator
password: rep-pass
superuser:
username: postgres
password: zalando
parameters:
archive_mode: "on"
wal_level: hot_standby
archive_command: mkdir -p ../wal_archive && test ! -f ../wal_archive/%f && cp %p ../wal_archive/%f
max_wal_senders: 10
wal_keep_segments: 8
archive_timeout: 1800s
max_replication_slots: 10
hot_standby: "on"
wal_log_hints: "on"
unix_socket_directories: '.'
tags:
nofailover: False
noloadbalance: False
clonefrom: False
nofailover: false
noloadbalance: false
clonefrom: false
requirements.txt
+1 -1
View File
@@ -3,7 +3,7 @@ psycopg2>=2.6.1
PyYAML
requests
six >= 1.7
kazoo>=2.2.1
kazoo==2.2.1
python-etcd==0.4.3
python-consul==0.6.0
click>=4.1
tests/test_api.py
+74 -33
View File
@@ -1,13 +1,12 @@
import json
import psycopg2
import unittest
from mock import Mock, patch
from patroni.api import RestApiHandler, RestApiServer
from patroni.dcs import Member
from patroni.dcs import ClusterConfig, 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
@@ -19,6 +18,7 @@ class MockPostgresql(object):
server_version = '999999'
sysid = 'dummysysid'
scope = 'dummy'
pending_restart = True
@staticmethod
def connection():
@@ -49,6 +49,8 @@ class MockHa(object):
class MockPatroni(object):
nap_time = 10
config = Mock()
postgresql = MockPostgresql()
ha = MockHa()
dcs = Mock()
@@ -56,6 +58,10 @@ class MockPatroni(object):
version = '0.00'
noloadbalance = Mock(return_value=False)
@staticmethod
def sighup_handler():
pass
class MockRequest(object):
@@ -70,17 +76,22 @@ class MockRestApiServer(RestApiServer):
def __init__(self, Handler, request):
self.socket = 0
self.serve_forever = Mock()
BaseHTTPServer.HTTPServer.__init__ = Mock()
MockRestApiServer._BaseServer__is_shut_down = Mock()
MockRestApiServer._BaseServer__shutdown_request = True
config = {'listen': '127.0.0.1:8008', 'auth': 'test:test', 'certfile': 'dumb'}
config = {'listen': '127.0.0.1:8008', 'auth': 'test:test'}
super(MockRestApiServer, self).__init__(MockPatroni(), config)
config['certfile'] = 'dumb'
self.reload_config(config)
Handler(MockRequest(request), ('0.0.0.0', 8080), self)
@patch('ssl.wrap_socket', Mock(return_value=0))
class TestRestApiHandler(unittest.TestCase):
_authorization = '\nAuthorization: Basic dGVzdDp0ZXN0'
def test_do_GET(self):
MockRestApiServer(RestApiHandler, 'GET /replica')
with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})):
@@ -100,17 +111,6 @@ class TestRestApiHandler(unittest.TestCase):
def test_do_OPTIONS(self):
self.assertIsNotNone(MockRestApiServer(RestApiHandler, '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, '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 = Mock(return_value=b'foo')
makefile.return_value.flush = Mock(side_effect=socket.error('foo'))
MockRestApiServer(RestApiHandler, 'OPTIONS / HTTP/1.0')
def test_do_GET_patroni(self):
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
@@ -118,8 +118,51 @@ class TestRestApiHandler(unittest.TestCase):
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /restart HTTP/1.0'))
MockRestApiServer(RestApiHandler, 'POST /restart HTTP/1.0\nAuthorization:')
@patch.object(MockHa, 'dcs')
def test_do_GET_config(self, mock_dcs):
mock_dcs.cluster.config.data = {}
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /config'))
mock_dcs.cluster.config = None
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /config'))
@patch.object(MockHa, 'dcs')
def test_do_PATCH_config(self, mock_dcs):
config = {'postgresql': {'use_slots': False, 'use_pg_rewind': True, 'parameters': {'wal_level': 'logical'}}}
mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, json.dumps(config))
request = 'PATCH /config HTTP/1.0' + self._authorization
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
request += '\nContent-Length: '
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request + '34\n\n{"postgresql":{"use_slots":false}}'))
config['ttl'] = 5
config['postgresql'].update({'use_slots': {'foo': True}, "parameters": None})
config = json.dumps(config)
request += str(len(config)) + '\n\n' + config
MockRestApiServer(RestApiHandler, request)
mock_dcs.set_config_value.return_value = False
MockRestApiServer(RestApiHandler, request)
@patch.object(MockHa, 'dcs')
def test_do_PUT_config(self, mock_dcs):
mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, '{}')
request = 'PUT /config HTTP/1.0' + self._authorization + '\nContent-Length: '
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request + '2\n\n{}'))
config = '{"foo": "bar"}'
request += str(len(config)) + '\n\n' + config
MockRestApiServer(RestApiHandler, request)
mock_dcs.set_config_value.return_value = False
MockRestApiServer(RestApiHandler, request)
mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, config)
MockRestApiServer(RestApiHandler, request)
@patch.object(MockPatroni, 'sighup_handler', Mock(side_effect=Exception))
def test_do_POST_reload(self):
with patch.object(MockPatroni, 'config') as mock_config:
mock_config.reload_local_configuration.return_value = False
MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization)
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization))
def test_do_POST_restart(self):
request = 'POST /restart HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0'
request = 'POST /restart HTTP/1.0' + self._authorization
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
with patch.object(MockHa, 'restart', Mock(side_effect=Exception)):
MockRestApiServer(RestApiHandler, request)
@@ -127,7 +170,7 @@ class TestRestApiHandler(unittest.TestCase):
@patch.object(MockHa, 'dcs')
def test_do_POST_reinitialize(self, dcs):
cluster = dcs.get_cluster.return_value
request = 'POST /reinitialize HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0'
request = 'POST /reinitialize HTTP/1.0' + self._authorization
MockRestApiServer(RestApiHandler, request)
cluster.is_unlocked.return_value = False
MockRestApiServer(RestApiHandler, request)
@@ -148,19 +191,20 @@ class TestRestApiHandler(unittest.TestCase):
def test_do_POST_failover(self, dcs):
cluster = dcs.get_cluster.return_value
request = 'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 0\n\n'
post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: '
MockRestApiServer(RestApiHandler, post + '7\n\n{"1":2}')
request = post + '0\n\n'
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'postgresql1'
MockRestApiServer(RestApiHandler, request)
request = 'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
'Content-Length: 25\n\n{"leader": "postgresql1"}'
MockRestApiServer(RestApiHandler, request)
MockRestApiServer(RestApiHandler, post + '25\n\n{"leader": "postgresql1"}')
cluster.leader.name = 'postgresql2'
request = 'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
'Content-Length: 53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'postgresql1'
@@ -186,24 +230,21 @@ class TestRestApiHandler(unittest.TestCase):
MockRestApiServer(RestApiHandler, request)
# Valid future date
request = 'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
'"postgresql1", "member": "postgresql2", "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2",' +\
' "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
MockRestApiServer(RestApiHandler, request)
with patch.object(MockPatroni, 'dcs') as d:
d.manual_failover.return_value = False
MockRestApiServer(RestApiHandler, request)
# Exception: No timezone specified
request = 'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 97\n\n{"leader": ' +\
'"postgresql1", "member": "postgresql2", "scheduled_at": "6016-02-15T18:13:30.568224"}'
request = post + '97\n\n{"leader": "postgresql1", "member": "postgresql2",' +\
' "scheduled_at": "6016-02-15T18:13:30.568224"}'
MockRestApiServer(RestApiHandler, request)
# Exception: Scheduled in the past
request = 'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
'"postgresql1", "member": "postgresql2", "scheduled_at": "1016-02-15T18:13:30.568224+01:00"}'
MockRestApiServer(RestApiHandler, request)
request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2", "scheduled_at": "'
MockRestApiServer(RestApiHandler, request + '1016-02-15T18:13:30.568224+01:00"}')
# Invalid date
request = 'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
'"postgresql1", "member": "postgresql2", "scheduled_at": "2010-02-29T18:13:30.568224+01:00"}'
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request + '2010-02-29T18:13:30.568224+01:00"}'))
tests/test_config.py
+74
View File
@@ -0,0 +1,74 @@
import os
import unittest
import sys
from mock import MagicMock, Mock, patch
from patroni.config import Config
from six.moves import builtins
class TestConfig(unittest.TestCase):
@patch('os.path.isfile', Mock(return_value=True))
@patch('json.load', Mock(side_effect=Exception))
@patch.object(builtins, 'open', MagicMock())
def setUp(self):
sys.argv = ['patroni.py']
os.environ[Config.PATRONI_CONFIG_VARIABLE] = 'restapi: {}\npostgresql: {data_dir: foo}'
self.config = Config()
def test_no_config(self):
self.assertRaises(SystemExit, Config)
@patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception))
def test_set_dynamic_configuration(self):
self.assertIsNone(self.config.set_dynamic_configuration({'foo': 'bar'}))
def test_reload_local_configuration(self):
os.environ.update({
'PATRONI_NAME': 'postgres0',
'PATRONI_NAMESPACE': '/patroni/',
'PATRONI_SCOPE': 'batman2',
'PATRONI_RESTAPI_USERNAME': 'username',
'PATRONI_RESTAPI_PASSWORD': 'password',
'PATRONI_RESTAPI_LISTEN': '0.0.0.0:8008',
'PATRONI_RESTAPI_CONNECT_ADDRESS': '127.0.0.1:8008',
'PATRONI_RESTAPI_CERTFILE': '/certfile',
'PATRONI_RESTAPI_KEYFILE': '/keyfile',
'PATRONI_POSTGRESQL_LISTEN': '0.0.0.0:5432',
'PATRONI_POSTGRESQL_CONNECT_ADDRESS': '127.0.0.1:5432',
'PATRONI_POSTGRESQL_DATA_DIR': 'data/postgres0',
'PATRONI_POSTGRESQL_PGPASS': '/tmp/pgpass0',
'PATRONI_ETCD_HOST': '127.0.0.1:2379',
'PATRONI_CONSUL_HOST': '127.0.0.1:8500',
'PATRONI_ZOOKEEPER_HOSTS': 'host1,host2',
'PATRONI_EXHIBITOR_HOSTS': 'host1,host2',
'PATRONI_EXHIBITOR_PORT': '8181',
'PATRONI_foo_HOSTS': '[host1,host2', # Exception in parse_list
'PATRONI_SUPERUSER_USERNAME': 'postgres',
'PATRONI_SUPERUSER_PASSWORD': 'zalando',
'PATRONI_REPLICATION_USERNAME': 'replicator',
'PATRONI_REPLICATION_PASSWORD': 'rep-pass',
'PATRONI_admin_PASSWORD': 'admin',
'PATRONI_admin_OPTIONS': 'createrole,createdb'
})
sys.argv = ['patroni.py', 'postgres0.yml']
config = Config()
with patch.object(Config, '_load_config_file', Mock(return_value={'restapi': {}})):
with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
self.assertRaises(Exception, config.reload_local_configuration, True)
self.assertTrue(config.reload_local_configuration(True))
self.assertTrue(config.reload_local_configuration())
@patch('tempfile.mkstemp', Mock(return_value=[3000, 'blabla']))
@patch('os.path.exists', Mock(return_value=True))
@patch('os.remove', Mock(side_effect=IOError))
@patch('os.close', Mock(side_effect=IOError))
@patch('os.rename', Mock(return_value=None))
@patch('json.dump', Mock())
def test_save_cache(self):
self.config.set_dynamic_configuration({'ttl': 30, 'postgresql': {'foo': 'bar'}})
with patch('os.fdopen', Mock(side_effect=IOError)):
self.config.save_cache()
with patch('os.fdopen', MagicMock()):
self.config.save_cache()
tests/test_consul.py
+16 -3
View File
@@ -51,14 +51,15 @@ class TestConsul(unittest.TestCase):
@patch.object(consul.Consul.KV, 'get', kv_get)
@patch.object(consul.Consul.KV, 'delete', Mock())
def setUp(self):
self.c = Consul('postgresql1', {'ttl': 30, 'scope': 'test', 'host': 'localhost:1'})
self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10})
self.c._base_path = '/service/good'
self.c._load_cluster()
@patch('time.sleep', Mock(side_effect=SleepException))
def test_create_or_restore_session(self):
@patch.object(consul.Consul.Session, 'create', Mock(side_effect=ConsulException))
def test_create_session(self):
self.c._session = None
self.assertRaises(SleepException, self.c.create_or_restore_session)
self.assertRaises(SleepException, self.c.create_session)
@patch.object(consul.Consul.Session, 'renew', Mock(side_effect=NotFound))
@patch.object(consul.Consul.Session, 'create', Mock(side_effect=ConsulException))
@@ -96,6 +97,10 @@ class TestConsul(unittest.TestCase):
def test_set_failover_value(self):
self.c.set_failover_value('')
@patch.object(consul.Consul.KV, 'put', Mock(return_value=True))
def test_set_config_value(self):
self.c.set_config_value('')
@patch.object(consul.Consul.KV, 'put', Mock(side_effect=ConsulException))
def test_write_leader_optime(self):
self.c.write_leader_optime('')
@@ -125,3 +130,11 @@ class TestConsul(unittest.TestCase):
self.c.watch(1)
with patch.object(consul.Consul.KV, 'get', Mock(side_effect=ConsulException)):
self.c.watch(1)
@patch.object(consul.Consul.Session, 'destroy', Mock(side_effect=ConsulException))
def test_set_ttl(self):
self.c.set_ttl(20)
self.assertTrue(self.c.watch(1))
def test_set_retry_timeout(self):
self.c.set_retry_timeout(10)
tests/test_ctl.py
+1 -1
View File
@@ -51,7 +51,7 @@ class TestCtl(unittest.TestCase):
self.runner = CliRunner()
with patch.object(etcd.Client, 'machines') as mock_machines:
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'}}, 'foo')
self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}}, 'foo')
@patch('psycopg2.connect', psycopg2_connect)
def test_get_cursor(self):
tests/test_etcd.py
+8 -1
View File
@@ -78,6 +78,8 @@ def etcd_read(self, key, **kwargs):
raise etcd.EtcdKeyNotFound
response = {"action": "get", "node": {"key": "/service/batman5", "dir": True, "nodes": [
{"key": "/service/batman5/config", "value": '{"foo": "bar"}',
"modifiedIndex": 1582, "createdIndex": 1582},
{"key": "/service/batman5/failover", "value": "",
"modifiedIndex": 1582, "createdIndex": 1582},
{"key": "/service/batman5/initialize", "value": "postgresql0",
@@ -191,7 +193,8 @@ 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', {'namespace': '/patroni/', 'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'})
self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'})
def test_base_path(self):
self.assertEquals(self.etcd._base_path, '/patroni/test')
@@ -254,3 +257,7 @@ class TestEtcd(unittest.TestCase):
def test_other_exceptions(self):
self.etcd.retry = Mock(side_effect=AttributeError('foo'))
self.assertRaises(EtcdError, self.etcd.cancel_initialization)
def test_set_ttl(self):
self.etcd.set_ttl(20)
self.assertTrue(self.etcd.watch(1))
tests/test_exhibitor.py
+2 -1
View File
@@ -23,7 +23,8 @@ class TestExhibitor(unittest.TestCase):
@patch('requests.get', requests_get)
@patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
def setUp(self):
self.e = Exhibitor('foo', {'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test'})
self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test',
'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
@patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True))
def test_get_cluster(self):
tests/test_ha.py
+33 -6
View File
@@ -1,9 +1,11 @@
import etcd
import unittest
import datetime
import etcd
import os
import pytz
import unittest
from mock import Mock, MagicMock, patch
from patroni.config import Config
from patroni.dcs import Cluster, Failover, Leader, Member, get_dcs
from patroni.exceptions import DCSError, PostgresException
from patroni.ha import Ha
@@ -20,7 +22,7 @@ def false(*args, **kwargs):
def get_cluster(initialize, leader, members, failover):
return Cluster(initialize, leader, 10, members, failover)
return Cluster(initialize, None, leader, 10, members, failover)
def get_cluster_not_initialized_without_leader():
@@ -48,6 +50,27 @@ def get_cluster_initialized_with_only_leader(failover=None):
class MockPatroni(object):
def __init__(self, p, d):
os.environ[Config.PATRONI_CONFIG_VARIABLE] = """
restapi:
listen: 0.0.0.0:8008
bootstrap:
users:
replicator:
password: rep-pass
options:
- replication
postgresql:
name: foo
data_dir: data/postgresql0
pg_rewind:
username: postgres
password: postgres
zookeeper:
exhibitor:
hosts: [localhost]
port: 8181
"""
self.config = Config()
self.postgresql = p
self.dcs = d
self.api = Mock()
@@ -87,13 +110,17 @@ class TestHa(unittest.TestCase):
with patch.object(etcd.Client, 'machines') as mock_machines:
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
self.p = Postgresql({'name': 'postgresql0', 'scope': 'dummy', 'listen': '127.0.0.1:5432',
'data_dir': 'data/postgresql0', 'superuser': {}, 'admin': {},
'replication': {'username': '', 'password': '', 'network': ''}})
'data_dir': 'data/postgresql0', 'retry_timeout': 10,
'authentication': {'superuser': {'username': 'foo', 'password': 'bar'},
'replication': {'username': '', 'password': ''}},
'parameters': {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'foo': 'bar',
'hot_standby': 'on', 'max_wal_senders': 5, 'wal_keep_segments': 8}})
self.p.set_state('running')
self.p.set_role('replica')
self.p.check_replication_lag = true
self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
self.e = get_dcs('foo', {'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'}})
self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test',
'name': 'foo', 'retry_timeout': 10}})
self.ha = Ha(MockPatroni(self.p, self.e))
self.ha._async_executor.run_async = run_async
self.ha.old_cluster = self.e.get_cluster()
tests/test_patroni.py
+25 -14
View File
@@ -1,13 +1,12 @@
import etcd
import os
import sys
import time
import unittest
import yaml
from mock import Mock, patch
from patroni.api import RestApiServer
from patroni.async_executor import AsyncExecutor
from patroni.exceptions import DCSError
from patroni import Patroni, main as _main
from six.moves import BaseHTTPServer
from test_etcd import SleepException, etcd_read, etcd_write
@@ -18,29 +17,36 @@ from test_postgresql import Postgresql, psycopg2_connect
@patch('subprocess.call', Mock(return_value=0))
@patch('psycopg2.connect', psycopg2_connect)
@patch.object(Postgresql, 'write_pg_hba', Mock())
@patch.object(Postgresql, '_write_postgresql_conf', Mock())
@patch.object(Postgresql, 'write_recovery_conf', Mock())
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
@patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
@patch.object(AsyncExecutor, 'run', Mock())
@patch.object(etcd.Client, 'write', etcd_write)
@patch.object(etcd.Client, 'read', etcd_read)
class TestPatroni(unittest.TestCase):
@patch.object(etcd.Client, 'read', etcd_read)
def setUp(self):
RestApiServer._BaseServer__is_shut_down = Mock()
RestApiServer._BaseServer__shutdown_request = True
RestApiServer.socket = 0
with patch.object(etcd.Client, 'machines') as mock_machines:
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
with open('postgres0.yml', 'r') as f:
config = yaml.load(f)
self.p = Patroni(config)
sys.argv = ['patroni.py', 'postgres0.yml']
self.p = Patroni()
@patch('patroni.dcs.AbstractDCS.get_cluster', Mock(side_effect=[None, DCSError('foo'), None]))
def test_load_dynamic_configuration(self):
self.p.config._dynamic_configuration = {}
self.p.load_dynamic_configuration()
self.p.load_dynamic_configuration()
@patch('time.sleep', Mock(side_effect=SleepException))
@patch.object(etcd.Client, 'delete', Mock())
@patch.object(etcd.Client, 'machines')
def test_patroni_main(self, mock_machines):
with patch('subprocess.call', Mock(return_value=1)):
_main()
sys.argv = ['patroni.py', 'postgres0.yml']
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
@@ -48,19 +54,19 @@ class TestPatroni(unittest.TestCase):
self.assertRaises(SleepException, _main)
with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
_main()
sys.argv = ['patroni.py']
# read the content of the yaml configuration file into the environment variable
# in order to test how does patroni handle the configuration passed from the environment.
with open('postgres0.yml', 'r') as f:
os.environ[Patroni.PATRONI_CONFIG_VARIABLE] = f.read()
with patch.object(Patroni, 'run', Mock(side_effect=SleepException())):
self.assertRaises(SleepException, _main)
del os.environ[Patroni.PATRONI_CONFIG_VARIABLE]
@patch('patroni.config.Config.save_cache', Mock())
@patch('patroni.config.Config.reload_local_configuration', Mock(return_value=True))
def test_run(self):
self.p.sighup_handler()
self.p.ha.dcs.watch = Mock(side_effect=SleepException)
self.p.api.start = Mock()
self.p.config._dynamic_configuration = {}
self.assertRaises(SleepException, self.p.run)
with patch('patroni.config.Config.set_dynamic_configuration', Mock(return_value=True)):
self.assertRaises(SleepException, self.p.run)
with patch('patroni.postgresql.Postgresql.data_directory_empty', Mock(return_value=False)):
self.assertRaises(SleepException, self.p.run)
def test_schedule_next_run(self):
self.p.ha.dcs.watch = Mock(return_value=True)
@@ -82,3 +88,8 @@ class TestPatroni(unittest.TestCase):
self.assertIsNone(self.p.replicatefrom)
self.p.tags['replicatefrom'] = 'foo'
self.assertEqual(self.p.replicatefrom, 'foo')
def test_reload_config(self):
self.p.reload_config()
self.p.get_tags = Mock(side_effect=Exception)
self.p.reload_config()
tests/test_postgresql.py
+99 -56
View File
@@ -5,7 +5,7 @@ import shutil
import subprocess
import unittest
from mock import Mock, MagicMock, PropertyMock, patch
from mock import Mock, MagicMock, PropertyMock, patch, mock_open
from patroni.dcs import Cluster, Leader, Member
from patroni.exceptions import PostgresException, PostgresConnectionException
from patroni.postgresql import Postgresql
@@ -34,19 +34,14 @@ class MockCursor(object):
self.results = [(False, )]
elif sql.startswith('SELECT to_char(pg_postmaster_start_time'):
self.results = [('', True, '', '', '', '', False)]
elif sql.startswith('SELECT name, setting'):
self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
('search_path', 'public', None, 'string', 'user'),
('port', '5433', None, 'integer', 'postmaster'),
('listen_addresses', '*', None, 'string', 'postmaster'),
('autovacuum', 'on', None, 'bool', 'sighup')]
else:
self.results = [(
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
)]
self.results = [(None, None, None, None, None, None, None, None, None, None)]
def fetchone(self):
return self.results[0]
@@ -140,6 +135,13 @@ Data page checksum version: 0
"""
def postmaster_opts_string(*args, **kwargs):
return '/usr/local/pgsql/bin/postgres "-D" "data/postgresql0" "--listen_addresses=127.0.0.1" \
"--port=5432" "--hot_standby=on" "--wal_keep_segments=8" "--wal_level=hot_standby" \
"--archive_command=mkdir -p ../wal_archive && cp %p ../wal_archive/%f" "--wal_log_hints=on" \
"--max_wal_senders=5" "--archive_timeout=1800s" "--archive_mode=on" "--max_replication_slots=5"\n'
def psycopg2_connect(*args, **kwargs):
return MockConnect()
@@ -151,25 +153,27 @@ def fake_listdir(path):
@patch('subprocess.call', Mock(return_value=0))
@patch('psycopg2.connect', psycopg2_connect)
class TestPostgresql(unittest.TestCase):
_PARAMETERS = {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'f.oo': 'bar',
'search_path': 'public', 'hot_standby': 'on', 'max_wal_senders': 5,
'wal_keep_segments': 8, 'wal_log_hints': 'on', 'max_locks_per_transaction': 64,
'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 0}
@patch('subprocess.call', Mock(return_value=0))
@patch('psycopg2.connect', psycopg2_connect)
@patch('os.rename', Mock())
@patch.object(Postgresql, 'get_major_version', Mock(return_value=9.4))
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
def setUp(self):
self.data_dir = 'data/test0'
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir)
self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir,
self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir, 'retry_timeout': 10,
'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432',
'pg_hba': ['host replication replicator 127.0.0.1/32 md5',
'hostssl all all 0.0.0.0/0 md5',
'host all all 0.0.0.0/0 md5'],
'superuser': {'username': 'test', 'password': 'test'},
'admin': {'username': 'admin', 'password': 'admin'},
'pg_rewind': {'username': 'admin', 'password': 'admin'},
'replication': {'username': 'replicator',
'password': 'rep-pass'},
'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'},
'authentication': {'superuser': {'username': 'test', 'password': 'test'},
'replication': {'username': 'replicator', 'password': 'rep-pass'}},
'use_pg_rewind': True,
'parameters': self._PARAMETERS,
'recovery_conf': {'foo': 'bar'},
'callbacks': {'on_start': 'true', 'on_stop': 'true',
'on_restart': 'true', 'on_role_change': 'true',
'on_reload': 'true'
@@ -185,49 +189,40 @@ class TestPostgresql(unittest.TestCase):
shutil.rmtree('data')
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())
with open(os.path.join(self.data_dir, 'pg_hba.conf')) as f:
lines = f.readlines()
assert 'host replication replicator 127.0.0.1/32 md5\n' in lines
assert 'host all all 0.0.0.0/0 md5\n' in lines
self.assertEquals(self.p.get_initdb_options([{'encoding': 'UTF8'}, 'data-checksums']),
['--encoding=UTF8', '--data-checksums'])
self.assertRaises(Exception, self.p.get_initdb_options, [{'pgdata': 'bar'}])
self.assertRaises(Exception, self.p.get_initdb_options, [{'foo': 'bar', 1: 2}])
self.assertRaises(Exception, self.p.get_initdb_options, [1])
@patch('os.path.exists', Mock(return_value=True))
@patch('os.unlink', Mock())
def test_delete_trigger_file(self):
self.p.delete_trigger_file()
def test_start(self):
@patch.object(Postgresql, 'is_running')
def test_start(self, mock_is_running):
mock_is_running.return_value = True
self.assertTrue(self.p.start())
self.p.is_running = false
mock_is_running.return_value = False
open(os.path.join(self.data_dir, 'postmaster.pid'), 'w').close()
pg_conf = os.path.join(self.data_dir, 'postgresql.conf')
open(pg_conf, 'w').close()
self.assertTrue(self.p.start())
with open(pg_conf) as f:
lines = f.readlines()
self.assertTrue("foo = 'bar'\n" in lines)
self.assertTrue("f.oo = 'bar'\n" in lines)
def test_stop(self):
@patch.object(Postgresql, 'is_running')
def test_stop(self, mock_is_running):
mock_is_running.return_value = True
self.assertTrue(self.p.stop())
with patch('subprocess.call', Mock(return_value=1)):
mock_is_running.return_value = False
self.assertTrue(self.p.stop())
self.p.is_running = Mock(return_value=True)
self.assertFalse(self.p.stop())
def test_restart(self):
self.p.start = false
self.p.is_running = false
self.assertFalse(self.p.restart())
self.assertEquals(self.p.state, 'restart failed (restarting)')
@@ -247,6 +242,7 @@ class TestPostgresql(unittest.TestCase):
@patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1))
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
@patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
def test_follow(self, mock_pg_rewind):
self.p.follow(None, None)
self.p.follow(self.leader, self.leader)
@@ -267,10 +263,6 @@ class TestPostgresql(unittest.TestCase):
@patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
def test_can_rewind(self):
tmp = self.p.pg_rewind
self.p.pg_rewind = None
self.assertFalse(self.p.can_rewind)
self.p.pg_rewind = tmp
with mock.patch('subprocess.call', MagicMock(return_value=1)):
self.assertFalse(self.p.can_rewind)
with mock.patch('subprocess.call', side_effect=OSError):
@@ -298,9 +290,10 @@ class TestPostgresql(unittest.TestCase):
with patch('subprocess.call', Mock(side_effect=Exception("foo"))):
self.assertEquals(self.p.create_replica(self.leader), 1)
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
def test_sync_replication_slots(self):
self.p.start()
cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem], None)
cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None)
self.p.sync_replication_slots(cluster)
self.p.query = Mock(side_effect=psycopg2.OperationalError)
self.p.schedule_load_slots = True
@@ -326,9 +319,11 @@ class TestPostgresql(unittest.TestCase):
def test_reload(self):
self.assertTrue(self.p.reload())
def test_is_healthy(self):
@patch.object(Postgresql, 'is_running')
def test_is_healthy(self, mock_is_running):
mock_is_running.return_value = True
self.assertTrue(self.p.is_healthy())
self.p.is_running = false
mock_is_running.return_value = False
self.assertFalse(self.p.is_healthy())
def test_promote(self):
@@ -339,6 +334,13 @@ class TestPostgresql(unittest.TestCase):
def test_last_operation(self):
self.assertEquals(self.p.last_operation(), '0')
@patch('os.path.isfile', Mock(return_value=True))
@patch('os.kill', Mock(side_effect=Exception))
@patch.object(builtins, 'open', mock_open(read_data='-999999999999999'))
@patch.object(Postgresql, '_version_file_exists', Mock(return_value=True))
def test_is_running(self):
self.assertFalse(self.p.is_running())
@patch('subprocess.Popen', Mock(side_effect=OSError))
def test_call_nowait(self):
self.assertFalse(self.p.call_nowait('on_start'))
@@ -346,6 +348,7 @@ class TestPostgresql(unittest.TestCase):
def test_non_existing_callback(self):
self.assertFalse(self.p.call_nowait('foobar'))
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
def test_is_leader_exception(self):
self.p.start()
self.p.query = Mock(side_effect=psycopg2.OperationalError("not supported"))
@@ -357,15 +360,23 @@ class TestPostgresql(unittest.TestCase):
@patch('os.rename', Mock())
@patch('os.path.isdir', Mock(return_value=True))
def test_move_data_directory(self):
self.p.is_running = false
self.p.move_data_directory()
with patch('os.rename', Mock(side_effect=OSError)):
self.p.move_data_directory()
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
def test_bootstrap(self):
with patch('subprocess.call', Mock(return_value=1)):
self.assertRaises(PostgresException, self.p.bootstrap)
self.p.bootstrap()
self.assertRaises(PostgresException, self.p.bootstrap, {})
self.p.bootstrap({'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}},
'pg_hba': ['host replication replicator 127.0.0.1/32 md5',
'hostssl all all 0.0.0.0/0 md5',
'host all all 0.0.0.0/0 md5']})
with open(os.path.join(self.data_dir, 'pg_hba.conf')) as f:
lines = f.readlines()
assert 'host replication replicator 127.0.0.1/32 md5\n' in lines
assert 'host all all 0.0.0.0/0 md5\n' in lines
@patch('patroni.postgresql.Postgresql.create_replica', Mock(return_value=0))
def test_clone(self):
@@ -396,6 +407,18 @@ class TestPostgresql(unittest.TestCase):
with patch('subprocess.check_output', Mock(side_effect=subprocess.CalledProcessError(1, ''))):
self.assertEquals(self.p.controldata(), {})
def test_read_postmaster_opts(self):
m = mock_open(read_data=postmaster_opts_string())
with patch.object(builtins, 'open', m):
data = self.p.read_postmaster_opts()
self.assertEquals(data['wal_level'], 'hot_standby')
self.assertEquals(int(data['max_replication_slots']), 5)
self.assertEqual(data.get('D'), None)
m.side_effect = IOError
data = self.p.read_postmaster_opts()
self.assertEqual(data, dict())
@patch('subprocess.Popen')
@patch.object(builtins, 'open', MagicMock(return_value=42))
def test_single_user_mode(self, subprocess_popen_mock):
@@ -471,3 +494,23 @@ class TestPostgresql(unittest.TestCase):
self.assertTrue(self.p.replica_method_can_work_without_replication_connection('foo'))
self.p.config['foo'] = {'command': 'bar'}
self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foo'))
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
def test_reload_config(self):
parameters = self._PARAMETERS.copy()
parameters.pop('f.oo')
self.p.reload_config({'retry_timeout': 10, 'listen': '*', 'parameters': parameters})
parameters['b.ar'] = 'bar'
self.p.reload_config({'retry_timeout': 10, 'listen': '*', 'parameters': parameters})
parameters['autovacuum'] = 'on'
self.p.reload_config({'retry_timeout': 10, 'listen': '*', 'parameters': parameters})
parameters['autovacuum'] = 'off'
parameters.pop('search_path')
self.p.reload_config({'retry_timeout': 10, 'listen': '*:5433', 'parameters': parameters})
@patch.object(Postgresql, '_version_file_exists', Mock(return_value=True))
def test_get_major_version(self):
with patch.object(builtins, 'open', mock_open(read_data='9.4')):
self.assertEquals(self.p.get_major_version(), 9.4)
with patch.object(builtins, 'open', Mock(side_effect=Exception)):
self.assertEquals(self.p.get_major_version(), 0.0)
tests/test_zookeeper.py
+17 -2
View File
@@ -13,6 +13,9 @@ class MockKazooClient(Mock):
leader = False
exists = True
def __init__(self, *args, **kwargs):
super(MockKazooClient, self).__init__()
@property
def client_id(self):
return (-1, '')
@@ -69,7 +72,7 @@ class MockKazooClient(Mock):
raise Exception
if path == '/service/test/members/bar' and value == b'retry':
return
if path == '/service/test/failover':
if path in ('/service/test/failover', '/service/test/config'):
if value == b'Exception':
raise Exception
elif value == b'ok':
@@ -93,11 +96,18 @@ class TestZooKeeper(unittest.TestCase):
@patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
def setUp(self):
self.zk = ZooKeeper('foo', {'hosts': ['localhost:2181'], 'scope': 'test'})
self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test',
'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
def test_session_listener(self):
self.zk.session_listener(KazooState.SUSPENDED)
def test_set_ttl(self):
self.zk.set_ttl(20)
def test_set_retry_timeout(self):
self.zk.set_retry_timeout(10)
def test_get_node(self):
self.assertIsNone(self.zk.get_node('/no_node'))
@@ -124,6 +134,11 @@ class TestZooKeeper(unittest.TestCase):
self.zk.set_failover_value('ok')
self.zk.set_failover_value('Exception')
def test_set_config_value(self):
self.zk.set_config_value('')
self.zk.set_config_value('ok')
self.zk.set_config_value('Exception')
def test_initialize(self):
self.assertFalse(self.zk.initialize())