diff --git a/README.rst b/README.rst index 922f1e25..edf0357e 100644 --- a/README.rst +++ b/README.rst @@ -75,7 +75,13 @@ run: YAML Configuration =============== -Go `here `__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml `__. +Go `here `__ for comprehensive information about settings for etcd, consul, and ZooKeeper. And for an example, see `postgres0.yml `__. + +========================= +Environment Configuration +========================= + +Go `here `__ for comprehensive information about configuring(overriding) settings via environment variables. =============== Replication Choices diff --git a/SETTINGS.rst b/SETTINGS.rst deleted file mode 100644 index 8e240989..00000000 --- a/SETTINGS.rst +++ /dev/null @@ -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. - diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst new file mode 100644 index 00000000..1db8dadf --- /dev/null +++ b/docs/ENVIRONMENT.rst @@ -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\_\_PASSWORD=''** +- **PATRONI\_\_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 `__ 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...' diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst new file mode 100644 index 00000000..86f21d5c --- /dev/null +++ b/docs/SETTINGS.rst @@ -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 `///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 `__ 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...']. diff --git a/features/environment.py b/features/environment.py index a8d2f408..3fc52a2d 100644 --- a/features/environment.py +++ b/features/environment.py @@ -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): diff --git a/features/patroni_api.feature b/features/patroni_api.feature index 31bbbdb2..6bbdbeb1 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -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 - diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index d94086b5..90c38564 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -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) diff --git a/patroni/__init__.py b/patroni/__init__.py index 70c7fdfb..c48eab40 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -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: diff --git a/patroni/api.py b/patroni/api.py index 526da8b4..7000fc2f 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -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() diff --git a/patroni/config.py b/patroni/config.py new file mode 100644 index 00000000..e749612e --- /dev/null +++ b/patroni/config.py @@ -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__PASSWORD=, PATRONI__OPTIONS= + # CREATE USER "" WITH 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] diff --git a/patroni/ctl.py b/patroni/ctl.py index 412fa354..499f6563 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -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) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index ceffecbc..c068ef8e 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -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` """ diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index a033aeda..887787f2 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -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 diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 47f0b636..7b2b7477 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -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: diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py index e5053809..56105af1 100644 --- a/patroni/dcs/exhibitor.py +++ b/patroni/dcs/exhibitor.py @@ -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(): diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 9ec12af3..dae08ffb 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -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")) diff --git a/patroni/ha.py b/patroni/ha.py index b12aa856..0352b270 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -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() diff --git a/patroni/postgresql.py b/patroni/postgresql.py index f357208f..aa3adf0c 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -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") diff --git a/patroni/utils.py b/patroni/utils.py index 0969d458..1b604406 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -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 diff --git a/postgres0.yml b/postgres0.yml index 067ab743..cb1000c5 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -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:///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 diff --git a/requirements.txt b/requirements.txt index 57ed4ea9..8b9b3c79 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/tests/test_api.py b/tests/test_api.py index 5871d240..8483299b 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -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"}')) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 00000000..a8a5a6c4 --- /dev/null +++ b/tests/test_config.py @@ -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() diff --git a/tests/test_consul.py b/tests/test_consul.py index 22665174..32e27462 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -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) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index c9bd1258..eb63e153 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -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): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 508cd480..ebdf6aa2 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -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)) diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py index be60e562..d9c0d6d3 100644 --- a/tests/test_exhibitor.py +++ b/tests/test_exhibitor.py @@ -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): diff --git a/tests/test_ha.py b/tests/test_ha.py index 5ceb3392..4f691212 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -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() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index d9daaedc..f9246cac 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -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() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 2c03faa9..7fa7279a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -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) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 163d999c..01555d8e 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -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())