diff --git a/patroni/__main__.py b/patroni/__main__.py index 2cc2f7d6..07bc0dc8 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -212,13 +212,15 @@ class Patroni(AbstractPatroniDaemon, Tags): the change and cache the new dynamic configuration values in ``patroni.dynamic.json`` file under Postgres data directory. """ + from patroni.postgresql.misc import PostgresqlRole + logger.info(self.ha.run_cycle()) if self.dcs.cluster and self.dcs.cluster.config and self.dcs.cluster.config.data \ and self.config.set_dynamic_configuration(self.dcs.cluster.config): self.reload_config() - if self.postgresql.role != 'uninitialized': + if self.postgresql.role != PostgresqlRole.UNINITIALIZED: self.config.save_cache() self.schedule_next_run() diff --git a/patroni/api.py b/patroni/api.py index f2effde5..7df9b257 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -30,7 +30,7 @@ from . import global_config, psycopg from .__main__ import Patroni from .dcs import Cluster from .exceptions import PostgresConnectionException, PostgresException -from .postgresql.misc import postgres_version_to_int, PostgresqlState +from .postgresql.misc import postgres_version_to_int, PostgresqlRole, PostgresqlState from .utils import cluster_as_json, deep_compare, enable_keepalive, parse_bool, \ parse_int, patch_config, Retry, RetryFailedError, split_host_port, tzutc, uri @@ -324,17 +324,19 @@ class RestApiHandler(BaseHTTPRequestHandler): is_lagging = leader_optime and leader_optime > replayed_location + max_replica_lag replica_status_code = 200 if not patroni.noloadbalance and not is_lagging and \ - response.get('role') == 'replica' and response.get('state') == PostgresqlState.RUNNING else 503 + response.get('role') == PostgresqlRole.REPLICA and response.get('state') == PostgresqlState.RUNNING else 503 if not cluster and response.get('pause'): - leader_status_code = 200 if response.get('role') in ('primary', 'standby_leader') else 503 - primary_status_code = 200 if response.get('role') == 'primary' else 503 - standby_leader_status_code = 200 if response.get('role') == 'standby_leader' else 503 + leader_status_code = 200 if response.get('role') in (PostgresqlRole.PRIMARY, + PostgresqlRole.STANDBY_LEADER) else 503 + primary_status_code = 200 if response.get('role') == PostgresqlRole.PRIMARY else 503 + standby_leader_status_code = 200 if response.get('role') == PostgresqlRole.STANDBY_LEADER else 503 elif patroni.ha.is_leader(): leader_status_code = 200 if config.is_standby_cluster: primary_status_code = replica_status_code = 503 - standby_leader_status_code = 200 if response.get('role') in ('replica', 'standby_leader') else 503 + standby_leader_status_code =\ + 200 if response.get('role') in (PostgresqlRole.REPLICA, PostgresqlRole.STANDBY_LEADER) else 503 else: primary_status_code = 200 standby_leader_status_code = 503 @@ -436,7 +438,7 @@ class RestApiHandler(BaseHTTPRequestHandler): """ patroni: Patroni = self.server.patroni - is_primary = patroni.postgresql.role == 'primary' and patroni.postgresql.is_running() + is_primary = patroni.postgresql.role == PostgresqlRole.PRIMARY and patroni.postgresql.is_running() # We can tolerate Patroni problems longer on the replica. # On the primary the liveness probe most likely will start failing only after the leader key expired. # It should not be a big problem because replicas will see that the primary is still alive via REST API call. @@ -584,7 +586,7 @@ class RestApiHandler(BaseHTTPRequestHandler): 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(labels, int(postgres['role'] == 'primary'))) + metrics.append("patroni_primary{0} {1}".format(labels, int(postgres['role'] == PostgresqlRole.PRIMARY))) metrics.append("# HELP patroni_xlog_location Current location of the Postgres" " transaction log, 0 if this node is not the leader.") @@ -593,11 +595,12 @@ class RestApiHandler(BaseHTTPRequestHandler): 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(labels, int(postgres['role'] == 'standby_leader'))) + metrics.append("patroni_standby_leader{0} {1}".format(labels, + int(postgres['role'] == PostgresqlRole.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(labels, int(postgres['role'] == 'replica'))) + metrics.append("patroni_replica{0} {1}".format(labels, int(postgres['role'] == PostgresqlRole.REPLICA))) metrics.append("# HELP patroni_sync_standby Value is 1 if this node is a sync standby, 0 otherwise.") metrics.append("# TYPE patroni_sync_standby gauge") @@ -910,7 +913,7 @@ class RestApiHandler(BaseHTTPRequestHandler): status_code = _ break elif k == 'role': - if request[k] not in ('primary', 'standby_leader', 'replica'): + if request[k] not in (PostgresqlRole.PRIMARY, PostgresqlRole.STANDBY_LEADER, PostgresqlRole.REPLICA): status_code = 400 data = "PostgreSQL role should be either primary, standby_leader, or replica" break @@ -1266,16 +1269,17 @@ class RestApiHandler(BaseHTTPRequestHandler): * ``state``: one of :class:`~patroni.postgresql.misc.PostgresqlState` or ``unknown``; * ``postmaster_start_time``: ``pg_postmaster_start_time()``; - * ``role``: ``replica`` or ``primary`` based on ``pg_is_in_recovery()`` output; + * ``role``: :class:`~patroni.postgresql.misc.PostgresqlRole.REPLICA` or + :class:`~patroni.postgresql.misc.PostgresqlRole.PRIMARY` based on ``pg_is_in_recovery()`` output; * ``server_version``: Postgres version without periods, e.g. ``150002`` for Postgres ``15.2``; * ``latest_end_lsn``: latest_end_lsn value from ``pg_stat_get_wal_receiver()``, only on replica nodes; * ``xlog``: dictionary. Its structure depends on ``role``: - * If ``primary``: + * If :class:`~patroni.postgresql.misc.PostgresqlRole.PRIMARY`: * ``location``: ``pg_current_wal_flush_lsn()`` - * If ``replica``: + * If :class:`~patroni.postgresql.misc.PostgresqlRole.REPLICA`: * ``received_location``: ``pg_wal_lsn_diff(pg_last_wal_receive_lsn(), '0/0')``; * ``replayed_location``: ``pg_wal_lsn_diff(pg_last_wal_replay_lsn(), '0/0)``; @@ -1327,7 +1331,7 @@ class RestApiHandler(BaseHTTPRequestHandler): result = { 'state': postgresql.state, 'postmaster_start_time': row[0], - 'role': 'replica' if row[1] == 0 else 'primary', + 'role': PostgresqlRole.REPLICA if row[1] == 0 else PostgresqlRole.PRIMARY, 'server_version': postgresql.server_version, 'xlog': ({ 'received_location': row[10] or row[4] or row[3], @@ -1338,10 +1342,10 @@ class RestApiHandler(BaseHTTPRequestHandler): }) } - if result['role'] == 'replica' and config.is_standby_cluster: + if result['role'] == PostgresqlRole.REPLICA and config.is_standby_cluster: result['role'] = postgresql.role - if result['role'] == 'replica' and config.is_synchronous_mode\ + if result['role'] == PostgresqlRole.REPLICA and config.is_synchronous_mode\ and cluster and cluster.sync.matches(postgresql.name): result['quorum_standby' if global_config.is_quorum_commit_mode else 'sync_standby'] = True diff --git a/patroni/ctl.py b/patroni/ctl.py index 5c363d1d..b32da0fd 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -29,6 +29,7 @@ import time from collections import defaultdict from contextlib import contextmanager +from enum import Enum from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING, Union from urllib.parse import urlparse @@ -64,7 +65,7 @@ from . import global_config from .config import Config from .dcs import AbstractDCS, Cluster, get_dcs as _get_dcs, Member from .exceptions import PatroniException -from .postgresql.misc import postgres_version_to_int, PostgresqlState +from .postgresql.misc import postgres_version_to_int, PostgresqlRole, PostgresqlState from .postgresql.mpp import get_mpp from .request import PatroniRequest from .utils import cluster_as_json, patch_config, polling_loop @@ -80,6 +81,20 @@ DCS_DEFAULTS: Dict[str, Dict[str, Any]] = { 'etcd3': {'port': 2379, 'template': "etcd3:\n host: '{host}:{port}'"}} +class CtlPostgresqlRole(str, Enum): + + LEADER = 'leader' + PRIMARY = 'primary' + STANDBY_LEADER = 'standby-leader' + REPLICA = 'replica' + STANDBY = 'standby' + ANY = 'any' + + def __repr__(self) -> str: + """Get a string representation of a :class:`CtlPostgresqlRole` member.""" + return self.value + + class PatroniCtlException(click.ClickException): """Raised upon issues faced by ``patronictl`` utility.""" @@ -289,7 +304,7 @@ arg_cluster_name = click.argument('cluster_name', required=False, option_default_citus_group = click.option('--group', required=False, type=int, help='Citus group', default=lambda: _get_configuration().get('citus', {}).get('group')) option_citus_group = click.option('--group', required=False, type=int, help='Citus group') -role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 'standby', 'any']) +role_choice = click.Choice([role.value for role in CtlPostgresqlRole]) @click.group(cls=click.Group) @@ -493,45 +508,42 @@ def watching(w: bool, watch: Optional[int], max_count: Optional[int] = None, cle yield 0 -def get_all_members(cluster: Cluster, group: Optional[int], role: str = 'leader') -> Iterator[Member]: +def get_all_members(cluster: Cluster, group: Optional[int], + role: CtlPostgresqlRole = CtlPostgresqlRole.LEADER) -> Iterator[Member]: """Get all cluster members that have the given *role*. :param cluster: the Patroni cluster. :param group: filter which Citus group we should get members from. If ``None`` get from all groups. - :param role: role to filter members. Can be one among: - - * ``primary``: the primary PostgreSQL instance; - * ``replica`` or ``standby``: a standby PostgreSQL instance; - * ``leader``: the leader of a Patroni cluster. Can also be used to get the leader of a Patroni standby cluster; - * ``standby-leader``: the leader of a Patroni standby cluster; - * ``any``: matches any node independent of its role. + :param role: role to filter members. One of :class:`CtlPostgresqlRole` values. :yields: members that have the given *role*. """ clusters = {0: cluster} if is_citus_cluster() and group is None: clusters.update(cluster.workers) - if role in ('leader', 'primary', 'standby-leader'): + if role in (CtlPostgresqlRole.LEADER, CtlPostgresqlRole.PRIMARY, CtlPostgresqlRole.STANDBY_LEADER): # In the DCS the members' role can be one among: ``primary``, ``master``, ``replica`` or ``standby_leader``. # ``primary`` and ``master`` are the same thing. - role = {'standby-leader': 'standby_leader'}.get(role, role) for cluster in clusters.values(): if cluster.leader is not None and cluster.leader.name and\ - (role == 'leader' - or cluster.leader.data.get('role') not in ('primary', 'master') and role == 'standby_leader' - or cluster.leader.data.get('role') != 'standby_leader' and role == 'primary'): + (role == CtlPostgresqlRole.LEADER + or cluster.leader.data.get('role') not in (PostgresqlRole.PRIMARY, PostgresqlRole.MASTER) + and role == CtlPostgresqlRole.STANDBY_LEADER + or cluster.leader.data.get('role') != PostgresqlRole.STANDBY_LEADER + and role == CtlPostgresqlRole.PRIMARY): yield cluster.leader.member return for cluster in clusters.values(): leader_name = (cluster.leader.member.name if cluster.leader else None) for m in cluster.members: - if role == 'any' or role in ('replica', 'standby') and m.name != leader_name: + if role == CtlPostgresqlRole.ANY or\ + role in (CtlPostgresqlRole.REPLICA, CtlPostgresqlRole.STANDBY) and m.name != leader_name: yield m def get_any_member(cluster: Cluster, group: Optional[int], - role: Optional[str] = None, member: Optional[str] = None) -> Optional[Member]: + role: Optional[CtlPostgresqlRole] = None, member: Optional[str] = None) -> Optional[Member]: """Get the first found cluster member that has the given *role*. :param cluster: the Patroni cluster. @@ -547,9 +559,9 @@ def get_any_member(cluster: Cluster, group: Optional[int], if member is not None: if role is not None: raise PatroniCtlException('--role and --member are mutually exclusive options') - role = 'any' + role = CtlPostgresqlRole.ANY elif role is None: - role = 'leader' + role = CtlPostgresqlRole.LEADER for m in get_all_members(cluster, group, role): if member is None or m.name == member: @@ -573,7 +585,8 @@ def get_all_members_leader_first(cluster: Cluster) -> Iterator[Member]: def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[str, Any], - role: Optional[str] = None, member_name: Optional[str] = None) -> Union['cursor', 'Cursor[Any]', None]: + role: Optional[CtlPostgresqlRole] = None, + member_name: Optional[str] = None) -> Union['cursor', 'Cursor[Any]', None]: """Get a cursor object to execute queries against a member that has the given *role* or *member_name*. .. note:: @@ -612,7 +625,7 @@ def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[ # If we want ``any`` node we are fine to return the cursor. ``None`` is similar to ``any`` at this point, as it's # been dealt with through :func:`get_any_member`. # If we want the Patroni leader node, :func:`get_any_member` already checks that for us - if role in (None, 'any', 'leader'): + if role in (None, CtlPostgresqlRole.ANY, CtlPostgresqlRole.LEADER): return cursor # If we want something other than ``any`` or ``leader``, then we do not rely only on the DCS information about @@ -621,7 +634,9 @@ def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[ row = cursor.fetchone() in_recovery = not row or row[0] - if in_recovery and role in ('replica', 'standby', 'standby-leader') or not in_recovery and role == 'primary': + if in_recovery and\ + role in (CtlPostgresqlRole.REPLICA, CtlPostgresqlRole.STANDBY, CtlPostgresqlRole.STANDBY_LEADER) or\ + not in_recovery and role == CtlPostgresqlRole.PRIMARY: return cursor conn.close() @@ -629,7 +644,7 @@ def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[ return None -def get_members(cluster: Cluster, cluster_name: str, member_names: List[str], role: str, +def get_members(cluster: Cluster, cluster_name: str, member_names: List[str], role: CtlPostgresqlRole, force: bool, action: str, ask_confirmation: bool = True, group: Optional[int] = None) -> List[Member]: """Get the list of members based on the given filters. @@ -752,7 +767,7 @@ def confirm_members_action(members: List[Member], force: bool, action: str, @click.option('--member', '-m', help='Generate a dsn for this member', type=str) @arg_cluster_name @option_citus_group -def dsn(cluster_name: str, group: Optional[int], role: Optional[str], member: Optional[str]) -> None: +def dsn(cluster_name: str, group: Optional[int], role: Optional[CtlPostgresqlRole], member: Optional[str]) -> None: """Process ``dsn`` command of ``patronictl`` utility. Get DSN to connect to *member*. @@ -798,7 +813,7 @@ def dsn(cluster_name: str, group: Optional[int], role: Optional[str], member: Op def query( cluster_name: str, group: Optional[int], - role: Optional[str], + role: Optional[CtlPostgresqlRole], member: Optional[str], w: bool, watch: Optional[int], @@ -865,7 +880,7 @@ def query( def query_member(cluster: Cluster, group: Optional[int], cursor: Union['cursor', 'Cursor[Any]', None], - member: Optional[str], role: Optional[str], command: str, + member: Optional[str], role: Optional[CtlPostgresqlRole], command: str, connect_parameters: Dict[str, Any]) -> Tuple[List[List[Any]], Optional[List[Any]]]: """Execute SQL *command* against a member. @@ -903,7 +918,7 @@ def query_member(cluster: Cluster, group: Optional[int], cursor: Union['cursor', if member is not None: message = f'No connection to member {member} is available' elif role is not None: - message = f'No connection to role {role} is available' + message = f'No connection to role {role!r} is available' else: message = 'No connection is available' logging.debug(message) @@ -1030,9 +1045,11 @@ def parse_scheduled(scheduled: Optional[str]) -> Optional[datetime.datetime]: @click.argument('cluster_name') @click.argument('member_names', nargs=-1) @option_citus_group -@click.option('--role', '-r', help='Reload only members with this role', type=role_choice, default='any') +@click.option('--role', '-r', help='Reload only members with this role', + type=role_choice, default=CtlPostgresqlRole.ANY) @option_force -def reload(cluster_name: str, member_names: List[str], group: Optional[int], force: bool, role: str) -> None: +def reload(cluster_name: str, member_names: List[str], group: Optional[int], + force: bool, role: CtlPostgresqlRole) -> None: """Process ``reload`` command of ``patronictl`` utility. Reload configuration of cluster members based on given filters. @@ -1067,7 +1084,8 @@ def reload(cluster_name: str, member_names: List[str], group: Optional[int], for @click.argument('cluster_name') @click.argument('member_names', nargs=-1) @option_citus_group -@click.option('--role', '-r', help='Restart only members with this role', type=role_choice, default='any') +@click.option('--role', '-r', help='Restart only members with this role', type=role_choice, + default=CtlPostgresqlRole.ANY) @click.option('--any', 'p_any', help='Restart a single member only', is_flag=True) @click.option('--scheduled', help='Timestamp of a scheduled restart in unambiguous format (e.g. ISO 8601)', default=None) @@ -1077,7 +1095,7 @@ def reload(cluster_name: str, member_names: List[str], group: Optional[int], for @click.option('--timeout', help='Return error and fail over if necessary when restarting takes longer than this.') @option_force def restart(cluster_name: str, group: Optional[int], member_names: List[str], - force: bool, role: str, p_any: bool, scheduled: Optional[str], version: Optional[str], + force: bool, role: CtlPostgresqlRole, p_any: bool, scheduled: Optional[str], version: Optional[str], pending: bool, timeout: Optional[str]) -> None: """Process ``restart`` command of ``patronictl`` utility. @@ -1184,7 +1202,8 @@ def reinit(cluster_name: str, group: Optional[int], member_names: List[str], for :param wait: wait for the operation to complete. """ cluster = get_dcs(cluster_name, group).get_cluster() - members = get_members(cluster, cluster_name, member_names, 'replica', force, 'reinitialize', group=group) + members = get_members(cluster, cluster_name, member_names, CtlPostgresqlRole.REPLICA, + force, 'reinitialize', group=group) wait_on_members: List[Member] = [] for member in members: @@ -1709,10 +1728,10 @@ def timestamp(precision: int = 6) -> str: @option_citus_group @click.argument('member_names', nargs=-1) @click.argument('target', type=click.Choice(['restart', 'switchover'])) -@click.option('--role', '-r', help='Flush only members with this role', type=role_choice, default='any') +@click.option('--role', '-r', help='Flush only members with this role', type=role_choice, default=CtlPostgresqlRole.ANY) @option_force def flush(cluster_name: str, group: Optional[int], - member_names: List[str], force: bool, role: str, target: str) -> None: + member_names: List[str], force: bool, role: CtlPostgresqlRole, target: str) -> None: """Process ``flush`` command of ``patronictl`` utility. Discard scheduled restart or switchover events. @@ -2193,7 +2212,7 @@ def version(cluster_name: str, group: Optional[int], member_names: List[str]) -> click.echo("") cluster = get_dcs(cluster_name, group).get_cluster() - for m in get_all_members(cluster, group, 'any'): + for m in get_all_members(cluster, group, CtlPostgresqlRole.ANY): if m.api_url: if not member_names or m.name in member_names: try: diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 1aeb5eae..0e6d14a1 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -25,6 +25,7 @@ from ..utils import deep_compare, parse_int, uri if TYPE_CHECKING: # pragma: no cover from ..config import Config from ..postgresql import Postgresql + from ..postgresql.misc import PostgresqlRole from ..postgresql.mpp import AbstractMPP slot_name_re = re.compile('^[a-z0-9_]{1,63}$') @@ -416,10 +417,13 @@ class Leader(NamedTuple): >>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote """ + from ..postgresql.misc import PostgresqlRole + version = self.member.patroni_version # 1.5.6 is the last version which doesn't expose checkpoint_after_promote: false if version and version > (1, 5, 6): - return self.data.get('role') in ('master', 'primary') and 'checkpoint_after_promote' not in self.data + return self.data.get('role') in (PostgresqlRole.MASTER, PostgresqlRole.PRIMARY) \ + and 'checkpoint_after_promote' not in self.data return None @@ -1010,7 +1014,8 @@ class Cluster(NamedTuple('Cluster', return {name: value for name, value in self.__permanent_slots.items() if self.is_logical_slot(value)} def get_replication_slots(self, postgresql: 'Postgresql', member: Tags, *, - role: Optional[str] = None, show_error: bool = False) -> Dict[str, Dict[str, Any]]: + role: Optional['PostgresqlRole'] = None, + show_error: bool = False) -> Dict[str, Dict[str, Any]]: """Lookup configured slot names in the DCS, report issues found and merge with permanent slots. Will log an error if: @@ -1019,7 +1024,8 @@ class Cluster(NamedTuple('Cluster', :param postgresql: reference to :class:`Postgresql` object. :param member: reference to an object implementing :class:`Tags` interface. - :param role: role of the node, if not set will be taken from *postgresql*. + :param role: role of the node, if not set will be taken from *postgresql* + One of :class:`~patroni.postgresql.misc.PostgresqlRole` values. :param show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found. :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks. @@ -1041,7 +1047,7 @@ class Cluster(NamedTuple('Cluster', return slots def _merge_permanent_slots(self, slots: Dict[str, Dict[str, Any]], permanent_slots: Dict[str, Any], - name: str, role: str, can_advance_slots: bool) -> List[str]: + name: str, role: 'PostgresqlRole', can_advance_slots: bool) -> List[str]: """Merge replication *slots* for members with *permanent_slots*. Perform validation of configured permanent slot name, skipping invalid names. @@ -1051,13 +1057,15 @@ class Cluster(NamedTuple('Cluster', :param slots: Slot names with existing attributes if known. :param name: name of this node. - :param role: role of the node -- ``primary``, ``standby_leader`` or ``replica``. + :param role: role of the node. One of :class:`~patroni.postgresql.misc.PostgresqlRole` values. :param permanent_slots: dictionary containing slot name key and slot information values. :param can_advance_slots: ``True`` if ``pg_replication_slot_advance()`` function is available, ``False`` otherwise. :returns: List of disabled permanent, logical slot names, if postgresql version < 11. """ + from ..postgresql.misc import PostgresqlRole + name = slot_name_from_member_name(name) topology = {slot_name_from_member_name(m.name): m.replicatefrom and slot_name_from_member_name(m.replicatefrom) for m in self.members} @@ -1084,7 +1092,8 @@ class Cluster(NamedTuple('Cluster', # case we should have the following: A(B: active, C: inactive) <- B (C: active) <- C # We don't consider the same situation on node B, because if node C doesn't exists, we will not # be able to know its `replicatefrom` tag value. - expected_active = not topology.get(slot_name) and role in ('primary', 'standby_leader') + expected_active = not topology.get(slot_name) and role in (PostgresqlRole.PRIMARY, + PostgresqlRole.STANDBY_LEADER) slots[slot_name] = {**value, 'expected_active': expected_active} continue @@ -1101,7 +1110,7 @@ class Cluster(NamedTuple('Cluster', logger.error("Bad value for slot '%s' in permanent_slots: %s", slot_name, permanent_slots[slot_name]) return disabled_permanent_logical_slots - def _get_permanent_slots(self, postgresql: 'Postgresql', tags: Tags, role: str) -> Dict[str, Any]: + def _get_permanent_slots(self, postgresql: 'Postgresql', tags: Tags, role: 'PostgresqlRole') -> Dict[str, Any]: """Get configured permanent replication slots. .. note:: @@ -1117,20 +1126,23 @@ class Cluster(NamedTuple('Cluster', :param postgresql: reference to :class:`Postgresql` object. :param tags: reference to an object implementing :class:`Tags` interface. - :param role: role of the node -- ``primary``, ``standby_leader`` or ``replica``. + :param role: role of the node. One of :class:`~patroni.postgresql.misc.PostgresqlRole` values. :returns: dictionary of permanent slot names mapped to attributes. """ + from ..postgresql.misc import PostgresqlRole + if not global_config.use_slots or tags.nofailover: return {} if global_config.is_standby_cluster or self.get_slot_name_on_primary(postgresql.name, tags) is None: - return self.permanent_physical_slots if postgresql.can_advance_slots or role == 'standby_leader' else {} + return self.permanent_physical_slots\ + if postgresql.can_advance_slots or role == PostgresqlRole.STANDBY_LEADER else {} - return self.__permanent_slots if postgresql.can_advance_slots or role == 'primary' \ + return self.__permanent_slots if postgresql.can_advance_slots or role == PostgresqlRole.PRIMARY \ else self.__permanent_logical_slots - def _get_members_slots(self, name: str, role: str, nofailover: bool, + def _get_members_slots(self, name: str, role: 'PostgresqlRole', nofailover: bool, can_advance_slots: bool) -> Dict[str, Dict[str, Any]]: """Get physical replication slots configuration for a given member. @@ -1162,6 +1174,8 @@ class Cluster(NamedTuple('Cluster', :returns: dictionary of physical replication slots that should exist on a given node. """ + from ..postgresql.misc import PostgresqlRole + if not global_config.use_slots: return {} @@ -1195,7 +1209,8 @@ class Cluster(NamedTuple('Cluster', # In case when retention of replication slots is possible the `expected_active` function # will be used to figure out whether the replication slot is expected to be active. # Otherwise it will be used to find replication slots that should exist on a current node. - expected_active = leader_filter if role in ('primary', 'standby_leader') else replica_filter + expected_active = leader_filter if role in (PostgresqlRole.PRIMARY, PostgresqlRole.STANDBY_LEADER) \ + else replica_filter if can_advance_slots and global_config.member_slots_ttl > 0: # if the node does only cascading and can't become the leader, we @@ -1239,7 +1254,9 @@ class Cluster(NamedTuple('Cluster', the node that we are checking permanent logical replication slots for. :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``. """ - role = 'replica' + from ..postgresql.misc import PostgresqlRole + + role = PostgresqlRole.REPLICA members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(postgresql.name, role, member.nofailover, postgresql.can_advance_slots) @@ -1260,10 +1277,13 @@ class Cluster(NamedTuple('Cluster', :param slots: slot names with LSN values. :returns: a :class:`dict` object that contains only slots that are known to be permanent. """ + from ..postgresql.misc import PostgresqlRole + if global_config.member_slots_ttl > 0: return slots - permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, RemoteMember('', {}), 'replica') + permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, RemoteMember('', {}), + PostgresqlRole.REPLICA) members_slots = {slot_name_from_member_name(m.name) for m in self.members} return {name: value for name, value in slots.items() if name in permanent_slots @@ -1279,7 +1299,9 @@ class Cluster(NamedTuple('Cluster', :returns: ``True`` if any detected replications slots are ``logical``, otherwise ``False``. """ - slots = self.get_replication_slots(postgresql, member, role='replica').values() + from ..postgresql.misc import PostgresqlRole + + slots = self.get_replication_slots(postgresql, member, role=PostgresqlRole.REPLICA).values() return any(v for v in slots if v.get("type") == "logical") def should_enforce_hot_standby_feedback(self, postgresql: 'Postgresql', member: Tags) -> bool: diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index db4738d7..b4c29b28 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -19,7 +19,7 @@ from consul import base, Check, ConsulException, NotFound from urllib3.exceptions import HTTPError from ..exceptions import DCSError -from ..postgresql.misc import PostgresqlState +from ..postgresql.misc import PostgresqlRole, PostgresqlState from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT from . import AbstractDCS, catch_return_false_exception, Cluster, ClusterConfig, \ @@ -533,8 +533,8 @@ class Consul(AbstractDCS): check['TLSServerName'] = self._service_check_tls_server_name tags = self._service_tags[:] tags.append(role) - if role == 'primary': - tags.append('master') + if data['role'] == PostgresqlRole.PRIMARY: + tags.append(PostgresqlRole.MASTER) self._previous_loop_service_tags = self._service_tags self._previous_loop_token = self._client.token @@ -552,7 +552,7 @@ class Consul(AbstractDCS): return self.deregister_service(params['service_id']) self._previous_loop_register_service = self._register_service - if role in ['primary', 'replica', 'standby-leader']: + if data['role'] in [PostgresqlRole.PRIMARY, PostgresqlRole.REPLICA, PostgresqlRole.STANDBY_LEADER]: if state != PostgresqlState.RUNNING: return return self.register_service(service_name, **params) diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 77fc9590..47e2cd8b 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -23,7 +23,7 @@ from urllib3.exceptions import HTTPError from ..collections import EMPTY_DICT from ..exceptions import DCSError -from ..postgresql.misc import PostgresqlState +from ..postgresql.misc import PostgresqlRole, PostgresqlState from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, iter_response_objects, \ keepalive_socket_options, Retry, RetryFailedError, tzutc, uri, USER_AGENT @@ -1319,9 +1319,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._standby_leader_label_value if data['role'] == 'standby_leader' else self._leader_label_value + role = self._standby_leader_label_value \ + if data['role'] == PostgresqlRole.STANDBY_LEADER else self._leader_label_value tmp_role = 'primary' - elif data['state'] == PostgresqlState.RUNNING and data['role'] != 'primary': + elif data['state'] == PostgresqlState.RUNNING and data['role'] != PostgresqlRole.PRIMARY: role = {'replica': self._follower_label_value}.get(data['role'], data['role']) tmp_role = data['role'] else: diff --git a/patroni/ha.py b/patroni/ha.py index 527c2072..9ba050c5 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -17,7 +17,7 @@ from .collections import CaseInsensitiveSet from .dcs import AbstractDCS, Cluster, Leader, Member, RemoteMember, Status, SyncState from .exceptions import DCSError, PatroniFatalException, PostgresConnectionException from .postgresql.callback_executor import CallbackAction -from .postgresql.misc import postgres_version_to_int, PostgresqlState +from .postgresql.misc import postgres_version_to_int, PostgresqlRole, PostgresqlState from .postgresql.postmaster import PostmasterProcess from .postgresql.rewind import Rewind from .quorum import QuorumStateResolver @@ -54,7 +54,8 @@ class _MemberStatus(Tags, NamedTuple('_MemberStatus', # If one of those is not in a response we want to count the node as not healthy/reachable wal: Dict[str, Any] = json.get('wal') or json['xlog'] # abuse difference in primary/replica response format - in_recovery = not (bool(wal.get('location')) or json.get('role') in ('master', 'primary')) + in_recovery = not (bool(wal.get('location')) + or json.get('role') in (PostgresqlRole.MASTER, PostgresqlRole.PRIMARY)) lsn = int(in_recovery and max(wal.get('received_location', 0), wal.get('replayed_location', 0))) return cls(member, True, in_recovery, lsn, json) @@ -474,7 +475,7 @@ class Ha(object): # Unfortunately such optimization isn't possible on the standby_leader, # therefore we will get the timeline from pg_control, either by calling # pg_control_checkpoint() on 9.6+ or by parsing the output of pg_controldata. - if self.state_handler.role == 'standby_leader': + if self.state_handler.role == PostgresqlRole.STANDBY_LEADER: timeline = pg_control_timeline or self.state_handler.pg_control_timeline() else: timeline = self.state_handler.replica_cached_timeline(self._leader_timeline) or 0 @@ -493,7 +494,7 @@ class Ha(object): ret = self.dcs.touch_member(data) if ret: new_state = (data['state'], data['role']) - if self._last_state != new_state and new_state == (PostgresqlState.RUNNING, 'primary'): + if self._last_state != new_state and new_state == (PostgresqlState.RUNNING, PostgresqlRole.PRIMARY): self.notify_mpp_coordinator('after_promote') self._last_state = new_state return ret @@ -560,7 +561,7 @@ class Ha(object): with self._async_response: # pretend that post_bootstrap was already executed self._async_response.complete(result) if result: - self.state_handler.set_role('standby_leader') + self.state_handler.set_role(PostgresqlRole.STANDBY_LEADER) return result @@ -635,7 +636,7 @@ class Ha(object): and data.get('Database cluster state') in ('in production', 'in crash recovery', 'shutting down', 'shut down')\ and self.state_handler.state == PostgresqlState.CRASHED\ - and self.state_handler.role == 'primary'\ + and self.state_handler.role == PostgresqlRole.PRIMARY\ and not self.state_handler.config.recovery_conf_exists(): # We know 100% that we were running as a primary a few moments ago, therefore could just start postgres msg = 'starting primary after failure' @@ -654,7 +655,7 @@ class Ha(object): self.load_cluster_from_dcs() - role = 'replica' + role = PostgresqlRole.REPLICA if self.has_lock() and not self.is_standby_cluster(): self._rewind.reset_state() # we want to later trigger CHECKPOINT after promote msg = "starting as readonly because i had the session lock" @@ -668,7 +669,7 @@ class Ha(object): if self.has_lock(): # in standby cluster msg = "starting as a standby leader because i had the session lock" - role = 'standby_leader' + role = PostgresqlRole.STANDBY_LEADER node_to_follow = self._get_node_to_follow(self.cluster) elif self.is_standby_cluster() and self.cluster.is_unlocked(): msg = "trying to follow a remote member because standby cluster is unhealthy" @@ -732,10 +733,10 @@ class Ha(object): if not (self._rewind.is_needed and self._rewind.can_rewind_or_reinitialize_allowed)\ or self.cluster.is_unlocked(): if is_leader: - self.state_handler.set_role('primary') + self.state_handler.set_role(PostgresqlRole.PRIMARY) return 'continue to run as primary without lock' - elif self.state_handler.role != 'standby_leader': - self.state_handler.set_role('replica') + elif self.state_handler.role != PostgresqlRole.STANDBY_LEADER: + self.state_handler.set_role(PostgresqlRole.REPLICA) if not node_to_follow: return 'no action. I am ({0})'.format(self.state_handler.name) @@ -755,10 +756,12 @@ class Ha(object): if not self.is_paused(): self.state_handler.handle_parameter_change() - role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica' + role = PostgresqlRole.STANDBY_LEADER \ + if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else PostgresqlRole.REPLICA # It might happen that leader key in the standby cluster references non-exiting member. # In this case it is safe to continue running without changing recovery.conf - if self.is_standby_cluster() and role == 'replica' and not (node_to_follow and node_to_follow.conn_url): + if self.is_standby_cluster() and role == PostgresqlRole.REPLICA \ + and not (node_to_follow and node_to_follow.conn_url): return 'continue following the old known standby leader' else: change_required, restart_required = self.state_handler.config.check_recovery_conf(node_to_follow) @@ -769,7 +772,7 @@ class Ha(object): else: self.state_handler.follow(node_to_follow, role, do_reload=True) self._rewind.trigger_check_diverged_lsn() - elif role == 'standby_leader' and self.state_handler.role != role: + elif role == PostgresqlRole.STANDBY_LEADER and self.state_handler.role != role: self.state_handler.set_role(role) self.state_handler.call_nowait(CallbackAction.ON_ROLE_CHANGE) @@ -1095,12 +1098,12 @@ class Ha(object): if self.state_handler.is_primary(): # Inform the state handler about its primary role. # It may be unaware of it if postgres is promoted manually. - self.state_handler.set_role('primary') + self.state_handler.set_role(PostgresqlRole.PRIMARY) self.process_sync_replication() self.update_cluster_history() self.state_handler.mpp_handler.sync_meta_data(self.cluster) return message - elif self.state_handler.role in ('primary', 'promoted'): + elif self.state_handler.role in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED): self.process_sync_replication() return message else: @@ -1108,7 +1111,7 @@ class Ha(object): # Somebody else updated sync state, it may be due to us losing the lock. To be safe, # postpone promotion until next cycle. TODO: trigger immediate retry of run_cycle. return 'Postponing promotion because synchronous replication state was updated by somebody else' - if self.state_handler.role not in ('primary', 'promoted'): + if self.state_handler.role not in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED): # reset failsafe state when promote self._failsafe.set_is_active(0) @@ -1156,7 +1159,7 @@ class Ha(object): :returns: the reason why caller shouldn't continue as a primary or the current value of received/replayed LSN. """ - if self.state_handler.state == PostgresqlState.RUNNING and self.state_handler.role == 'primary': + if self.state_handler.state == PostgresqlState.RUNNING and self.state_handler.role == PostgresqlRole.PRIMARY: return 'Running as a leader' self._failsafe.update(data) return self._last_wal_lsn @@ -1570,7 +1573,7 @@ class Ha(object): # location, we can remove the leader key and allow them to start leader race. time.sleep(1) # give replicas some more time to catch up if self.is_failover_possible(cluster_lsn=checkpoint_location): - self.state_handler.set_role('demoted') + self.state_handler.set_role(PostgresqlRole.DEMOTED) with self._async_executor: self.release_leader_key_voluntarily(prev_location) status['released'] = True @@ -1586,7 +1589,7 @@ class Ha(object): on_shutdown=on_shutdown if mode_control['release'] else None, before_shutdown=before_shutdown if mode == 'graceful' else None, stop_timeout=self.primary_stop_timeout()) - self.state_handler.set_role('demoted') + self.state_handler.set_role(PostgresqlRole.DEMOTED) self.set_is_leader(False) if mode_control['release']: @@ -1758,7 +1761,7 @@ class Ha(object): # enforce anything, since the leader is not a primary # So just remind the role. msg = 'no action. I am ({0}), the standby leader with the lock'.format(self.state_handler.name) \ - if self.state_handler.role == 'standby_leader' else \ + if self.state_handler.role == PostgresqlRole.STANDBY_LEADER else \ 'promoted self to a standby leader because i had the session lock' return self.enforce_follow_remote_member(msg) else: @@ -1958,7 +1961,7 @@ class Ha(object): self.state_handler.cancellable.cancel() return 'lost leader before promote' - if self.state_handler.role == 'primary': + if self.state_handler.role == PostgresqlRole.PRIMARY: logger.info('Demoting primary during %s', self._async_executor.scheduled_action) if self._async_executor.scheduled_action in ('restart', 'starting primary after failure'): # Restart needs a special interlocking cancel because postmaster may be just started in a @@ -1984,8 +1987,8 @@ class Ha(object): if not self.state_handler.is_running(): self.watchdog.disable() if self.has_lock(): - if self.state_handler.role in ('primary', 'standby_leader'): - self.state_handler.set_role('demoted') + if self.state_handler.role in (PostgresqlRole.PRIMARY, PostgresqlRole.STANDBY_LEADER): + self.state_handler.set_role(PostgresqlRole.DEMOTED) self.state_handler.call_nowait(CallbackAction.ON_ROLE_CHANGE) self._delete_leader() return 'removed leader key after trying and failing to start postgres' @@ -2010,7 +2013,7 @@ class Ha(object): if not self.state_handler.is_primary(): return 'waiting for end of recovery after bootstrap' - self.state_handler.set_role('primary') + self.state_handler.set_role(PostgresqlRole.PRIMARY) ret = self._async_executor.try_run_async('post_bootstrap', self.state_handler.bootstrap.post_bootstrap, args=(self.patroni.config['bootstrap'], self._async_response)) return ret or 'running post_bootstrap' @@ -2160,7 +2163,7 @@ class Ha(object): data_directory_error = e if not data_directory_is_accessible or data_directory_is_empty: - self.state_handler.set_role('uninitialized') + self.state_handler.set_role(PostgresqlRole.UNINITIALIZED) self.state_handler.stop('immediate', stop_timeout=self.patroni.config['retry_timeout']) # In case datadir went away while we were primary self.watchdog.disable() diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 5c3e8214..cc5f55b4 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -28,7 +28,7 @@ from .callback_executor import CallbackAction, CallbackExecutor from .cancellable import CancellableSubprocess from .config import ConfigHandler, mtime from .connection import ConnectionPool, get_connection_cursor -from .misc import parse_history, parse_lsn, postgres_major_version_to_int, PostgresqlState +from .misc import parse_history, parse_lsn, postgres_major_version_to_int, PostgresqlRole, PostgresqlState from .mpp import AbstractMPP from .postmaster import PostmasterProcess from .slots import SlotsHandler @@ -95,7 +95,7 @@ class Postgresql(object): self.mpp_handler = mpp.get_handler_impl(self) self._bin_dir = config.get('bin_dir') or '' self._role_lock = Lock() - self.set_role('uninitialized') + self.set_role(PostgresqlRole.UNINITIALIZED) self.config = ConfigHandler(self, config) self.config.check_directories() @@ -142,20 +142,20 @@ class Postgresql(object): # we know that PostgreSQL is accepting connections and can read some GUC's from pg_settings self.config.load_current_server_parameters() - self.set_role('primary' if self.is_primary() else 'replica') + self.set_role(PostgresqlRole.PRIMARY if self.is_primary() else PostgresqlRole.REPLICA) hba_saved = self.config.replace_pg_hba() ident_saved = self.config.replace_pg_ident() - if self.major_version < 120000 or self.role == 'primary': + if self.major_version < 120000 or self.role == PostgresqlRole.PRIMARY: # If PostgreSQL is running as a primary or we run PostgreSQL that is older than 12 we can # call reload_config() once again (the first call happened in the ConfigHandler constructor), # so that it can figure out if config files should be updated and pg_ctl reload executed. self.config.reload_config(config, sighup=bool(hba_saved or ident_saved)) elif hba_saved or ident_saved: self.reload() - elif not self.is_running() and self.role == 'primary': - self.set_role('demoted') + elif not self.is_running() and self.role == PostgresqlRole.PRIMARY: + self.set_role(PostgresqlRole.DEMOTED) @property def create_replica_methods(self) -> List[str]: @@ -237,7 +237,7 @@ class Postgresql(object): " pg_catalog.pg_stat_get_activity(w.pid)" " WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name) if global_config.is_synchronous_mode - and self.role in ('primary', 'promoted') else "'on', '', NULL") + and self.role in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED) else "'on', '', NULL") if self._major_version >= 90600: filter_failover = ' WHERE NOT failover' if self._major_version >= 170000 else '' @@ -252,7 +252,7 @@ class Postgresql(object): if self._major_version >= 130000 else "NULL") extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END, {0}, slot_name, " "conninfo, status, {1} FROM pg_catalog.pg_stat_get_wal_receiver()").format(written_lsn, extra) - if self.role == 'standby_leader': + if self.role == PostgresqlRole.STANDBY_LEADER: extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()" else: extra = "0" + extra @@ -366,13 +366,13 @@ class Postgresql(object): self._sysid = data.get('Database system identifier', '') return self._sysid - def get_postgres_role_from_data_directory(self) -> str: + def get_postgres_role_from_data_directory(self) -> PostgresqlRole: if self.data_directory_empty() or not self.controldata(): - return 'uninitialized' + return PostgresqlRole.UNINITIALIZED elif self.config.recovery_conf_exists(): - return 'replica' + return PostgresqlRole.REPLICA else: - return 'primary' + return PostgresqlRole.PRIMARY @property def server_version(self) -> int: @@ -592,7 +592,7 @@ class Postgresql(object): 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 == 'primary') + return bool(self.is_running() and self.role == PostgresqlRole.PRIMARY) def replay_paused(self) -> bool: return self._cluster_info_state_get('replay_paused') or False @@ -694,7 +694,7 @@ class Postgresql(object): if self.callback and cb_type in self.callback: cmd = self.callback[cb_type] - role = 'primary' if self.role == 'promoted' else self.role + role = PostgresqlRole.PRIMARY if self.role == PostgresqlRole.PROMOTED else self.role try: cmd = shlex.split(self.callback[cb_type]) + [cb_type, role, self.scope] self._callback_executor.call(cmd) @@ -702,11 +702,11 @@ class Postgresql(object): logger.exception('callback %s %r %s %s failed', cmd, cb_type, role, self.scope) @property - def role(self) -> str: + def role(self) -> PostgresqlRole: with self._role_lock: return self._role - def set_role(self, value: str) -> None: + def set_role(self, value: PostgresqlRole) -> None: with self._role_lock: self._role = value @@ -747,7 +747,7 @@ class Postgresql(object): return False def start(self, timeout: Optional[float] = None, task: Optional[CriticalTask] = None, - block_callbacks: bool = False, role: Optional[str] = None, + block_callbacks: bool = False, role: Optional[PostgresqlRole] = None, after_start: Optional[Callable[..., Any]] = None) -> Optional[bool]: """Start PostgreSQL @@ -1032,7 +1032,7 @@ class Postgresql(object): return self.state == PostgresqlState.RUNNING def restart(self, timeout: Optional[float] = None, task: Optional[CriticalTask] = None, - block_callbacks: bool = False, role: Optional[str] = None, + block_callbacks: bool = False, role: Optional[PostgresqlRole] = None, before_shutdown: Optional[Callable[..., Any]] = None, after_start: Optional[Callable[..., Any]] = None) -> Optional[bool]: """Restarts PostgreSQL. @@ -1140,14 +1140,15 @@ class Postgresql(object): logger.exception('Failed to read and parse %s', (history_path,)) return history - def follow(self, member: Union[Leader, Member, None], role: str = 'replica', + def follow(self, member: Union[Leader, Member, None], role: PostgresqlRole = PostgresqlRole.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. :param member: The member to follow - :param role: The desired role, normally 'replica', but could also be a 'standby_leader' + :param role: The desired role, one of :class:`~misc.PostgresqlRole` values, normally + :class:`~misc.PostgresqlRole.REPLICA`, but could also be a :class:`~misc.PostgresqlRole.STANDBY_LEADER` :param timeout: start timeout, how long should the `start()` method wait for postgres accepting connections :param do_reload: indicates that after updating postgresql.conf we just need to do a reload instead of restart @@ -1165,8 +1166,9 @@ class Postgresql(object): # and we know for sure that postgres was already running before, we will only execute on_role_change # callback and prevent execution of on_restart/on_start callback. # If the role remains the same (replica or standby_leader), we will execute on_start or on_restart - change_role = self.cb_called and (self.role in ('primary', 'demoted') - or not {'standby_leader', 'replica'} - {self.role, role}) + change_role = self.cb_called and \ + (self.role in (PostgresqlRole.PRIMARY, PostgresqlRole.DEMOTED) + or not {PostgresqlRole.STANDBY_LEADER, PostgresqlRole.REPLICA} - {self.role, role}) if change_role: self.__cb_pending = CallbackAction.NOOP @@ -1191,7 +1193,7 @@ class Postgresql(object): for _ in polling_loop(wait_seconds): data = self.controldata() if data.get('Database cluster state') == 'in production': - self.set_role('primary') + self.set_role(PostgresqlRole.PRIMARY) return True def _pre_promote(self) -> bool: @@ -1226,7 +1228,7 @@ class Postgresql(object): def promote(self, wait_seconds: int, task: CriticalTask, before_promote: Optional[Callable[..., Any]] = None) -> Optional[bool]: - if self.role in ('promoted', 'primary'): + if self.role in (PostgresqlRole.PROMOTED, PostgresqlRole.PRIMARY): return True ret = self._pre_promote() @@ -1250,7 +1252,7 @@ class Postgresql(object): ret = self.pg_ctl('promote', '-W') if ret: - self.set_role('promoted') + self.set_role(PostgresqlRole.PROMOTED) self.call_nowait(CallbackAction.ON_ROLE_CHANGE) ret = self._wait_promote(wait_seconds) return ret @@ -1358,7 +1360,7 @@ class Postgresql(object): logger.exception("Could not rename data directory %s", self._data_dir) def remove_data_directory(self) -> None: - self.set_role('uninitialized') + self.set_role(PostgresqlRole.UNINITIALIZED) logger.info('Removing data directory: %s', self._data_dir) try: if os.path.islink(self._data_dir): diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 1960ed00..10377145 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -20,7 +20,7 @@ from ..psycopg import parse_conninfo from ..utils import compare_values, get_postgres_version, is_subpath, \ maybe_convert_from_base_unit, parse_bool, parse_int, split_host_port, uri, validate_directory from ..validator import EnumValidator, IntValidator -from .misc import get_major_from_minor_version, postgres_version_to_int, PostgresqlState +from .misc import get_major_from_minor_version, postgres_version_to_int, PostgresqlRole, PostgresqlState from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value if TYPE_CHECKING: # pragma: no cover @@ -1062,7 +1062,7 @@ class ConfigHandler(object): synchronous_standby_names = self._server_parameters.get('synchronous_standby_names') if synchronous_standby_names is None: if global_config.is_synchronous_mode_strict\ - and self._postgresql.role in ('primary', 'promoted'): + and self._postgresql.role in (PostgresqlRole.PRIMARY, PostgresqlRole.PROMOTED): parameters['synchronous_standby_names'] = '*' else: parameters.pop('synchronous_standby_names', None) @@ -1330,7 +1330,7 @@ class ConfigHandler(object): As a workaround we will start it with the values from controldata and set `pending_restart` to true as an indicator that current values of parameters are not matching expectations.""" - if self._postgresql.role == 'primary': + if self._postgresql.role == PostgresqlRole.PRIMARY: return self._server_parameters options_mapping = { diff --git a/patroni/postgresql/misc.py b/patroni/postgresql/misc.py index 3fca51cf..a57a8a0b 100644 --- a/patroni/postgresql/misc.py +++ b/patroni/postgresql/misc.py @@ -34,6 +34,22 @@ class PostgresqlState(str, Enum): return self.value +class PostgresqlRole(str, Enum): + """Possible values of :attr:`Postgresql.role`.""" + + PRIMARY = 'primary' + MASTER = 'master' + STANDBY_LEADER = 'standby_leader' + REPLICA = 'replica' + DEMOTED = 'demoted' + UNINITIALIZED = 'uninitialized' + PROMOTED = 'promoted' + + def __repr__(self) -> str: + """Get a string representation of a :class:`PostgresqlRole` member.""" + return self.value + + def postgres_version_to_int(pg_version: str) -> int: """Convert the server_version to integer diff --git a/patroni/postgresql/mpp/citus.py b/patroni/postgresql/mpp/citus.py index a00de868..106052e4 100644 --- a/patroni/postgresql/mpp/citus.py +++ b/patroni/postgresql/mpp/citus.py @@ -9,7 +9,7 @@ from urllib.parse import urlparse from ...dcs import Cluster from ...psycopg import connect, ProgrammingError, quote_ident from ...utils import parse_int -from ..misc import PostgresqlState +from ..misc import PostgresqlRole, PostgresqlState from . import AbstractMPP, AbstractMPPHandler if TYPE_CHECKING: # pragma: no cover @@ -76,7 +76,7 @@ class PgDistNode: :returns: ``True`` if this object represents the ``primary``. """ - return self.role in ('primary', 'demoted') + return self.role in (PostgresqlRole.PRIMARY, PostgresqlRole.DEMOTED) def as_tuple(self, include_nodeid: bool = False) -> Tuple[str, int, str, Optional[int]]: """Helper method to compare two :class:`PgDistGroup` objects. @@ -478,7 +478,7 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread): for groupid, worker in cluster.workers.items(): leader = worker.leader if leader and leader.conn_url\ - and leader.data.get('role') in ('master', 'primary')\ + and leader.data.get('role') in (PostgresqlRole.MASTER, PostgresqlRole.PRIMARY)\ and leader.data.get('state') == PostgresqlState.RUNNING: self.add_task('after_promote', groupid, worker, leader.name, leader.conn_url) diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 6505e853..bf71ab61 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -14,7 +14,7 @@ from ..collections import EMPTY_DICT from ..dcs import Leader, RemoteMember from . import Postgresql from .connection import get_connection_cursor -from .misc import format_lsn, fsync_dir, parse_history, parse_lsn +from .misc import format_lsn, fsync_dir, parse_history, parse_lsn, PostgresqlRole logger = logging.getLogger(__name__) @@ -223,7 +223,8 @@ class Rewind(object): if local_timeline is None or local_lsn is None: return - if isinstance(leader, Leader) and leader.member.data.get('role') not in ('master', 'primary'): + if isinstance(leader, Leader) and leader.member.data.get('role') not in (PostgresqlRole.MASTER, + PostgresqlRole.PRIMARY): return # We want to use replication credentials when connecting to the "postgres" database in case if diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 9420c68e..8dec8e13 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -18,7 +18,7 @@ from ..file_perm import pg_perm from ..psycopg import OperationalError from ..tags import Tags from .connection import get_connection_cursor -from .misc import format_lsn, fsync_dir +from .misc import format_lsn, fsync_dir, PostgresqlRole if TYPE_CHECKING: # pragma: no cover from psycopg import Cursor @@ -693,7 +693,7 @@ class SlotsHandler: leader = cluster.leader if not leader: return - slots = cluster.get_replication_slots(self._postgresql, tags, role='replica') + slots = cluster.get_replication_slots(self._postgresql, tags, role=PostgresqlRole.REPLICA) copy_slots: Dict[str, Dict[str, Any]] = {} with self._get_leader_connection_cursor(leader) as cur: try: diff --git a/tests/test_api.py b/tests/test_api.py index 65d1c20e..aed7a2d3 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -14,7 +14,7 @@ from patroni.dcs import ClusterConfig, Member from patroni.exceptions import PostgresConnectionException from patroni.ha import _MemberStatus from patroni.postgresql.config import get_param_diff -from patroni.postgresql.misc import PostgresqlState +from patroni.postgresql.misc import PostgresqlRole, PostgresqlState from patroni.psycopg import OperationalError from patroni.utils import RetryFailedError, tzutc @@ -51,7 +51,7 @@ class MockPostgresql: connection_pool = MockConnectionPool() name = 'test' state = PostgresqlState.RUNNING - role = 'primary' + role = PostgresqlRole.PRIMARY server_version = 90625 major_version = 90600 sysid = 'dummysysid' @@ -214,21 +214,22 @@ class TestRestApiHandler(unittest.TestCase): MockRestApiServer(RestApiHandler, 'GET /read-only') with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={})): MockRestApiServer(RestApiHandler, 'GET /replica') - with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})): + with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})): MockRestApiServer(RestApiHandler, 'GET /replica') with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'state': PostgresqlState.RUNNING})): MockRestApiServer(RestApiHandler, 'GET /health') MockRestApiServer(RestApiHandler, 'GET /leader') with patch.object(RestApiHandler, 'get_postgresql_status', - Mock(return_value={'role': 'replica', 'sync_standby': True})): + Mock(return_value={'role': PostgresqlRole.REPLICA, 'sync_standby': True})): MockRestApiServer(RestApiHandler, 'GET /synchronous') MockRestApiServer(RestApiHandler, 'GET /read-only-sync') with patch.object(RestApiHandler, 'get_postgresql_status', - Mock(return_value={'role': 'replica', 'quorum_standby': True})): + Mock(return_value={'role': PostgresqlRole.REPLICA, 'quorum_standby': True})): MockRestApiServer(RestApiHandler, 'GET /quorum') MockRestApiServer(RestApiHandler, 'GET /read-only-quorum') - with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})): + with patch.object(RestApiHandler, 'get_postgresql_status', + Mock(return_value={'role': PostgresqlRole.REPLICA})): MockRestApiServer(RestApiHandler, 'GET /asynchronous') with patch.object(MockHa, 'is_leader', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /replica') @@ -237,7 +238,7 @@ class TestRestApiHandler(unittest.TestCase): with patch.object(global_config.__class__, 'is_standby_cluster', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /standby_leader') MockPatroni.dcs.cluster = None - with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})): + with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})): MockRestApiServer(RestApiHandler, 'GET /primary') with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /primary') @@ -264,7 +265,7 @@ class TestRestApiHandler(unittest.TestCase): 'tag_key1=true&tag_key2=false&' 'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2') # - with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})): + with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})): MockRestApiServer(RestApiHandler, 'GET /primary?lag=1M&' 'tag_key1=true&tag_key2=false&' 'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag') @@ -278,7 +279,8 @@ class TestRestApiHandler(unittest.TestCase): 'tag_key1=true&tag_key2=false&' 'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2') # - with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'standby_leader'})): + with patch.object(RestApiHandler, 'get_postgresql_status', + Mock(return_value={'role': PostgresqlRole.STANDBY_LEADER})): MockRestApiServer(RestApiHandler, 'GET /standby_leader?lag=1M&' 'tag_key1=true&tag_key2=false&' 'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag') @@ -305,7 +307,7 @@ class TestRestApiHandler(unittest.TestCase): 'tag_key1=true&tag_key2=false&' 'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag&tag_key6=RandomTag2') # - with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})): + with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': PostgresqlRole.PRIMARY})): MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M&' 'tag_key1=true&tag_key2=false&' 'tag_key3=1&tag_key4=1.4&tag_key5=RandomTag') @@ -461,27 +463,29 @@ class TestRestApiHandler(unittest.TestCase): request = make_request(schedule=future_restart_time.isoformat(), role='unknown', postgres_version='9.5.3') MockRestApiServer(RestApiHandler, request) # wrong version - request = make_request(schedule=future_restart_time.isoformat(), role='primary', postgres_version='9.5.3.1') + request = make_request(schedule=future_restart_time.isoformat(), + role=PostgresqlRole.PRIMARY, postgres_version='9.5.3.1') MockRestApiServer(RestApiHandler, request) # unknown filter request = make_request(schedule=future_restart_time.isoformat(), batman='lives') MockRestApiServer(RestApiHandler, request) # incorrect schedule - request = make_request(schedule='2016-08-42 12:45TZ+1', role='primary') + request = make_request(schedule='2016-08-42 12:45TZ+1', role=PostgresqlRole.PRIMARY) MockRestApiServer(RestApiHandler, request) # everything fine, but the schedule is missing - request = make_request(role='primary', postgres_version='9.5.2') + request = make_request(role=PostgresqlRole.PRIMARY, postgres_version='9.5.2') MockRestApiServer(RestApiHandler, request) for retval in (True, False): with patch.object(MockHa, 'schedule_future_restart', Mock(return_value=retval)): request = make_request(schedule=future_restart_time.isoformat()) MockRestApiServer(RestApiHandler, request) with patch.object(MockHa, 'restart', Mock(return_value=(retval, "foo"))): - request = make_request(role='primary', postgres_version='9.5.2') + request = make_request(role=PostgresqlRole.PRIMARY, postgres_version='9.5.2') MockRestApiServer(RestApiHandler, request) with patch.object(global_config.__class__, 'is_paused', PropertyMock(return_value=True)): - MockRestApiServer(RestApiHandler, make_request(schedule='2016-08-42 12:45TZ+1', role='primary')) + MockRestApiServer(RestApiHandler, + make_request(schedule='2016-08-42 12:45TZ+1', role=PostgresqlRole.PRIMARY)) # Valid timeout MockRestApiServer(RestApiHandler, make_request(timeout='60s')) # Invalid timeout diff --git a/tests/test_callback_executor.py b/tests/test_callback_executor.py index e76c9657..fdd80c85 100644 --- a/tests/test_callback_executor.py +++ b/tests/test_callback_executor.py @@ -1,10 +1,12 @@ import unittest +from unittest import mock from unittest.mock import Mock, patch import psutil -from patroni.postgresql.callback_executor import CallbackExecutor +from patroni.postgresql.callback_executor import CallbackAction, CallbackExecutor +from patroni.postgresql.misc import PostgresqlRole class TestCallbackExecutor(unittest.TestCase): @@ -14,7 +16,7 @@ class TestCallbackExecutor(unittest.TestCase): mock_popen.return_value.children.return_value = [] mock_popen.return_value.is_running.return_value = True - callback = ['test.sh', 'on_start', 'replica', 'foo'] + callback = ['test.sh', CallbackAction.ON_START, PostgresqlRole.REPLICA, 'foo'] ce = CallbackExecutor() ce._kill_children = Mock(side_effect=Exception) ce._invoke_excepthook = Mock() @@ -22,6 +24,8 @@ class TestCallbackExecutor(unittest.TestCase): ce.join() self.assertIsNone(ce.call(callback)) + mock_popen.assert_called_with(['test.sh', 'on_start', 'replica', 'foo'], close_fds=True) + mock_popen.reset_mock() mock_popen.return_value.kill.side_effect = psutil.AccessDenied() self.assertIsNone(ce.call(callback)) @@ -38,5 +42,7 @@ class TestCallbackExecutor(unittest.TestCase): self.assertIsNone(ce.call(callback)) mock_popen.side_effect = [Mock()] - self.assertIsNone(ce.call(['test.sh', 'on_reload', 'replica', 'foo'])) + self.assertIsNone(ce.call(['test.sh', CallbackAction.ON_RELOAD, PostgresqlRole.REPLICA, 'foo'])) + self.assertEqual(mock_popen.call_args_list[-2], + mock.call(['test.sh', 'on_reload', 'replica', 'foo'], close_fds=True)) ce.join() diff --git a/tests/test_consul.py b/tests/test_consul.py index 8fb38602..0dce3042 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -9,7 +9,7 @@ from consul import ConsulException, NotFound from patroni.dcs import get_dcs from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulClient, ConsulError, \ ConsulInternalError, HTTPClient, InvalidSession, InvalidSessionTTL, RetryFailedError -from patroni.postgresql.misc import PostgresqlState +from patroni.postgresql.misc import PostgresqlRole, PostgresqlState from patroni.postgresql.mpp import get_mpp from . import SleepException @@ -257,7 +257,8 @@ class TestConsul(unittest.TestCase): @patch.object(Agent.Service, 'register', Mock(side_effect=(False, True, True, True))) @patch.object(Agent.Service, 'deregister', Mock(return_value=True)) def test_update_service(self): - d = {'role': 'replica', 'api_url': 'http://a/t', 'conn_url': 'pg://c:1', 'state': PostgresqlState.RUNNING} + d = {'role': PostgresqlRole.REPLICA, 'api_url': 'http://a/t', 'conn_url': 'pg://c:1', + 'state': PostgresqlState.RUNNING} self.assertIsNone(self.c.update_service({}, {})) self.assertFalse(self.c.update_service({}, d)) self.assertTrue(self.c.update_service(d, d)) @@ -269,7 +270,7 @@ class TestConsul(unittest.TestCase): d['state'] = PostgresqlState.RUNNING d['role'] = 'bla' self.assertIsNone(self.c.update_service({}, d)) - d['role'] = 'primary' + d['role'] = PostgresqlRole.PRIMARY self.assertTrue(self.c.update_service({}, d)) @patch.object(KV, 'put', Mock(side_effect=ConsulException)) @@ -281,7 +282,7 @@ class TestConsul(unittest.TestCase): self.c.refresh_session = Mock(return_value=False) - d = {'role': 'replica', 'api_url': 'http://a/t', + d = {'role': PostgresqlRole.REPLICA, 'api_url': 'http://a/t', 'conn_url': 'pg://c:1', 'state': PostgresqlState.RUNNING} # Changing register_service from True to False calls deregister() diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 93aab2f4..7cedb3cc 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -11,6 +11,7 @@ import etcd from click.testing import CliRunner from prettytable import PrettyTable +from patroni.ctl import CtlPostgresqlRole from patroni.postgresql.misc import PostgresqlState try: @@ -52,7 +53,7 @@ def get_default_config(*args): @patch('patroni.ctl.load_config', get_default_config) @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) class TestCtl(unittest.TestCase): - TEST_ROLES = ('primary', 'leader') + TEST_ROLES = (CtlPostgresqlRole.PRIMARY, CtlPostgresqlRole.LEADER) @patch('socket.getaddrinfo', socket_getaddrinfo) def setUp(self): @@ -89,14 +90,16 @@ class TestCtl(unittest.TestCase): self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role=role)) # MockCursor returns pg_is_in_recovery as false - self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role='replica')) + self.assertIsNone( + get_cursor(get_cluster_initialized_with_leader(), None, {}, role=CtlPostgresqlRole.REPLICA)) - self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, role='any')) + self.assertIsNotNone( + get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, role=CtlPostgresqlRole.ANY)) # Mutually exclusive options with self.assertRaises(PatroniCtlException) as e: get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, member_name='other', - role='replica') + role=CtlPostgresqlRole.REPLICA) self.assertEqual(str(e.exception), '--role and --member are mutually exclusive options') @@ -301,7 +304,8 @@ class TestCtl(unittest.TestCase): self.assertTrue('False' in str(rows)) with patch.object(MockCursor, 'execute', Mock(side_effect=OperationalError('bla'))): - rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, + CtlPostgresqlRole.REPLICA, 'SELECT pg_catalog.pg_is_in_recovery()', {}) with patch('patroni.ctl.get_cursor', Mock(return_value=None)): # No role nor member given -- generic message @@ -313,11 +317,13 @@ class TestCtl(unittest.TestCase): self.assertTrue('No connection to member foo' in str(rows)) # Role given -- message pointing to role - rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, + CtlPostgresqlRole.REPLICA, 'SELECT pg_catalog.pg_is_in_recovery()', {}) self.assertTrue('No connection to role replica' in str(rows)) with patch('patroni.ctl.get_cursor', Mock(side_effect=OperationalError('bla'))): - rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, + CtlPostgresqlRole.REPLICA, 'SELECT pg_catalog.pg_is_in_recovery()', {}) def test_dsn(self): result = self.runner.invoke(ctl, ['dsn', 'alpha']) @@ -471,12 +477,12 @@ class TestCtl(unittest.TestCase): self.assertEqual(len(r), 1) self.assertEqual(r[0].name, 'leader') - r = list(get_all_members(get_cluster_initialized_with_leader(), None, role='replica')) + r = list(get_all_members(get_cluster_initialized_with_leader(), None, role=CtlPostgresqlRole.REPLICA)) self.assertEqual(len(r), 1) self.assertEqual(r[0].name, 'other') self.assertEqual(len(list(get_all_members(get_cluster_initialized_without_leader(), - None, role='replica'))), 2) + None, role=CtlPostgresqlRole.REPLICA))), 2) def test_members(self): result = self.runner.invoke(ctl, ['list']) diff --git a/tests/test_ha.py b/tests/test_ha.py index 599a8b25..ca7ac9f5 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -18,7 +18,7 @@ from patroni.postgresql.bootstrap import Bootstrap from patroni.postgresql.callback_executor import CallbackAction from patroni.postgresql.cancellable import CancellableSubprocess from patroni.postgresql.config import ConfigHandler -from patroni.postgresql.misc import PostgresqlState +from patroni.postgresql.misc import PostgresqlRole, PostgresqlState from patroni.postgresql.postmaster import PostmasterProcess from patroni.postgresql.rewind import Rewind from patroni.postgresql.slots import SlotsHandler @@ -59,7 +59,7 @@ def get_cluster_bootstrapping_without_leader(cluster_config=None): def get_cluster_initialized_without_leader(leader=False, failover=None, sync=None, cluster_config=None, failsafe=False): m1 = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', 'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4, - 'role': 'primary', 'state': 'running'}) + 'role': PostgresqlRole.PRIMARY, 'state': 'running'}) leader = Leader(0, 0, m1 if leader else Member(0, '', 28, {})) m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', 'api_url': 'http://127.0.0.1:8011/patroni', @@ -214,7 +214,7 @@ class TestHa(PostgresInit): def setUp(self): super(TestHa, self).setUp() self.p.set_state(PostgresqlState.RUNNING) - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.p.postmaster_start_time = MagicMock(return_value=str(postmaster_start_time)) self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False) self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test', @@ -242,9 +242,9 @@ class TestHa(PostgresInit): with patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value='streaming')): self.ha.touch_member() self.p.timeline_wal_position = Mock(return_value=(0, 1, 1)) - self.p.set_role('standby_leader') + self.p.set_role(PostgresqlRole.STANDBY_LEADER) self.ha.touch_member() - self.p.set_role('primary') + self.p.set_role(PostgresqlRole.PRIMARY) self.ha.dcs.touch_member = true self.ha.touch_member() @@ -299,7 +299,7 @@ class TestHa(PostgresInit): self.p.follow = false self.p.is_running = false self.p.name = 'leader' - self.p.set_role('demoted') + self.p.set_role(PostgresqlRole.DEMOTED) self.p.controldata = lambda: {'Database cluster state': 'shut down', 'Database system identifier': SYSID} self.ha.cluster = get_cluster_initialized_with_leader() self.assertEqual(self.ha.run_cycle(), 'starting as readonly because i had the session lock') @@ -308,7 +308,7 @@ class TestHa(PostgresInit): self.p.start = false self.p.is_running = false self.p.name = 'leader' - self.p.set_role('primary') + self.p.set_role(PostgresqlRole.PRIMARY) self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID} self.ha.cluster = get_cluster_initialized_with_leader() self.assertEqual(self.ha.run_cycle(), 'starting primary after failure') @@ -346,7 +346,7 @@ class TestHa(PostgresInit): def test_recover_with_rewind(self): self.p.is_running = false self.ha.cluster = get_cluster_initialized_with_leader() - self.ha.cluster.leader.member.data.update(version='2.0.2', role='primary') + self.ha.cluster.leader.member.data.update(version='2.0.2', role=PostgresqlRole.PRIMARY) self.ha._rewind.pg_rewind = true self.ha._rewind.check_leader_is_not_in_recovery = true with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)): @@ -414,7 +414,7 @@ class TestHa(PostgresInit): def test_long_promote(self): self.ha.has_lock = true self.p.is_primary = false - self.p.set_role('primary') + self.p.set_role(PostgresqlRole.PRIMARY) self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock') def test_demote_after_failing_to_obtain_lock(self): @@ -579,7 +579,7 @@ class TestHa(PostgresInit): def test_update_failsafe(self): self.assertRaises(Exception, self.ha.update_failsafe, {}) - self.p.set_role('primary') + self.p.set_role(PostgresqlRole.PRIMARY) self.assertEqual(self.ha.update_failsafe({}), 'Running as a leader') def test_call_failsafe_member(self): @@ -715,7 +715,7 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'updated leader lock during restart') self.ha.update_lock = false - self.p.set_role('primary') + self.p.set_role(PostgresqlRole.PRIMARY) with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)), \ patch('patroni.async_executor.CriticalTask.result', PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True), \ @@ -907,7 +907,7 @@ class TestHa(PostgresInit): def test_manual_failover_process_no_leader(self): self.p.is_primary = false - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) # failover to another member, fetch_node_status for candidate fails with patch('patroni.ha.logger.warning') as mock_warning: @@ -917,7 +917,7 @@ class TestHa(PostgresInit): ('%s: member %s is %s', 'manual failover', 'leader', 'not reachable')) # failover to another member, candidate is accessible, in_recovery - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.ha.fetch_node_status = get_node_status() self.assertEqual(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node') @@ -928,7 +928,7 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') # failover to me but I am set to nofailover. In no case I should be elected as a leader - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None)) self.ha.patroni.nofailover = True self.assertEqual(self.ha.run_cycle(), 'following a different leader because I am not allowed to promote') @@ -952,7 +952,7 @@ class TestHa(PostgresInit): def test_manual_switchover_process_no_leader(self): self.p.is_primary = false - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) # I was the leader, other members are healthy self.ha.fetch_node_status = get_node_status() @@ -989,7 +989,7 @@ class TestHa(PostgresInit): # manual failover to our node (postgresql0), # which name is not in sync nodes list (some sync nodes are available) - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None), sync=('leader1', 'other')) self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['leader1']), @@ -1045,7 +1045,7 @@ class TestHa(PostgresInit): # switchover to me, I am not leader self.p.is_primary = false - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, 'leader', self.p.name, None)) self.assertEqual(self.ha.run_cycle(), 'PAUSE: promoted self to leader by acquiring session lock') @@ -1117,9 +1117,9 @@ class TestHa(PostgresInit): def test_post_recover(self, mock_call_nowait): self.p.is_running = false self.ha.has_lock = true - self.p.set_role('primary') + self.p.set_role(PostgresqlRole.PRIMARY) self.assertEqual(self.ha.post_recover(), 'removed leader key after trying and failing to start postgres') - self.assertEqual(self.p.role, 'demoted') + self.assertEqual(self.p.role, PostgresqlRole.DEMOTED) mock_call_nowait.assert_called_once_with(CallbackAction.ON_ROLE_CHANGE) self.ha.has_lock = false self.assertEqual(self.ha.post_recover(), 'failed to start postgres') @@ -1167,7 +1167,7 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), "restart scheduled") def test_restart_matches(self): - self.p._role = 'replica' + self.p._role = PostgresqlRole.REPLICA self.p._connection.server_version = 90500 self.p._pending_restart = True self.assertFalse(self.ha.restart_matches("primary", "9.5.0", True)) @@ -1195,7 +1195,7 @@ class TestHa(PostgresInit): self.ha._leader_timeline = 1 self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock') self.assertEqual(self.ha.run_cycle(), 'no action. I am (leader), the standby leader with the lock') - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.p.config.check_recovery_conf = Mock(return_value=(True, False)) self.assertEqual(self.ha.run_cycle(), 'promoted self to a standby leader because i had the session lock') @@ -1221,7 +1221,7 @@ class TestHa(PostgresInit): @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True)) def test_process_unhealthy_standby_cluster_as_cascade_replica(self): self.p.is_primary = false - self.p.name = 'replica' + self.p.name = PostgresqlRole.REPLICA self.ha.cluster = get_standby_cluster_initialized_with_only_leader() self.assertTrue(self.ha.run_cycle().startswith('running pg_rewind from remote_member:')) @@ -1458,7 +1458,7 @@ class TestHa(PostgresInit): mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock() self.p.is_primary = false - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.ha.has_lock = true mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) self.p.name = 'leader' @@ -1472,7 +1472,7 @@ class TestHa(PostgresInit): mock_set_sync.reset_mock() # When we just became primary nobody is sync - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) mock_write_sync.return_value = False self.assertTrue(self.ha.enforce_primary_role('msg', 'promote msg') != 'promote msg') mock_set_sync.assert_not_called() @@ -1481,7 +1481,7 @@ class TestHa(PostgresInit): self.ha.is_synchronous_mode = true self.p.is_primary = false - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.p.name = 'other' self.ha.cluster = get_cluster_initialized_without_leader(sync=('leader', 'other2')) mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty()) @@ -1512,7 +1512,7 @@ class TestHa(PostgresInit): self.p.name = 'other' self.p.is_primary = false - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) mock_restart = self.p.restart = Mock(return_value=True) self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) self.ha.touch_member = Mock(return_value=True) @@ -1627,7 +1627,7 @@ class TestHa(PostgresInit): self.p.data_directory_empty = Mock(side_effect=OSError(5, "Input/output error: '{}'".format(self.p.data_dir))) self.assertEqual(self.ha.run_cycle(), 'released leader key voluntarily as data dir not accessible and currently leader') - self.assertEqual(self.p.role, 'uninitialized') + self.assertEqual(self.p.role, PostgresqlRole.UNINITIALIZED) # as has_lock is mocked out, we need to fake the leader key release self.ha.has_lock = false @@ -1651,7 +1651,7 @@ class TestHa(PostgresInit): self.p.is_primary = false self.ha.run_cycle() exit_mock.assert_called_once_with(1) - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.ha.dcs.initialize = Mock() with patch.object(Postgresql, 'cb_called', PropertyMock(return_value=True)): self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') @@ -1687,7 +1687,7 @@ class TestHa(PostgresInit): self.ha.is_paused = true self.p.data_directory_empty = true self.assertEqual(self.ha.run_cycle(), 'PAUSE: running with empty data directory') - self.assertEqual(self.p.role, 'uninitialized') + self.assertEqual(self.p.role, PostgresqlRole.UNINITIALIZED) @patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True)) def test_sysid_no_match_in_pause(self): @@ -1742,7 +1742,7 @@ class TestHa(PostgresInit): self.p._major_version = 90500 self.ha.cluster = get_cluster_initialized_without_leader(sync=('other', self.p.name + ',foo')) self.p.is_primary = false - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=None) # Postgres 9.5, write_sync_state to DCS failed self.assertEqual(self.ha.run_cycle(), @@ -1764,7 +1764,7 @@ class TestHa(PostgresInit): self.p._major_version = 90600 mock_set_sync.reset_mock() mock_write_sync.reset_mock() - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) # Postgres 9.6, with quorum commit we avoid updating /sync key and put some nodes to ssn self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') self.assertEqual(mock_write_sync.call_count, 0) @@ -1773,7 +1773,7 @@ class TestHa(PostgresInit): self.p._major_version = 150000 mock_set_sync.reset_mock() - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.p.name = 'nonsync' self.ha.fetch_node_status = get_node_status() # Postgres 15, with quorum commit. Non-sync node promoted we avoid updating /sync key and put some nodes to ssn diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 48828266..6553c37f 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -15,7 +15,7 @@ from patroni.dcs import get_dcs from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \ K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, Retry, \ RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME -from patroni.postgresql.misc import PostgresqlState +from patroni.postgresql.misc import PostgresqlRole, PostgresqlState from patroni.postgresql.mpp import get_mpp from . import MockResponse, SleepException @@ -322,19 +322,19 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): mock_patch_namespaced_pod.return_value.metadata.resource_version = '10' self.k._name = 'p-1' - self.k.touch_member({'role': 'replica', 'state': PostgresqlState.INITDB}) + self.k.touch_member({'role': PostgresqlRole.REPLICA, 'state': PostgresqlState.INITDB}) self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar') - self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'replica'}) + self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.REPLICA}) self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], None) - self.k.touch_member({'role': 'replica', 'state': PostgresqlState.CUSTOM_BOOTSTRAP}) + self.k.touch_member({'role': PostgresqlRole.REPLICA, 'state': PostgresqlState.CUSTOM_BOOTSTRAP}) self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar') - self.k.touch_member({'role': 'replica', 'state': PostgresqlState.BOOTSTRAP_STARTING}) + self.k.touch_member({'role': PostgresqlRole.REPLICA, 'state': PostgresqlState.BOOTSTRAP_STARTING}) self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar') - self.k.touch_member({'state': PostgresqlState.STOPPED, 'role': 'primary'}) + self.k.touch_member({'state': PostgresqlState.STOPPED, 'role': PostgresqlRole.PRIMARY}) self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], None) self.k._role_label = 'isMaster' @@ -343,26 +343,26 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k._standby_leader_label_value = 'false' self.k._tmp_role_label = 'tmp_role' - self.k.touch_member({'state': PostgresqlState.CREATING_REPLICA, 'role': 'replica'}) + self.k.touch_member({'state': PostgresqlState.CREATING_REPLICA, 'role': PostgresqlRole.REPLICA}) self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], 'bar') - self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'replica'}) + self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.REPLICA}) self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['foo'], None) 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.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], PostgresqlRole.REPLICA) mock_patch_namespaced_pod.rest_mock() self.k._name = 'p-0' - self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'standby_leader'}) + self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.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'], 'primary') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], PostgresqlRole.PRIMARY) mock_patch_namespaced_pod.rest_mock() - self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'primary'}) + self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.PRIMARY}) mock_patch_namespaced_pod.assert_called() self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'true') - self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'primary') + self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], PostgresqlRole.PRIMARY) def test_initialize(self): self.k.initialize() @@ -495,7 +495,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes): self.assertEqual(('create_config_service failed',), mock_logger_exception.call_args[0]) mock_logger_exception.reset_mock() - self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': 'replica'}) + self.k.touch_member({'state': PostgresqlState.RUNNING, 'role': PostgresqlRole.REPLICA}) mock_logger_exception.assert_called_once() self.assertEqual(('create_config_service failed',), mock_logger_exception.call_args[0]) diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 53010d56..ba4e8c21 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -20,7 +20,7 @@ from patroni.dcs.etcd import AbstractEtcdClientWithFailover from patroni.exceptions import DCSError from patroni.postgresql import Postgresql from patroni.postgresql.config import ConfigHandler -from patroni.postgresql.misc import PostgresqlState +from patroni.postgresql.misc import PostgresqlRole, PostgresqlState from . import psycopg_connect, SleepException from .test_etcd import etcd_read, etcd_write @@ -163,7 +163,7 @@ class TestPatroni(unittest.TestCase): @patch.object(Postgresql, 'state', PropertyMock(return_value=PostgresqlState.RUNNING)) @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False)) def test_run(self): - self.p.postgresql.set_role('replica') + self.p.postgresql.set_role(PostgresqlRole.REPLICA) self.p.sighup_handler() self.p.ha.dcs.watch = Mock(side_effect=SleepException) self.p.api.start = Mock() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 3ab78cac..5e0aa13a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -22,7 +22,7 @@ from patroni.postgresql import PgIsReadyStatus, Postgresql from patroni.postgresql.bootstrap import Bootstrap from patroni.postgresql.callback_executor import CallbackAction from patroni.postgresql.config import _false_validator, get_param_diff -from patroni.postgresql.misc import PostgresqlState +from patroni.postgresql.misc import PostgresqlRole, PostgresqlState from patroni.postgresql.postmaster import PostmasterProcess from patroni.postgresql.validator import _get_postgres_guc_validators, _load_postgres_gucs_validators, \ _read_postgres_gucs_validators_file, Bool, Enum, EnumBool, Integer, InvalidGucValidatorsFile, \ @@ -446,7 +446,7 @@ class TestPostgresql(BaseTestPostgresql): task = CriticalTask() self.assertTrue(self.p.promote(0, task)) - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.p.config._config['pre_promote'] = 'test' with patch('patroni.postgresql.cancellable.CancellableSubprocess.is_cancelled', PropertyMock(return_value=1)): self.assertFalse(self.p.promote(0, task)) @@ -480,7 +480,7 @@ class TestPostgresql(BaseTestPostgresql): @patch('shlex.split', Mock(side_effect=OSError)) def test_call_nowait(self): - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START)) self.p.bootstrapping = True self.assertIsNone(self.p.call_nowait(CallbackAction.ON_START)) @@ -508,7 +508,7 @@ class TestPostgresql(BaseTestPostgresql): @patch('os.path.exists', Mock(return_value=True)) @patch.object(Postgresql, 'controldata', Mock()) def test_get_postgres_role_from_data_directory(self): - self.assertEqual(self.p.get_postgres_role_from_data_directory(), 'replica') + self.assertEqual(self.p.get_postgres_role_from_data_directory(), PostgresqlRole.REPLICA) @patch('os.remove', Mock()) @patch('shutil.rmtree', Mock()) @@ -852,7 +852,8 @@ class TestPostgresql(BaseTestPostgresql): def test_get_primary_timeline(self): self.assertEqual(self.p.get_primary_timeline(), 1) - @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='replica')) + @patch.object(Postgresql, 'get_postgres_role_from_data_directory', + Mock(return_value=PostgresqlRole.REPLICA)) @patch.object(Postgresql, 'is_running', Mock(return_value=False)) @patch.object(Bootstrap, 'running_custom_bootstrap', PropertyMock(return_value=True)) @patch('patroni.postgresql.config.logger') @@ -893,7 +894,7 @@ class TestPostgresql(BaseTestPostgresql): @patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))) def test_received_timeline(self): - self.p.set_role('standby_leader') + self.p.set_role(PostgresqlRole.STANDBY_LEADER) self.p.reset_cluster_info_state(None) self.assertRaises(PostgresConnectionException, self.p.received_timeline) diff --git a/tests/test_slots.py b/tests/test_slots.py index 8be9864e..7a72b6e1 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -8,7 +8,7 @@ from unittest.mock import Mock, patch, PropertyMock from patroni import global_config, psycopg from patroni.dcs import Cluster, ClusterConfig, Member, Status, SyncState from patroni.postgresql import Postgresql -from patroni.postgresql.misc import fsync_dir, PostgresqlState +from patroni.postgresql.misc import fsync_dir, PostgresqlRole, PostgresqlState from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler from patroni.tags import Tags @@ -52,21 +52,22 @@ class TestSlotsHandler(BaseTestPostgresql): global_config.update(cluster) with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)): self.s.sync_replication_slots(cluster, self.tags) - self.p.set_role('standby_leader') + self.p.set_role(PostgresqlRole.STANDBY_LEADER) with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \ patch.object(global_config.__class__, 'is_standby_cluster', PropertyMock(return_value=True)), \ patch('patroni.postgresql.slots.logger.debug') as mock_debug: self.s.sync_replication_slots(cluster, self.tags) mock_debug.assert_called_once() - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \ patch.object(global_config.__class__, 'is_paused', PropertyMock(return_value=True)), \ patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop: config.data['slots'].pop('ls') self.s.sync_replication_slots(cluster, self.tags) mock_drop.assert_not_called() - self.p.set_role('primary') - with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')): + self.p.set_role(PostgresqlRole.PRIMARY) + with mock.patch('patroni.postgresql.Postgresql.role', + new_callable=PropertyMock(return_value=PostgresqlRole.REPLICA)): self.s.sync_replication_slots(cluster, self.tags) with patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock: alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'}) @@ -79,7 +80,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.assertTrue("test.3" in ca, "non matching {0}".format(ca)) with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)): self.s.sync_replication_slots(cluster, self.tags) - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.s.sync_replication_slots(cluster, self.tags) def test_cascading_replica_sync_replication_slots(self): @@ -91,7 +92,7 @@ class TestSlotsHandler(BaseTestPostgresql): }) cluster = Cluster(True, config, self.leader, Status(0, {'ls': 10}, []), [self.me, self.other, self.leadermem, cascading_replica], None, SyncState.empty(), None, None) - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) with patch.object(Postgresql, '_query') as mock_query, \ patch.object(Postgresql, 'is_primary', Mock(return_value=False)): mock_query.return_value = [('ls', 'logical', 104, 'b', 'a', 5, 12345, 105)] @@ -147,10 +148,10 @@ class TestSlotsHandler(BaseTestPostgresql): # sanity for primary self.p.name = self.leadermem.name self.assertEqual( - cluster._get_permanent_slots(self.p, self.leadermem, 'primary'), + cluster._get_permanent_slots(self.p, self.leadermem, PostgresqlRole.PRIMARY), {'foo': {'type': 'logical', 'database': 'a', 'plugin': 'b'}, 'bar': {'type': 'physical'}}) self.assertEqual( - cluster._get_members_slots(self.p.name, 'primary', False, True), + cluster._get_members_slots(self.p.name, PostgresqlRole.PRIMARY, False, True), {'test_3': {'type': 'physical', 'lsn': 98, 'expected_active': False}, 'test_4': {'type': 'physical', 'lsn': 98, 'expected_active': True}}) @@ -158,7 +159,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.p.name = nostream_node.name # permanent logical slots are not allowed on nostream node self.assertEqual( - cluster._get_permanent_slots(self.p, nostream_node, 'replica'), + cluster._get_permanent_slots(self.p, nostream_node, PostgresqlRole.REPLICA), {'bar': {'type': 'physical'}}) self.assertEqual( cluster.get_slot_name_on_primary(self.p.name, nostream_node), @@ -166,7 +167,7 @@ class TestSlotsHandler(BaseTestPostgresql): # check cascade member-slot existence on nostream node self.assertEqual( - cluster._get_members_slots(nostream_node.name, 'replica', False, True), + cluster._get_members_slots(nostream_node.name, PostgresqlRole.REPLICA, False, True), {'leader': {'type': 'physical', 'lsn': 99, 'expected_active': False}, 'test_3': {'type': 'physical', 'lsn': 98, 'expected_active': True}, 'test_4': {'type': 'physical', 'lsn': 98, 'expected_active': False}}) @@ -174,7 +175,7 @@ class TestSlotsHandler(BaseTestPostgresql): # cascade also does not entitled to have logical slot on itself ... self.p.name = cascade_node.name self.assertEqual( - cluster._get_permanent_slots(self.p, cascade_node, 'replica'), + cluster._get_permanent_slots(self.p, cascade_node, PostgresqlRole.REPLICA), {'bar': {'type': 'physical'}}) # ... and member-slot on primary self.assertEqual( @@ -184,7 +185,7 @@ class TestSlotsHandler(BaseTestPostgresql): # simple replica must have every permanent slot ... self.p.name = stream_node.name self.assertEqual( - cluster._get_permanent_slots(self.p, stream_node, 'replica'), + cluster._get_permanent_slots(self.p, stream_node, PostgresqlRole.REPLICA), {'foo': {'type': 'logical', 'database': 'a', 'plugin': 'b'}, 'bar': {'type': 'physical'}}) # ... and member-slot on primary self.assertEqual( @@ -219,7 +220,7 @@ class TestSlotsHandler(BaseTestPostgresql): @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) def test__ensure_logical_slots_replica(self): - self.p.set_role('replica') + self.p.set_role(PostgresqlRole.REPLICA) self.cluster.status.slots['ls'] = 12346 with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)): self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), []) @@ -321,7 +322,7 @@ class TestSlotsHandler(BaseTestPostgresql): self.assertTrue(mock_query.call_args[0][0].startswith('WITH slots AS (SELECT slot_name, active')) @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) - @patch.object(Postgresql, 'role', PropertyMock(return_value='replica')) + @patch.object(Postgresql, 'role', PropertyMock(return_value=PostgresqlRole.REPLICA)) def test_advance_physical_slots(self): config = ClusterConfig(1, {'slots': {'blabla': {'type': 'physical'}, 'leader': None}}, 1) cluster = Cluster(True, config, self.leader, Status(0, {'blabla': 12346}, []), @@ -363,7 +364,7 @@ class TestSlotsHandler(BaseTestPostgresql): mock_drop.assert_not_called() @patch.object(Postgresql, 'is_primary', Mock(return_value=False)) - @patch.object(Postgresql, 'role', PropertyMock(return_value='replica')) + @patch.object(Postgresql, 'role', PropertyMock(return_value=PostgresqlRole.REPLICA)) @patch.object(TestTags, 'tags', PropertyMock(return_value={'nofailover': True})) def test_slots_nofailover_tag(self): self.p.name = self.leadermem.name