mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-31 16:49:46 +00:00
Merge branch 'master' of github.com:zalando/patroni into feature/citus-secondaries
This commit is contained in:
Alexander Kukushkin
+1
-1
@@ -33,7 +33,7 @@ nosetests.xml
|
||||
coverage.xml
|
||||
htmlcov
|
||||
junit.xml
|
||||
features/output
|
||||
features/output*
|
||||
dummy
|
||||
|
||||
# Translations
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
+33
-27
@@ -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",
|
||||
|
||||
@@ -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.
|
||||
|
||||
+5
-2
@@ -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.
|
||||
|
||||
+50
-29
@@ -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``;
|
||||
|
||||
+1
-1
@@ -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`.
|
||||
|
||||
+8
-4
@@ -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
|
||||
|
||||
+38
-30
@@ -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.
|
||||
|
||||
@@ -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')
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
+10
-11
@@ -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:
|
||||
|
||||
@@ -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*.
|
||||
"""
|
||||
|
||||
+43
-37
@@ -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
|
||||
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
sphinx>=4
|
||||
sphinx_rtd_theme>1
|
||||
sphinxcontrib-apidoc
|
||||
sphinx-github-style
|
||||
pyyaml
|
||||
sphinx-github-style<1.0.3
|
||||
|
||||
+1
-1
@@ -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()'):
|
||||
|
||||
+34
-8
@@ -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')
|
||||
|
||||
@@ -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'])
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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))
|
||||
|
||||
+3
-2
@@ -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'])
|
||||
|
||||
|
||||
Reference in New Issue
Block a user