diff --git a/.gitignore b/.gitignore index 23cbf17a..c902c6eb 100644 --- a/.gitignore +++ b/.gitignore @@ -33,7 +33,7 @@ nosetests.xml coverage.xml htmlcov junit.xml -features/output +features/output* dummy # Translations diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index 00595cf7..e3762aca 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -115,7 +115,7 @@ Kubernetes - **PATRONI\_KUBERNETES\_ROLE\_LABEL**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``. - **PATRONI\_KUBERNETES\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `master`. Default value is `master`. - **PATRONI\_KUBERNETES\_FOLLOWER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is `replica`. Default value is `replica`. -- **PATRONI\_KUBERNETES\_STANDBY\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``. +- **PATRONI\_KUBERNETES\_STANDBY\_LEADER\_LABEL\_VALUE**: (optional) value of the pod label when Postgres role is ``standby_leader``. Default value is ``master``. - **PATRONI\_KUBERNETES\_TMP\_ROLE\_LABEL**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary. - **PATRONI\_KUBERNETES\_USE\_ENDPOINTS**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state. - **PATRONI\_KUBERNETES\_POD\_IP**: (optional) IP address of the pod Patroni is running in. This value is required when `PATRONI_KUBERNETES_USE_ENDPOINTS` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted. diff --git a/docs/conf.py b/docs/conf.py index 2228c72e..f94b9660 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -54,6 +54,13 @@ apidoc_output_dir = 'modules' apidoc_excluded_paths = excludes apidoc_separate_modules = True +# Include autodoc for all members, including private ones and the ones that are missing a docstring. +autodoc_default_options = { + "members": True, + "undoc-members": True, + "private-members": True, +} + # Add any paths that contain templates here, relative to this directory. templates_path = ['_templates'] @@ -280,6 +287,14 @@ def doctree_read(app, doctree): toc_tree_node['entries'].remove(e) +def autodoc_skip(app, what, name, obj, would_skip, options): + """Include autodoc of ``__init__`` methods, which are skipped by default.""" + if name == "__init__": + return False + return would_skip + + + # A possibility to have an own stylesheet, to add new rules or override existing ones # For the latter case, the CSS specificity of the rules should be higher than the default ones def setup(app): @@ -292,3 +307,4 @@ def setup(app): app.connect('builder-inited', builder_inited) app.connect('env-get-outdated', env_get_outdated) app.connect('doctree-read', doctree_read) + app.connect("autodoc-skip-member", autodoc_skip) diff --git a/docs/rest_api.rst b/docs/rest_api.rst index 6e97ade0..e49f6ee0 100644 --- a/docs/rest_api.rst +++ b/docs/rest_api.rst @@ -131,7 +131,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -178,7 +179,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -223,7 +225,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -267,7 +270,8 @@ The ``GET /patroni`` is used by Patroni during the leader race. It also could be "database_system_identifier": "7268616322854375442", "patroni": { "version": "3.1.0", - "scope": "demo" + "scope": "demo", + "name": "patroni1" } } @@ -279,70 +283,70 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e # HELP patroni_version Patroni semver without periods. \ # TYPE patroni_version gauge - patroni_version{scope="batman"} 020103 + patroni_version{scope="batman",name="patroni1"} 020103 # HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise. # TYPE patroni_postgres_running gauge - patroni_postgres_running{scope="batman"} 1 + patroni_postgres_running{scope="batman",name="patroni1"} 1 # HELP patroni_postmaster_start_time Epoch seconds since Postgres started. # TYPE patroni_postmaster_start_time gauge - patroni_postmaster_start_time{scope="batman"} 1657656955.179243 + patroni_postmaster_start_time{scope="batman",name="patroni1"} 1657656955.179243 # HELP patroni_master Value is 1 if this node is the leader, 0 otherwise. # TYPE patroni_master gauge - patroni_master{scope="batman"} 1 + patroni_master{scope="batman",name="patroni1"} 1 # HELP patroni_primary Value is 1 if this node is the leader, 0 otherwise. # TYPE patroni_primary gauge - patroni_primary{scope="batman"} 1 + patroni_primary{scope="batman",name="patroni1"} 1 # HELP patroni_xlog_location Current location of the Postgres transaction log, 0 if this node is not the leader. # TYPE patroni_xlog_location counter - patroni_xlog_location{scope="batman"} 22320573386952 + patroni_xlog_location{scope="batman",name="patroni1"} 22320573386952 # HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise. # TYPE patroni_standby_leader gauge - patroni_standby_leader{scope="batman"} 0 + patroni_standby_leader{scope="batman",name="patroni1"} 0 # HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise. # TYPE patroni_replica gauge - patroni_replica{scope="batman"} 0 + patroni_replica{scope="batman",name="patroni1"} 0 # HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise. # TYPE patroni_sync_standby gauge - patroni_sync_standby{scope="batman"} 0 + patroni_sync_standby{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_received_location Current location of the received Postgres transaction log, 0 if this node is not a replica. # TYPE patroni_xlog_received_location counter - patroni_xlog_received_location{scope="batman"} 0 + patroni_xlog_received_location{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_replayed_location Current location of the replayed Postgres transaction log, 0 if this node is not a replica. # TYPE patroni_xlog_replayed_location counter - patroni_xlog_replayed_location{scope="batman"} 0 + patroni_xlog_replayed_location{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed Postgres transaction log, 0 if null. # TYPE patroni_xlog_replayed_timestamp gauge - patroni_xlog_replayed_timestamp{scope="batman"} 0 + patroni_xlog_replayed_timestamp{scope="batman",name="patroni1"} 0 # HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise. # TYPE patroni_xlog_paused gauge - patroni_xlog_paused{scope="batman"} 0 + patroni_xlog_paused{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise. # TYPE patroni_postgres_streaming gauge - patroni_postgres_streaming{scope="batman"} 1 + patroni_postgres_streaming{scope="batman",name="patroni1"} 1 # HELP patroni_postgres_in_archive_recovery Value is 1 if Postgres is replicating from archive, 0 otherwise. # TYPE patroni_postgres_in_archive_recovery gauge - patroni_postgres_in_archive_recovery{scope="batman"} 0 + patroni_postgres_in_archive_recovery{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise. # TYPE patroni_postgres_server_version gauge - patroni_postgres_server_version {scope="batman"} 140004 + patroni_postgres_server_version{scope="batman",name="patroni1"} 140004 # HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked. # TYPE patroni_cluster_unlocked gauge - patroni_cluster_unlocked{scope="batman"} 0 + patroni_cluster_unlocked{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise. # TYPE patroni_postgres_timeline counter - patroni_failsafe_mode_is_active{scope="batman"} 0 + patroni_failsafe_mode_is_active{scope="batman",name="patroni1"} 0 # HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise. # TYPE patroni_postgres_timeline counter - patroni_postgres_timeline{scope="batman"} 24 + patroni_postgres_timeline{scope="batman",name="patroni1"} 24 # HELP patroni_dcs_last_seen Epoch timestamp when DCS was last contacted successfully by Patroni. # TYPE patroni_dcs_last_seen gauge - patroni_dcs_last_seen{scope="batman"} 1677658321 + patroni_dcs_last_seen{scope="batman",name="patroni1"} 1677658321 # HELP patroni_pending_restart Value is 1 if the node needs a restart, 0 otherwise. # TYPE patroni_pending_restart gauge - patroni_pending_restart{scope="batman"} 1 + patroni_pending_restart{scope="batman",name="patroni1"} 1 # HELP patroni_is_paused Value is 1 if auto failover is disabled, 0 otherwise. # TYPE patroni_is_paused gauge - patroni_is_paused{scope="batman"} 1 + patroni_is_paused{scope="batman",name="patroni1"} 1 Cluster status endpoints @@ -381,6 +385,7 @@ Cluster status endpoints "lag": 0 } ], + "scope": "demo", "scheduled_switchover": { "at": "2023-09-24T10:36:00+02:00", "from": "patroni1", @@ -489,8 +494,9 @@ Let's check that the node processed this configuration. First of all it should s "location": 2197818976 }, "patroni": { + "version": "1.0", "scope": "batman", - "version": "1.0" + "name": "patroni1" }, "state": "running", "role": "master", diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst index f0cb8ef4..d9283192 100644 --- a/docs/yaml_configuration.rst +++ b/docs/yaml_configuration.rst @@ -171,7 +171,7 @@ Kubernetes - **role\_label**: (optional) name of the label containing role (master or replica or other custom value). Patroni will set this label on the pod it runs in. Default value is ``role``. - **leader\_label\_value**: (optional) value of the pod label when Postgres role is ``master``. Default value is ``master``. - **follower\_label\_value**: (optional) value of the pod label when Postgres role is ``replica``. Default value is ``replica``. -- **standby\_leader\_label\_value**: (optional) value of the pod label when Postgres role is ``standby-leader``. Default value is ``standby-leader``. +- **standby\_leader\_label\_value**: (optional) value of the pod label when Postgres role is ``standby_leader``. Default value is ``master``. - **tmp_\role\_label**: (optional) name of the temporary label containing role (master or replica). Value of this label will always use the default of corresponding role. Set only when necessary. - **use\_endpoints**: (optional) if set to true, Patroni will use Endpoints instead of ConfigMaps to run leader elections and keep cluster state. - **pod\_ip**: (optional) IP address of the pod Patroni is running in. This value is required when `use_endpoints` is enabled and is used to populate the leader endpoint subsets when the pod's PostgreSQL is promoted. diff --git a/features/Dockerfile b/features/Dockerfile index 86c6041d..7f52c2af 100644 --- a/features/Dockerfile +++ b/features/Dockerfile @@ -27,8 +27,8 @@ RUN set -ex \ && apt-get update \ && apt-get reinstall init-system-helpers \ && apt-get install -y \ - python3-pip \ python3-dev \ + python3-venv \ rsync \ curl \ gcc \ @@ -40,7 +40,9 @@ RUN set -ex \ net-tools \ iputils-ping \ && rm -rf /var/cache/apt \ - && python3 -m pip install --no-cache-dir tox \ + \ + && python3 -m venv /tox \ + && /tox/bin/pip install --no-cache-dir tox>=4 \ \ && mkdir -p "$PGHOME" \ && sed -i "s|/var/lib/postgresql.*|$PGHOME:/bin/bash|" /etc/passwd \ @@ -50,6 +52,7 @@ RUN set -ex \ && curl -sL "$ETCDURL/etcd-v$ETCDVERSION-linux-$(dpkg --print-architecture).tar.gz" \ | tar xz -C /usr/local/bin --strip=1 --wildcards --no-anchored etcd etcdctl +ENV PATH="/tox/bin:$PATH" # This Dockerfile syntax only works with docker buildx and the syntax # line at the top of this file. diff --git a/patroni/api.py b/patroni/api.py index b01c24d8..c5320506 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -198,7 +198,11 @@ class RestApiHandler(BaseHTTPRequestHandler): 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} + response['patroni'] = { + 'version': patroni.version, + 'scope': patroni.postgresql.scope, + 'name': patroni.postgresql.name + } if patroni.scheduled_restart: response['scheduled_restart'] = patroni.scheduled_restart.copy() del response['scheduled_restart']['postmaster_start_time'] @@ -449,7 +453,10 @@ class RestApiHandler(BaseHTTPRequestHandler): """ cluster = self.server.patroni.dcs.get_cluster(True) global_config = self.server.patroni.config.get_global_config(cluster) - self._write_json_response(200, cluster_as_json(cluster, global_config)) + + response = cluster_as_json(cluster, global_config) + response['scope'] = self.server.patroni.postgresql.scope + self._write_json_response(200, response) def do_GET_history(self) -> None: """Handle a ``GET`` request to ``/history`` path. @@ -526,113 +533,113 @@ class RestApiHandler(BaseHTTPRequestHandler): metrics: List[str] = [] - scope_label = '{{scope="{0}"}}'.format(patroni.postgresql.scope) + labels = f'{{scope="{patroni.postgresql.scope}",name="{patroni.postgresql.name}"}}' metrics.append("# HELP patroni_version Patroni semver without periods.") metrics.append("# TYPE patroni_version gauge") padded_semver = ''.join([x.zfill(2) for x in patroni.version.split('.')]) # 2.0.2 => 020002 - metrics.append("patroni_version{0} {1}".format(scope_label, padded_semver)) + metrics.append("patroni_version{0} {1}".format(labels, padded_semver)) metrics.append("# HELP patroni_postgres_running Value is 1 if Postgres is running, 0 otherwise.") metrics.append("# TYPE patroni_postgres_running gauge") - metrics.append("patroni_postgres_running{0} {1}".format(scope_label, int(postgres['state'] == 'running'))) + metrics.append("patroni_postgres_running{0} {1}".format(labels, int(postgres['state'] == 'running'))) metrics.append("# HELP patroni_postmaster_start_time Epoch seconds since Postgres started.") metrics.append("# TYPE patroni_postmaster_start_time gauge") postmaster_start_time = postgres.get('postmaster_start_time') postmaster_start_time = (postmaster_start_time - epoch).total_seconds() if postmaster_start_time else 0 - metrics.append("patroni_postmaster_start_time{0} {1}".format(scope_label, postmaster_start_time)) + metrics.append("patroni_postmaster_start_time{0} {1}".format(labels, postmaster_start_time)) metrics.append("# HELP patroni_master Value is 1 if this node is the leader, 0 otherwise.") metrics.append("# TYPE patroni_master gauge") - metrics.append("patroni_master{0} {1}".format(scope_label, int(postgres['role'] in ('master', 'primary')))) + metrics.append("patroni_master{0} {1}".format(labels, int(postgres['role'] in ('master', 'primary')))) metrics.append("# HELP patroni_primary Value is 1 if this node is the leader, 0 otherwise.") metrics.append("# TYPE patroni_primary gauge") - metrics.append("patroni_primary{0} {1}".format(scope_label, int(postgres['role'] in ('master', 'primary')))) + metrics.append("patroni_primary{0} {1}".format(labels, int(postgres['role'] in ('master', 'primary')))) metrics.append("# HELP patroni_xlog_location Current location of the Postgres" " transaction log, 0 if this node is not the leader.") metrics.append("# TYPE patroni_xlog_location counter") - metrics.append("patroni_xlog_location{0} {1}".format(scope_label, postgres.get('xlog', {}).get('location', 0))) + metrics.append("patroni_xlog_location{0} {1}".format(labels, postgres.get('xlog', {}).get('location', 0))) metrics.append("# HELP patroni_standby_leader Value is 1 if this node is the standby_leader, 0 otherwise.") metrics.append("# TYPE patroni_standby_leader gauge") - metrics.append("patroni_standby_leader{0} {1}".format(scope_label, int(postgres['role'] == 'standby_leader'))) + metrics.append("patroni_standby_leader{0} {1}".format(labels, int(postgres['role'] == 'standby_leader'))) metrics.append("# HELP patroni_replica Value is 1 if this node is a replica, 0 otherwise.") metrics.append("# TYPE patroni_replica gauge") - metrics.append("patroni_replica{0} {1}".format(scope_label, int(postgres['role'] == 'replica'))) + metrics.append("patroni_replica{0} {1}".format(labels, int(postgres['role'] == 'replica'))) metrics.append("# HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise.") metrics.append("# TYPE patroni_sync_standby gauge") - metrics.append("patroni_sync_standby{0} {1}".format(scope_label, int(postgres.get('sync_standby', False)))) + metrics.append("patroni_sync_standby{0} {1}".format(labels, int(postgres.get('sync_standby', False)))) metrics.append("# HELP patroni_xlog_received_location Current location of the received" " Postgres transaction log, 0 if this node is not a replica.") metrics.append("# TYPE patroni_xlog_received_location counter") metrics.append("patroni_xlog_received_location{0} {1}" - .format(scope_label, postgres.get('xlog', {}).get('received_location', 0))) + .format(labels, postgres.get('xlog', {}).get('received_location', 0))) metrics.append("# HELP patroni_xlog_replayed_location Current location of the replayed" " Postgres transaction log, 0 if this node is not a replica.") metrics.append("# TYPE patroni_xlog_replayed_location counter") metrics.append("patroni_xlog_replayed_location{0} {1}" - .format(scope_label, postgres.get('xlog', {}).get('replayed_location', 0))) + .format(labels, postgres.get('xlog', {}).get('replayed_location', 0))) metrics.append("# HELP patroni_xlog_replayed_timestamp Current timestamp of the replayed" " Postgres transaction log, 0 if null.") metrics.append("# TYPE patroni_xlog_replayed_timestamp gauge") replayed_timestamp = postgres.get('xlog', {}).get('replayed_timestamp') replayed_timestamp = (replayed_timestamp - epoch).total_seconds() if replayed_timestamp else 0 - metrics.append("patroni_xlog_replayed_timestamp{0} {1}".format(scope_label, replayed_timestamp)) + metrics.append("patroni_xlog_replayed_timestamp{0} {1}".format(labels, replayed_timestamp)) metrics.append("# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.") metrics.append("# TYPE patroni_xlog_paused gauge") metrics.append("patroni_xlog_paused{0} {1}" - .format(scope_label, int(postgres.get('xlog', {}).get('paused', False) is True))) + .format(labels, int(postgres.get('xlog', {}).get('paused', False) is True))) if postgres.get('server_version', 0) >= 90600: metrics.append("# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.") metrics.append("# TYPE patroni_postgres_streaming gauge") metrics.append("patroni_postgres_streaming{0} {1}" - .format(scope_label, int(postgres.get('replication_state') == 'streaming'))) + .format(labels, int(postgres.get('replication_state') == 'streaming'))) metrics.append("# HELP patroni_postgres_in_archive_recovery Value is 1" " if Postgres is replicating from archive, 0 otherwise.") metrics.append("# TYPE patroni_postgres_in_archive_recovery gauge") metrics.append("patroni_postgres_in_archive_recovery{0} {1}" - .format(scope_label, int(postgres.get('replication_state') == 'in archive recovery'))) + .format(labels, int(postgres.get('replication_state') == 'in archive recovery'))) metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.") metrics.append("# TYPE patroni_postgres_server_version gauge") - metrics.append("patroni_postgres_server_version {0} {1}".format(scope_label, postgres.get('server_version', 0))) + metrics.append("patroni_postgres_server_version {0} {1}".format(labels, postgres.get('server_version', 0))) metrics.append("# HELP patroni_cluster_unlocked Value is 1 if the cluster is unlocked, 0 if locked.") metrics.append("# TYPE patroni_cluster_unlocked gauge") - metrics.append("patroni_cluster_unlocked{0} {1}".format(scope_label, int(postgres.get('cluster_unlocked', 0)))) + metrics.append("patroni_cluster_unlocked{0} {1}".format(labels, int(postgres.get('cluster_unlocked', 0)))) metrics.append("# HELP patroni_failsafe_mode_is_active Value is 1 if failsafe mode is active, 0 if inactive.") metrics.append("# TYPE patroni_failsafe_mode_is_active gauge") metrics.append("patroni_failsafe_mode_is_active{0} {1}" - .format(scope_label, int(postgres.get('failsafe_mode_is_active', 0)))) + .format(labels, int(postgres.get('failsafe_mode_is_active', 0)))) metrics.append("# HELP patroni_postgres_timeline Postgres timeline of this node (if running), 0 otherwise.") metrics.append("# TYPE patroni_postgres_timeline counter") - metrics.append("patroni_postgres_timeline{0} {1}".format(scope_label, postgres.get('timeline', 0))) + metrics.append("patroni_postgres_timeline{0} {1}".format(labels, postgres.get('timeline', 0))) metrics.append("# HELP patroni_dcs_last_seen Epoch timestamp when DCS was last contacted successfully" " by Patroni.") metrics.append("# TYPE patroni_dcs_last_seen gauge") - metrics.append("patroni_dcs_last_seen{0} {1}".format(scope_label, postgres.get('dcs_last_seen', 0))) + metrics.append("patroni_dcs_last_seen{0} {1}".format(labels, postgres.get('dcs_last_seen', 0))) metrics.append("# HELP patroni_pending_restart Value is 1 if the node needs a restart, 0 otherwise.") metrics.append("# TYPE patroni_pending_restart gauge") metrics.append("patroni_pending_restart{0} {1}" - .format(scope_label, int(patroni.postgresql.pending_restart))) + .format(labels, int(patroni.postgresql.pending_restart))) metrics.append("# HELP patroni_is_paused Value is 1 if auto failover is disabled, 0 otherwise.") metrics.append("# TYPE patroni_is_paused gauge") - metrics.append("patroni_is_paused{0} {1}".format(scope_label, int(postgres.get('pause', 0)))) + metrics.append("patroni_is_paused{0} {1}".format(labels, int(postgres.get('pause', 0)))) self.write_response(200, '\n'.join(metrics) + '\n', content_type='text/plain') @@ -1368,6 +1375,9 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): def query(self, sql: str, *params: Any) -> List[Tuple[Any, ...]]: """Execute *sql* query with *params* and optionally return results. + .. note:: + Prefer to use own connection to postgres and fallback to ``heartbeat`` when own isn't available. + :param sql: the SQL statement to be run. :param params: positional arguments to be used as parameters for *sql*. @@ -1377,10 +1387,21 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): :class:`psycopg.Error`: if had issues while executing *sql*. :class:`~patroni.exceptions.PostgresConnectionException`: if had issues while connecting to the database. """ + # We first try to get a heartbeat connection because it is always required for the main thread. try: - return self.patroni.postgresql.query(sql, *params, retry=False) - except RetryFailedError as e: - raise PostgresConnectionException(str(e)) + heartbeat_connection = self.patroni.postgresql.connection_pool.get('heartbeat') + heartbeat_connection.get() # try to open psycopg connection to postgres + except psycopg.Error as exc: + raise PostgresConnectionException('connection problems') from exc + + try: + connection = self.patroni.postgresql.connection_pool.get('restapi') + connection.get() # try to open psycopg connection to postgres + except psycopg.Error: + logger.debug('restapi connection to postgres is not available') + connection = heartbeat_connection + + return connection.query(sql, *params) @staticmethod def _set_fd_cloexec(fd: socket.socket) -> None: @@ -1546,7 +1567,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): :param listen: IP and port to bind REST API to. It should be a string in the format ``host:port``, where ``host`` can be a hostname or IP address. It is the value of ``restapi.listen`` setting. :param ssl_options: dictionary that may contain the following keys, depending on what has been configured in - ``restapi` section: + ``restapi`` section: * ``certfile``: path to PEM certificate. If given, will start in HTTPS mode; * ``keyfile``: path to key of ``certfile``; diff --git a/patroni/config.py b/patroni/config.py index 55c8f4bf..65991b0e 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -257,7 +257,7 @@ class Config(object): * file or directory path passed as command-line argument (*configfile*), if it exists and the file or files found in the directory can be parsed (see :meth:`~Config._load_config_path`), otherwise - * YAML file passed via the environment variable (see :cvar:`PATRONI_CONFIG_VARIABLE`), if the referenced + * YAML file passed via the environment variable (see :attr:`PATRONI_CONFIG_VARIABLE`), if the referenced file exists and can be parsed, otherwise * from configuration values defined as environment variables, see :meth:`~Config._build_environment_configuration`. diff --git a/patroni/ctl.py b/patroni/ctl.py index b1d63d2b..3f49130e 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -264,7 +264,9 @@ role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 's @click.option('-k', '--insecure', is_flag=True, help='Allow connections to SSL sites without certs') @click.pass_context def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: bool) -> None: - """Entry point of ``patronictl`` utility. + """Command-line interface for interacting with Patroni. + \f + Entry point of ``patronictl`` utility. Load the configuration file. @@ -1561,9 +1563,11 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, rows.append([member.get(n.lower().replace(' ', '_'), '') for n in columns]) title = 'Citus cluster' if is_citus_cluster else 'Cluster' - group_title = '' if group is None else 'group: {0}, '.format(group) - title_details = group_title and ' ({0}{1})'.format(group_title, initialize) - title = ' {0}: {1}{2} '.format(title, name, title_details) + title_details = f' ({initialize})' + if is_citus_cluster: + title_details = '' if group is None else f' (group: {group}, {initialize})' + + title = f' {title}: {name}{title_details} ' print_output(columns, rows, {'Group': 'r', 'Lag in MB': 'r', 'TL': 'r'}, fmt, title) if fmt not in ('pretty', 'topology'): # Omit service info when using machine-readable formats diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index d28a59ce..4cef65ea 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -903,8 +903,16 @@ class Cluster(NamedTuple('Cluster', @property def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]: - """Dictionary of permanent replication slots.""" - return self.config and self.config.permanent_slots or {} + """Dictionary of permanent replication slots with their known LSN.""" + ret = deepcopy(self.config.permanent_slots if self.config else {}) + # If primary reported flush LSN for permanent slots we want to enrich our structure with it + for name, lsn in (self.slots or {}).items(): + if name in ret: + if not ret[name]: + ret[name] = {} + if isinstance(ret[name], dict): + ret[name]['lsn'] = lsn + return ret @property def __permanent_physical_slots(self) -> Dict[str, Any]: @@ -929,7 +937,6 @@ class Cluster(NamedTuple('Cluster', Will log an error if: - * Conflicting slot names between members are found * Any logical slots are disabled, due to version compatibility, and *show_error* is ``True``. :param my_name: name of this node. @@ -942,21 +949,9 @@ class Cluster(NamedTuple('Cluster', :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks. """ - slot_members: List[str] = self._get_slot_members(my_name, role) - - slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'} - for name in slot_members} - - if len(slots) < len(slot_members): - # Find which names are conflicting for a nicer error message - slot_conflicts: Dict[str, List[str]] = defaultdict(list) - for name in slot_members: - slot_conflicts[slot_name_from_member_name(name)].append(name) - logger.error("Following cluster members share a replication slot name: %s", - "; ".join(f"{', '.join(v)} map to {k}" - for k, v in slot_conflicts.items() if len(v) > 1)) - + slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role) permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover) + disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots( slots, permanent_slots, my_name, major_version) @@ -1016,7 +1011,7 @@ class Cluster(NamedTuple('Cluster', return disabled_permanent_logical_slots def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]: - """Get configured permanent slot names. + """Get configured permanent replication slots. .. note:: Permanent replication slots are only considered if ``use_slots`` configuration is enabled. @@ -1042,35 +1037,48 @@ class Cluster(NamedTuple('Cluster', return self.__permanent_slots if role in ('master', 'primary') else self.__permanent_logical_slots - def _get_slot_members(self, my_name: str, role: str) -> List[str]: - """Get a list of member names that have replication slots sourcing from this node. + def _get_members_slots(self, my_name: str, role: str) -> Dict[str, Dict[str, str]]: + """Get physical replication slots configuration for members that sourcing from this node. If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on the current primary, because that member would replicate from elsewhere. We still create the slot if the ``replicatefrom`` destination member is currently not a member of the cluster (fallback to the primary), or if ``replicatefrom`` destination member happens to be the current primary. + Will log an error if: + + * Conflicting slot names between members are found + :param my_name: name of this node. :param role: role of this node, if this is a ``primary`` or ``standby_leader`` return list of members replicating from this node. If not then return a list of members replicating as cascaded replicas from this node. - :returns: list of member names. + :returns: dictionary of physical replication slots that should exist on a given node. """ if not self.use_slots: - return [] + return {} + + # we always want to exclude the member with our name from the list + members = filter(lambda m: m.name != my_name, self.members) if role in ('master', 'primary', 'standby_leader'): - slot_members = [m.name for m in self.members - if m.name != my_name - and (m.replicatefrom is None - or m.replicatefrom == my_name - or not self.has_member(m.replicatefrom))] + members = [m for m in members if m.replicatefrom is None + or m.replicatefrom == my_name or not self.has_member(m.replicatefrom)] else: # only manage slots for replicas that replicate from this one, except for the leader among them - slot_members = [m.name for m in self.members - if m.replicatefrom == my_name and m.name != self.leader_name] - return slot_members + members = [m for m in members if m.replicatefrom == my_name and m.name != self.leader_name] + + slots = {slot_name_from_member_name(m.name): {'type': 'physical'} for m in members} + if len(slots) < len(members): + # Find which names are conflicting for a nicer error message + slot_conflicts: Dict[str, List[str]] = defaultdict(list) + for member in members: + slot_conflicts[slot_name_from_member_name(member.name)].append(member.name) + logger.error("Following cluster members share a replication slot name: %s", + "; ".join(f"{', '.join(v)} map to {k}" + for k, v in slot_conflicts.items() if len(v) > 1)) + return slots def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool: """Check if the given member node has permanent ``logical`` replication slots configured. diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 5dd69c86..b2a6c478 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -141,6 +141,36 @@ class HTTPClient(object): class ConsulClient(base.Consul): def __init__(self, *args: Any, **kwargs: Any) -> None: + """ + Consul client with Patroni customisations. + + .. note:: + + Parameters, *token*, *cert* and *ca_cert* are not passed to the parent class :class:`consul.base.Consul`. + + Original class documentation, + + *token* is an optional ``ACL token``. If supplied it will be used by + default for all requests made with this client session. It's still + possible to override this token by passing a token explicitly for a + request. + + *consistency* sets the consistency mode to use by default for all reads + that support the consistency option. It's still possible to override + this by passing explicitly for a given request. *consistency* can be + either 'default', 'consistent' or 'stale'. + + *dc* is the datacenter that this agent will communicate with. + By default, the datacenter of the host is used. + + *verify* is whether to verify the SSL certificate for HTTPS requests + + *cert* client side certificates for HTTPS requests + + :param args: positional arguments to pass to :class:`consul.base.Consul` + :param kwargs: keyword arguments, with *cert*, *ca_cert* and *token* removed, passed to + :class:`consul.base.Consul` + """ self._cert = kwargs.pop('cert', None) self._ca_cert = kwargs.pop('ca_cert', None) self.token = kwargs.get('token') diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index fec544b0..a88f4b23 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -756,7 +756,7 @@ class Kubernetes(AbstractDCS): self._role_label = config.get('role_label', 'role') self._leader_label_value = config.get('leader_label_value', 'master') self._follower_label_value = config.get('follower_label_value', 'replica') - self._standby_leader_label_value = config.get('standby_leader_label_value', 'standby-leader') + self._standby_leader_label_value = config.get('standby_leader_label_value', 'master') self._tmp_role_label = config.get('tmp_role_label') self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME super(Kubernetes, self).__init__({**config, 'namespace': ''}) @@ -1140,6 +1140,13 @@ class Kubernetes(AbstractDCS): """Unused""" raise NotImplementedError # pragma: no cover + def write_leader_optime(self, last_lsn: int) -> None: + """Write value for WAL LSN to ``optime`` annotation of the leader object. + + :param last_lsn: absolute WAL LSN in bytes. + """ + self.patch_or_create(self.leader_path, {self._OPTIME: str(last_lsn)}, patch=True, retry=False) + def _update_leader_with_retry(self, annotations: Dict[str, Any], resource_version: Optional[str], ips: List[str]) -> bool: retry = self._retry.copy() @@ -1269,13 +1276,10 @@ class Kubernetes(AbstractDCS): def touch_member(self, data: Dict[str, Any]) -> bool: cluster = self.cluster if cluster and cluster.leader and cluster.leader.name == self._name: - role = self._leader_label_value + role = self._standby_leader_label_value if data['role'] == 'standby_leader' else self._leader_label_value tmp_role = 'master' elif data['state'] == 'running' and data['role'] not in ('master', 'primary'): - role = { - 'replica': self._follower_label_value, - 'standby-leader': self._standby_leader_label_value, - }.get(data['role'], data['role']) + role = {'replica': self._follower_label_value}.get(data['role'], data['role']) tmp_role = data['role'] else: role = None diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 79141256..315bf8c7 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -14,7 +14,7 @@ from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Tuple from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value from ..collections import CaseInsensitiveDict, CaseInsensitiveSet from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name -from ..exceptions import PatroniFatalException +from ..exceptions import PatroniFatalException, PostgresConnectionException from ..file_perm import pg_perm from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath from ..validator import IntValidator, EnumValidator @@ -623,7 +623,24 @@ class ConfigHandler(object): 'recovery_target_action', 'standby_mode', self._triggerfile_wrong_name}) return CaseInsensitiveSet(self._RECOVERY_PARAMETERS - skip_params) - def _read_recovery_params(self) -> Tuple[Optional[CaseInsensitiveDict], Optional[bool]]: + def _read_recovery_params(self) -> Tuple[Optional[CaseInsensitiveDict], bool]: + """Read current recovery parameters values. + + .. note:: + We query Postgres only if we detected that Postgresql was restarted + or when at least one of the following files was updated: + + * ``postgresql.conf``; + * ``postgresql.auto.conf``; + * ``passfile`` that is used in the ``primary_conninfo``. + + :returns: a tuple with two elements: + + * :class:`CaseInsensitiveDict` object with current values of recovery parameters, + or ``None`` if no configuration files were updated; + + * ``True`` if new values of recovery parameters were queried, ``False`` otherwise. + """ if self._postgresql.is_starting(): return None, False @@ -644,11 +661,20 @@ class ConfigHandler(object): self._postgresql_conf_mtime = pg_conf_mtime self._auto_conf_mtime = auto_conf_mtime self._postmaster_ctime = postmaster_ctime - except Exception: + except Exception as exc: + if all((isinstance(exc, PostgresConnectionException), + self._postgresql_conf_mtime == pg_conf_mtime, + self._auto_conf_mtime == auto_conf_mtime, + self._passfile_mtime == passfile_mtime, + self._postmaster_ctime != postmaster_ctime)): + # We detected that the connection to postgres fails, but the process creation time of the postmaster + # doesn't match the old value. It is an indicator that Postgres crashed and either doing crash + # recovery or down. In this case we return values like nothing changed in the config. + return None, False values = None return values, True - def _read_recovery_params_pre_v12(self) -> Tuple[Optional[CaseInsensitiveDict], Optional[bool]]: + def _read_recovery_params_pre_v12(self) -> Tuple[Optional[CaseInsensitiveDict], bool]: recovery_conf_mtime = mtime(self._recovery_conf) passfile_mtime = mtime(self._passfile) if self._passfile else False if recovery_conf_mtime == self._recovery_conf_mtime and passfile_mtime == self._passfile_mtime: diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 51a8fc5a..7391f543 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -434,7 +434,7 @@ class SlotsHandler: self._advance = SlotsAdvanceThread(self) return self._advance.schedule(slots) - def _ensure_logical_slots_replica(self, cluster: Cluster, slots: Dict[str, Any]) -> List[str]: + def _ensure_logical_slots_replica(self, slots: Dict[str, Any]) -> List[str]: """Update logical *slots* on replicas. If the logical slot already exists, copy state information into the replication slots structure stored in the @@ -444,7 +444,6 @@ class SlotsHandler: As logical slots can only be created when the primary is available, pass the list of slots that need to be copied back to the caller. They will be created on replicas with :meth:`SlotsHandler.copy_logical_slots`. - :param cluster: object containing stateful information for the cluster. :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot if the value is a dictionary with the key ``type`` and a value of ``logical``. @@ -459,15 +458,16 @@ class SlotsHandler: continue # If the logical already exists, copy some information about it into the original structure - if self._replication_slots.get(name, {}).get('datoid'): + if name in self._replication_slots and compare_slots(value, self._replication_slots[name]): self._copy_items(self._replication_slots[name], value) - if cluster.slots and name in cluster.slots: + if 'lsn' in value: # The slot has feedback in DCS try: # Skip slots that don't need to be advanced - if value['confirmed_flush_lsn'] < int(cluster.slots[name]): - advance_slots[value['database']][name] = int(cluster.slots[name]) + if value['confirmed_flush_lsn'] < int(value['lsn']): + advance_slots[value['database']][name] = int(value['lsn']) except Exception as e: - logger.error('Failed to parse "%s": %r', cluster.slots[name], e) - elif cluster.slots and name in cluster.slots: # We want to copy only slots with feedback in a DCS + logger.error('Failed to parse "%s": %r', value['lsn'], e) + elif name not in self._replication_slots and 'lsn' in value: + # We want to copy only slots with feedback in a DCS create_slots.append(name) # Slots to be copied from the primary should be removed from the *slots* structure, @@ -512,10 +512,9 @@ class SlotsHandler: if self._postgresql.is_primary(): self._logical_slots_processing_queue.clear() self._ensure_logical_slots_primary(slots) - elif cluster.slots and slots: + else: self.check_logical_slots_readiness(cluster, replicatefrom) - - ret = self._ensure_logical_slots_replica(cluster, slots) + ret = self._ensure_logical_slots_replica(slots) self._replication_slots = slots except Exception: diff --git a/patroni/postgresql/validator.py b/patroni/postgresql/validator.py index d568fa74..384db08d 100644 --- a/patroni/postgresql/validator.py +++ b/patroni/postgresql/validator.py @@ -290,7 +290,7 @@ def _load_postgres_gucs_validators() -> None: Any problem faced while reading or parsing files will be logged as a ``WARNING`` by the child function, and the corresponding file or validator will be ignored. - By default Patroni only ships the file ``0_postgres.yml``, which contains Community Postgres GUCs validators, but + By default, Patroni only ships the file ``0_postgres.yml``, which contains Community Postgres GUCs validators, but that behavior can be extended. For example: if a vendor wants to add GUC validators to Patroni for covering a custom Postgres build, then they can create their custom YAML files under ``available_parameters`` directory. @@ -300,8 +300,10 @@ def _load_postgres_gucs_validators() -> None: writes them to ``postgresql.conf`` if running PG 12 and above). Then, each of these sections, if specified, may contain one or more attributes with the following structure: + * key: the name of a GUC; * value: a list of validators. Each item in the list must contain a ``type`` attribute, which must be one among: + * ``Bool``; or * ``Integer``; or * ``Real``; or @@ -313,6 +315,7 @@ def _load_postgres_gucs_validators() -> None: class in this module. .. seealso:: + * :class:`Bool`; * :class:`Integer`; * :class:`Real`; @@ -325,61 +328,62 @@ def _load_postgres_gucs_validators() -> None: This is a sample content for an YAML file based on Postgres GUCs, showing each of the supported types and sections: - ```yaml - parameters: - archive_command: - - type: String - version_from: 90300 - version_till: null - archive_mode: - - type: Bool - version_from: 90300 - version_till: 90500 - - type: EnumBool - version_from: 90500 - version_till: null - possible_values: - - always - archive_timeout: - - type: Integer - version_from: 90300 - version_till: null - min_val: 0 - max_val: 1073741823 - unit: s - autovacuum_vacuum_cost_delay: - - type: Integer - version_from: 90300 - version_till: 120000 - min_val: -1 - max_val: 100 - unit: ms - - type: Real - version_from: 120000 - version_till: null - min_val: -1 - max_val: 100 - unit: ms - client_min_messages: - - type: Enum - version_from: 90300 - version_till: null - possible_values: - - debug5 - - debug4 - - debug3 - - debug2 - - debug1 - - log - - notice - - warning - - error - recovery_parameters: - archive_cleanup_command: - - type: String - version_from: 90300 - version_till: null - ``` + .. code-block:: yaml + + parameters: + archive_command: + - type: String + version_from: 90300 + version_till: null + archive_mode: + - type: Bool + version_from: 90300 + version_till: 90500 + - type: EnumBool + version_from: 90500 + version_till: null + possible_values: + - always + archive_timeout: + - type: Integer + version_from: 90300 + version_till: null + min_val: 0 + max_val: 1073741823 + unit: s + autovacuum_vacuum_cost_delay: + - type: Integer + version_from: 90300 + version_till: 120000 + min_val: -1 + max_val: 100 + unit: ms + - type: Real + version_from: 120000 + version_till: null + min_val: -1 + max_val: 100 + unit: ms + client_min_messages: + - type: Enum + version_from: 90300 + version_till: null + possible_values: + - debug5 + - debug4 + - debug3 + - debug2 + - debug1 + - log + - notice + - warning + - error + recovery_parameters: + archive_cleanup_command: + - type: String + version_from: 90300 + version_till: null + """ conf_dir = os.path.join( os.path.dirname(os.path.abspath(__file__)), @@ -434,13 +438,15 @@ def _transform_parameter_value(validators: MutableMapping[str, Tuple[_Transforma :param value: value of the Postgres GUC. :param available_gucs: a set of all GUCs available in Postgres *version*. Each item is the name of a Postgres GUC. Used for a couple purposes: + * Disallow writing GUCs to ``postgresql.conf`` (or ``recovery.conf``) that does not exist in Postgres *version*; * Avoid ignoring GUC *name* if it does not have a validator in *validators*, but is a valid GUC in Postgres - *version*. + *version*. :returns: the return value may be one among: - * *value* transformed to the expected format for GUC *name* in Postgres *version*, if *name* is present in - *available_gucs* and has a validator in *validators* for the corresponding Postgres *version*; or + + * *value* transformed to the expected format for GUC *name* in Postgres *version*, if *name* is present + in *available_gucs* and has a validator in *validators* for the corresponding Postgres *version*; or * The own *value* if *name* is present in *available_gucs* but not in *validators*; or * ``None`` if *name* is not present in *available_gucs*. """ diff --git a/patroni/validator.py b/patroni/validator.py index ddfb2c07..bc308f71 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -333,15 +333,17 @@ class Case(object): """Create a :class:`Case` object. :param schema: the schema for validating a set of attributes that may be available in the configuration. - Each key is the configuration that is available in a given scope and that should be validated, and the - related value is the validation function or expected type. + Each key is the configuration that is available in a given scope and that should be validated, + and the related value is the validation function or expected type. :Example: - Case({ - "host": validate_host_port, - "url": str, - }) + .. code-block:: python + + Case({ + "host": validate_host_port, + "url": str, + }) That will check that ``host`` configuration, if given, is valid based on :func:`validate_host_port`, and will also check that ``url`` configuration, if given, is a ``str`` instance. @@ -363,14 +365,16 @@ class Or(object): :Example: - Or("host", "hosts"): Case({ - "host": validate_host_port, - "hosts": Or(comma_separated_host_port, [validate_host_port]), - }) + .. code-block:: python - The outer :class:`Or` is used to define that ``host`` and ``hosts`` are possible options in this scope. - The inner :class`Or` in the ``hosts`` key value is used to define that ``hosts`` option is valid if either of - :func:`comma_separated_host_port` or :func:`validate_host_port` succeed to validate it. + Or("host", "hosts"): Case({ + "host": validate_host_port, + "hosts": Or(comma_separated_host_port, [validate_host_port]), + }) + + The outer :class:`Or` is used to define that ``host`` and ``hosts`` are possible options in this scope. + The inner :class`Or` in the ``hosts`` key value is used to define that ``hosts`` option is valid if either + of :func:`comma_separated_host_port` or :func:`validate_host_port` succeed to validate it. """ self.args = args @@ -535,32 +539,34 @@ class Schema(object): :Example: - Schema({ - "application_name": str, - "bind": { - "host": validate_host, - "port": int, - }, - "aliases": [str], - Optional("data_directory"): "/var/lib/myapp", - Or("log_to_file", "log_to_db"): Case({ - "log_to_file": bool, - "log_to_db": bool, - }), - "version": Or(int, float), - }) + .. code-block:: python - This sample schema defines that your YAML configuration follows these rules: + Schema({ + "application_name": str, + "bind": { + "host": validate_host, + "port": int, + }, + "aliases": [str], + Optional("data_directory"): "/var/lib/myapp", + Or("log_to_file", "log_to_db"): Case({ + "log_to_file": bool, + "log_to_db": bool, + }), + "version": Or(int, float), + }) - * It must contain an ``application_name`` entry which value should be a :class:`str` instance; - * It must contain a ``bind.host`` entry which value should be valid as per function ``validate_host``; - * It must contain a ``bind.port`` entry which value should be an :class:`int` instance; - * It must contain a ``aliases`` entry which value should be a :class:`list` of :class:`str` instances; - * It may optionally contain a ``data_directory`` entry, with a value which should be a string; - * It must contain at least one of ``log_to_file`` or ``log_to_db``, with a value which should be a - :class:`bool` instance; - * It must contain a ``version`` entry which value should be either an :class:`int` or a :class:`float` - instance. + This sample schema defines that your YAML configuration follows these rules: + + * It must contain an ``application_name`` entry which value should be a :class:`str` instance; + * It must contain a ``bind.host`` entry which value should be valid as per function ``validate_host``; + * It must contain a ``bind.port`` entry which value should be an :class:`int` instance; + * It must contain a ``aliases`` entry which value should be a :class:`list` of :class:`str` instances; + * It may optionally contain a ``data_directory`` entry, with a value which should be a string; + * It must contain at least one of ``log_to_file`` or ``log_to_db``, with a value which should be a + :class:`bool` instance; + * It must contain a ``version`` entry which value should be either an :class:`int` or a :class:`float` + instance. """ self.validator = validator diff --git a/requirements.docs.txt b/requirements.docs.txt index 1a71f097..afb47378 100644 --- a/requirements.docs.txt +++ b/requirements.docs.txt @@ -1,5 +1,4 @@ sphinx>=4 sphinx_rtd_theme>1 sphinxcontrib-apidoc -sphinx-github-style -pyyaml +sphinx-github-style<1.0.3 diff --git a/tests/__init__.py b/tests/__init__.py index f3aee63b..f70aafb1 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -104,7 +104,7 @@ class MockCursor(object): elif sql.startswith('SELECT slot_name, slot_type, datname, plugin, catalog_xmin'): self.results = [('ls', 'logical', 'a', 'b', 100, 500, b'123456')] elif sql.startswith('SELECT slot_name'): - self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)] + self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'b', 'a', 5, 100, 500)] elif sql.startswith('WITH slots AS (SELECT slot_name, active'): self.results = [(False, True)] if self.rowcount == 1 else [] elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'): diff --git a/tests/test_api.py b/tests/test_api.py index f433ca18..fa9a6280 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -11,9 +11,12 @@ from socketserver import ThreadingMixIn from patroni.api import RestApiHandler, RestApiServer from patroni.config import GlobalConfig from patroni.dcs import ClusterConfig, Member +from patroni.exceptions import PostgresConnectionException from patroni.ha import _MemberStatus +from patroni.psycopg import OperationalError from patroni.utils import RetryFailedError, tzutc +from . import MockConnect, psycopg_connect from .test_ha import get_cluster_initialized_without_leader @@ -21,8 +24,29 @@ future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5) postmaster_start_time = datetime.datetime.now(tzutc) -class MockPostgresql(object): +class MockConnection: + @staticmethod + def get(*args): + return psycopg_connect() + + @staticmethod + def query(sql, *params): + return [(postmaster_start_time, 0, '', 0, '', False, postmaster_start_time, 'streaming', None, + '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' + + '"state":"streaming","sync_state":"async","sync_priority":0}]')] + + +class MockConnectionPool: + + @staticmethod + def get(*args): + return MockConnection() + + +class MockPostgresql: + + connection_pool = MockConnectionPool() name = 'test' state = 'running' role = 'primary' @@ -54,12 +78,6 @@ class MockPostgresql(object): def replication_state_from_parameters(*args): return 'streaming' - @staticmethod - def query(sql, *params, retry=False): - return [(postmaster_start_time, 0, '', 0, '', False, postmaster_start_time, 'streaming', None, - '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' - + '"state":"streaming","sync_state":"async","sync_priority":0}]')] - class MockWatchdog(object): is_healthy = False @@ -487,7 +505,7 @@ class TestRestApiHandler(unittest.TestCase): @patch('time.sleep', Mock()) def test_RestApiServer_query(self): - with patch.object(MockPostgresql, 'query', Mock(side_effect=RetryFailedError('bla'))): + with patch.object(MockConnection, 'query', Mock(side_effect=RetryFailedError('bla'))): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) @patch('time.sleep', Mock()) @@ -659,3 +677,11 @@ class TestRestApiServer(unittest.TestCase): def test_get_certificate_serial_number(self): self.assertIsNone(self.srv.get_certificate_serial_number()) + + def test_query(self): + with patch.object(MockConnection, 'get', Mock(side_effect=OperationalError)): + self.assertRaises(PostgresConnectionException, self.srv.query, 'SELECT 1') + with patch.object(MockConnection, 'get', Mock(side_effect=[MockConnect(), OperationalError])), \ + patch.object(MockConnection, 'query') as mock_query: + self.srv.query('SELECT 1') + mock_query.assert_called_once_with('SELECT 1') diff --git a/tests/test_ctl.py b/tests/test_ctl.py index f4f08296..b1468075 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -402,9 +402,19 @@ class TestCtl(unittest.TestCase): @patch('patroni.ctl.get_dcs') def test_members(self, mock_get_dcs): mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + result = self.runner.invoke(ctl, ['list']) assert '127.0.0.1' in result.output assert result.exit_code == 0 + assert 'Citus cluster: alpha -' in result.output + + result = self.runner.invoke(ctl, ['list', '--group', '0']) + assert 'Citus cluster: alpha (group: 0, 12345678901) -' in result.output + + with patch('patroni.ctl.load_config', Mock(return_value={'scope': 'alpha'})): + result = self.runner.invoke(ctl, ['list']) + assert 'Cluster: alpha (12345678901) -' in result.output + with patch('patroni.ctl.load_config', Mock(return_value={})): self.runner.invoke(ctl, ['list']) diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 694cd505..4f9f418c 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -308,13 +308,15 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): mock_patch_namespaced_pod.assert_called() self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'replica') - - self.k.touch_member({'state': 'running', 'role': 'standby-leader'}) - mock_patch_namespaced_pod.assert_called() - self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') - self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'standby-leader') + mock_patch_namespaced_pod.rest_mock() self.k._name = 'p-0' + self.k.touch_member({'role': 'standby_leader'}) + mock_patch_namespaced_pod.assert_called() + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'master') + mock_patch_namespaced_pod.rest_mock() + self.k.touch_member({'role': 'primary'}) mock_patch_namespaced_pod.assert_called() self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'true') @@ -434,6 +436,10 @@ class TestKubernetesEndpoints(BaseTestKubernetes): mock_logger_exception.assert_called_once() self.assertEqual(('create_config_service failed',), mock_logger_exception.call_args[0]) + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True) + def test_write_leader_optime(self): + self.k.write_leader_optime(12345) + def mock_watch(*args): return urllib3.HTTPResponse() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 29b25ee4..9ab3f52d 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -310,6 +310,17 @@ class TestPostgresql(BaseTestPostgresql): self.p.config.write_postgresql_conf() self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) + + # Config files changed, but can't connect to postgres + mock_get_pg_settings.side_effect = PostgresConnectionException('') + with patch('patroni.postgresql.config.mtime', mock_mtime): + self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) + + # Config files didn't change, but postgres crashed or in crash recovery + with patch.object(MockPostmaster, 'create_time', Mock(return_value=1234568), create=True): + self.assertEqual(self.p.config.check_recovery_conf(None), (False, False)) + + # Any other exception raised when executing the query mock_get_pg_settings.side_effect = Exception with patch('patroni.postgresql.config.mtime', mock_mtime): self.assertEqual(self.p.config.check_recovery_conf(None), (True, True)) diff --git a/tests/test_slots.py b/tests/test_slots.py index add0fdf7..a35d465c 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -32,9 +32,9 @@ class TestSlotsHandler(BaseTestPostgresql): self.p._global_config = GlobalConfig({}) self.s = self.p.slots_handler self.p.start() - config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1) + config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}, 'ls2': None}}, 1) self.cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem], - None, SyncState.empty(), None, {'ls': 12345}, None) + None, SyncState.empty(), None, {'ls': 12345, 'ls2': 12345}, None) def test_sync_replication_slots(self): config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'}, @@ -123,6 +123,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls']) self.cluster.slots['ls'] = 'a' self.assertEqual(self.s.sync_replication_slots(self.cluster, False), []) + self.cluster.config.data['slots']['ls']['database'] = 'b' with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True): self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])