mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Enable pyright strict mode (#2652)
- added pyrightconfig.json with typeCheckingMode=strict - added type hints to all files except api.py - added type stubs for dns, etcd, consul, kazoo, pysyncobj and other modules - added type stubs for psycopg2 and urllib3 with some little fixes - fixes most of the issues reported by pyright - remaining issues will be addressed later, along with enabling CI linting task
This commit is contained in:
Alexander Kukushkin
+163
-142
@@ -12,7 +12,7 @@ from datetime import datetime
|
||||
from dateutil import tz
|
||||
from psutil import TimeoutExpired
|
||||
from threading import current_thread, Lock
|
||||
from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING
|
||||
from typing import Any, Callable, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING
|
||||
|
||||
from .bootstrap import Bootstrap
|
||||
from .callback_executor import CallbackAction, CallbackExecutor
|
||||
@@ -25,11 +25,14 @@ from .postmaster import PostmasterProcess
|
||||
from .slots import SlotsHandler
|
||||
from .sync import SyncHandler
|
||||
from .. import psycopg
|
||||
from ..dcs import Cluster, Member
|
||||
from ..async_executor import CriticalTask
|
||||
from ..dcs import Cluster, Leader, Member
|
||||
from ..exceptions import PostgresConnectionException
|
||||
from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from psycopg import Connection as Connection3, Cursor
|
||||
from psycopg2 import connection as connection3, cursor
|
||||
from ..config import GlobalConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -59,13 +62,15 @@ class Postgresql(object):
|
||||
"pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, "
|
||||
"pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()")
|
||||
|
||||
def __init__(self, config):
|
||||
self.name = config['name']
|
||||
self.scope = config['scope']
|
||||
self._data_dir = config['data_dir']
|
||||
def __init__(self, config: Dict[str, Any]) -> None:
|
||||
self.name: str = config['name']
|
||||
self.scope: str = config['scope']
|
||||
self._data_dir: str = config['data_dir']
|
||||
self._database = config.get('database', 'postgres')
|
||||
self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
|
||||
self._pg_control = os.path.join(self._data_dir, 'global', 'pg_control')
|
||||
self.connection_string: str
|
||||
self.proxy_url: Optional[str]
|
||||
self._major_version = self.get_major_version()
|
||||
self._global_config = None
|
||||
|
||||
@@ -92,7 +97,7 @@ class Postgresql(object):
|
||||
|
||||
self.cancellable = CancellableSubprocess()
|
||||
|
||||
self._sysid = None
|
||||
self._sysid = ''
|
||||
self.retry = Retry(max_tries=-1, deadline=config['retry_timeout'] / 2.0, max_delay=1,
|
||||
retry_exceptions=PostgresConnectionException)
|
||||
|
||||
@@ -102,7 +107,7 @@ class Postgresql(object):
|
||||
|
||||
self._role_lock = Lock()
|
||||
self.set_role(self.get_postgres_role_from_data_directory())
|
||||
self._state_entry_timestamp = None
|
||||
self._state_entry_timestamp = 0
|
||||
|
||||
self._cluster_info_state = {}
|
||||
self._has_permanent_logical_slots = True
|
||||
@@ -126,35 +131,35 @@ class Postgresql(object):
|
||||
self.set_role('demoted')
|
||||
|
||||
@property
|
||||
def create_replica_methods(self):
|
||||
return self.config.get('create_replica_methods', []) or self.config.get('create_replica_method', [])
|
||||
def create_replica_methods(self) -> List[str]:
|
||||
return self.config.get('create_replica_methods', []) or self.config.get('create_replica_method', []) or []
|
||||
|
||||
@property
|
||||
def major_version(self):
|
||||
def major_version(self) -> int:
|
||||
return self._major_version
|
||||
|
||||
@property
|
||||
def database(self):
|
||||
def database(self) -> str:
|
||||
return self._database
|
||||
|
||||
@property
|
||||
def data_dir(self):
|
||||
def data_dir(self) -> str:
|
||||
return self._data_dir
|
||||
|
||||
@property
|
||||
def callback(self):
|
||||
return self.config.get('callbacks') or {}
|
||||
def callback(self) -> Dict[str, str]:
|
||||
return self.config.get('callbacks', {}) or {}
|
||||
|
||||
@property
|
||||
def wal_dir(self):
|
||||
def wal_dir(self) -> str:
|
||||
return os.path.join(self._data_dir, 'pg_' + self.wal_name)
|
||||
|
||||
@property
|
||||
def wal_name(self):
|
||||
def wal_name(self) -> str:
|
||||
return 'wal' if self._major_version >= 100000 else 'xlog'
|
||||
|
||||
@property
|
||||
def lsn_name(self):
|
||||
def lsn_name(self) -> str:
|
||||
return 'lsn' if self._major_version >= 100000 else 'location'
|
||||
|
||||
@property
|
||||
@@ -163,7 +168,7 @@ class Postgresql(object):
|
||||
return self._major_version >= 90600
|
||||
|
||||
@property
|
||||
def cluster_info_query(self):
|
||||
def cluster_info_query(self) -> str:
|
||||
"""Returns the monitoring query with a fixed number of fields.
|
||||
|
||||
The query text is constructed based on current state in DCS and PostgreSQL version:
|
||||
@@ -206,7 +211,7 @@ class Postgresql(object):
|
||||
|
||||
return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra)
|
||||
|
||||
def _version_file_exists(self):
|
||||
def _version_file_exists(self) -> bool:
|
||||
return not self.data_directory_empty() and os.path.isfile(self._version_file)
|
||||
|
||||
def get_major_version(self) -> int:
|
||||
@@ -221,11 +226,11 @@ class Postgresql(object):
|
||||
logger.exception('Failed to read PG_VERSION from %s', self._data_dir)
|
||||
return 0
|
||||
|
||||
def pgcommand(self, cmd):
|
||||
def pgcommand(self, cmd: str) -> str:
|
||||
"""Returns path to the specified PostgreSQL command"""
|
||||
return os.path.join(self._bin_dir, cmd)
|
||||
|
||||
def pg_ctl(self, cmd, *args, **kwargs):
|
||||
def pg_ctl(self, cmd: str, *args: str, **kwargs: Any) -> bool:
|
||||
"""Builds and executes pg_ctl command
|
||||
|
||||
:returns: `!True` when return_code == 0, otherwise `!False`"""
|
||||
@@ -244,7 +249,7 @@ class Postgresql(object):
|
||||
initdb = [self.pgcommand('initdb')] + list(args) + [self.data_dir]
|
||||
return subprocess.call(initdb, **kwargs) == 0
|
||||
|
||||
def pg_isready(self):
|
||||
def pg_isready(self) -> str:
|
||||
"""Runs pg_isready to see if PostgreSQL is accepting connections.
|
||||
|
||||
:returns: 'ok' if PostgreSQL is up, 'reject' if starting up, 'no_resopnse' if not up."""
|
||||
@@ -267,25 +272,25 @@ class Postgresql(object):
|
||||
3: STATE_UNKNOWN}
|
||||
return return_codes.get(ret, STATE_UNKNOWN)
|
||||
|
||||
def reload_config(self, config, sighup=False):
|
||||
def reload_config(self, config: Dict[str, Any], sighup: bool = False) -> None:
|
||||
self.config.reload_config(config, sighup)
|
||||
self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout'] / 2.0
|
||||
|
||||
@property
|
||||
def pending_restart(self):
|
||||
def pending_restart(self) -> bool:
|
||||
return self._pending_restart
|
||||
|
||||
def set_pending_restart(self, value):
|
||||
def set_pending_restart(self, value: bool) -> None:
|
||||
self._pending_restart = value
|
||||
|
||||
@property
|
||||
def sysid(self):
|
||||
def sysid(self) -> str:
|
||||
if not self._sysid and not self.bootstrapping:
|
||||
data = self.controldata()
|
||||
self._sysid = data.get('Database system identifier', "")
|
||||
self._sysid = data.get('Database system identifier', '')
|
||||
return self._sysid
|
||||
|
||||
def get_postgres_role_from_data_directory(self):
|
||||
def get_postgres_role_from_data_directory(self) -> str:
|
||||
if self.data_directory_empty() or not self.controldata():
|
||||
return 'uninitialized'
|
||||
elif self.config.recovery_conf_exists():
|
||||
@@ -294,24 +299,24 @@ class Postgresql(object):
|
||||
return 'master'
|
||||
|
||||
@property
|
||||
def server_version(self):
|
||||
def server_version(self) -> int:
|
||||
return self._connection.server_version
|
||||
|
||||
def connection(self):
|
||||
def connection(self) -> Union['connection3', 'Connection3[Any]']:
|
||||
return self._connection.get()
|
||||
|
||||
def set_connection_kwargs(self, kwargs):
|
||||
def set_connection_kwargs(self, kwargs: Dict[str, Any]) -> None:
|
||||
self._connection.set_conn_kwargs(kwargs.copy())
|
||||
self.citus_handler.set_conn_kwargs(kwargs.copy())
|
||||
|
||||
def _query(self, sql, *params):
|
||||
def _query(self, sql: str, *params: Any) -> Union['Cursor[Any]', 'cursor']:
|
||||
"""We are always using the same cursor, therefore this method is not thread-safe!!!
|
||||
You can call it from different threads only if you are holding explicit `AsyncExecutor` lock,
|
||||
because the main thread is always holding this lock when running HA cycle."""
|
||||
cursor = None
|
||||
try:
|
||||
cursor = self._connection.cursor()
|
||||
cursor.execute(sql, params or None)
|
||||
cursor.execute(sql.encode('utf-8'), params or None)
|
||||
return cursor
|
||||
except psycopg.Error as e:
|
||||
if cursor and cursor.connection.closed == 0:
|
||||
@@ -327,7 +332,7 @@ class Postgresql(object):
|
||||
raise RetryFailedError('cluster is being restarted')
|
||||
raise PostgresConnectionException('connection problems')
|
||||
|
||||
def query(self, sql, *args, **kwargs):
|
||||
def query(self, sql: str, *args: Any, **kwargs: Any) -> Union['Cursor[Any]', 'cursor']:
|
||||
if not kwargs.get('retry', True):
|
||||
return self._query(sql, *args)
|
||||
try:
|
||||
@@ -335,22 +340,22 @@ class Postgresql(object):
|
||||
except RetryFailedError as e:
|
||||
raise PostgresConnectionException(str(e))
|
||||
|
||||
def pg_control_exists(self):
|
||||
def pg_control_exists(self) -> bool:
|
||||
return os.path.isfile(self._pg_control)
|
||||
|
||||
def data_directory_empty(self):
|
||||
def data_directory_empty(self) -> bool:
|
||||
if self.pg_control_exists():
|
||||
return False
|
||||
return data_directory_is_empty(self._data_dir)
|
||||
|
||||
def replica_method_options(self, method):
|
||||
return deepcopy(self.config.get(method, {}))
|
||||
def replica_method_options(self, method: str) -> Dict[str, Any]:
|
||||
return deepcopy(self.config.get(method, {}) or {})
|
||||
|
||||
def replica_method_can_work_without_replication_connection(self, method):
|
||||
return method != 'basebackup' and (self.replica_method_options(method).get('no_master')
|
||||
or self.replica_method_options(method).get('no_leader'))
|
||||
def replica_method_can_work_without_replication_connection(self, method: str) -> bool:
|
||||
return method != 'basebackup' and bool(self.replica_method_options(method).get('no_master')
|
||||
or self.replica_method_options(method).get('no_leader'))
|
||||
|
||||
def can_create_replica_without_replication_connection(self, replica_methods=None):
|
||||
def can_create_replica_without_replication_connection(self, replica_methods: Optional[List[str]]) -> bool:
|
||||
""" go through the replication methods to see if there are ones
|
||||
that does not require a working replication connection.
|
||||
"""
|
||||
@@ -359,10 +364,10 @@ class Postgresql(object):
|
||||
return any(self.replica_method_can_work_without_replication_connection(m) for m in replica_methods)
|
||||
|
||||
@property
|
||||
def enforce_hot_standby_feedback(self):
|
||||
def enforce_hot_standby_feedback(self) -> bool:
|
||||
return self._enforce_hot_standby_feedback
|
||||
|
||||
def set_enforce_hot_standby_feedback(self, value):
|
||||
def set_enforce_hot_standby_feedback(self, value: bool) -> None:
|
||||
# If we enable or disable the hot_standby_feedback we need to update postgresql.conf and reload
|
||||
if self._enforce_hot_standby_feedback != value:
|
||||
self._enforce_hot_standby_feedback = value
|
||||
@@ -370,7 +375,7 @@ class Postgresql(object):
|
||||
self.config.write_postgresql_conf()
|
||||
self.reload()
|
||||
|
||||
def reset_cluster_info_state(self, cluster: Union[Cluster, None], nofailover: Optional[bool] = None,
|
||||
def reset_cluster_info_state(self, cluster: Union[Cluster, None], nofailover: bool = False,
|
||||
global_config: Optional['GlobalConfig'] = None) -> None:
|
||||
"""Reset monitoring query cache.
|
||||
|
||||
@@ -395,7 +400,7 @@ class Postgresql(object):
|
||||
|
||||
self._global_config = global_config
|
||||
|
||||
def _cluster_info_state_get(self, name):
|
||||
def _cluster_info_state_get(self, name: str) -> Optional[Any]:
|
||||
if not self._cluster_info_state:
|
||||
try:
|
||||
result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone()
|
||||
@@ -417,62 +422,63 @@ class Postgresql(object):
|
||||
|
||||
return self._cluster_info_state.get(name)
|
||||
|
||||
def replayed_location(self):
|
||||
def replayed_location(self) -> Optional[int]:
|
||||
return self._cluster_info_state_get('replayed_location')
|
||||
|
||||
def received_location(self):
|
||||
def received_location(self) -> Optional[int]:
|
||||
return self._cluster_info_state_get('received_location')
|
||||
|
||||
def slots(self):
|
||||
return self._cluster_info_state_get('slots')
|
||||
def slots(self) -> Dict[str, int]:
|
||||
return self._cluster_info_state_get('slots') or {}
|
||||
|
||||
def primary_slot_name(self):
|
||||
def primary_slot_name(self) -> Optional[str]:
|
||||
return self._cluster_info_state_get('slot_name')
|
||||
|
||||
def primary_conninfo(self):
|
||||
def primary_conninfo(self) -> Optional[str]:
|
||||
return self._cluster_info_state_get('conninfo')
|
||||
|
||||
def received_timeline(self):
|
||||
def received_timeline(self) -> Optional[int]:
|
||||
return self._cluster_info_state_get('received_tli')
|
||||
|
||||
def synchronous_commit(self) -> str:
|
||||
""":returns: "synchronous_commit" GUC value."""
|
||||
return self._cluster_info_state_get('synchronous_commit')
|
||||
return self._cluster_info_state_get('synchronous_commit') or 'on'
|
||||
|
||||
def synchronous_standby_names(self) -> str:
|
||||
""":returns: "synchronous_standby_names" GUC value."""
|
||||
return self._cluster_info_state_get('synchronous_standby_names')
|
||||
return self._cluster_info_state_get('synchronous_standby_names') or ''
|
||||
|
||||
def pg_stat_replication(self) -> List[Dict[str, Any]]:
|
||||
""":returns: a result set of 'SELECT * FROM pg_stat_replication'."""
|
||||
return self._cluster_info_state_get('pg_stat_replication') or []
|
||||
|
||||
def is_leader(self):
|
||||
def is_leader(self) -> bool:
|
||||
try:
|
||||
return bool(self._cluster_info_state_get('timeline'))
|
||||
except PostgresConnectionException:
|
||||
logger.warning('Failed to determine PostgreSQL state from the connection, falling back to cached role')
|
||||
return bool(self.is_running() and self.role in ('master', 'primary'))
|
||||
|
||||
def replay_paused(self):
|
||||
return self._cluster_info_state_get('replay_paused')
|
||||
def replay_paused(self) -> bool:
|
||||
return self._cluster_info_state_get('replay_paused') or False
|
||||
|
||||
def resume_wal_replay(self):
|
||||
def resume_wal_replay(self) -> None:
|
||||
self._query('SELECT pg_catalog.pg_{0}_replay_resume()'.format(self.wal_name))
|
||||
|
||||
def handle_parameter_change(self):
|
||||
def handle_parameter_change(self) -> None:
|
||||
if self.major_version >= 140000 and not self.is_starting() and self.replay_paused():
|
||||
logger.info('Resuming paused WAL replay for PostgreSQL 14+')
|
||||
self.resume_wal_replay()
|
||||
|
||||
def pg_control_timeline(self):
|
||||
def pg_control_timeline(self) -> Optional[int]:
|
||||
try:
|
||||
|
||||
return int(self.controldata().get("Latest checkpoint's TimeLineID"))
|
||||
return int(self.controldata().get("Latest checkpoint's TimeLineID", ""))
|
||||
except (TypeError, ValueError):
|
||||
logger.exception('Failed to parse timeline from pg_controldata output')
|
||||
|
||||
def parse_wal_record(self, timeline, lsn):
|
||||
def parse_wal_record(self, timeline: str,
|
||||
lsn: str) -> Union[Tuple[str, str, str, str], Tuple[None, None, None, None]]:
|
||||
out, err = self.waldump(timeline, lsn, 1)
|
||||
if out and not err:
|
||||
match = re.match(r'^rmgr:\s+(.+?)\s+len \(rec/tot\):\s+\d+/\s+\d+, tx:\s+\d+, '
|
||||
@@ -482,7 +488,7 @@ class Postgresql(object):
|
||||
return match.groups()
|
||||
return None, None, None, None
|
||||
|
||||
def latest_checkpoint_location(self):
|
||||
def latest_checkpoint_location(self) -> Optional[int]:
|
||||
"""Returns checkpoint location for the cleanly shut down primary.
|
||||
But, if we know that the checkpoint was written to the new WAL
|
||||
due to the archive_mode=on, we will return the LSN of prev wal record (SWITCH)."""
|
||||
@@ -490,25 +496,25 @@ class Postgresql(object):
|
||||
data = self.controldata()
|
||||
timeline = data.get("Latest checkpoint's TimeLineID")
|
||||
lsn = checkpoint_lsn = data.get('Latest checkpoint location')
|
||||
if data.get('Database cluster state') == 'shut down' and lsn and timeline:
|
||||
if data.get('Database cluster state') == 'shut down' and lsn and timeline and checkpoint_lsn:
|
||||
try:
|
||||
checkpoint_lsn = parse_lsn(checkpoint_lsn)
|
||||
rm_name, lsn, prev, desc = self.parse_wal_record(timeline, lsn)
|
||||
desc = desc.strip().lower()
|
||||
if rm_name == 'XLOG' and parse_lsn(lsn) == checkpoint_lsn and prev and\
|
||||
desc = str(desc).strip().lower()
|
||||
if rm_name == 'XLOG' and lsn and parse_lsn(lsn) == checkpoint_lsn and prev and\
|
||||
desc.startswith('checkpoint') and desc.endswith('shutdown'):
|
||||
_, lsn, _, desc = self.parse_wal_record(timeline, prev)
|
||||
prev = parse_lsn(prev)
|
||||
# If the cluster is shutdown with archive_mode=on, WAL is switched before writing the checkpoint.
|
||||
# In this case we want to take the LSN of previous record (switch) as the last known WAL location.
|
||||
if parse_lsn(lsn) == prev and desc.strip() in ('xlog switch', 'SWITCH'):
|
||||
if lsn and parse_lsn(lsn) == prev and str(desc).strip() in ('xlog switch', 'SWITCH'):
|
||||
return prev
|
||||
except Exception as e:
|
||||
logger.error('Exception when parsing WAL pg_%sdump output: %r', self.wal_name, e)
|
||||
if isinstance(checkpoint_lsn, int):
|
||||
return checkpoint_lsn
|
||||
|
||||
def is_running(self):
|
||||
def is_running(self) -> Optional[PostmasterProcess]:
|
||||
"""Returns PostmasterProcess if one is running on the data directory or None. If most recently seen process
|
||||
is running updates the cached process based on pid file."""
|
||||
if self._postmaster_proc:
|
||||
@@ -523,7 +529,7 @@ class Postgresql(object):
|
||||
return self._postmaster_proc
|
||||
|
||||
@property
|
||||
def cb_called(self):
|
||||
def cb_called(self) -> bool:
|
||||
return self.__cb_called
|
||||
|
||||
def call_nowait(self, cb_type: CallbackAction) -> None:
|
||||
@@ -544,31 +550,31 @@ class Postgresql(object):
|
||||
logger.exception('callback %s %r %s %s failed', cmd, cb_type, role, self.scope)
|
||||
|
||||
@property
|
||||
def role(self):
|
||||
def role(self) -> str:
|
||||
with self._role_lock:
|
||||
return self._role
|
||||
|
||||
def set_role(self, value):
|
||||
def set_role(self, value: str) -> None:
|
||||
with self._role_lock:
|
||||
self._role = value
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
def state(self) -> str:
|
||||
with self._state_lock:
|
||||
return self._state
|
||||
|
||||
def set_state(self, value):
|
||||
def set_state(self, value: str) -> None:
|
||||
with self._state_lock:
|
||||
self._state = value
|
||||
self._state_entry_timestamp = time.time()
|
||||
|
||||
def time_in_state(self):
|
||||
def time_in_state(self) -> float:
|
||||
return time.time() - self._state_entry_timestamp
|
||||
|
||||
def is_starting(self):
|
||||
def is_starting(self) -> bool:
|
||||
return self.state == 'starting'
|
||||
|
||||
def wait_for_port_open(self, postmaster, timeout):
|
||||
def wait_for_port_open(self, postmaster: PostmasterProcess, timeout: float) -> bool:
|
||||
"""Waits until PostgreSQL opens ports."""
|
||||
for _ in polling_loop(timeout):
|
||||
if self.cancellable.is_cancelled:
|
||||
@@ -588,7 +594,9 @@ class Postgresql(object):
|
||||
logger.warning("Timed out waiting for PostgreSQL to start")
|
||||
return False
|
||||
|
||||
def start(self, timeout=None, task=None, block_callbacks=False, role=None, after_start=None):
|
||||
def start(self, timeout: Optional[float] = None, task: Optional[CriticalTask] = None,
|
||||
block_callbacks: bool = False, role: Optional[str] = None,
|
||||
after_start: Optional[Callable[..., Any]] = None) -> Optional[bool]:
|
||||
"""Start PostgreSQL
|
||||
|
||||
Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion
|
||||
@@ -650,7 +658,7 @@ class Postgresql(object):
|
||||
start_timeout = timeout
|
||||
if not start_timeout:
|
||||
try:
|
||||
start_timeout = float(self.config.get('pg_ctl_timeout', 60))
|
||||
start_timeout = float(self.config.get('pg_ctl_timeout', 60) or 0)
|
||||
except ValueError:
|
||||
start_timeout = 60
|
||||
|
||||
@@ -668,7 +676,8 @@ class Postgresql(object):
|
||||
else:
|
||||
return None
|
||||
|
||||
def checkpoint(self, connect_kwargs=None, timeout=None):
|
||||
def checkpoint(self, connect_kwargs: Optional[Dict[str, Any]] = None,
|
||||
timeout: Optional[float] = None) -> Optional[str]:
|
||||
check_not_is_in_recovery = connect_kwargs is not None
|
||||
connect_kwargs = connect_kwargs or self.config.local_connect_kwargs
|
||||
for p in ['connect_timeout', 'options']:
|
||||
@@ -680,15 +689,17 @@ class Postgresql(object):
|
||||
cur.execute("SET statement_timeout = 0")
|
||||
if check_not_is_in_recovery:
|
||||
cur.execute('SELECT pg_catalog.pg_is_in_recovery()')
|
||||
if cur.fetchone()[0]:
|
||||
row = cur.fetchone()
|
||||
if not row or row[0]:
|
||||
return 'is_in_recovery=true'
|
||||
cur.execute('CHECKPOINT')
|
||||
except psycopg.Error:
|
||||
logger.exception('Exception during CHECKPOINT')
|
||||
return 'not accessible or not healty'
|
||||
|
||||
def stop(self, mode='fast', block_callbacks=False, checkpoint=None,
|
||||
on_safepoint=None, on_shutdown=None, before_shutdown=None, stop_timeout=None):
|
||||
def stop(self, mode: str = 'fast', block_callbacks: bool = False, checkpoint: Optional[bool] = None,
|
||||
on_safepoint: Optional[Callable[..., Any]] = None, on_shutdown: Optional[Callable[[int], Any]] = None,
|
||||
before_shutdown: Optional[Callable[..., Any]] = None, stop_timeout: Optional[int] = None) -> bool:
|
||||
"""Stop PostgreSQL
|
||||
|
||||
Supports a callback when a safepoint is reached. A safepoint is when no user backend can return a successful
|
||||
@@ -716,7 +727,9 @@ class Postgresql(object):
|
||||
self.set_state('stop failed')
|
||||
return success
|
||||
|
||||
def _do_stop(self, mode, block_callbacks, checkpoint, on_safepoint, on_shutdown, before_shutdown, stop_timeout):
|
||||
def _do_stop(self, mode: str, block_callbacks: bool, checkpoint: bool,
|
||||
on_safepoint: Optional[Callable[..., Any]], on_shutdown: Optional[Callable[..., Any]],
|
||||
before_shutdown: Optional[Callable[..., Any]], stop_timeout: Optional[int]) -> Tuple[bool, bool]:
|
||||
postmaster = self.is_running()
|
||||
if not postmaster:
|
||||
if on_safepoint:
|
||||
@@ -774,7 +787,8 @@ class Postgresql(object):
|
||||
|
||||
return True, True
|
||||
|
||||
def terminate_postmaster(self, postmaster, mode, stop_timeout):
|
||||
def terminate_postmaster(self, postmaster: PostmasterProcess, mode: str,
|
||||
stop_timeout: Optional[int]) -> Optional[bool]:
|
||||
if mode in ['fast', 'smart']:
|
||||
try:
|
||||
success = postmaster.signal_stop('immediate', self.pgcommand('pg_ctl'))
|
||||
@@ -787,13 +801,13 @@ class Postgresql(object):
|
||||
logger.warning("Sending SIGKILL to Postmaster and its children")
|
||||
return postmaster.signal_kill()
|
||||
|
||||
def terminate_starting_postmaster(self, postmaster):
|
||||
def terminate_starting_postmaster(self, postmaster: PostmasterProcess) -> None:
|
||||
"""Terminates a postmaster that has not yet opened ports or possibly even written a pid file. Blocks
|
||||
until the process goes away."""
|
||||
postmaster.signal_stop('immediate', self.pgcommand('pg_ctl'))
|
||||
postmaster.wait()
|
||||
|
||||
def _wait_for_connection_close(self, postmaster):
|
||||
def _wait_for_connection_close(self, postmaster: PostmasterProcess) -> None:
|
||||
try:
|
||||
with self.connection().cursor() as cur:
|
||||
while postmaster.is_running(): # Need a timeout here?
|
||||
@@ -802,17 +816,17 @@ class Postgresql(object):
|
||||
except psycopg.Error:
|
||||
pass
|
||||
|
||||
def reload(self, block_callbacks=False):
|
||||
def reload(self, block_callbacks: bool = False) -> bool:
|
||||
ret = self.pg_ctl('reload')
|
||||
if ret and not block_callbacks:
|
||||
self.call_nowait(CallbackAction.ON_RELOAD)
|
||||
return ret
|
||||
|
||||
def check_for_startup(self):
|
||||
def check_for_startup(self) -> bool:
|
||||
"""Checks PostgreSQL status and returns if PostgreSQL is in the middle of startup."""
|
||||
return self.is_starting() and not self.check_startup_state_changed()
|
||||
|
||||
def check_startup_state_changed(self):
|
||||
def check_startup_state_changed(self) -> bool:
|
||||
"""Checks if PostgreSQL has completed starting up or failed or still starting.
|
||||
|
||||
Should only be called when state == 'starting'
|
||||
@@ -847,7 +861,7 @@ class Postgresql(object):
|
||||
|
||||
return True
|
||||
|
||||
def wait_for_startup(self, timeout=None):
|
||||
def wait_for_startup(self, timeout: float = 0) -> Optional[bool]:
|
||||
"""Waits for PostgreSQL startup to complete or fail.
|
||||
|
||||
:returns: True if start was successful, False otherwise"""
|
||||
@@ -862,8 +876,10 @@ class Postgresql(object):
|
||||
|
||||
return self.state == 'running'
|
||||
|
||||
def restart(self, timeout=None, task=None, block_callbacks=False,
|
||||
role=None, before_shutdown=None, after_start=None):
|
||||
def restart(self, timeout: Optional[float] = None, task: Optional[CriticalTask] = None,
|
||||
block_callbacks: bool = False, role: Optional[str] = None,
|
||||
before_shutdown: Optional[Callable[..., Any]] = None,
|
||||
after_start: Optional[Callable[..., Any]] = None) -> Optional[bool]:
|
||||
"""Restarts PostgreSQL.
|
||||
|
||||
When timeout parameter is set the call will block either until PostgreSQL has started, failed to start or
|
||||
@@ -880,13 +896,13 @@ class Postgresql(object):
|
||||
self.set_state('restart failed ({0})'.format(self.state))
|
||||
return ret
|
||||
|
||||
def is_healthy(self):
|
||||
def is_healthy(self) -> bool:
|
||||
if not self.is_running():
|
||||
logger.warning('Postgresql is not running.')
|
||||
return False
|
||||
return True
|
||||
|
||||
def get_guc_value(self, name):
|
||||
def get_guc_value(self, name: str) -> Optional[str]:
|
||||
cmd = [self.pgcommand('postgres'), '-D', self._data_dir, '-C', name,
|
||||
'--config-file={}'.format(self.config.postgresql_conf)]
|
||||
try:
|
||||
@@ -896,7 +912,7 @@ class Postgresql(object):
|
||||
except Exception as e:
|
||||
logger.error('Failed to execute %s: %r', cmd, e)
|
||||
|
||||
def controldata(self):
|
||||
def controldata(self) -> Dict[str, str]:
|
||||
""" return the contents of pg_controldata, or non-True value if pg_controldata call failed """
|
||||
# Don't try to call pg_controldata during backup restore
|
||||
if self._version_file_exists() and self.state != 'creating replica':
|
||||
@@ -911,11 +927,11 @@ class Postgresql(object):
|
||||
logger.exception("Error when calling pg_controldata")
|
||||
return {}
|
||||
|
||||
def waldump(self, timeline, lsn, limit):
|
||||
def waldump(self, timeline: Union[int, str], lsn: str, limit: int) -> Tuple[Optional[bytes], Optional[bytes]]:
|
||||
cmd = self.pgcommand('pg_{0}dump'.format(self.wal_name))
|
||||
env = {**os.environ, 'LANG': 'C', 'LC_ALL': 'C', 'PGDATA': self._data_dir}
|
||||
try:
|
||||
waldump = subprocess.Popen([cmd, '-t', str(timeline), '-s', str(lsn), '-n', str(limit)],
|
||||
waldump = subprocess.Popen([cmd, '-t', str(timeline), '-s', lsn, '-n', str(limit)],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
||||
out, err = waldump.communicate()
|
||||
waldump.wait()
|
||||
@@ -925,22 +941,24 @@ class Postgresql(object):
|
||||
return None, None
|
||||
|
||||
@contextmanager
|
||||
def get_replication_connection_cursor(self, host=None, port=5432, **kwargs):
|
||||
def get_replication_connection_cursor(self, host: Optional[str] = None, port: int = 5432,
|
||||
**kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
|
||||
conn_kwargs = self.config.replication.copy()
|
||||
conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
|
||||
connect_timeout=3, replication=1, options='-c statement_timeout=2000')
|
||||
with get_connection_cursor(**conn_kwargs) as cur:
|
||||
yield cur
|
||||
|
||||
def get_replica_timeline(self):
|
||||
def get_replica_timeline(self) -> Optional[int]:
|
||||
try:
|
||||
with self.get_replication_connection_cursor(**self.config.local_replication_address) as cur:
|
||||
cur.execute('IDENTIFY_SYSTEM')
|
||||
return cur.fetchone()[1]
|
||||
row = cur.fetchone()
|
||||
return row[1] if row else None
|
||||
except Exception:
|
||||
logger.exception('Can not fetch local timeline and lsn from replication connection')
|
||||
|
||||
def replica_cached_timeline(self, primary_timeline):
|
||||
def replica_cached_timeline(self, primary_timeline: Optional[int]) -> Optional[int]:
|
||||
if not self._cached_replica_timeline or not primary_timeline\
|
||||
or self._cached_replica_timeline != primary_timeline:
|
||||
self._cached_replica_timeline = self.get_replica_timeline()
|
||||
@@ -948,29 +966,26 @@ class Postgresql(object):
|
||||
|
||||
def get_primary_timeline(self) -> int:
|
||||
""":returns: current timeline if postgres is running as a primary or 0."""
|
||||
return self._cluster_info_state_get('timeline')
|
||||
return self._cluster_info_state_get('timeline') or 0
|
||||
|
||||
def get_history(self, timeline):
|
||||
def get_history(self, timeline: int) -> List[Union[Tuple[int, int, str], Tuple[int, int, str, str, str]]]:
|
||||
history_path = os.path.join(self.wal_dir, '{0:08X}.history'.format(timeline))
|
||||
history_mtime = mtime(history_path)
|
||||
history: List[Union[Tuple[int, int, str], Tuple[int, int, str, str, str]]] = []
|
||||
if history_mtime:
|
||||
try:
|
||||
with open(history_path, 'r') as f:
|
||||
history = f.read()
|
||||
history = list(parse_history(history))
|
||||
history_content = f.read()
|
||||
history = list(parse_history(history_content))
|
||||
if history[-1][0] == timeline - 1:
|
||||
history_mtime = datetime.fromtimestamp(history_mtime).replace(tzinfo=tz.tzlocal())
|
||||
history[-1].append(history_mtime.isoformat())
|
||||
history[-1].append(self.name)
|
||||
return history
|
||||
history[-1] = history[-1][:3] + (history_mtime.isoformat(), self.name)
|
||||
except Exception:
|
||||
logger.exception('Failed to read and parse %s', (history_path,))
|
||||
return history
|
||||
|
||||
def follow(self,
|
||||
member: Member,
|
||||
role: Optional[str] = 'replica',
|
||||
timeout: Optional[float] = None,
|
||||
do_reload: Optional[bool] = False) -> Optional[bool]:
|
||||
def follow(self, member: Union[Leader, Member, None], role: str = 'replica',
|
||||
timeout: Optional[float] = None, do_reload: bool = False) -> Optional[bool]:
|
||||
"""Reconfigure postgres to follow a new member or use different recovery parameters.
|
||||
|
||||
Method may call `on_role_change` callback if role is changing.
|
||||
@@ -1016,14 +1031,14 @@ class Postgresql(object):
|
||||
self.call_nowait(CallbackAction.ON_ROLE_CHANGE)
|
||||
return ret
|
||||
|
||||
def _wait_promote(self, wait_seconds):
|
||||
def _wait_promote(self, wait_seconds: int) -> Optional[bool]:
|
||||
for _ in polling_loop(wait_seconds):
|
||||
data = self.controldata()
|
||||
if data.get('Database cluster state') == 'in production':
|
||||
self.set_role('master')
|
||||
return True
|
||||
|
||||
def _pre_promote(self):
|
||||
def _pre_promote(self) -> bool:
|
||||
"""
|
||||
Runs a fencing script after the leader lock is acquired but before the replica is promoted.
|
||||
If the script exits with a non-zero code, promotion does not happen and the leader key is removed from DCS.
|
||||
@@ -1053,7 +1068,8 @@ class Postgresql(object):
|
||||
except Exception as e:
|
||||
logger.error('Exception when calling `%s`: %r', cmd, e)
|
||||
|
||||
def promote(self, wait_seconds, task, before_promote=None, on_success=None):
|
||||
def promote(self, wait_seconds: int, task: CriticalTask, before_promote: Optional[Callable[..., Any]] = None,
|
||||
on_success: Optional[Callable[..., Any]] = None) -> Optional[bool]:
|
||||
if self.role in ('promoted', 'master', 'primary'):
|
||||
return True
|
||||
|
||||
@@ -1086,43 +1102,48 @@ class Postgresql(object):
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def _wal_position(is_leader, wal_position, received_location, replayed_location):
|
||||
def _wal_position(is_leader: bool, wal_position: int,
|
||||
received_location: Optional[int], replayed_location: Optional[int]) -> int:
|
||||
return wal_position if is_leader else max(received_location or 0, replayed_location or 0)
|
||||
|
||||
def timeline_wal_position(self):
|
||||
def timeline_wal_position(self) -> Tuple[int, int, Optional[int]]:
|
||||
# This method could be called from different threads (simultaneously with some other `_query` calls).
|
||||
# If it is called not from main thread we will create a new cursor to execute statement.
|
||||
if current_thread().ident == self.__thread_ident:
|
||||
timeline = self._cluster_info_state_get('timeline')
|
||||
wal_position = self._cluster_info_state_get('wal_position')
|
||||
timeline = self._cluster_info_state_get('timeline') or 0
|
||||
wal_position = self._cluster_info_state_get('wal_position') or 0
|
||||
replayed_location = self.replayed_location()
|
||||
received_location = self.received_location()
|
||||
pg_control_timeline = self._cluster_info_state_get('pg_control_timeline')
|
||||
else:
|
||||
with self.connection().cursor() as cursor:
|
||||
cursor.execute(self.cluster_info_query)
|
||||
(timeline, wal_position, replayed_location,
|
||||
received_location, _, pg_control_timeline) = cursor.fetchone()[:6]
|
||||
cursor.execute(self.cluster_info_query.encode('utf-8'))
|
||||
row = cursor.fetchone()
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
assert row is not None
|
||||
(timeline, wal_position, replayed_location, received_location, _, pg_control_timeline) = row[:6]
|
||||
|
||||
wal_position = self._wal_position(timeline, wal_position, received_location, replayed_location)
|
||||
wal_position = self._wal_position(bool(timeline), wal_position, received_location, replayed_location)
|
||||
return (timeline, wal_position, pg_control_timeline)
|
||||
|
||||
def postmaster_start_time(self):
|
||||
def postmaster_start_time(self) -> Optional[str]:
|
||||
try:
|
||||
query = "SELECT " + self.POSTMASTER_START_TIME
|
||||
if current_thread().ident == self.__thread_ident:
|
||||
return self.query(query).fetchone()[0].isoformat(sep=' ')
|
||||
with self.connection().cursor() as cursor:
|
||||
cursor.execute(query)
|
||||
return cursor.fetchone()[0].isoformat(sep=' ')
|
||||
row = self.query(query).fetchone()
|
||||
else:
|
||||
with self.connection().cursor() as cursor:
|
||||
cursor.execute(query)
|
||||
row = cursor.fetchone()
|
||||
return row[0].isoformat(sep=' ') if row else None
|
||||
except psycopg.Error:
|
||||
return None
|
||||
|
||||
def last_operation(self):
|
||||
return self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position'),
|
||||
def last_operation(self) -> int:
|
||||
return self._wal_position(self.is_leader(), self._cluster_info_state_get('wal_position') or 0,
|
||||
self.received_location(), self.replayed_location())
|
||||
|
||||
def configure_server_parameters(self):
|
||||
def configure_server_parameters(self) -> None:
|
||||
self._major_version = self.get_major_version()
|
||||
self.config.setup_server_parameters()
|
||||
|
||||
@@ -1135,9 +1156,9 @@ class Postgresql(object):
|
||||
self.configure_server_parameters()
|
||||
return self._major_version > 0
|
||||
|
||||
def pg_wal_realpath(self):
|
||||
def pg_wal_realpath(self) -> Dict[str, str]:
|
||||
"""Returns a dict containing the symlink (key) and target (value) for the wal directory"""
|
||||
links = {}
|
||||
links: Dict[str, str] = {}
|
||||
for pg_wal_dir in ('pg_xlog', 'pg_wal'):
|
||||
pg_wal_path = os.path.join(self._data_dir, pg_wal_dir)
|
||||
if os.path.exists(pg_wal_path) and os.path.islink(pg_wal_path):
|
||||
@@ -1145,9 +1166,9 @@ class Postgresql(object):
|
||||
links[pg_wal_path] = pg_wal_realpath
|
||||
return links
|
||||
|
||||
def pg_tblspc_realpaths(self):
|
||||
def pg_tblspc_realpaths(self) -> Dict[str, str]:
|
||||
"""Returns a dict containing the symlink (key) and target (values) for the tablespaces"""
|
||||
links = {}
|
||||
links: Dict[str, str] = {}
|
||||
pg_tblsp_dir = os.path.join(self._data_dir, 'pg_tblspc')
|
||||
if os.path.exists(pg_tblsp_dir):
|
||||
for tsdn in os.listdir(pg_tblsp_dir):
|
||||
@@ -1157,7 +1178,7 @@ class Postgresql(object):
|
||||
links[pg_tsp_path] = pg_tsp_rpath
|
||||
return links
|
||||
|
||||
def move_data_directory(self):
|
||||
def move_data_directory(self) -> None:
|
||||
if os.path.isdir(self._data_dir) and not self.is_running():
|
||||
try:
|
||||
postfix = 'failed'
|
||||
@@ -1191,7 +1212,7 @@ class Postgresql(object):
|
||||
except OSError:
|
||||
logger.exception("Could not rename data directory %s", self._data_dir)
|
||||
|
||||
def remove_data_directory(self):
|
||||
def remove_data_directory(self) -> None:
|
||||
self.set_role('uninitialized')
|
||||
logger.info('Removing data directory: %s', self._data_dir)
|
||||
try:
|
||||
@@ -1219,7 +1240,7 @@ class Postgresql(object):
|
||||
logger.exception('Could not remove data directory %s', self._data_dir)
|
||||
self.move_data_directory()
|
||||
|
||||
def schedule_sanity_checks_after_pause(self):
|
||||
def schedule_sanity_checks_after_pause(self) -> None:
|
||||
"""
|
||||
After coming out of pause we have to:
|
||||
1. configure server parameters if necessary
|
||||
@@ -1229,4 +1250,4 @@ class Postgresql(object):
|
||||
self.ensure_major_version_is_known()
|
||||
self.slots_handler.schedule()
|
||||
self.citus_handler.schedule_cache_rebuild()
|
||||
self._sysid = None
|
||||
self._sysid = ''
|
||||
|
||||
Reference in New Issue
Block a user