patroni.dcs package

Submodules

Module contents

Abstract classes for Distributed Configuration Store.

class patroni.dcs.AbstractDCS(config: Dict[str, Any], mpp: AbstractMPP)

Bases: ABC

Abstract representation of DCS modules.

Implementations of a concrete DCS class, using appropriate backend client interfaces, must include the following methods and properties.

Functional methods that are critical in their timing, required to complete within retry_timeout period in order to prevent the DCS considered inaccessible, each perform construction of complex data objects:

_postgresql_cluster_loader():
method which processes the structure of data stored in the DCS used to build the Cluster object with all relevant associated data.

_mpp_cluster_loader():
Similar to above but specifically representing MPP group and workers information.

_load_cluster():
main method for calling specific loader method to build the Cluster object representing the state and topology of the cluster.

Functional methods that are critical in their timing and must be written with ACID transaction properties in mind:

attempt_to_acquire_leader():
method used in the leader race to attempt to acquire the leader lock by creating the leader key in the DCS, if it does not exist.

_update_leader():
method to update leader key in DCS. Relies on Compare-And-Set to ensure the Primary lock key is updated. If this fails to update within the retry_timeout window the Primary will be demoted.

Functional method that relies on Compare-And-Create to ensure only one member creates the relevant key:

initialize():
method used in the race for cluster initialization which creates the initialize key in the DCS.

DCS backend getter and setter methods and properties:

take_leader(): method to create a new leader key in the DCS.

set_ttl(): method for setting TTL value in DCS.

ttl(): property which returns the current TTL.

set_retry_timeout(): method for setting retry_timeout in DCS backend.

_write_leader_optime(): compatibility method to write WAL LSN to DCS.

_write_status(): method to write WAL LSN for slots to the DCS.

_write_failsafe(): method to write cluster topology to the DCS, used by failsafe mechanism.

touch_member(): method to update individual member key in the DCS.

set_history_value(): method to set the history key in the DCS.

DCS setter methods using Compare-And-Set which although important are less critical if they fail, attempts can be retried or may result in warning log messages:

set_failover_value(): method to create and/or update the failover key in the DCS.

set_config_value(): method to create and/or update the failover key in the DCS.

set_sync_state_value(): method to set the synchronous state sync key in the DCS.

DCS data and key removal methods:

delete_sync_state():
likewise, a method to remove synchronous state sync key from the DCS.

delete_cluster():
method which will remove cluster information from the DCS. Used only from patronictl.

_delete_leader():
method relies on CAS, used by a member that is the current leader, to remove the leader key in the DCS.

cancel_initialization():
method to remove the initialize key for the cluster from the DCS.

If either of the sync_state set or delete methods fail, although not critical, this may result in Synchronous replication key updated by someone else messages being logged.

Care should be taken to consult each abstract method for any additional information and requirements such as expected exceptions that should be raised in certain conditions and the object types for arguments and return from methods and properties.

_CONFIG = 'config'

_FAILOVER = 'failover'

_FAILSAFE = 'failsafe'

_HISTORY = 'history'

_INITIALIZE = 'initialize'

_LEADER = 'leader'

_LEADER_OPTIME = 'optime/leader'

_MEMBERS = 'members/'

_OPTIME = 'optime'

_STATUS = 'status'

_SYNC = 'sync'

__get_postgresql_cluster(path: str | None = None) → Cluster

Low level method to load a Cluster object from DCS.

Parameters:: path – optional client path in DCS backend to load from.
Returns:: a loaded Cluster instance.

__init__(config: Dict[str, Any], mpp: AbstractMPP) → None

Prepare DCS paths, MPP object, initial values for state information and processing dependencies.

Parameters:

config – dict, reference to config section of selected DCS. i.e.: zookeeper for zookeeper, etcd for etcd, etc…
mpp – an object implementing AbstractMPP interface.

_abc_impl = <_abc._abc_data object>

_build_retain_slots(cluster: Cluster, slots: Dict[str, int] | None) → List[str] | None

Handle retention policy of physical replication slots for cluster members.

When the member key is missing we want to keep its replication slot for a while, so that WAL segments will not be already absent when it comes back online. It is being solved by storing the list of replication slots representing members in the retain_slots field of the /status key.

This method handles retention policy by keeping the list of such replication slots in memory and removing names when they were observed longer than member_slots_ttl ago.

Parameters:

cluster – Cluster object with information about the current cluster state.
slots – slot names with LSN values that exist on the leader node and consists of slots for cluster members and permanent replication slots.

Returns:

the list of replication slots to be written to /status key or None.

abstractmethod _delete_leader(leader: Leader) → bool

Remove leader key from DCS.

This method should remove leader key if current instance is the leader.

Parameters:: leader – Leader object with information about the leader.
Returns:: True if successfully committed to DCS.

_get_mpp_cluster() → Cluster

Load MPP cluster from DCS.

Returns:: A MPP Cluster instance for the coordinator with workers clusters in the Cluster.workers dict.

abstractmethod _load_cluster(path: str, loader: Callable[[Any], Cluster | Dict[int, Cluster]]) → Cluster | Dict[int, Cluster]

Main abstract method that implements the loading of Cluster instance.

Note

