gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Ignore D401 in flake8-docstrings (#2627)

Browse Source 
* Ignore D401 in flake8-docstrings
* Fix newly reported flake8 issues, ignore the old W503 rule
* rely on concatenation of adjecent strings
* Format behave scripts
* Reformat ha.py according to new rules

Co-authored-by: Alexander Kukushkin <[email protected]>
This commit is contained in:
Polina Bungina
2023-04-03 09:52:22 +02:00
committed by GitHub
co-authored by Alexander Kukushkin
parent 6f357a4e17
commit 3fe2a7868a
37 changed files with 282 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/environment.py
+19 -19
View File
@@ -191,7 +191,7 @@ class PatroniController(AbstractController):
config['raft'] = {'data_dir': self._output_dir, 'self_addr': 'localhost:' + os.environ['RAFT_PORT']}
host = config['restapi']['listen'].rsplit(':', 1)[0]
config['restapi']['listen'] = config['restapi']['connect_address'] = '{0}:{1}'.format(host, 8008+int(name[-1]))
config['restapi']['listen'] = config['restapi']['connect_address'] = '{}:{}'.format(host, 8008 + int(name[-1]))
host = config['postgresql']['listen'].rsplit(':', 1)[0]
config['postgresql']['listen'] = config['postgresql']['connect_address'] = '{0}:{1}'.format(host, self.__PORT)
@@ -251,9 +251,9 @@ class PatroniController(AbstractController):
'parameters': {
'wal_keep_segments': 100,
'archive_mode': 'on',
'archive_command': (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT +
' --mode archive ' +
'--dirname {} --filename %f --pathname %p').format(
'archive_command': (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT
+ ' --mode archive '
+ '--dirname {} --filename %f --pathname %p').format(
os.path.join(self._work_directory, 'data', 'wal_archive'))
}
}
@@ -801,7 +801,7 @@ class PatroniPoolController(object):
raise Exception # this one should never happen because the previous line will always raise and exception
except Exception as e:
self._context.postgres_supports_ssl = isinstance(e, subprocess.CalledProcessError)\
and 'SSL is not supported by this build' not in e.output.decode()
and 'SSL is not supported by this build' not in e.output.decode()
@property
def patroni_path(self):
@@ -853,8 +853,8 @@ class PatroniPoolController(object):
'bootstrap': {
'method': 'pg_basebackup',
'pg_basebackup': {
'command': " ".join(self.BACKUP_SCRIPT +
['--walmethod=stream', '--dbname="{0}"'.format(f.backup_source)])
'command': " ".join(self.BACKUP_SCRIPT
+ ['--walmethod=stream', '--dbname="{0}"'.format(f.backup_source)])
},
'dcs': {
'postgresql': {
@@ -867,9 +867,9 @@ class PatroniPoolController(object):
'postgresql': {
'parameters': {
'archive_mode': 'on',
'archive_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode archive ' +
'--dirname {} --filename %f --pathname %p').format(
os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/'))
'archive_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode archive '
+ '--dirname {} --filename %f --pathname %p')
.format(os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/'))
},
'authentication': {
'superuser': {'password': 'zalando1'},
@@ -885,13 +885,13 @@ class PatroniPoolController(object):
'bootstrap': {
'method': 'backup_restore',
'backup_restore': {
'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' +
os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')),
'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir='
+ os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')),
'recovery_conf': {
'recovery_target_action': 'promote',
'recovery_target_timeline': 'latest',
'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' +
'--dirname {} --filename %f --pathname %p').format(
'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore '
+ '--dirname {} --filename %f --pathname %p').format(
os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/'))
}
}
@@ -910,14 +910,14 @@ class PatroniPoolController(object):
'scope': cluster_name,
'postgresql': {
'recovery_conf': {
'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' +
'--dirname {} --filename %f --pathname %p').format(
os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/'))
'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore '
+ '--dirname {} --filename %f --pathname %p')
.format(os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/'))
},
'create_replica_methods': ['no_leader_bootstrap'],
'no_leader_bootstrap': {
'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' +
os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')),
'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir='
+ os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')),
'no_leader': '1'
}
}
features/steps/basic_replication.py
+1 -1
View File
@@ -38,7 +38,7 @@ def toggle_wal_replay(context, action, pg_name):
# pause or resume the wal replay process
try:
version = context.pctl.query(pg_name, "SHOW server_version_num").fetchone()[0]
wal_name = 'xlog' if int(version)/10000 < 10 else 'wal'
wal_name = 'xlog' if int(version) / 10000 < 10 else 'wal'
context.pctl.query(pg_name, "SELECT pg_{0}_replay_{1}()".format(wal_name, action))
except pg.Error as e:
assert False, "Error during {0} wal recovery on {1}: {2}".format(action, pg_name, e)
features/steps/citus.py
+3 -3
View File
@@ -35,7 +35,7 @@ def check_group_member(context, name, group, key, value, time_limit):
except Exception:
pass
time.sleep(1)
assert False, ("{0} in a group {1} does not have {2}={3} (found {4}) in dcs" +
assert False, ("{0} in a group {1} does not have {2}={3} (found {4}) in dcs"
" after {5} seconds").format(name, group, key, value, response, time_limit)
@@ -93,7 +93,7 @@ def thread_is_alive(context):
@step("I stop a thread")
def stop_insert_thread(context):
context.thread_stop_event.set()
context.thread.join(1*context.timeout_multiplier)
context.thread.join(1 * context.timeout_multiplier)
assert not context.thread.is_alive(), "Thread is still alive"
@@ -114,4 +114,4 @@ def check_transaction(context, name):
@step("a transaction finishes in {timeout:d} seconds")
def check_transaction_timeout(context, timeout):
assert (datetime.now(tzutc) - context.xact_start).seconds > timeout,\
"a transaction finished earlier than in {0} seconds".format(timeout)
"a transaction finished earlier than in {0} seconds".format(timeout)
features/steps/standby_cluster.py
+2 -2
View File
@@ -16,8 +16,8 @@ def start_patroni(context, name, cluster_name):
"postgresql": {
"callbacks": callbacks(context, name),
"backup_restore": {
"command": (context.pctl.PYTHON + " features/backup_restore.py --sourcedir=" +
os.path.join(context.pctl.patroni_path, 'data', 'basebackup').replace('\\', '/'))}
"command": (context.pctl.PYTHON + " features/backup_restore.py --sourcedir="
+ os.path.join(context.pctl.patroni_path, 'data', 'basebackup').replace('\\', '/'))}
}
})
patroni/api.py
+3 -3
View File
@@ -338,7 +338,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
metrics.append("# TYPE patroni_is_paused gauge")
metrics.append("patroni_is_paused{0} {1}".format(scope_label, int(postgres.get('pause', 0))))
self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain')
self._write_response(200, '\n'.join(metrics) + '\n', content_type='text/plain')
def _read_json_content(self, body_is_optional=False):
if 'content-length' not in self.headers:
@@ -537,7 +537,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
def poll_failover_result(self, leader, candidate, action):
timeout = max(10, self.server.patroni.dcs.loop_wait)
for _ in range(0, timeout*2):
for _ in range(0, timeout * 2):
time.sleep(1)
try:
cluster = self.server.patroni.dcs.get_cluster()
@@ -861,7 +861,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
HTTPServer.__init__(self, info[0][-1][:2], RestApiHandler)
except socket.error:
logger.error(
"Couldn't start a service on '%s:%s', please check your `restapi.listen` configuration", host, port)
"Couldn't start a service on '%s:%s', please check your `restapi.listen` configuration", host, port)
raise
def __initialize(self, listen, ssl_options):
patroni/config.py
+2 -2
View File
@@ -321,8 +321,8 @@ class Config(object):
@staticmethod
def _process_postgresql_parameters(parameters, is_local=False):
return {name: value for name, value in (parameters or {}).items()
if name not in ConfigHandler.CMDLINE_OPTIONS or
not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)}
if name not in ConfigHandler.CMDLINE_OPTIONS
or not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)}
def _safe_copy_dynamic_configuration(self, dynamic_configuration):
config = deepcopy(self.__DEFAULT_CONFIG)
patroni/ctl.py
+5 -5
View File
@@ -249,9 +249,9 @@ def get_all_members(obj, cluster, group, role='leader'):
role = {'primary': 'master', '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') != 'master' and role == 'standby_leader' or
cluster.leader.data.get('role') != 'standby_leader' and role == 'master'):
(role == 'leader'
or cluster.leader.data.get('role') != 'master' and role == 'standby_leader'
or cluster.leader.data.get('role') != 'standby_leader' and role == 'master'):
yield cluster.leader.member
return
@@ -879,7 +879,7 @@ def output_members(obj, cluster, name, extended=False, fmt='pretty', group=None)
member.update(cluster=name, member=member['name'], group=g,
host=member.get('host', ''), tl=member.get('timeline', ''),
role=member['role'].replace('_', ' ').title(),
lag_in_mb=round(lag/1024/1024) if isinstance(lag, int) else lag,
lag_in_mb=round(lag / 1024 / 1024) if isinstance(lag, int) else lag,
pending_restart='*' if member.get('pending_restart') else '')
if append_port and member['host'] and member.get('port'):
@@ -1254,7 +1254,7 @@ def edit_config(obj, cluster_name, group, force, quiet, kvpairs, pgkvpairs, appl
after_editing, changed_data = apply_yaml_file(changed_data, apply_filename)
if kvpairs or pgkvpairs:
all_pairs = list(kvpairs) + ['postgresql.parameters.'+v.lstrip() for v in pgkvpairs]
all_pairs = list(kvpairs) + ['postgresql.parameters.' + v.lstrip() for v in pgkvpairs]
after_editing, changed_data = apply_config_changes(before_editing, changed_data, all_pairs)
# If no changes were specified on the command line invoke editor
patroni/dcs/__init__.py
+8 -8
View File
@@ -353,8 +353,8 @@ class ClusterConfig(namedtuple('ClusterConfig', 'index,data,modify_index')):
@property
def permanent_slots(self):
return isinstance(self.data, dict) and (
self.data.get('permanent_replication_slots') or
self.data.get('permanent_slots') or self.data.get('slots')
self.data.get('permanent_replication_slots')
or self.data.get('permanent_slots') or self.data.get('slots')
) or {}
@property
@@ -546,15 +546,15 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,'
# primary), or if replicatefrom destination member happens to be the current primary
use_slots = self.use_slots
if role in ('master', 'primary', 'standby_leader'):
slot_members = [m.name for m in self.members if use_slots and m.name != my_name and
(m.replicatefrom is None or m.replicatefrom == my_name or
not self.has_member(m.replicatefrom))]
slot_members = [m.name for m in self.members if use_slots and m.name != my_name
and (m.replicatefrom is None or m.replicatefrom == my_name
or not self.has_member(m.replicatefrom))]
permanent_slots = self.__permanent_slots if use_slots and \
role in ('master', 'primary') else self.__permanent_physical_slots
else:
# only manage slots for replicas that replicate from this one, except for the leader among them
slot_members = [m.name for m in self.members if use_slots and
m.replicatefrom == my_name and m.name != self.leader_name]
slot_members = [m.name for m in self.members if use_slots
and m.replicatefrom == my_name and m.name != self.leader_name]
permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {}
slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members}
@@ -590,7 +590,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,'
if major_version < 110000:
disabled_permanent_logical_slots.append(name)
elif name in slots:
logger.error("Permanent logical replication slot {'%s': %s} is conflicting with" +
logger.error("Permanent logical replication slot {'%s': %s} is conflicting with"
" physical replication slot for cluster member", name, value)
else:
slots[name] = value
patroni/dcs/consul.py
+5 -5
View File
@@ -15,7 +15,7 @@ from urllib3.exceptions import HTTPError
from urllib.parse import urlencode, urlparse, quote
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
from ..exceptions import DCSError
from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT
@@ -63,7 +63,7 @@ class HTTPClient(object):
self._ttl = None
def set_read_timeout(self, timeout):
self._read_timeout = timeout/3.0
self._read_timeout = timeout / 3.0
@property
def ttl(self):
@@ -113,7 +113,7 @@ class HTTPClient(object):
# supplied maximum wait time to spread out the wake up time of any concurrent requests. This adds
# up to wait / 16 additional time to the maximum duration. Since our goal is actually getting a
# response rather read timeout we will add to the timeout a slightly bigger value.
kwargs['timeout'] = timeout + max(timeout/15.0, 1)
kwargs['timeout'] = timeout + max(timeout / 15.0, 1)
else:
kwargs['timeout'] = self._read_timeout
kwargs['headers'] = (headers or {}).copy()
@@ -267,7 +267,7 @@ class Consul(AbstractDCS):
self._register_service = should_register_service
def set_ttl(self, ttl):
if self._client.http.set_ttl(ttl/2.0): # Consul multiplies the TTL by 2x
if self._client.http.set_ttl(ttl / 2.0): # Consul multiplies the TTL by 2x
self._session = None
self.__do_not_watch = True
@@ -282,7 +282,7 @@ class Consul(AbstractDCS):
def adjust_ttl(self):
try:
settings = self._client.agent.self()
min_ttl = (settings['Config']['SessionTTLMin'] or 10000000000)/1000000000.0
min_ttl = (settings['Config']['SessionTTLMin'] or 10000000000) / 1000000000.0
logger.warning('Changing Session TTL from %s to %s', self._client.http.ttl, min_ttl)
self._client.http.set_ttl(min_ttl)
except Exception:
patroni/dcs/etcd.py
+6 -6
View File
@@ -22,7 +22,7 @@ from urllib3 import Timeout
from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
from ..exceptions import DCSError
from ..request import get as requests_get
from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT
@@ -148,7 +148,7 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
kwargs.update(retries=0, timeout=timeout)
else:
_, per_node_timeout, per_node_retries = self._calculate_timeouts(etcd_nodes)
connect_timeout = max(1, per_node_timeout/2)
connect_timeout = max(1, per_node_timeout / 2)
kwargs.update(timeout=Timeout(connect=connect_timeout, total=per_node_timeout), retries=per_node_retries)
return kwargs
@@ -233,8 +233,8 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
# whether the key didn't received an update or there is a network problem.
elif i + 1 < len(machines_cache):
self.set_base_uri(machines_cache[i + 1])
if (isinstance(fields, dict) and fields.get("wait") == "true" and
isinstance(e, (ReadTimeoutError, ProtocolError))):
if (isinstance(fields, dict) and fields.get("wait") == "true"
and isinstance(e, (ReadTimeoutError, ProtocolError))):
logger.debug("Watch timed out.")
raise etcd.EtcdWatchTimedOut("Watch timed out: {0}".format(e), cause=e)
logger.error("Request to server %s failed: %r", base_uri, e)
@@ -287,7 +287,7 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
retry.sleep_func(sleeptime)
retry.update_delay()
# We still have some time left. Partially reduce `machines_cache` and retry request
kwargs.update(timeout=Timeout(connect=max(1, timeout/2), total=timeout), retries=retries)
kwargs.update(timeout=Timeout(connect=max(1, timeout / 2), total=timeout), retries=retries)
machines_cache = machines_cache[:nodes]
@staticmethod
@@ -584,7 +584,7 @@ class AbstractEtcd(AbstractDCS):
ttl = int(ttl)
ret = self._ttl != ttl
self._ttl = ttl
self._client.set_machines_cache_ttl(ttl*10)
self._client.set_machines_cache_ttl(ttl * 10)
return ret
@property
patroni/dcs/etcd3.py
+5 -5
View File
@@ -14,7 +14,7 @@ from threading import Condition, Lock, Thread
from urllib3.exceptions import ReadTimeoutError, ProtocolError
from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors
from ..exceptions import DCSError, PatroniException
from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT
@@ -411,15 +411,15 @@ class KVCache(Thread):
new_value = kv.get('value')
value_changed = old_value != new_value and \
(key == self._leader_key or key in (self._optime_key, self._status_key) and new_value is not None or
key == self._config_key and old_value is not None and new_value is not None)
(key == self._leader_key or key in (self._optime_key, self._status_key) and new_value is not None
or key == self._config_key and old_value is not None and new_value is not None)
if value_changed:
logger.debug('%s changed from %s to %s', key, old_value, new_value)
# We also want to wake up HA loop on replicas if leader optime (or status key) was updated
if value_changed and (key not in (self._optime_key, self._status_key) or
(self.get(self._leader_key) or {}).get('value') != self._name):
if value_changed and (key not in (self._optime_key, self._status_key)
or (self.get(self._leader_key) or {}).get('value') != self._name):
self._dcs.event.set()
def _process_message(self, message):
patroni/dcs/kubernetes.py
+10 -9
View File
@@ -20,10 +20,10 @@ from typing import Any, Dict, List, Optional
from urllib3.exceptions import HTTPError
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re
TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re
from ..exceptions import DCSError
from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\
Retry, RetryFailedError, tzutc, uri, USER_AGENT
Retry, RetryFailedError, tzutc, uri, USER_AGENT
logger = logging.getLogger(__name__)
@@ -143,7 +143,7 @@ class K8sConfig(object):
if user.get('token'):
self._make_headers(token=user['token'])
elif 'username' in user and 'password' in user:
self._headers = self._make_headers(basic_auth=':'.join((user['username'], user['password'])))
self._headers = self._make_headers(basic_auth=':'.join((user['username'], user['password'])))
@property
def server(self):
@@ -175,7 +175,7 @@ class K8sObject(object):
if isinstance(value, dict):
# we know that `annotations` and `labels` are dicts and therefore don't want to convert them into K8sObject
return value if parent in {'annotations', 'labels'} and \
all(isinstance(v, str) for v in value.values()) else cls(value)
all(isinstance(v, str) for v in value.values()) else cls(value)
elif isinstance(value, list):
return [cls._wrap(None, v) for v in value]
else:
@@ -264,7 +264,7 @@ class K8sClient(object):
def _get_api_servers(self, api_servers_cache):
_, per_node_timeout, per_node_retries = self._calculate_timeouts(len(api_servers_cache))
kwargs = {'headers': self._make_headers({}), 'preload_content': True, 'retries': per_node_retries,
'timeout': urllib3.Timeout(connect=max(1, per_node_timeout/2.0), total=per_node_timeout)}
'timeout': urllib3.Timeout(connect=max(1, per_node_timeout / 2.0), total=per_node_timeout)}
path = self._API_URL_PREFIX + 'default/endpoints/kubernetes'
for base_uri in api_servers_cache:
try:
@@ -382,7 +382,7 @@ class K8sClient(object):
retries = 0
else:
_, timeout, retries = self._calculate_timeouts(api_servers)
timeout = urllib3.Timeout(connect=max(1, timeout/2.0), total=timeout)
timeout = urllib3.Timeout(connect=max(1, timeout / 2.0), total=timeout)
kwargs.update(retries=retries, timeout=timeout)
while True:
@@ -405,7 +405,8 @@ class K8sClient(object):
retry.sleep_func(sleeptime)
retry.update_delay()
# We still have some time left. Partially reduce `api_servers_cache` and retry request
kwargs.update(timeout=urllib3.Timeout(connect=max(1, timeout/2.0), total=timeout), retries=retries)
kwargs.update(timeout=urllib3.Timeout(connect=max(1, timeout / 2.0), total=timeout),
retries=retries)
api_servers_cache = api_servers_cache[:nodes]
def call_api(self, method, path, headers=None, body=None, _retry=None,
@@ -503,7 +504,7 @@ class CoreV1ApiProxy(object):
# to start worrying (send keepalive messages). Finally, the connection should be
# considered as dead if we received nothing from the socket after the ttl seconds.
self._api_client.pool_manager.connection_pool_kw['socket_options'] = \
list(keepalive_socket_options(ttl, int(loop_wait + retry_timeout)))
list(keepalive_socket_options(ttl, int(loop_wait + retry_timeout)))
self._api_client.set_read_timeout(retry_timeout)
self._api_client.set_api_servers_cache_ttl(loop_wait)
@@ -905,7 +906,7 @@ class Kubernetes(AbstractDCS):
# get synchronization state
sync = nodes.get(path + self._SYNC)
metadata = sync and sync.metadata
sync = SyncState.from_node(metadata and metadata.resource_version, metadata and metadata.annotations)
sync = SyncState.from_node(metadata and metadata.resource_version, metadata and metadata.annotations)
return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe)
patroni/dcs/raft.py
+3 -3
View File
@@ -47,7 +47,7 @@ class SyncObjUtility(object):
def __init__(self, otherNodes, conf, retry_timeout=10):
self._nodes = otherNodes
self._utility = TcpUtility(conf.password, retry_timeout/max(1, len(otherNodes)))
self._utility = TcpUtility(conf.password, retry_timeout / max(1, len(otherNodes)))
def executeCommand(self, command):
try:
@@ -83,8 +83,8 @@ class DynMemberSyncObj(SyncObj):
def getMembers(self, args, callback):
callback([{'addr': node.id, 'leader': node == self._getLeader(), 'status': CONNECTION_STATE.CONNECTED
if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] +
[{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None)
if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes]
+ [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None)
def _onTick(self, timeToWait=0.0):
super(DynMemberSyncObj, self)._onTick(timeToWait)
patroni/dcs/zookeeper.py
+11 -10
View File
@@ -28,7 +28,7 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler):
self.set_connect_timeout(connect_timeout)
def set_connect_timeout(self, connect_timeout):
self._connect_timeout = max(1.0, connect_timeout/2.0) # try to connect to zookeeper node during loop_wait/2
self._connect_timeout = max(1.0, connect_timeout / 2.0) # try to connect to zookeeper node during loop_wait/2
def create_connection(self, *args, **kwargs):
"""This method is trying to establish connection with one of the zookeeper nodes.
@@ -43,11 +43,11 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler):
args = list(args)
if len(args) == 0: # kazoo 2.6.0 slightly changed the way how it calls create_connection method
kwargs['timeout'] = max(self._connect_timeout, kwargs.get('timeout', self._connect_timeout*10)/10.0)
kwargs['timeout'] = max(self._connect_timeout, kwargs.get('timeout', self._connect_timeout * 10) / 10.0)
elif len(args) == 1:
args.append(self._connect_timeout)
else:
args[1] = max(self._connect_timeout, args[1]/10.0)
args[1] = max(self._connect_timeout, args[1] / 10.0)
return super(PatroniSequentialThreadingHandler, self).create_connection(*args, **kwargs)
def select(self, *args, **kwargs):
@@ -134,7 +134,7 @@ class ZooKeeper(AbstractDCS):
`write_leader_optime()` methods, which also may hang..."""
ret = self._orig_kazoo_connect(*args)
return max(self.loop_wait - 2, 2)*1000, ret[1]
return max(self.loop_wait - 2, 2) * 1000, ret[1]
def session_listener(self, state):
if state in [KazooState.SUSPENDED, KazooState.LOST]:
@@ -385,12 +385,13 @@ class ZooKeeper(AbstractDCS):
member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
member_data = self.__last_member_data or member and member.data
# We want to notify leader if some important fields in the member key changed by removing ZNode
if member and (self._client.client_id is not None and member.session != self._client.client_id[0] or
not (deep_compare(member_data.get('tags', {}), data.get('tags', {})) and
(member_data.get('state') == data.get('state') or
'running' not in (member_data.get('state'), data.get('state'))) and
member_data.get('version') == data.get('version') and
member_data.get('checkpoint_after_promote') == data.get('checkpoint_after_promote'))):
if member and (self._client.client_id is not None and member.session != self._client.client_id[0]
or not (deep_compare(member_data.get('tags', {}), data.get('tags', {}))
and (member_data.get('state') == data.get('state')
or 'running' not in (member_data.get('state'), data.get('state')))
and member_data.get('version') == data.get('version')
and member_data.get('checkpoint_after_promote')
== data.get('checkpoint_after_promote'))):
try:
self._client.delete_async(self.member_path).get(timeout=1)
except NoNodeError:
patroni/ha.py
+17 -17
View File
@@ -377,14 +377,13 @@ class Ha(object):
return 'failed to acquire initialize lock'
else:
create_replica_methods = self.global_config.get_standby_cluster_config().get('create_replica_methods', []) \
if self.is_standby_cluster() else None
if self.is_standby_cluster() else None
can_bootstrap = self.state_handler.can_create_replica_without_replication_connection(create_replica_methods)
concurrent_bootstrap = self.cluster.initialize == ""
if can_bootstrap and not concurrent_bootstrap:
msg = 'bootstrap (without leader)'
return self._async_executor.try_run_async(msg, self.clone) or 'trying to ' + msg
return 'waiting for {0}leader to bootstrap'.format(
'standby_' if self.is_standby_cluster() else '')
return 'waiting for {0}leader to bootstrap'.format('standby_' if self.is_standby_cluster() else '')
def bootstrap_standby_leader(self):
""" If we found 'standby' key in the configuration, we need to bootstrap
@@ -651,7 +650,7 @@ class Ha(object):
if self.touch_member():
# Primary should notice the updated value during the next cycle. We will wait double that, if primary
# hasn't noticed the value by then not disabling sync replication is not likely to matter.
for _ in polling_loop(timeout=self.dcs.loop_wait*2, interval=2):
for _ in polling_loop(timeout=self.dcs.loop_wait * 2, interval=2):
try:
if not self.is_sync_standby(self.dcs.get_cluster()):
break
@@ -970,7 +969,7 @@ class Ha(object):
if self.state_handler.is_leader():
# in pause leader is the healthiest only when no initialize or sysid matches with initialize!
return not self.is_paused() or not self.cluster.initialize\
or self.state_handler.sysid == self.cluster.initialize
or self.state_handler.sysid == self.cluster.initialize
if self.is_paused():
return False
@@ -1035,10 +1034,11 @@ class Ha(object):
PostgreSQL as quickly as possible without regard for data durability. May only be called synchronously.
"""
mode_control = {
'offline': dict(stop='fast', checkpoint=False, release=False, offline=True, async_req=False),
'graceful': dict(stop='fast', checkpoint=True, release=True, offline=False, async_req=False),
'immediate': dict(stop='immediate', checkpoint=False, release=True, offline=False, async_req=True),
'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async_req=True),
'offline': dict(stop='fast', checkpoint=False, release=False, offline=True, async_req=False), # noqa: E241,E501
'graceful': dict(stop='fast', checkpoint=True, release=True, offline=False, async_req=False), # noqa: E241,E501
'immediate': dict(stop='immediate', checkpoint=False, release=True, offline=False, async_req=True), # noqa: E241,E501
'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async_req=True), # noqa: E241,E501
}[mode]
logger.info('Demoting self (%s)', mode)
@@ -1267,10 +1267,10 @@ class Ha(object):
if self.is_standby_cluster():
return self.follow('cannot be a real primary in a standby cluster',
'no action. I am ({0}), a secondary, and following a standby leader ({1})'.format(
self.state_handler.name, lock_owner), refresh=False)
self.state_handler.name, lock_owner), refresh=False)
return self.follow('demoting self because I do not have the lock and I was a leader',
'no action. I am ({0}), a secondary, and following a leader ({1})'.format(
self.state_handler.name, lock_owner), refresh=False)
self.state_handler.name, lock_owner), refresh=False)
def evaluate_scheduled_restart(self):
if self._async_executor.busy: # Restart already in progress
@@ -1287,8 +1287,8 @@ class Ha(object):
self.delete_future_restart()
return None
if (restart_data and
self.should_run_scheduled_action('restart', restart_data['schedule'], self.delete_future_restart)):
if restart_data\
and self.should_run_scheduled_action('restart', restart_data['schedule'], self.delete_future_restart):
try:
ret, message = self.restart(restart_data, run_async=True)
if not ret:
@@ -1336,8 +1336,8 @@ class Ha(object):
return ret
def future_restart_scheduled(self):
return self.patroni.scheduled_restart.copy() if (self.patroni.scheduled_restart and
isinstance(self.patroni.scheduled_restart, dict)) else None
return self.patroni.scheduled_restart.copy()\
if (self.patroni.scheduled_restart and isinstance(self.patroni.scheduled_restart, dict)) else None
def restart_scheduled(self):
return self._async_executor.scheduled_action == 'restart'
@@ -1637,7 +1637,7 @@ class Ha(object):
if self.has_lock():
self.release_leader_key_voluntarily()
return 'released leader key voluntarily as data dir {0} and currently leader'.format(
'empty' if data_directory_is_accessible else 'not accessible')
'empty' if data_directory_is_accessible else 'not accessible')
if not data_directory_is_accessible:
return 'data directory is not accessible: {0}'.format(data_directory_error)
@@ -1827,7 +1827,7 @@ class Ha(object):
# XXX: what about when Patroni is started as the wrong user that has access to the watchdog device
# but cannot shut down PostgreSQL. Root would be the obvious example. Would be nice to not kill the
# system due to a bad config.
logger.error("PostgreSQL shutdown failed, leader key not removed." +
logger.error("PostgreSQL shutdown failed, leader key not removed.%s",
(" Leaving watchdog running." if self.watchdog.is_running else ""))
def watch(self, timeout):
patroni/postgresql/__init__.py
+18 -18
View File
@@ -93,11 +93,11 @@ class Postgresql(object):
self.cancellable = CancellableSubprocess()
self._sysid = None
self.retry = Retry(max_tries=-1, deadline=config['retry_timeout']/2.0, max_delay=1,
self.retry = Retry(max_tries=-1, deadline=config['retry_timeout'] / 2.0, max_delay=1,
retry_exceptions=PostgresConnectionException)
# Retry 'pg_is_in_recovery()' only once
self._is_leader_retry = Retry(max_tries=1, deadline=config['retry_timeout']/2.0, max_delay=1,
self._is_leader_retry = Retry(max_tries=1, deadline=config['retry_timeout'] / 2.0, max_delay=1,
retry_exceptions=PostgresConnectionException)
self._role_lock = Lock()
@@ -173,21 +173,21 @@ class Postgresql(object):
If some conditions are not satisfied we simply put static values instead. E.g., NULL, 0, '', and so on."""
extra = ", " + (("pg_catalog.current_setting('synchronous_commit'), " +
extra = ", " + (("pg_catalog.current_setting('synchronous_commit'), "
"pg_catalog.current_setting('synchronous_standby_names'), "
"(SELECT pg_catalog.json_agg(r.*) FROM (SELECT w.pid as pid, application_name, sync_state," +
" pg_catalog.pg_{0}_{1}_diff(write_{1}, '0/0')::bigint AS write_lsn," +
" pg_catalog.pg_{0}_{1}_diff(flush_{1}, '0/0')::bigint AS flush_lsn," +
" pg_catalog.pg_{0}_{1}_diff(replay_{1}, '0/0')::bigint AS replay_lsn " +
"FROM pg_catalog.pg_stat_get_wal_senders() w," +
" pg_catalog.pg_stat_get_activity(w.pid)" +
"(SELECT pg_catalog.json_agg(r.*) FROM (SELECT w.pid as pid, application_name, sync_state,"
" pg_catalog.pg_{0}_{1}_diff(write_{1}, '0/0')::bigint AS write_lsn,"
" pg_catalog.pg_{0}_{1}_diff(flush_{1}, '0/0')::bigint AS flush_lsn,"
" pg_catalog.pg_{0}_{1}_diff(replay_{1}, '0/0')::bigint AS replay_lsn "
"FROM pg_catalog.pg_stat_get_wal_senders() w,"
" pg_catalog.pg_stat_get_activity(w.pid)"
" WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name)
if (not self._global_config or self._global_config.is_synchronous_mode)
and self.role in ('master', 'primary', 'promoted') else "'on', '', NULL")
if self._major_version >= 90600:
extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, " +
"plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" +
extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
"plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
" AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
@@ -253,7 +253,7 @@ class Postgresql(object):
def reload_config(self, config, sighup=False):
self.config.reload_config(config, sighup)
self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout']/2.0
self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout'] / 2.0
@property
def pending_restart(self):
@@ -331,8 +331,8 @@ class Postgresql(object):
return deepcopy(self.config.get(method, {}))
def replica_method_can_work_without_replication_connection(self, method):
return method != 'basebackup' and (self.replica_method_options(method).get('no_master') or
self.replica_method_options(method).get('no_leader'))
return method != 'basebackup' and (self.replica_method_options(method).get('no_master')
or self.replica_method_options(method).get('no_leader'))
def can_create_replica_without_replication_connection(self, replica_methods=None):
""" go through the replication methods to see if there are ones
@@ -374,8 +374,8 @@ class Postgresql(object):
# We want to enable hot_standby_feedback if the replica is supposed
# to have a logical slot or in case if it is the cascading replica.
self.set_enforce_hot_standby_feedback(
self._has_permanent_logical_slots or
cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
self._has_permanent_logical_slots
or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
self._global_config = global_config
@@ -971,8 +971,8 @@ 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 ('master', 'primary', 'demoted') or
not {'standby_leader', 'replica'} - {self.role, role})
change_role = self.cb_called and (self.role in ('master', 'primary', 'demoted')
or not {'standby_leader', 'replica'} - {self.role, role})
if change_role:
self.__cb_pending = CallbackAction.NOOP
patroni/postgresql/cancellable.py
+1 -1
View File
@@ -29,7 +29,7 @@ class CancellableExecutor(object):
self._process_cmd = cmd
self._process = psutil.Popen(cmd, *args, **kwargs)
except Exception:
return logger.exception('Failed to execute %s', cmd)
return logger.exception('Failed to execute %s', cmd)
return True
def _kill_process(self):
patroni/postgresql/citus.py
+1 -1
View File
@@ -326,7 +326,7 @@ class CitusHandler(Thread):
task = self.add_task(event['type'], event['group'],
cluster.leader.conn_url,
event['timeout'], event['cooldown']*1000)
event['timeout'], event['cooldown'] * 1000)
if task and event['type'] == 'before_demote':
task.wait()
patroni/postgresql/config.py
+9 -9
View File
@@ -9,11 +9,11 @@ import time
from urllib.parse import urlparse, parse_qsl, unquote
from .validator import CaseInsensitiveDict, recovery_parameters,\
transform_postgresql_parameter_value, transform_recovery_parameter_value
transform_postgresql_parameter_value, transform_recovery_parameter_value
from ..dcs import RemoteMember, slot_name_from_member_name
from ..exceptions import PatroniFatalException
from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, \
validate_directory, is_subpath
validate_directory, is_subpath
logger = logging.getLogger(__name__)
@@ -520,8 +520,8 @@ class ConfigHandler(object):
def build_recovery_params(self, member):
recovery_params = CaseInsensitiveDict({p: v for p, v in self.get('recovery_conf', {}).items()
if not p.lower().startswith('recovery_target') and
p.lower() not in ('primary_conninfo', 'primary_slot_name')})
if not p.lower().startswith('recovery_target')
and p.lower() not in ('primary_conninfo', 'primary_slot_name')})
recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'})
if self._postgresql.major_version >= 120000:
# on pg12 we want to protect from following params being set in one of included files
@@ -864,8 +864,8 @@ class ConfigHandler(object):
self._postgresql.citus_handler.adjust_postgres_gucs(parameters)
ret = CaseInsensitiveDict({k: v for k, v in parameters.items() if not self._postgresql.major_version or
self._postgresql.major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 90100))[2]})
ret = CaseInsensitiveDict({k: v for k, v in parameters.items() if not self._postgresql.major_version
or self._postgresql.major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 90100))[2]})
ret.update({k: os.path.join(self._config_dir, ret[k]) for k in ('hba_file', 'ident_file') if k in ret})
return ret
@@ -924,8 +924,8 @@ class ConfigHandler(object):
def _get_pg_settings(self, names):
return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context, sourcefile'
+ ' FROM pg_catalog.pg_settings ' +
' WHERE pg_catalog.lower(name) = ANY(%s)'),
+ ' FROM pg_catalog.pg_settings '
+ ' WHERE pg_catalog.lower(name) = ANY(%s)'),
[n.lower() for n in names])}
@staticmethod
@@ -1123,7 +1123,7 @@ class ConfigHandler(object):
@property
def rewind_credentials(self):
return self._config['authentication'].get('rewind', self._superuser) \
if self._postgresql.major_version >= 110000 else self._superuser
if self._postgresql.major_version >= 110000 else self._superuser
@property
def ident_file(self):
patroni/postgresql/rewind.py
+8 -7
View File
@@ -78,9 +78,9 @@ class Rewind(object):
def check_leader_has_run_checkpoint(conn_kwargs):
try:
with get_connection_cursor(connect_timeout=3, options='-c statement_timeout=2000', **conn_kwargs) as cur:
cur.execute("SELECT NOT pg_catalog.pg_is_in_recovery()" +
" AND ('x' || pg_catalog.substr(pg_catalog.pg_walfile_name(" +
" pg_catalog.pg_current_wal_lsn()), 1, 8))::bit(32)::int = timeline_id" +
cur.execute("SELECT NOT pg_catalog.pg_is_in_recovery()"
" AND ('x' || pg_catalog.substr(pg_catalog.pg_walfile_name("
" pg_catalog.pg_current_wal_lsn()), 1, 8))::bit(32)::int = timeline_id"
" FROM pg_catalog.pg_control_checkpoint()")
if not cur.fetchone()[0]:
return 'leader has not run a checkpoint yet'
@@ -384,9 +384,10 @@ class Rewind(object):
if self._postgresql.major_version < 120000 else self._postgresql.get_guc_value('restore_command')
# Until v15 pg_rewind expected postgresql.conf to be inside $PGDATA, which is not the case on e.g. Debian
pg_rewind_can_restore = restore_command and (self._postgresql.major_version >= 150000 or
(self._postgresql.major_version >= 130000 and
self._postgresql.config._config_dir == self._postgresql.data_dir))
pg_rewind_can_restore = restore_command and (self._postgresql.major_version >= 150000
or (self._postgresql.major_version >= 130000
and self._postgresql.config._config_dir
== self._postgresql.data_dir))
cmd = [self._postgresql.pgcommand('pg_rewind')]
if pg_rewind_can_restore:
@@ -439,7 +440,7 @@ class Rewind(object):
# superuser credentials match rewind_credentials if the latter are not provided or we run 10 or older
if self._postgresql.config.superuser == self._postgresql.config.rewind_credentials:
leader_status = self._postgresql.checkpoint(
self._conn_kwargs(leader, self._postgresql.config.superuser))
self._conn_kwargs(leader, self._postgresql.config.superuser))
else: # we run 11+ and have a dedicated pg_rewind user
leader_status = self.check_leader_has_run_checkpoint(r)
if leader_status: # we tried to run/check for a checkpoint on the remote leader, but it failed
patroni/postgresql/slots.py
+12 -12
View File
@@ -14,8 +14,8 @@ logger = logging.getLogger(__name__)
def compare_slots(s1, s2, dbid='database'):
return s1['type'] == s2['type'] and (s1['type'] == 'physical' or
s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
return s1['type'] == s2['type'] and (s1['type'] == 'physical'
or s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
class SlotsAdvanceThread(Thread):
@@ -180,11 +180,11 @@ class SlotsHandler(object):
def drop_replication_slot(self, name):
"""Returns a tuple(active, dropped)"""
cursor = self._query(('WITH slots AS (SELECT slot_name, active' +
' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),' +
' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),' +
' true AS dropped FROM slots WHERE not active) ' +
'SELECT active, COALESCE(dropped, false) FROM slots' +
cursor = self._query(('WITH slots AS (SELECT slot_name, active'
' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),'
' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),'
' true AS dropped FROM slots WHERE not active) '
'SELECT active, COALESCE(dropped, false) FROM slots'
' FULL OUTER JOIN dropped ON true'), name)
return cursor.fetchone() if cursor.rowcount == 1 else (False, False)
@@ -216,8 +216,8 @@ class SlotsHandler(object):
for name, value in slots.items():
if name not in self._replication_slots and value['type'] == 'physical':
try:
self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" +
" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})"
" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots"
" WHERE slot_type = 'physical' AND slot_name = %s)").format(
immediately_reserve), name, name)
except Exception:
@@ -247,8 +247,8 @@ class SlotsHandler(object):
with self.get_local_connection_cursor(dbname=database) as cur:
for name, value in values.items():
try:
cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" +
" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" +
cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)"
" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots"
" WHERE slot_type = 'logical' AND slot_name = %s)",
(name, value['plugin'], name))
except Exception as e:
@@ -378,7 +378,7 @@ class SlotsHandler(object):
if compare_slots(slot, slots[r[0]]):
create_slots[r[0]] = slot
else:
logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: ' +
logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: '
'configuration=%s, slot on the primary=%s', r[0], slots[r[0]], slot)
except Exception as e:
logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
patroni/postgresql/sync.py
+2 -2
View File
@@ -243,8 +243,8 @@ class SyncHandler(object):
else:
sync_param = next(iter(value), None)
if not (self._postgresql.config.set_synchronous_standby_names(sync_param) and
self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']:
if not (self._postgresql.config.set_synchronous_standby_names(sync_param)
and self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']:
return
time.sleep(0.1) # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms
patroni/validator.py
+83 -83
View File
@@ -666,7 +666,7 @@ def _get_type_name(python_type: Any) -> str:
"""
return {str: 'a string', int: 'an integer', float: 'a number',
bool: 'a boolean', list: 'an array', dict: 'a dictionary'}.get(
python_type, getattr(python_type, __name__, "unknown type"))
python_type, getattr(python_type, __name__, "unknown type"))
def assert_(condition: bool, message: OptionalType[str] = "Wrong value") -> None:
@@ -699,89 +699,89 @@ validate_etcd = {
}
schema = Schema({
"name": str,
"scope": str,
"restapi": {
"listen": validate_host_port_listen,
"connect_address": validate_connect_address
},
Optional("bootstrap"): {
"dcs": {
Optional("ttl"): int,
Optional("loop_wait"): int,
Optional("retry_timeout"): int,
Optional("maximum_lag_on_failover"): int
"name": str,
"scope": str,
"restapi": {
"listen": validate_host_port_listen,
"connect_address": validate_connect_address
},
Optional("bootstrap"): {
"dcs": {
Optional("ttl"): int,
Optional("loop_wait"): int,
Optional("retry_timeout"): int,
Optional("maximum_lag_on_failover"): int
},
"pg_hba": [str],
"initdb": [Or(str, dict)]
},
Or(*available_dcs): Case({
"consul": {
Or("host", "url"): Case({
"host": validate_host_port,
"url": str})
},
"etcd": validate_etcd,
"etcd3": validate_etcd,
"exhibitor": {
"hosts": [str],
"port": lambda i: assert_(int(i) <= 65535),
Optional("pool_interval"): int
},
"raft": {
"self_addr": validate_connect_address,
Optional("bind_addr"): validate_host_port_listen,
"partner_addrs": validate_host_port_list,
Optional("data_dir"): str,
Optional("password"): str
},
"zookeeper": {
"hosts": Or(comma_separated_host_port, [validate_host_port]),
},
"kubernetes": {
"labels": {},
Optional("namespace"): str,
Optional("scope_label"): str,
Optional("role_label"): str,
Optional("use_endpoints"): bool,
Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address),
Optional("ports"): [{"name": str, "port": int}],
Optional("retriable_http_codes"): Or(int, [int]),
},
}),
Optional("citus"): {
"database": str,
"group": int
},
"postgresql": {
"listen": validate_host_port_listen_multiple_hosts,
"connect_address": validate_connect_address,
Optional("proxy_address"): validate_connect_address,
"authentication": {
"replication": userattributes,
"superuser": userattributes,
"rewind": userattributes
"pg_hba": [str],
"initdb": [Or(str, dict)]
},
"data_dir": validate_data_dir,
Optional("bin_dir"): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup",
"postgres", "pg_isready"]),
Optional("parameters"): {
Optional("unix_socket_directories"): lambda s: assert_(all([isinstance(s, str), len(s)]))
Or(*available_dcs): Case({
"consul": {
Or("host", "url"): Case({
"host": validate_host_port,
"url": str})
},
"etcd": validate_etcd,
"etcd3": validate_etcd,
"exhibitor": {
"hosts": [str],
"port": lambda i: assert_(int(i) <= 65535),
Optional("pool_interval"): int
},
"raft": {
"self_addr": validate_connect_address,
Optional("bind_addr"): validate_host_port_listen,
"partner_addrs": validate_host_port_list,
Optional("data_dir"): str,
Optional("password"): str
},
"zookeeper": {
"hosts": Or(comma_separated_host_port, [validate_host_port]),
},
"kubernetes": {
"labels": {},
Optional("namespace"): str,
Optional("scope_label"): str,
Optional("role_label"): str,
Optional("use_endpoints"): bool,
Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address),
Optional("ports"): [{"name": str, "port": int}],
Optional("retriable_http_codes"): Or(int, [int]),
},
}),
Optional("citus"): {
"database": str,
"group": int
},
Optional("pg_hba"): [str],
Optional("pg_ident"): [str],
Optional("pg_ctl_timeout"): int,
Optional("use_pg_rewind"): bool
},
Optional("watchdog"): {
Optional("mode"): lambda m: assert_(m in ["off", "automatic", "required"]),
Optional("device"): str
},
Optional("tags"): {
Optional("nofailover"): bool,
Optional("clonefrom"): bool,
Optional("noloadbalance"): bool,
Optional("replicatefrom"): str,
Optional("nosync"): bool
}
"postgresql": {
"listen": validate_host_port_listen_multiple_hosts,
"connect_address": validate_connect_address,
Optional("proxy_address"): validate_connect_address,
"authentication": {
"replication": userattributes,
"superuser": userattributes,
"rewind": userattributes
},
"data_dir": validate_data_dir,
Optional("bin_dir"): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup",
"postgres", "pg_isready"]),
Optional("parameters"): {
Optional("unix_socket_directories"): lambda s: assert_(all([isinstance(s, str), len(s)]))
},
Optional("pg_hba"): [str],
Optional("pg_ident"): [str],
Optional("pg_ctl_timeout"): int,
Optional("use_pg_rewind"): bool
},
Optional("watchdog"): {
Optional("mode"): lambda m: assert_(m in ["off", "automatic", "required"]),
Optional("device"): str
},
Optional("tags"): {
Optional("nofailover"): bool,
Optional("clonefrom"): bool,
Optional("noloadbalance"): bool,
Optional("replicatefrom"): str,
Optional("nosync"): bool
}
})
tests/test_api.py
+2 -2
View File
@@ -398,8 +398,8 @@ class TestRestApiHandler(unittest.TestCase):
with patch.object(MockHa, 'is_failsafe_mode', Mock(return_value=False), create=True):
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization))
with patch.object(MockHa, 'is_failsafe_mode', Mock(return_value=True), create=True):
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization +
'\nContent-Length: 9\n\n{"a":"b"}'))
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization
+ '\nContent-Length: 9\n\n{"a":"b"}'))
@patch.object(MockPatroni, 'sighup_handler', Mock())
def test_do_POST_reload(self):
tests/test_bootstrap.py
+1 -1
View File
@@ -132,7 +132,7 @@ class TestBootstrap(BaseTestPostgresql):
@patch.object(CancellableSubprocess, 'call')
@patch.object(Postgresql, 'get_major_version', Mock(return_value=90600))
@patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'in production'}))
@patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'in production'}))
def test_custom_bootstrap(self, mock_cancellable_subprocess_call):
self.p.config._config.pop('pg_hba')
config = {'method': 'foo', 'foo': {'command': 'bar'}}
tests/test_consul.py
+5 -5
View File
@@ -4,7 +4,7 @@ import unittest
from consul import ConsulException, NotFound
from mock import Mock, PropertyMock, patch
from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \
ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError
ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError
from . import SleepException
@@ -26,12 +26,12 @@ def kv_get(self, key, **kwargs):
'ModifyIndex': 2621, 'Session': 'fd4f44fe-2cac-bba5-a60b-304b51ff39b7', 'Value': b'postgresql1'},
{'CreateIndex': 6156, 'Flags': 0, 'Key': key + 'members/postgresql0', 'LockIndex': 1,
'ModifyIndex': 6156, 'Session': '782e6da4-ed02-3aef-7963-99a90ed94b53',
'Value': ('postgres://replicator:[email protected]:5432/postgres' +
'?application_name=http://127.0.0.1:8008/patroni').encode('utf-8')},
'Value': ('postgres://replicator:[email protected]:5432/postgres'
+ '?application_name=http://127.0.0.1:8008/patroni').encode('utf-8')},
{'CreateIndex': 2630, 'Flags': 0, 'Key': key + 'members/postgresql1', 'LockIndex': 1,
'ModifyIndex': 2630, 'Session': 'fd4f44fe-2cac-bba5-a60b-304b51ff39b7',
'Value': ('postgres://replicator:[email protected]:5433/postgres' +
'?application_name=http://127.0.0.1:8009/patroni').encode('utf-8')},
'Value': ('postgres://replicator:[email protected]:5433/postgres'
+ '?application_name=http://127.0.0.1:8009/patroni').encode('utf-8')},
{'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'optime/leader', 'LockIndex': 0,
'ModifyIndex': 6429, 'Value': b'4496294792'},
{'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'sync', 'LockIndex': 0,
tests/test_ctl.py
+1 -1
View File
@@ -470,7 +470,7 @@ class TestCtl(unittest.TestCase):
scheduled_at = datetime.now(tzutc) + timedelta(seconds=600)
mock_get_dcs.return_value.get_cluster = Mock(
return_value=get_cluster_initialized_with_leader(Failover(1, 'a', 'b', scheduled_at)))
return_value=get_cluster_initialized_with_leader(Failover(1, 'a', 'b', scheduled_at)))
result = self.runner.invoke(ctl, ['flush', 'dummy', 'switchover'])
assert result.output.startswith('Success: ')
tests/test_etcd.py
+4 -4
View File
@@ -61,13 +61,13 @@ def etcd_read(self, key, **kwargs):
"modifiedIndex": 1582, "createdIndex": 1582},
{"key": "/service/batman5/members", "dir": True, "nodes": [
{"key": "/service/batman5/members/postgresql1",
"value": "postgres://replicator:[email protected]:5434/postgres" +
"?application_name=http://127.0.0.1:8009/patroni",
"value": "postgres://replicator:[email protected]:5434/postgres"
+ "?application_name=http://127.0.0.1:8009/patroni",
"expiration": "2015-05-15T09:10:59.949384522Z", "ttl": 21,
"modifiedIndex": 20727, "createdIndex": 20727},
{"key": "/service/batman5/members/postgresql0",
"value": "postgres://replicator:[email protected]:5433/postgres" +
"?application_name=http://127.0.0.1:8008/patroni",
"value": "postgres://replicator:[email protected]:5433/postgres"
+ "?application_name=http://127.0.0.1:8008/patroni",
"expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30,
"modifiedIndex": 20730, "createdIndex": 20730}],
"modifiedIndex": 1581, "createdIndex": 1581},
tests/test_etcd3.py
+1 -1
View File
@@ -6,7 +6,7 @@ import urllib3
from mock import Mock, patch
from patroni.dcs.etcd import DnsCachingResolver
from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\
InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3
InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3
from threading import Thread
from . import SleepException, MockResponse
tests/test_kubernetes.py
+2 -2
View File
@@ -8,8 +8,8 @@ import unittest
from mock import Mock, PropertyMock, mock_open, patch
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
K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\
Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
from threading import Thread
from . import MockResponse, SleepException
tests/test_postgresql.py
+16 -16
View File
@@ -357,27 +357,27 @@ class TestPostgresql(BaseTestPostgresql):
self.assertEqual(self.p.latest_checkpoint_location(), 28163096)
# 9.3 and 9.4 format
mock_popen.return_value.communicate.side_effect = [
(b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' +
b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' +
b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None),
(b'rmgr: Transaction len (rec/tot): 64/ 96, tx: 726, lsn: 0/01ADBBB8, prev 0/01ADBB70, ' +
b'bkp: 0000, desc: commit: 2021-02-26 11:19:37.900918 CET; inval msgs: catcache 11 catcache 10', None)]
(b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, '
+ b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi'
+ b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None),
(b'rmgr: Transaction len (rec/tot): 64/ 96, tx: 726, lsn: 0/01ADBBB8, prev 0/01ADBB70, '
+ b'bkp: 0000, desc: commit: 2021-02-26 11:19:37.900918 CET; inval msgs: catcache 11 catcache 10', None)]
self.assertEqual(self.p.latest_checkpoint_location(), 28163096)
mock_popen.return_value.communicate.side_effect = [
(b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' +
b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' +
b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None),
(b'rmgr: XLOG len (rec/tot): 0/ 32, tx: 0, lsn: 0/01ADBBB8, prev 0/01ADBBA0, ' +
b'bkp: 0000, desc: xlog switch ', None)]
(b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, '
+ b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi'
+ b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None),
(b'rmgr: XLOG len (rec/tot): 0/ 32, tx: 0, lsn: 0/01ADBBB8, prev 0/01ADBBA0, '
+ b'bkp: 0000, desc: xlog switch ', None)]
self.assertEqual(self.p.latest_checkpoint_location(), 28163000)
# 9.5+ format
mock_popen.return_value.communicate.side_effect = [
(b'rmgr: XLOG len (rec/tot): 114/ 114, tx: 0, lsn: 0/01ADBC18, prev 0/018260F8, ' +
b'desc: CHECKPOINT_SHUTDOWN redo 0/1825ED8; tli 1; prev tli 1; fpw true; xid 0:494; oid 16387; multi 1' +
b'; offset 0; oldest xid 479 in DB 1; oldest multi 1 in DB 1; oldest/newest commit timestamp xid: 0/0;' +
b' oldest running xid 0; shutdown', None),
(b'rmgr: XLOG len (rec/tot): 24/ 24, tx: 0, lsn: 0/018260F8, prev 0/01826080, ' +
b'desc: SWITCH ', None)]
(b'rmgr: XLOG len (rec/tot): 114/ 114, tx: 0, lsn: 0/01ADBC18, prev 0/018260F8, '
+ b'desc: CHECKPOINT_SHUTDOWN redo 0/1825ED8; tli 1; prev tli 1; fpw true; xid 0:494; oid 16387; multi 1'
+ b'; offset 0; oldest xid 479 in DB 1; oldest multi 1 in DB 1; oldest/newest commit timestamp xid: 0/0;'
+ b' oldest running xid 0; shutdown', None),
(b'rmgr: XLOG len (rec/tot): 24/ 24, tx: 0, lsn: 0/018260F8, prev 0/01826080, '
+ b'desc: SWITCH ', None)]
self.assertEqual(self.p.latest_checkpoint_location(), 25321720)
def test_reload(self):
tests/test_raft.py
+1 -1
View File
@@ -5,7 +5,7 @@ import time
from mock import Mock, PropertyMock, patch
from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL,\
Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
from pysyncobj import SyncObjConf, FAIL_REASON
tests/test_rewind.py
+6 -6
View File
@@ -20,8 +20,8 @@ class MockThread(object):
def mock_cancellable_call(*args, **kwargs):
communicate = kwargs.pop('communicate', None)
if isinstance(communicate, dict):
communicate.update(stdout=b'', stderr=b'pg_rewind: error: could not open file ' +
b'"data/postgresql0/pg_xlog/000000010000000000000003": No such file')
communicate.update(stdout=b'', stderr=b'pg_rewind: error: could not open file '
+ b'"data/postgresql0/pg_xlog/000000010000000000000003": No such file')
return 1
@@ -222,7 +222,7 @@ class TestRewind(BaseTestPostgresql):
@patch('patroni.postgresql.rewind.logger.info')
def test_archive_ready_wals(self, mock_logger_info):
with patch('os.listdir', Mock(side_effect=OSError)), \
patch.object(Postgresql, 'get_guc_value', Mock(side_effect=['on', 'command %f'])):
patch.object(Postgresql, 'get_guc_value', Mock(side_effect=['on', 'command %f'])):
self.r._archive_ready_wals()
mock_logger_info.assert_not_called()
@@ -232,19 +232,19 @@ class TestRewind(BaseTestPostgresql):
'on', '',
]
with patch.object(Postgresql, 'get_guc_value', Mock(side_effect=get_guc_value_res)):
for _ in range(len(get_guc_value_res)//2):
for _ in range(len(get_guc_value_res) // 2):
self.r._archive_ready_wals()
mock_logger_info.assert_not_called()
with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])):
# successful archive_command call
with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)):
with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)):
get_guc_value_res = [
'on', 'command %f',
'always', 'command %f',
]
with patch.object(Postgresql, 'get_guc_value', Mock(side_effect=get_guc_value_res)):
for _ in range(len(get_guc_value_res)//2):
for _ in range(len(get_guc_value_res) // 2):
self.r._archive_ready_wals()
mock_logger_info.assert_called_once()
self.assertEqual(('Trying to archive %s: %s',
tests/test_validator.py
+6 -6
View File
@@ -23,7 +23,7 @@ config = {
"loop_wait": 1000,
"retry_timeout": 1000,
"maximum_lag_on_failover": 1000
},
},
"pg_hba": ["string"],
"initdb": ["string", {"key": "value"}]
},
@@ -49,7 +49,7 @@ config = {
"password": "12345"
},
"zookeeper": {
"hosts": "127.0.0.1:3379,127.0.0.1:3380"
"hosts": "127.0.0.1:3379,127.0.0.1:3380"
},
"kubernetes": {
"namespace": "string",
@@ -85,10 +85,10 @@ config = {
"device": "string"
},
"tags": {
"nofailover": False,
"clonefrom": False,
"noloadbalance": False,
"nosync": False
"nofailover": False,
"clonefrom": False,
"noloadbalance": False,
"nosync": False
}
}
tests/test_watchdog.py
+1 -1
View File
@@ -37,7 +37,7 @@ def mock_ioctl(fd, op, arg=None, mutate_flag=False):
sys.stderr.write("Get support\n")
assert (mutate_flag is True)
arg.options = sum(map(linuxwd.WDIOF.get, ['SETTIMEOUT', 'KEEPALIVEPING']))
arg.identity = (ctypes.c_ubyte*32)(*map(ord, 'Mock Watchdog'))
arg.identity = (ctypes.c_ubyte * 32)(*map(ord, 'Mock Watchdog'))
elif op == linuxwd.WDIOC_GETTIMEOUT:
arg.value = dev.timeout
elif op == linuxwd.WDIOC_SETTIMEOUT:
tests/test_zookeeper.py
+1 -1
View File
@@ -8,7 +8,7 @@ from kazoo.protocol.states import KeeperState, ZnodeStat
from kazoo.retry import RetryFailedError
from mock import Mock, PropertyMock, patch
from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\
PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
class MockKazooClient(Mock):
tox.ini
+1
View File
@@ -1,2 +1,3 @@
[flake8]
max-line-length=120
ignore=D401,W503