Internally this method should call the loader method that will build Cluster object which represents current state and topology of the cluster in DCS. This method supposed to be called only by the get_cluster() method.

Parameters:

path – the path in DCS where to load Cluster(s) from.
loader – one of _postgresql_cluster_loader() or _mpp_cluster_loader().

Raise:

DCSError in case of communication problems with DCS. If the current node was running as a primary and exception raised, instance would be demoted.

abstractmethod _mpp_cluster_loader(path: Any) → Dict[int, Cluster]

Load and build all PostgreSQL clusters from a single MPP cluster.

Parameters:: path – the path in DCS where to load Cluster(s) from.
Returns:: all MPP groups as dict, with group IDs as keys and Cluster objects as values.

abstractmethod _postgresql_cluster_loader(path: Any) → Cluster

Load and build the Cluster object from DCS, which represents a single PostgreSQL cluster.

Parameters:: path – the path in DCS where to load Cluster from.
Returns:: Cluster instance.

_set_loop_wait(loop_wait: int) → None

Set new loop_wait value.

Parameters:: loop_wait – value to set.

abstractmethod _update_leader(leader: Leader) → bool

Update leader key (or session) ttl.

Note

You have to use CAS (Compare And Swap) operation in order to update leader key, for example for etcd prevValue parameter must be used.

If update fails due to DCS not being accessible or because it is not able to process requests (hopefully temporary), the DCSError exception should be raised.

Parameters:: leader – a reference to a current leader object.
Returns:: True if leader key (or session) has been updated successfully.

abstractmethod _write_failsafe(value: str) → bool

Write current cluster topology to DCS that will be used by failsafe mechanism (if enabled).

Parameters:: value – failsafe topology serialized in JSON format.
Returns:: True if successfully committed to DCS.

abstractmethod _write_leader_optime(last_lsn: str) → bool

Write current WAL LSN into /optime/leader key in DCS.

Parameters:: last_lsn – absolute WAL LSN in bytes.
Returns:: True if successfully committed to DCS.

abstractmethod _write_status(value: str) → bool

Write current WAL LSN and confirmed_flush_lsn of permanent slots into the /status key in DCS.

Parameters:: value – status serialized in JSON format.
Returns:: True if successfully committed to DCS.

acquire_leader_lock() → bool

Attempt to acquire leader lock.

Note

This method wraps attempt_to_acquire_leader(): and is used to reset retention time of physical replication slots that representing members of the cluster when current node is to be promoted to the leader.

Returns:: True if the leader key has been created successfully.

abstractmethod attempt_to_acquire_leader() → bool

Attempt to acquire leader lock.

Note

This method should create /leader key with the value _name.

The key must be created atomically. In case the key already exists it should not be overwritten and False must be returned.

If key creation fails due to DCS not being accessible or because it is not able to process requests (hopefully temporary), the DCSError exception should be raised.

Returns:: True if key has been created successfully.

abstractmethod cancel_initialization() → bool

Removes the initialize key for a cluster.

Returns:: True if successfully committed to DCS.

client_path(path: str) → str

Construct the absolute key name from appropriate parts for the DCS type.

Parameters:: path – The key name within the current Patroni cluster.
Returns:: absolute key name for the current Patroni cluster.

property cluster: Cluster | None: Cached DCS cluster information that has not yet expired.

property config_path: str: Get the client path for config.

abstractmethod delete_cluster() → bool

Delete cluster from DCS.

Returns:: True if successfully committed to DCS.

delete_leader(leader: Leader | None, last_lsn: int | None = None) → bool

Update optime/leader and voluntarily remove leader key from DCS.

This method should remove leader key if current instance is the leader.

Parameters:

leader – Leader object with information about the leader.
last_lsn – latest checkpoint location in bytes.

Returns:

boolean result of called abstract _delete_leader().

abstractmethod delete_sync_state(version: Any | None = None) → bool

Delete the synchronous state from DCS.

Parameters:: version – for conditional deletion of the key/object.
Returns:: True if delete successful.

property failover_path: str: Get the client path for failover.

property failsafe: Dict[str, str] | None: Stored value of _last_failsafe.

property failsafe_path: str: Get the client path for failsafe.

get_cluster() → Cluster

Retrieve a fresh view of DCS.

Note

Stores copy of time, status and failsafe values for comparison in DCS update decisions. Caching is required to avoid overhead placed upon the REST API.

Returns either a PostgreSQL or MPP implementation of Cluster depending on availability.

Returns:

get_mpp_coordinator() → Cluster | None

Load the PostgreSQL cluster for the MPP Coordinator.

Note

This method is only executed on the worker nodes to find the coordinator.

Returns:: Select Cluster instance associated with the MPP Coordinator group ID.

property history_path: str: Get the client path for history.

abstractmethod initialize(create_new: bool = True, sysid: str = '') → bool

Race for cluster initialization.

This method should atomically create initialize key and return True, otherwise it should return False.

Parameters:

create_new – False if the key should already exist (in the case we are setting the system_id).
sysid – PostgreSQL cluster system identifier, if specified, is written to the key.

Returns:

True if key has been created successfully.

property initialize_path: str: Get the client path for initialize.

is_mpp_coordinator() → bool

Cluster instance has a Coordinator group ID.

Returns:: True if the given node is running as the MPP Coordinator.

property last_seen: int: The time recorded when the DCS was last reachable.

property leader_optime_path: str: Get the client path for optime/leader (legacy key, superseded by status).

property leader_path: str: Get the client path for leader.

property loop_wait: int: The recorded value for cluster HA loop wait time in seconds.

manual_failover(leader: str | None, candidate: str | None, scheduled_at: datetime | None = None, version: Any | None = None) → bool

Prepare dictionary with given values and set /failover key in DCS.

Parameters:

leader – value to set for leader.
candidate – value to set for member.
scheduled_at – value converted to ISO date format for scheduled_at.
version – for conditional update of the key/object.

Returns:

True if successfully committed to DCS.

property member_path: str: Get the client path for member representing this node.

property members_path: str: Get the client path for members.

property mpp: AbstractMPP: Get the effective underlying MPP, if any has been configured.

reload_config(config: Config | Dict[str, Any]) → None

Load and set relevant values from configuration.

Sets loop_wait, ttl and retry_timeout properties.

Parameters:: config – Loaded configuration information object or dictionary of key value pairs.

reset_cluster() → None: Clear cached state of DCS.

abstractmethod set_config_value(value: str, version: Any | None = None) → bool

Create or update /config key in DCS.

Parameters:

value – new value to set in the config key.
version – for conditional update of the key/object.

Returns:

True if successfully committed to DCS.

abstractmethod set_failover_value(value: str, version: Any | None = None) → bool

Create or update /failover key.

Parameters:

value – value to set.
version – for conditional update of the key/object.

Returns:

True if successfully committed to DCS.

abstractmethod set_history_value(value: str) → bool

Set value for history in DCS.

Parameters:: value – new value of history key/object.
Returns:: True if successfully committed to DCS.

abstractmethod set_retry_timeout(retry_timeout: int) → None: Set the new value for retry_timeout.

abstractmethod set_sync_state_value(value: str, version: Any | None = None) → Any | bool

Set synchronous state in DCS.

Parameters:

value – the new value of /sync key.
version – for conditional update of the key/object.

Returns:

version of the new object or False in case of error.

abstractmethod set_ttl(ttl: int) → bool | None: Set the new ttl value for DCS keys.

property status_path: str: Get the client path for status.

property sync_path: str: Get the client path for sync.

static sync_state(leader: str | None, sync_standby: Collection[str] | None, quorum: int | None) → Dict[str, Any]

Build sync_state dictionary.

Parameters:

leader – name of the leader node that manages /sync key.
sync_standby – collection of currently known synchronous standby node names.
quorum – if the node from sync_standby list is doing a leader race it should see at least quorum other nodes from the sync_standby + leader list

Returns:

dictionary that later could be serialized to JSON or saved directly to DCS.

abstractmethod take_leader() → bool

Establish a new leader in DCS.

Note

This method should create leader key with value of _name and ttl of ttl.

Since it could be called only on initial cluster bootstrap it could create this key regardless, overwriting the key if necessary.

Returns:: True if successfully committed to DCS.

abstractmethod touch_member(data: Dict[str, Any]) → bool

Update member key in DCS.

Note

This method should create or update key with the name with /members/ + _name and the value of data in a given DCS.

Parameters:: data – information about an instance (including connection strings).
Returns:: True if successfully committed to DCS.

abstract property ttl: int: Get current ttl value.

update_leader(cluster: Cluster, last_lsn: int | None, slots: Dict[str, int] | None = None, failsafe: Dict[str, str] | None = None) → bool

Update leader key (or session) ttl, /status, and /failsafe keys.

Parameters:

cluster – Cluster object with information about the current cluster state.
last_lsn – absolute WAL LSN in bytes.
slots – dictionary with permanent slots confirmed_flush_lsn.
failsafe – if defined dictionary passed to write_failsafe().

Returns:

True if leader key (or session) has been updated successfully.

watch(leader_version: Any | None, timeout: float) → bool

Sleep if the current node is a leader, otherwise, watch for changes of leader key with a given timeout.

Parameters:

leader_version – version of a leader key.
timeout – timeout in seconds.

Returns:

if True this will reschedule the next run of the HA cycle.

write_failsafe(value: Dict[str, str]) → None

Write the /failsafe key in DCS.

Parameters:: value – dictionary value to set, consisting of the name and api_url of members.

write_leader_optime(last_lsn: int) → None

Write value for WAL LSN to optime/leader key in DCS.

Note

This method abstracts away the required data structure of write_status(), so it is not needed in the caller. However, the optime/leader is only written in write_status() when the cluster has members with a Patroni version that is old enough to require it (i.e. the old Patroni version doesn’t understand the new format).

Parameters:: last_lsn – absolute WAL LSN in bytes.

write_status(value: Dict[str, Any]) → None

Write cluster status to DCS if changed.

Note

The DCS key /status was introduced in Patroni version 2.1.0. Previous to this the position of last known leader LSN was stored in optime/leader. This method has detection for backwards compatibility of members with a version older than this.

Parameters:: value – JSON serializable dictionary with current WAL LSN and confirmed_flush_lsn of permanent slots.

Write the new synchronous state to DCS.

Calls sync_state() to build a dictionary and then calls DCS specific set_sync_state_value().

Parameters:

leader – name of the leader node that manages /sync key.
sync_standby – collection of currently known synchronous standby node names.
version – for conditional update of the key/object.
quorum – if the node from sync_standby list is doing a leader race it should see at least quorum other nodes from the sync_standby + leader list

Returns:

the new SyncState object or None.

class patroni.dcs.Cluster(*args: Any, **kwargs: Any)

Bases: Cluster

Immutable object (namedtuple) which represents PostgreSQL or MPP cluster.

Note

We are using an old-style attribute declaration here because otherwise it is not possible to override __new__ method. Without it the workers by default gets always the same dict object that could be mutated.

Consists of the following fields:

Variables:

initialize – shows whether this cluster has initialization key stored in DC or not.
config – global dynamic configuration, reference to ClusterConfig object.
leader – Leader object which represents current leader of the cluster.
status – Status object which represents the /status key.
members – list of:class:` Member` objects, all PostgreSQL cluster members including leader
failover – reference to Failover object.
sync – reference to SyncState object, last observed synchronous replication state.
history – reference to TimelineHistory object.
failsafe – failsafe topology. Node is allowed to become the leader only if its name is found in this list.
workers – dictionary of workers of the MPP cluster, optional. Each key representing the group and the corresponding value is a Cluster instance.

property __permanent_logical_slots: Dict[str, Any]: Dictionary of permanent logical replication slots.

property __permanent_slots: Dict[str, Dict[str, Any] | Any]: Dictionary of permanent replication slots with their known LSN.

_get_members_slots(name: str, role: PostgresqlRole, nofailover: bool, can_advance_slots: bool) → Dict[str, Dict[str, Any]]

Get physical replication slots configuration for a given member.

There are following situations possible:

If the nostream tag is set on the member - we should not have the replication slot for it on the current primary or any other member even if replicatefrom is set, because nostream disables WAL streaming.

PostgreSQL is 11 and newer and configuration allows retention of member replication slots. In this case we want to have replication slots for every member except the case when we have nofailover tag set.

PostgreSQL is older than 11 or configuration doesn’t allow member slots retention. In this case we want:

On primary have replication slots for all members that don’t have replicatefrom tag pointing to the existing member.

On replica node have replication slots only for members which replicatefrom tag pointing to us.

Will log an error if:

Conflicting slot names between members are found

Parameters:

name – name of this node.
role – role of this node, primary, standby_leader, or replica.
nofailover – True if this node is tagged to not be a failover candidate, False otherwise.
can_advance_slots – True if pg_replication_slot_advance() function is available, False otherwise.

Returns:

dictionary of physical replication slots that should exist on a given node.

_get_permanent_slots(postgresql: Postgresql, tags: Tags, role: PostgresqlRole) → Dict[str, Any]

Get configured permanent replication slots.

Note

Permanent replication slots are only considered if use_slots configuration is enabled. A node that is not supposed to become a leader (nofailover) will not have permanent replication slots. Also node with disabled streaming (nostream) and its cascading followers must not have permanent logical slots due to lack of feedback from node to primary, which makes them unsafe to use.

In a standby cluster we only support physical replication slots.

The returned dictionary for a non-standby cluster always contains permanent logical replication slots in order to show a warning if they are not supported by PostgreSQL before v11.

Parameters:

postgresql – reference to Postgresql object.
tags – reference to an object implementing Tags interface.
role – role of the node. One of PostgresqlRole values.

Returns:

dictionary of permanent slot names mapped to attributes.

_has_permanent_logical_slots(postgresql: Postgresql, member: Tags) → bool

Check if the given member node has permanent logical replication slots configured.

Parameters:

postgresql – reference to a Postgresql object.
member – reference to an object implementing Tags interface for the node that we are checking permanent logical replication slots for.

Returns:

True if any detected replications slots are logical, otherwise False.

_merge_permanent_slots(slots: Dict[str, Dict[str, Any]], permanent_slots: Dict[str, Any], 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.

Will update slots in-line based on type of slot, physical or logical, and name of node. Type is assumed to be physical if there are no attributes stored as the slot value.

Parameters:

slots – Slot names with existing attributes if known.
name – name of this node.
role – role of the node. One of PostgresqlRole values.
permanent_slots – dictionary containing slot name key and slot information values.
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.

static empty() → Cluster: Produce an empty Cluster instance.

get_clone_member(exclude_name: str) → Member | Leader | None

Get member or leader object to use as clone source.

Parameters:: exclude_name – name of a member name to exclude.
Returns:: a randomly selected candidate member from available running members that are configured to as viable sources for cloning (has tag clonefrom in configuration). If no member is appropriate the current leader is used.

get_member(member_name: str, fallback_to_leader: bool = True) → Member | Leader | None

Get Member object by name or the Leader.

Parameters:

member_name – name of the member to retrieve.
fallback_to_leader – if True return the Leader instead if the member cannot be found.

Returns:

the Member if found or Leader object.

get_replication_slots(postgresql: Postgresql, member: Tags, *, role: PostgresqlRole | None = 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:

Any logical slots are disabled, due to version compatibility, and show_error is True.

Parameters:

postgresql – reference to Postgresql object.
member – reference to an object implementing Tags interface.
role – role of the node, if not set will be taken from postgresql One of PostgresqlRole values.
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.

get_slot_name_on_primary(name: str, tags: Tags) → str | None

Get the name of physical replication slot for this node on the primary.

Note

P <– I <– L

In case of cascading replication we have to check not our physical slot, but slot of the replica that connects us to the primary.

Parameters:

name – name of the member node to check.
tags – reference to an object implementing Tags interface.

Returns:

the slot name on the primary that is in use for physical replication on this node.

has_member(member_name: str) → bool

Check if the given member name is present in the cluster.

Parameters:: member_name – name to look up in the members.
Returns:: True if the member name is found.

has_permanent_slots(postgresql: Postgresql, member: Tags) → bool

Check if our node has permanent replication slots configured.

Parameters:

postgresql – reference to Postgresql object.
member – reference to an object implementing Tags interface for the node that we are checking permanent logical replication slots for.

Returns:

True if there are permanent replication slots configured, otherwise False.

is_empty()

Validate definition of all attributes of this Cluster instance.

Returns:: True if all attributes of the current Cluster are unpopulated.

static is_logical_slot(value: Any) → bool

Check whether provided configuration is for permanent logical replication slot.

Parameters:: value – configuration of the permanent replication slot.
Returns:: True if value is a logical replication slot, otherwise False.

static is_physical_slot(value: Any) → bool

Check whether provided configuration is for permanent physical replication slot.

Parameters:: value – configuration of the permanent replication slot.
Returns:: True if value is a physical replication slot, otherwise False.

is_unlocked() → bool

Check if the cluster does not have the leader.

Returns:: True if a leader name is not defined.

property leader_name: str | None: The name of the leader if defined otherwise None.

maybe_filter_permanent_slots(postgresql: Postgresql, slots: Dict[str, int]) → Dict[str, int]

Filter out all non-permanent slots from provided slots dict.

Note

In case if retention of replication slots for members is enabled we will not do any filtering, because we need to publish LSN values for members replication slots, so that other nodes can use them to advance LSN, like they do it for permanent slots.

Parameters:

postgresql – reference to Postgresql object.
slots – slot names with LSN values.

Returns:

a dict object that contains only slots that are known to be permanent.

property min_version: Tuple[int, ...] | None: Lowest Patroni software version found in known members of the cluster.

property permanent_physical_slots: Dict[str, Any]: Dictionary of permanent physical replication slots.

should_enforce_hot_standby_feedback(postgresql: Postgresql, member: Tags) → bool

Determine whether hot_standby_feedback should be enabled for the given member.

The hot_standby_feedback must be enabled if the current replica has logical slots, or it is working as a cascading replica for the other node that has logical slots.

Parameters:

postgresql – reference to a Postgresql object.
member – reference to an object implementing Tags interface for the node that we are checking permanent logical replication slots for.

Returns:

True if this node or any member replicating from this node has permanent logical slots, otherwise False.

property slots: Dict[str, int]

{"slot_name": int}.

Note

We are trying to be foolproof here and for values that can’t be parsed to int will return 0.

Type:: State of permanent replication slots on the primary in the format

property timeline: int

Get the cluster history index from the history.

Returns:

If the recorded history is empty assume timeline is 1, if it is not defined or the stored history is not formatted as expected 0 is returned and an error will be logged. Otherwise, the last number stored incremented by 1 is returned.

Example:

No history provided: >>> Cluster(0, 0, 0, Status.empty(), 0, 0, 0, 0, None, {}).timeline 0

Empty history assume timeline is 1: >>> Cluster(0, 0, 0, Status.empty(), 0, 0, 0, TimelineHistory.from_node(1, ‘[]’), None, {}).timeline 1

Invalid history format, a string of a, returns 0: >>> Cluster(0, 0, 0, Status.empty(), 0, 0, 0, TimelineHistory.from_node(1, ‘[[“a”]]’), None, {}).timeline 0

History as a list of strings: >>> history = TimelineHistory.from_node(1, ‘[[“3”, “2”, “1”]]’) >>> Cluster(0, 0, 0, Status.empty(), 0, 0, 0, history, None, {}).timeline 4

class patroni.dcs.ClusterConfig(version: int | str, data: Dict[str, Any], modify_version: int | str)

Bases: NamedTuple

Immutable object (namedtuple) which represents cluster configuration.

Variables:

version – version number for the object.
data – dictionary of configuration information.
modify_version – modified version number.

_asdict(): Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('version', 'data', 'modify_version')

classmethod _make(iterable): Make a new ClusterConfig object from a sequence or iterable

_replace(**kwds): Return a new ClusterConfig object replacing specified fields with new values

data: Dict[str, Any]: Alias for field number 1

static from_node(version: int | str, value: str, modify_version: int | str | None = None) → ClusterConfig

Factory method to parse value as configuration information.

Parameters:

version – version number for object.
value – raw JSON serialized data, if not parsable replaced with an empty dictionary.
modify_version – optional modify version number, use version if not provided.

Returns:

constructed ClusterConfig instance.

Example:

>>> ClusterConfig.from_node(1, '{') is None
False

modify_version: int | str: Alias for field number 2

version: int | str: Alias for field number 0

class patroni.dcs.Failover(version: int | str, leader: str | None, candidate: str | None, scheduled_at: datetime | None)

Bases: NamedTuple

Immutable object (namedtuple) representing configuration information required for failover/switchover capability.

Variables:

version – version of the object.
leader – name of the leader. If value isn’t empty we treat it as a switchover from the specified node.
candidate – the name of the member node to be considered as a failover candidate.
scheduled_at – in the case of a switchover the datetime object to perform the scheduled switchover.

Example:

>>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader"}'))
True

>>> 'Failover' in str(Failover.from_node(1, {"leader": "cluster_leader"}))
True

>>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader", "member": "cluster_candidate"}'))
True

>>> Failover.from_node(1, 'null') is None
False

>>> n = '''{"leader": "cluster_leader", "member": "cluster_candidate",
...         "scheduled_at": "2016-01-14T10:09:57.1394Z"}'''

>>> 'tzinfo=' in str(Failover.from_node(1, n))
True

>>> Failover.from_node(1, None) is None
False

>>> Failover.from_node(1, '{}') is None
False

>>> 'abc' in Failover.from_node(1, 'abc:def')
True

_asdict(): Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('version', 'leader', 'candidate', 'scheduled_at')

classmethod _make(iterable): Make a new Failover object from a sequence or iterable

_replace(**kwds): Return a new Failover object replacing specified fields with new values

candidate: str | None: Alias for field number 2

static from_node(version: int | str, value: str | Dict[str, str]) → Failover

Factory method to parse value as failover configuration.

Parameters:

version – version number for the object.
value – JSON serialized data or a dictionary of configuration. Can also be a colon : delimited list of leader, followed by candidate name (legacy format). If scheduled_at key is defined the value will be parsed by dateutil.parser.parse().

Returns:

constructed Failover information object

leader: str | None: Alias for field number 1

scheduled_at: datetime | None: Alias for field number 3

version: int | str: Alias for field number 0

class patroni.dcs.Leader(version: int | str, session: int | float | str | None, member: Member)

Bases: NamedTuple

Immutable object (namedtuple) which represents leader key.

Consists of the following fields:

Variables:

version – modification version of a leader key in a Configuration Store
session – either session id or just ttl in seconds
member – reference to a Member object which represents current leader (see Cluster.members)

_asdict(): Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('version', 'session', 'member')

classmethod _make(iterable): Make a new Leader object from a sequence or iterable

_replace(**kwds): Return a new Leader object replacing specified fields with new values

property checkpoint_after_promote: bool | None

Determine whether a checkpoint has occurred for this leader after promotion.

Returns:

True if the role is master or primary and checkpoint_after_promote is not set, False if not a master or primary or if the checkpoint hasn’t occurred. If the version of Patroni is older than 1.5.6, return None.

Example:

>>> Leader(1, '', Member.from_node(1, '', '', '{"version":"z"}')).checkpoint_after_promote

conn_kwargs(auth: Dict[str, str] | None = None) → Dict[str, str]

Connection keyword arguments.

Parameters:: auth – an optional dictionary containing authentication information.
Returns:: the result of the called Member.conn_kwargs() method.

property conn_url: str | None: Connection URL value of the Member instance.

property data: Dict[str, Any]: Data value of the Member instance.

member: Member: Alias for field number 2

property name: str: The leader “member” name.

session: int | float | str | None: Alias for field number 1

property timeline: int | None: Timeline value of data.

version: int | str: Alias for field number 0

class patroni.dcs.Member(version: int | str, name: str, session: int | float | str | None, data: Dict[str, Any])

Bases: Tags, Member

Immutable object (namedtuple) which represents single member of PostgreSQL cluster.

Note

We are using an old-style attribute declaration here because otherwise it is not possible to override __new__ method in the RemoteMember class.

Note

These two keys in data are always written to the DCS, but care is taken to maintain consistency and resilience from data that is read:

conn_url: connection string containing host, user and password which could be used to access this member. api_url: REST API url of patroni instance

Consists of the following fields:

Variables:

version – modification version of a given member key in a Configuration Store.
name – name of PostgreSQL cluster member.
session – either session id or just ttl in seconds.
data – dictionary containing arbitrary data i.e. conn_url, api_url, xlog_location, state, role, tags, etc…

_abc_impl = <_abc._abc_data object>

property api_url: str | None: The api_url value from data if defined.

property clonefrom: bool: True if both clonefrom tag is True and a connection URL is defined.

conn_kwargs(auth: Any | None = None) → Dict[str, Any]

Give keyword arguments used for PostgreSQL connection settings.

Parameters:: auth – Authentication properties - can be defined as anything supported by the psycopg2 or psycopg modules. Converts a key of username to user if supplied.
Returns:: A dictionary containing a merge of default parameter keys host, port and dbname, with the contents of data conn_kwargs key. If those are not defined will parse and reform connection parameters from conn_url. One of these two attributes needs to have data defined to construct the output dictionary. Finally, auth parameters are merged with the dictionary before returned.

property conn_url: str | None: The conn_url value from data if defined or constructed from conn_kwargs.

static from_node(version: int | str, name: str, session: int | float | str | None, value: str) → Member

Factory method for instantiating Member from a JSON serialised string or object.

Parameters:

version – modification version of a given member key in a Configuration Store.
name – name of PostgreSQL cluster member.
session – either session id or just ttl in seconds.
value – JSON encoded string containing arbitrary data i.e. conn_url, api_url, xlog_location, state, role, tags, etc. OR a connection URL starting with postgres://.

Returns:

an Member instance built with the given arguments.

Example:

>>> Member.from_node(-1, '', '', '{"conn_url": "postgres://foo@bar/postgres"}') is not None
True

>>> Member.from_node(-1, '', '', '{')
Member(version=-1, name='', session='', data={})

get_endpoint_url(endpoint: str | None = None) → str

Get URL from member api_url and endpoint.

Parameters:: endpoint – URL path of REST API.
Returns:: full URL for this REST API.

property is_running: bool: True if the member state is RUNNING.

property lsn: int | None: Current LSN (receive/flush/replay).

property patroni_version: Tuple[int, ...] | None

The version string value from data converted to tuple.

Example:

>>> Member.from_node(1, '', '', '{"version":"1.2.3"}').patroni_version
(1, 2, 3)

property receive_lsn: int | None

property replay_lsn: int | None

property state: str: The state value of data.

property tags: Dict[str, Any]: The tags value from data if defined, otherwise an empty dictionary.

class patroni.dcs.RemoteMember(name: str, data: Dict[str, Any])

Bases: Member

Represents a remote member (typically a primary) for a standby cluster.

Variables:: ALLOWED_KEYS – Controls access to relevant key names that could be in stored data.

ALLOWED_KEYS: Tuple[str, ...] = ('primary_slot_name', 'create_replica_methods', 'restore_command', 'archive_cleanup_command', 'recovery_min_apply_delay', 'no_replication_slot')

_abc_impl = <_abc._abc_data object>

exception patroni.dcs.ReturnFalseException

Bases: Exception

Exception to be caught by the catch_return_false_exception() decorator.

class patroni.dcs.Status(last_lsn: int, slots: Dict[str, int] | None, retain_slots: List[str])

Bases: NamedTuple

Immutable object (namedtuple) which represents /status key.

Consists of the following fields:

Variables:

last_lsn – int object containing position of last known leader LSN.
slots – state of permanent replication slots on the primary in the format: {"slot_name": int}.
retain_slots – list physical replication slots for members that exist in the cluster.

_asdict(): Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('last_lsn', 'slots', 'retain_slots')

classmethod _make(iterable): Make a new Status object from a sequence or iterable

_replace(**kwds): Return a new Status object replacing specified fields with new values

static empty() → Status

Construct an empty Status instance.

Returns:: empty Status object.

static from_node(value: str | Dict[str, Any] | None) → Status

Factory method to parse value as Status object.

Parameters:: value – JSON serialized string or dict object.
Returns:: constructed Status object.

is_empty()

Validate definition of all attributes of this Status instance.

Returns:: True if all attributes of the current Status are unpopulated.

last_lsn: int: Alias for field number 0

retain_slots: List[str]: Alias for field number 2

slots: Dict[str, int] | None: Alias for field number 1

class patroni.dcs.SyncState(version: int | str | None, leader: str | None, sync_standby: str | None, quorum: int)

Bases: NamedTuple

Immutable object (namedtuple) which represents last observed synchronous replication state.

Variables:

version – modification version of a synchronization key in a Configuration Store.
leader – reference to member that was leader.
sync_standby – synchronous standby list (comma delimited) which are last synchronized to leader.
quorum – if the node from sync_standby list is doing a leader race it should see at least quorum other nodes from the sync_standby + leader list.

_asdict(): Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('version', 'leader', 'sync_standby', 'quorum')

classmethod _make(iterable): Make a new SyncState object from a sequence or iterable

_replace(**kwds): Return a new SyncState object replacing specified fields with new values

static _str_to_list(value: str) → List[str]

Splits a string by comma and returns list of strings.

Parameters:: value – a comma separated string.
Returns:: list of non-empty strings after splitting an input value by comma.

static empty(version: int | str | None = None) → SyncState

Construct an empty SyncState instance.

Parameters:: version – optional version number.
Returns:: empty synchronisation state object.

static from_node(version: int | str | None, value: str | Dict[str, Any] | None) → SyncState

Factory method to parse value as synchronisation state information.

Parameters:

version – optional version number for the object.
value – (optionally JSON serialised) synchronisation state information

Returns:

constructed SyncState object.

Example:

>>> SyncState.from_node(1, None).leader is None
True

>>> SyncState.from_node(1, '{}').leader is None
True

>>> SyncState.from_node(1, '{').leader is None
True

>>> SyncState.from_node(1, '[]').leader is None
True

>>> SyncState.from_node(1, '{"leader": "leader"}').leader == "leader"
True

>>> SyncState.from_node(1, {"leader": "leader"}).leader == "leader"
True

property is_empty: bool: True if /sync key is not valid (doesn’t have a leader).

leader: str | None: Alias for field number 1

leader_matches(name: str | None) → bool

Compare the given name to stored leader value.

Returns:: True if name is matching the leader value.

matches(name: str | None, check_leader: bool = False) → bool

Checks if node is presented in the /sync state.

Since PostgreSQL does case-insensitive checks for synchronous_standby_name we do it also.

Parameters:

name – name of the node.
check_leader – by default the name is searched for only in members, a value of True will include the leader to list.

Returns:

True if the /sync key not is_empty() and the given name is among those presented in the sync state.

Example:

>>> s = SyncState(1, 'foo', 'bar,zoo', 0)

>>> s.matches('foo')
False

>>> s.matches('fOo', True)
True

>>> s.matches('Bar')
True

>>> s.matches('zoO')
True

>>> s.matches('baz')
False

>>> s.matches(None)
False

>>> SyncState.empty(1).matches('foo')
False

property members: List[str]: sync_standby and leader as list or an empty list if object considered empty.

quorum: int: Alias for field number 3

sync_standby: str | None: Alias for field number 2

version: int | str | None: Alias for field number 0

property voters: List[str]: sync_standby as list or an empty list if undefined or object considered empty.

class patroni.dcs.TimelineHistory(version: int | str, value: Any, lines: List[Tuple[int, int, str] | Tuple[int, int, str, str] | Tuple[int, int, str, str, str]])

Bases: NamedTuple

Object representing timeline history file.

Note

The content held in lines deserialized from value are lines parsed from PostgreSQL timeline history files, consisting of the timeline number, the LSN where the timeline split and any other string held in the file. The files are parsed by parse_history().

Variables:

version – version number of the file.
value – raw JSON serialised data consisting of parsed lines from history files.
lines – List of Tuple parsed lines from history files.

_asdict(): Return a new dict which maps field names to their values.

_field_defaults = {}

_fields = ('version', 'value', 'lines')

classmethod _make(iterable): Make a new TimelineHistory object from a sequence or iterable

_replace(**kwds): Return a new TimelineHistory object replacing specified fields with new values

static from_node(version: int | str, value: str) → TimelineHistory

Parse the given JSON serialized string as a list of timeline history lines.

Parameters:

version – version number
value – JSON serialized string, consisting of parsed lines of PostgreSQL timeline history files, see TimelineHistory.

Returns:

composed timeline history object using parsed lines.

Example:

If the passed value argument is not parsed an empty list of lines is returned:

>>> h = TimelineHistory.from_node(1, 2)

>>> h.lines
[]

lines: List[Tuple[int, int, str] | Tuple[int, int, str, str] | Tuple[int, int, str, str, str]]: Alias for field number 2

value: Any: Alias for field number 1

version: int | str: Alias for field number 0

patroni.dcs.catch_return_false_exception(func: Callable[[...], Any]) → Any

Decorator function for catching functions raising ReturnFalseException.

Parameters:: func – function to be wrapped.
Returns:: wrapped function.

patroni.dcs.dcs_modules() → List[str]

Get names of DCS modules, depending on execution environment.

Returns:: list of known module names with absolute python module path namespace, e.g. patroni.dcs.etcd.

patroni.dcs.get_dcs(config: Config | Dict[str, Any]) → AbstractDCS

Attempt to load a Distributed Configuration Store from known available implementations.

Note

Using the list of available DCS classes returned by iter_classes() attempt to dynamically instantiate the class that implements a DCS using the abstract class AbstractDCS.

Basic top-level configuration parameters retrieved from config are propagated to the DCS specific config before being passed to the module DCS class.

If no module is found to satisfy configuration then report and log an error. This will cause Patroni to exit.

:raises PatroniFatalException: if a load of all available DCS modules have been tried and none succeeded.

Parameters:: config – object or dictionary with Patroni configuration. This is normally a representation of the main Patroni
Returns:: The first successfully loaded DCS module which is an implementation of AbstractDCS.

patroni.dcs.iter_dcs_classes(config: Config | Dict[str, Any] | None = None) → Iterator[Tuple[str, Type[AbstractDCS]]]

Attempt to import DCS modules that are present in the given configuration.

Note

If a module successfully imports we can assume that all its requirements are installed.

Parameters:: config – configuration information with possible DCS names as keys. If given, only attempt to import DCS modules defined in the configuration. Else, if None, attempt to import any supported DCS module.
Returns:: an iterator of tuples, each containing the module name and the imported DCS class object.

patroni.dcs.parse_connection_string(value: str) → Tuple[str, str | None]

Split and rejoin a URL string into a connection URL and an API URL.

Note

Original Governor stores connection strings for each cluster members in a following format:

postgres://{username}:{password}@{connect_address}/postgres

Since each of our patroni instances provides their own REST API endpoint, it’s good to store this information in DCS along with PostgreSQL connection string. In order to not introduce new keys and be compatible with original Governor we decided to extend original connection string in a following way:

postgres://{username}:{password}@{connect_address}/postgres?application_name={api_url}

This way original Governor could use such connection string as it is, because of feature of libpq library.

Parameters:: value – The URL string to split.
Returns:: the connection string stored in DCS split into two parts, conn_url and api_url.

patroni.dcs.slot_name_from_member_name(member_name: str) → str

Translate member name to valid PostgreSQL slot name.

Note

PostgreSQL’s replication slot names must be valid PostgreSQL names. This function maps the wider space of member names to valid PostgreSQL names. Names have their case lowered, dashes and periods common in hostnames are replaced with underscores, other characters are encoded as their unicode codepoint. Name is truncated to 64 characters. Multiple different member names may map to a single slot name.

Parameters:: member_name – The string to convert to a slot name.
Returns:: The string converted using the rules described above.