diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py index 6480f66a..b089bde0 100644 --- a/.github/workflows/install_deps.py +++ b/.github/workflows/install_deps.py @@ -110,7 +110,7 @@ def install_etcd(): def install_postgres(): - version = os.environ.get('PGVERSION', '15.1-1') + version = os.environ.get('PGVERSION', '16.1-1') platform = {'darwin': 'osx', 'win32': 'windows-x64', 'cygwin': 'windows-x64'}[sys.platform] if platform == 'osx': return subprocess.call(['brew', 'install', 'expect', 'postgresql@{0}'.format(version.split('.')[0])]) diff --git a/.github/workflows/mapping.py b/.github/workflows/mapping.py index f75efec4..279438b0 100644 --- a/.github/workflows/mapping.py +++ b/.github/workflows/mapping.py @@ -1 +1 @@ -versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '15'} +versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '14', 'kubernetes': '15'} diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py index 9a078e4f..cece186f 100644 --- a/.github/workflows/run_tests.py +++ b/.github/workflows/run_tests.py @@ -30,7 +30,7 @@ def main(): unbuffer = ['timeout', '900', 'unbuffer'] else: if sys.platform == 'darwin': - version = os.environ.get('PGVERSION', '15.1-1') + version = os.environ.get('PGVERSION', '16.1-1') path = '/usr/local/opt/postgresql@{0}/bin:.'.format(version.split('.')[0]) unbuffer = ['unbuffer'] else: diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index aafe6b56..f1f55e2a 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -85,7 +85,7 @@ jobs: env: DCS: ${{ matrix.dcs }} ETCDVERSION: 3.4.23 - PGVERSION: 15.1-1 # for windows and macos + PGVERSION: 16.1-1 # for windows and macos strategy: fail-fast: false matrix: diff --git a/Dockerfile b/Dockerfile index b74cdf24..d4fddfea 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine ## It has all the necessary components to play/debug with a single node appliance, running etcd -ARG PG_MAJOR=15 +ARG PG_MAJOR=16 ARG COMPRESS=false ARG PGHOME=/home/postgres ARG PGDATA=$PGHOME/data diff --git a/Dockerfile.citus b/Dockerfile.citus index 5f0164b4..6f02215b 100644 --- a/Dockerfile.citus +++ b/Dockerfile.citus @@ -1,6 +1,6 @@ ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine ## It has all the necessary components to play/debug with a single node appliance, running etcd -ARG PG_MAJOR=15 +ARG PG_MAJOR=16 ARG COMPRESS=false ARG PGHOME=/home/postgres ARG PGDATA=$PGHOME/data @@ -40,7 +40,7 @@ RUN set -ex \ echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \ && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \ && apt-get update -y \ - && apt-get -y install postgresql-$PG_MAJOR-citus-11.3; \ + && apt-get -y install postgresql-$PG_MAJOR-citus-12.1; \ fi \ \ # Cleanup all locales but en_US.UTF-8 diff --git a/features/environment.py b/features/environment.py index a0367657..db10c93d 100644 --- a/features/environment.py +++ b/features/environment.py @@ -1073,6 +1073,8 @@ def before_all(context): context.keyfile = os.path.join(context.pctl.output_dir, 'patroni.key') context.certfile = os.path.join(context.pctl.output_dir, 'patroni.crt') try: + if sys.platform == 'darwin' and 'GITHUB_ACTIONS' in os.environ: + raise Exception with open(os.devnull, 'w') as null: ret = subprocess.call(['openssl', 'req', '-nodes', '-new', '-x509', '-subj', '/CN=batman.patroni', '-addext', 'subjectAltName=IP:127.0.0.1', '-keyout', context.keyfile, diff --git a/kubernetes/Dockerfile b/kubernetes/Dockerfile index 29a683bd..e41bf1cd 100644 --- a/kubernetes/Dockerfile +++ b/kubernetes/Dockerfile @@ -1,4 +1,4 @@ -FROM postgres:15 +FROM postgres:16 LABEL maintainer="Alexander Kukushkin " RUN export DEBIAN_FRONTEND=noninteractive \ diff --git a/kubernetes/Dockerfile.citus b/kubernetes/Dockerfile.citus index f9564521..7af9e5ae 100644 --- a/kubernetes/Dockerfile.citus +++ b/kubernetes/Dockerfile.citus @@ -1,4 +1,4 @@ -FROM postgres:15 +FROM postgres:16 LABEL maintainer="Alexander Kukushkin " RUN export DEBIAN_FRONTEND=noninteractive \ @@ -11,7 +11,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \ ## Make sure we have a en_US.UTF-8 locale available && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \ && if [ $(dpkg --print-architecture) = 'arm64' ]; then \ - apt-get install -y postgresql-server-dev-15 \ + apt-get install -y postgresql-server-dev-16 \ gcc make autoconf \ libc6-dev flex libcurl4-gnutls-dev \ libicu-dev libkrb5-dev liblz4-dev \ @@ -24,7 +24,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \ echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \ && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \ && apt-get update -y \ - && apt-get -y install postgresql-15-citus-12.0; \ + && apt-get -y install postgresql-16-citus-12.1; \ fi \ && pip3 install --break-system-packages setuptools \ && pip3 install --break-system-packages 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \ @@ -38,7 +38,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \ && chmod 664 /etc/passwd \ # Clean up && apt-get remove -y git python3-pip python3-wheel \ - postgresql-server-dev-15 gcc make autoconf \ + postgresql-server-dev-16 gcc make autoconf \ libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \ libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \ && apt-get autoremove -y \ diff --git a/patroni/__main__.py b/patroni/__main__.py index 229ccfb9..2c253da4 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -68,7 +68,7 @@ class Patroni(AbstractPatroniDaemon, Tags): self.watchdog = Watchdog(self.config) self.load_dynamic_configuration() - self.postgresql = Postgresql(self.config['postgresql']) + self.postgresql = Postgresql(self.config['postgresql'], self.dcs.mpp) self.api = RestApiServer(self, self.config['restapi']) self.ha = Ha(self) diff --git a/patroni/api.py b/patroni/api.py index 5761d359..5c9d6603 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -1153,8 +1153,8 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_POST_citus(self) -> None: """Handle a ``POST`` request to ``/citus`` path. - Call :func:`~patroni.postgresql.CitusHandler.handle_event` to handle the request, then write a response with - HTTP status code ``200``. + Call :func:`~patroni.postgresql.mpp.AbstractMPPHandler.handle_event` to handle the request, + then write a response with HTTP status code ``200``. .. note:: If unable to parse the request body, then the request is silently discarded. diff --git a/patroni/ctl.py b/patroni/ctl.py index 3e981b45..6a5fe0e6 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -51,6 +51,7 @@ from .config import Config from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member from .exceptions import PatroniException from .postgresql.misc import postgres_version_to_int +from .postgresql.mpp import get_mpp from .utils import cluster_as_json, patch_config, polling_loop from .request import PatroniRequest from .version import __version__ @@ -313,7 +314,7 @@ def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: config = load_config(config_file, dcs_url) # backward compatibility for configuration file where ctl section is not defined config.setdefault('ctl', {})['insecure'] = config.get('ctl', {}).get('insecure') or insecure - ctx.obj = {'__config': config} + ctx.obj = {'__config': config, '__mpp': get_mpp(config)} def is_citus_cluster() -> bool: @@ -321,7 +322,7 @@ def is_citus_cluster() -> bool: :returns: ``True`` if configuration has ``citus`` section, otherwise ``False``. """ - return bool(_get_configuration().get('citus')) + return click.get_current_context().obj['__mpp'].is_enabled() def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: @@ -340,12 +341,13 @@ def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: config = _get_configuration() config.update({'scope': scope, 'patronictl': True}) if group is not None: - config['citus'] = {'group': group} + config['citus'] = {'group': group, 'database': 'postgres'} config.setdefault('name', scope) try: dcs = _get_dcs(config) if is_citus_cluster() and group is None: dcs.is_citus_coordinator = lambda: True + click.get_current_context().obj['__mpp'] = dcs.mpp return dcs except PatroniException as e: raise PatroniCtlException(str(e)) @@ -1185,7 +1187,7 @@ def reinit(cluster_name: str, group: Optional[int], member_names: List[str], for def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[int], - leader: Optional[str], candidate: Optional[str], + switchover_leader: Optional[str], candidate: Optional[str], force: bool, scheduled: Optional[str] = None) -> None: """Perform a failover or a switchover operation in the cluster. @@ -1199,7 +1201,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i :param cluster_name: name of the Patroni cluster. :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be prompted for filling it -- unless *force* is ``True``, in which case an exception is raised. - :param leader: name of the current leader member. + :param switchover_leader: name of the leader member passed as switchover option. :param candidate: name of a standby member to be promoted. Nodes that are tagged with ``nofailover`` cannot be used. :param force: perform the failover or switchover without asking for confirmations. :param scheduled: timestamp when the switchover should be scheduled to occur. If ``now`` perform immediately. @@ -1208,10 +1210,11 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i :class:`PatroniCtlException`: if: * Patroni is running on a Citus cluster, but no *group* was specified; or * a switchover was requested by the cluster has no leader; or - * *leader* does not match the current leader of the cluster; or + * *switchover_leader* does not match the current leader of the cluster; or * cluster has no candidates available for the operation; or * no *candidate* is given for a failover operation; or - * *leader* and *candidate* are the same; or + * current leader and *candidate* are the same; or + * *candidate* is tagged as nofailover; or * *candidate* is not a member of the cluster; or * trying to schedule a switchover in a cluster that is in maintenance mode; or * user aborts the operation. @@ -1231,23 +1234,24 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i config = global_config.from_cluster(cluster) + cluster_leader = cluster.leader and cluster.leader.name # leader has to be be defined for switchover only if action == 'switchover': - if cluster.leader is None or not cluster.leader.name: + if not cluster_leader: raise PatroniCtlException('This cluster has no leader') - if leader is None: + if switchover_leader is None: if force: - leader = cluster.leader.name + switchover_leader = cluster_leader else: prompt = 'Standby Leader' if config.is_standby_cluster else 'Primary' - leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name)) + switchover_leader = click.prompt(prompt, type=str, default=cluster_leader) - if cluster.leader.name != leader: - raise PatroniCtlException(f'Member {leader} is not the leader of cluster {cluster_name}') + if cluster_leader != switchover_leader: + raise PatroniCtlException(f'Member {switchover_leader} is not the leader of cluster {cluster_name}') # excluding members with nofailover tag - candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover] + candidate_names = [str(m.name) for m in cluster.members if m.name != cluster_leader and not m.nofailover] # We sort the names for consistent output to the client candidate_names.sort() @@ -1260,10 +1264,10 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i if action == 'failover' and not candidate: raise PatroniCtlException('Failover could be performed only to a specific candidate') - if candidate == leader: - raise PatroniCtlException(action.title() + ' target and source are the same.') - if candidate and candidate not in candidate_names: + if candidate == cluster_leader: + raise PatroniCtlException( + f'Member {candidate} is already the leader of cluster {cluster_name}') raise PatroniCtlException( f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover') @@ -1272,7 +1276,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i config.is_synchronous_mode, not cluster.sync.is_empty, not cluster.sync.matches(candidate, True))): - if click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}'): + if not click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}?'): raise PatroniCtlException('Aborting ' + action) scheduled_at_str = None @@ -1292,7 +1296,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i failover_value = {'candidate': candidate} if action == 'switchover': - failover_value['leader'] = leader + failover_value['leader'] = switchover_leader if scheduled_at_str: failover_value['scheduled_at'] = scheduled_at_str @@ -1300,7 +1304,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i # By now we have established that the leader exists and the candidate exists if not force: - demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else '' + demote_msg = f', demoting current leader {cluster_leader}' if cluster_leader else '' if scheduled_at_str: # only switchover can be scheduled if not click.confirm(f'Are you sure you want to schedule switchover of cluster ' @@ -1334,7 +1338,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i logging.exception(r) logging.warning('Failing over to DCS') click.echo('{0} Could not {1} using Patroni api, falling back to DCS'.format(timestamp(), action)) - dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at) + dcs.manual_failover(switchover_leader, candidate, scheduled_at=scheduled_at) output_members(cluster, cluster_name, group=group) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 28c3734f..a210615b 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -25,10 +25,9 @@ from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from ..config import Config from ..postgresql import Postgresql + from ..postgresql.mpp import AbstractMPP SLOT_ADVANCE_AVAILABLE_VERSION = 110000 -CITUS_COORDINATOR_GROUP_ID = 0 -citus_group_re = re.compile('^(0|[1-9][0-9]*)$') slot_name_re = re.compile('^[a-z0-9_]{1,63}$') logger = logging.getLogger(__name__) @@ -130,10 +129,9 @@ def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS': p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', 'patronictl', 'ttl', 'retry_timeout') if p in config}) - # From citus section we only need "group" parameter, but will propagate everything just in case. - if isinstance(config.get('citus'), dict): - config[name].update(config['citus']) - return dcs_class(config[name]) + + from patroni.postgresql.mpp import get_mpp + return dcs_class(config[name], get_mpp(config)) available_implementations = ', '.join(sorted([n for n, _ in iter_dcs_classes()])) raise PatroniFatalException("Can not find suitable configuration of distributed configuration store\n" @@ -1338,15 +1336,15 @@ class AbstractDCS(abc.ABC): _SYNC = 'sync' _FAILSAFE = 'failsafe' - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: 'AbstractMPP') -> None: """Prepare DCS paths, Citus group ID, initial values for state information and processing dependencies. :ivar config: :class:`dict`, reference to config section of selected DCS. i.e.: ``zookeeper`` for zookeeper, ``etcd`` for etcd, etc... """ + self._mpp = mpp self._name = config['name'] self._base_path = re.sub('/+', '/', '/'.join(['', config.get('namespace', 'service'), config['scope']])) - self._citus_group = str(config['group']) if isinstance(config.get('group'), int) else None self._set_loop_wait(config.get('loop_wait', 10)) self._ctl = bool(config.get('patronictl', False)) @@ -1359,6 +1357,11 @@ class AbstractDCS(abc.ABC): self._last_failsafe: Optional[Dict[str, str]] = {} self.event = Event() + @property + def mpp(self) -> 'AbstractMPP': + """Get the effective underlying MPP, if any has been configured.""" + return self._mpp + def client_path(self, path: str) -> str: """Construct the absolute key name from appropriate parts for the DCS type. @@ -1367,8 +1370,8 @@ class AbstractDCS(abc.ABC): :returns: absolute key name for the current Patroni cluster. """ components = [self._base_path] - if self._citus_group: - components.append(self._citus_group) + if self._mpp.is_enabled(): + components.append(str(self._mpp.group)) components.append(path.lstrip('/')) return '/'.join(components) @@ -1522,9 +1525,9 @@ class AbstractDCS(abc.ABC): def is_citus_coordinator(self) -> bool: """:class:`Cluster` instance has a Citus Coordinator group ID. - :returns: ``True`` if the given node is running as Citus Coordinator (``group=0``). + :returns: ``True`` if the given node is running as the MPP Coordinator. """ - return self._citus_group == str(CITUS_COORDINATOR_GROUP_ID) + return self._mpp.is_coordinator() def get_citus_coordinator(self) -> Optional[Cluster]: """Load the Patroni cluster for the Citus Coordinator. @@ -1532,10 +1535,10 @@ class AbstractDCS(abc.ABC): .. note:: This method is only executed on the worker nodes (``group!=0``) to find the coordinator. - :returns: Select :class:`Cluster` instance associated with the Citus Coordinator group ID. + :returns: Select :class:`Cluster` instance associated with the MPP Coordinator group ID. """ try: - return self.__get_patroni_cluster(f'{self._base_path}/{CITUS_COORDINATOR_GROUP_ID}/') + return self.__get_patroni_cluster(f'{self._base_path}/{self._mpp.coordinator_group_id}/') except Exception as e: logger.error('Failed to load Citus coordinator cluster from %s: %r', self.__class__.__name__, e) return None @@ -1549,7 +1552,7 @@ class AbstractDCS(abc.ABC): groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(groups, dict) - cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty()) + cluster = groups.pop(self._mpp.coordinator_group_id, Cluster.empty()) cluster.workers.update(groups) return cluster diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 27cab778..19d65306 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -16,8 +16,9 @@ from urllib.parse import urlencode, urlparse, quote from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -232,8 +233,8 @@ def service_name_from_scope_name(scope_name: str) -> str: class Consul(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(Consul, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Consul, self).__init__(config, mpp) self._base_path = self._base_path[1:] self._scope = config['scope'] self._session = None @@ -435,7 +436,7 @@ class Consul(AbstractDCS): clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict) for node in results or []: key = node['Key'][len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): node['Value'] = (node['Value'] or b'').decode('utf-8') clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} @@ -666,7 +667,7 @@ class Consul(AbstractDCS): if ret: # We have no other choise, only read after write :( if not retry.ensure_deadline(0.5): return False - _, ret = self.retry(self._client.kv.get, self.sync_path) + _, ret = self.retry(self._client.kv.get, self.sync_path, consistency='consistent') if ret and (ret.get('Value') or b'').decode('utf-8') == value: return ret['ModifyIndex'] return False diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 3be699a6..b9d3e0ab 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -22,8 +22,9 @@ from urllib3 import Timeout from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..request import get as requests_get from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover @@ -470,9 +471,9 @@ class EtcdClient(AbstractEtcdClientWithFailover): class AbstractEtcd(AbstractDCS): - def __init__(self, config: Dict[str, Any], client_cls: Type[AbstractEtcdClientWithFailover], + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP, client_cls: Type[AbstractEtcdClientWithFailover], retry_errors_cls: Union[Type[Exception], Tuple[Type[Exception], ...]]) -> None: - super(AbstractEtcd, self).__init__(config) + super(AbstractEtcd, self).__init__(config, mpp) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, retry_exceptions=retry_errors_cls) self._ttl = int(config.get('ttl') or 30) @@ -645,8 +646,8 @@ def catch_etcd_errors(func: Callable[..., Any]) -> Any: class Etcd(AbstractEtcd): - def __init__(self, config: Dict[str, Any]) -> None: - super(Etcd, self).__init__(config, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Etcd, self).__init__(config, mpp, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) self.__do_not_watch = False @property @@ -726,7 +727,7 @@ class Etcd(AbstractEtcd): clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict) for node in result.leaves: key = node.key[len(result.key):].lstrip('/').split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(result.etcd_index, nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index ea7e52f2..7cc2a115 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -16,9 +16,10 @@ from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union from . import ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, catch_return_false_exception, citus_group_re + TimelineHistory, catch_return_false_exception from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry from ..exceptions import DCSError, PatroniException +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT logger = logging.getLogger(__name__) @@ -671,8 +672,9 @@ class PatroniEtcd3Client(Etcd3Client): class Etcd3(AbstractEtcd): - def __init__(self, config: Dict[str, Any]) -> None: - super(Etcd3, self).__init__(config, PatroniEtcd3Client, (DeadlineExceeded, Unavailable, FailedPrecondition)) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Etcd3, self).__init__(config, mpp, PatroniEtcd3Client, + (DeadlineExceeded, Unavailable, FailedPrecondition)) self.__do_not_watch = False self._lease = None self._last_lease_refresh = 0 @@ -796,7 +798,7 @@ class Etcd3(AbstractEtcd): path = self._base_path + '/' for node in self._client.get_cluster(path): key = node['key'][len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py index 2b06073b..03d23575 100644 --- a/patroni/dcs/exhibitor.py +++ b/patroni/dcs/exhibitor.py @@ -7,6 +7,7 @@ from typing import Any, Callable, Dict, List, Union from . import Cluster from .zookeeper import ZooKeeper +from ..postgresql.mpp import AbstractMPP from ..request import get as requests_get from ..utils import uri @@ -66,10 +67,10 @@ class ExhibitorEnsembleProvider(object): class Exhibitor(ZooKeeper): - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: interval = config.get('poll_interval', 300) self._ensemble_provider = ExhibitorEnsembleProvider(config['hosts'], config['port'], poll_interval=interval) - super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts}) + super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts}, mpp) def _load_cluster( self, path: str, loader: Callable[[str], Union[Cluster, Dict[int, Cluster]]] diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index aee87bd3..343b496c 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -19,9 +19,9 @@ from urllib3.exceptions import HTTPError from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING -from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re +from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \ Retry, RetryFailedError, tzutc, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover @@ -748,7 +748,7 @@ class Kubernetes(AbstractDCS): _CITUS_LABEL = 'citus-group' - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: self._labels = deepcopy(config['labels']) self._labels[config.get('scope_label', 'cluster-name')] = config['scope'] self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items()) @@ -759,9 +759,9 @@ class Kubernetes(AbstractDCS): self._standby_leader_label_value = config.get('standby_leader_label_value', 'master') self._tmp_role_label = config.get('tmp_role_label') self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME - super(Kubernetes, self).__init__({**config, 'namespace': ''}) - if self._citus_group: - self._labels[self._CITUS_LABEL] = self._citus_group + super(Kubernetes, self).__init__({**config, 'namespace': ''}, mpp) + if self._mpp.is_enabled(): + self._labels[self._CITUS_LABEL] = str(self._mpp.group) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, retry_exceptions=KubernetesRetriableException) @@ -944,12 +944,12 @@ class Kubernetes(AbstractDCS): for name, pod in path['pods'].items(): group = pod.metadata.labels.get(self._CITUS_LABEL) - if group and citus_group_re.match(group): + if group and self._mpp.group_re.match(group): clusters[group]['pods'][name] = pod for name, kind in path['nodes'].items(): group = kind.metadata.labels.get(self._CITUS_LABEL) - if group and citus_group_re.match(group): + if group and self._mpp.group_re.match(group): clusters[group]['nodes'][name] = kind return {int(group): self._cluster_from_nodes(group, value['nodes'], value['pods'].values()) for group, value in clusters.items()} @@ -976,12 +976,12 @@ class Kubernetes(AbstractDCS): def _load_cluster( self, path: str, loader: Callable[[Any], Union[Cluster, Dict[int, Cluster]]] ) -> Union[Cluster, Dict[int, Cluster]]: - group = self._citus_group if path == self.client_path('') else None + group = str(self._mpp.group) if self._mpp.is_enabled() and path == self.client_path('') else None return self.__load_cluster(group, loader) def get_citus_coordinator(self) -> Optional[Cluster]: try: - ret = self.__load_cluster(str(CITUS_COORDINATOR_GROUP_ID), self._cluster_loader) + ret = self.__load_cluster(str(self._mpp.coordinator_group_id), self._cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(ret, Cluster) return ret diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index 98c48f44..0528cfb0 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -12,9 +12,9 @@ from pysyncobj.transport import TCPTransport, CONNECTION_STATE from pysyncobj.utility import TcpUtility from typing import Any, Callable, Collection, Dict, List, Optional, Set, Union, TYPE_CHECKING -from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, citus_group_re +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import validate_directory if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -285,8 +285,8 @@ class KVStoreTTL(DynMemberSyncObj): class Raft(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(Raft, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Raft, self).__init__(config, mpp) self._ttl = int(config.get('ttl') or 30) ready_event = threading.Event() @@ -387,7 +387,7 @@ class Raft(AbstractDCS): response = self._sync_obj.get(path, recursive=True) for key, value in (response or {}).items(): key = key[len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = value return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 3704b579..6bf77ae4 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -12,9 +12,9 @@ from kazoo.retry import RetryFailedError from kazoo.security import ACL, make_acl from typing import Any, Callable, Dict, List, Optional, Union, Tuple, TYPE_CHECKING -from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, citus_group_re +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -87,8 +87,8 @@ class PatroniKazooClient(KazooClient): class ZooKeeper(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(ZooKeeper, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(ZooKeeper, self).__init__(config, mpp) hosts: Union[str, List[str]] = config.get('hosts', []) if isinstance(hosts, list): @@ -261,7 +261,7 @@ class ZooKeeper(AbstractDCS): def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: ret: Dict[int, Cluster] = {} for node in self.get_children(path): - if citus_group_re.match(node): + if self._mpp.group_re.match(node): ret[int(node)] = self._cluster_loader(path + node + '/') return ret diff --git a/patroni/ha.py b/patroni/ha.py index ab1bc433..dea78163 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -332,7 +332,7 @@ class Ha(object): if coordinator and coordinator.leader and coordinator.leader.conn_url: try: data = {'type': event, - 'group': self.state_handler.citus_handler.group(), + 'group': self.state_handler.citus_handler.group, 'leader': self.state_handler.name, 'timeout': self.dcs.ttl, 'cooldown': self.patroni.config['retry_timeout']} @@ -847,7 +847,7 @@ class Ha(object): self.state_handler.set_role('master') self.process_sync_replication() self.update_cluster_history() - self.state_handler.citus_handler.sync_pg_dist_node(self.cluster) + self.state_handler.citus_handler.sync_meta_data(self.cluster) return message elif self.state_handler.role in ('master', 'promoted', 'primary'): self.process_sync_replication() @@ -1849,10 +1849,9 @@ class Ha(object): logger.fatal('system ID mismatch, node %s belongs to a different cluster: %s != %s', self.state_handler.name, self.cluster.initialize, data_sysid) sys.exit(1) - elif self.cluster.is_unlocked() and not self.is_paused(): + elif self.cluster.is_unlocked() and not self.is_paused() and not self.state_handler.cb_called: # "bootstrap", but data directory is not empty - if not self.state_handler.cb_called and self.state_handler.is_running() \ - and not self.state_handler.is_primary(): + if self.state_handler.is_running() and not self.state_handler.is_primary(): self._join_aborted = True logger.error('No initialize key in DCS and PostgreSQL is running as replica, aborting start') logger.error('Please first start Patroni on the node running as primary') diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index c87c29fa..f6e3ab54 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -19,8 +19,8 @@ from .callback_executor import CallbackAction, CallbackExecutor from .cancellable import CancellableSubprocess from .config import ConfigHandler, mtime from .connection import ConnectionPool, get_connection_cursor -from .citus import CitusHandler from .misc import parse_history, parse_lsn, postgres_major_version_to_int +from .mpp import AbstractMPP from .postmaster import PostmasterProcess from .slots import SlotsHandler from .sync import SyncHandler @@ -63,7 +63,7 @@ class Postgresql(object): "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, " "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()") - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: self.name: str = config['name'] self.scope: str = config['scope'] self._data_dir: str = config['data_dir'] @@ -80,7 +80,7 @@ class Postgresql(object): self._pending_restart = False self.connection_pool = ConnectionPool() self._connection = self.connection_pool.get('heartbeat') - self.citus_handler = CitusHandler(self, config.get('citus')) + self.citus_handler = mpp.get_handler_impl(self) self.config = ConfigHandler(self, config) self.config.check_directories() diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 271bbdfe..7dfa8934 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -1098,6 +1098,12 @@ class ConfigHandler(object): local_connection_address_changed = True else: logger.info('Changed %s from %s to %s', r[0], r[1], new_value) + elif r[0] in self._server_parameters \ + and not compare_values(r[3], r[2], r[1], self._server_parameters[r[0]]): + # Check if any parameter was set back to the current pg_settings value + # We can use pg_settings value here, as it is proved to be equal to new_value + logger.info('Changed %s from %s to %s', r[0], self._server_parameters[r[0]], r[1]) + conf_changed = True for param, value in changes.items(): if '.' in param: # Check that user-defined-paramters have changed (parameters with period in name) diff --git a/patroni/postgresql/mpp/__init__.py b/patroni/postgresql/mpp/__init__.py new file mode 100644 index 00000000..3494793b --- /dev/null +++ b/patroni/postgresql/mpp/__init__.py @@ -0,0 +1,296 @@ +"""Abstract classes for MPP handler. + +MPP stands for Massively Parallel Processing, and Citus belongs to this architecture. Currently, Citus is the only +supported MPP cluster. However, we may consider adapting other databases such as TimescaleDB, GPDB, etc. into Patroni. +""" +import abc + +from typing import Any, Dict, Iterator, Optional, Union, Tuple, Type, TYPE_CHECKING + +from ...dcs import Cluster +from ...dynamic_loader import iter_classes +from ...exceptions import PatroniException + +if TYPE_CHECKING: # pragma: no cover + from .. import Postgresql + from ...config import Config + + +class AbstractMPP(abc.ABC): + """An abstract class which should be passed to :class:`AbstractDCS`. + + .. note:: + We create :class:`AbstractMPP` and :class:`AbstractMPPHandler` to solve the chicken-egg initialization problem. + When initializing DCS, we dynamically create an object implementing :class:`AbstractMPP`, later this object is + used to instantiate an object implementing :class:`AbstractMPPHandler`. + """ + + group_re: Any # re.Pattern[str] + + def __init__(self, config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`AbstractMPP`. + + :param config: configuration of MPP section. + """ + self._config = config + + def is_enabled(self) -> bool: + """Check if MPP is enabled for a given MPP. + + .. note:: + We just check that the :attr:`_config` object isn't empty and expect + it to be empty only in case of :class:`Null`. + + :returns: ``True`` if MPP is enabled, otherwise ``False``. + """ + return bool(self._config) + + @staticmethod + @abc.abstractmethod + def validate_config(config: Any) -> bool: + """Check whether provided config is good for a given MPP. + + :param config: configuration of MPP section. + + :returns: ``True`` is config passes validation, otherwise ``False``. + """ + + @property + @abc.abstractmethod + def group(self) -> Any: + """The group for a given MPP implementation.""" + + @property + @abc.abstractmethod + def coordinator_group_id(self) -> Any: + """The group id of the coordinator PostgreSQL cluster.""" + + def is_coordinator(self) -> bool: + """Check whether this node is running in the coordinator PostgreSQL cluster. + + :returns: ``True`` if MPP is enabled and the group id of this node + matches with the :attr:`coordinator_group_id`, otherwise ``False``. + """ + return self.is_enabled() and self.group == self.coordinator_group_id + + def is_worker(self) -> bool: + """Check whether this node is running as a MPP worker PostgreSQL cluster. + + :returns: ``True`` if MPP is enabled and this node is known to be not running + as the coordinator PostgreSQL cluster, otherwise ``False``. + """ + return self.is_enabled() and not self.is_coordinator() + + def _get_handler_cls(self) -> Iterator[Type['AbstractMPPHandler']]: + """Find Handler classes inherited from a class type of this object. + + :yields: handler classes for this object. + """ + for cls in self.__class__.__subclasses__(): + if issubclass(cls, AbstractMPPHandler) and cls.__name__.startswith(self.__class__.__name__): + yield cls + + def get_handler_impl(self, postgresql: 'Postgresql') -> 'AbstractMPPHandler': + """Find and instantiate Handler implementation of this object. + + :param postgresql: a reference to :class:`Postgresql` object. + + :raises: + :exc:`PatroniException`: if the Handler class haven't been found. + + :returns: an instantiated class that implements Handler for this object. + """ + for cls in self._get_handler_cls(): + return cls(postgresql, self._config) + raise PatroniException(f'Failed to initialize {self.__class__.__name__}Handler object') + + +class AbstractMPPHandler(AbstractMPP): + """An abstract class which defines interfaces that should be implemented by real handlers.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`AbstractMPPHandler`. + + :param postgresql: a reference to :class:`Postgresql` object. + :param config: configuration of MPP section. + """ + super().__init__(config) + self._postgresql = postgresql + + @abc.abstractmethod + def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None: + """Handle an event sent from a worker node. + + :param cluster: the currently known cluster state from DCS. + :param event: the event to be handled. + """ + + @abc.abstractmethod + def sync_meta_data(self, cluster: Cluster) -> None: + """Sync meta data on the coordinator. + + :param cluster: the currently known cluster state from DCS. + """ + + @abc.abstractmethod + def on_demote(self) -> None: + """On demote handler. + + Is called when the primary was demoted. + """ + + @abc.abstractmethod + def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ + + @abc.abstractmethod + def bootstrap(self) -> None: + """Bootstrap handler. + + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ + + @abc.abstractmethod + def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: + """Adjust GUCs in the current PostgreSQL configuration. + + :param parameters: dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value. + """ + + @abc.abstractmethod + def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: ``True`` if the replication slots should not be removed, otherwise ``False``. + """ + + +class Null(AbstractMPP): + """Dummy implementation of :class:`AbstractMPP`.""" + + def __init__(self) -> None: + """Init method for :class:`Null`.""" + super().__init__({}) + + @staticmethod + def validate_config(config: Any) -> bool: + """Check whether provided config is good for :class:`Null`. + + :returns: always ``True``. + """ + return True + + @property + def group(self) -> None: + """The group for :class:`Null`. + + :returns: always ``None``. + """ + return None + + @property + def coordinator_group_id(self) -> None: + """The group id of the coordinator PostgreSQL cluster. + + :returns: always ``None``. + """ + return None + + +class NullHandler(Null, AbstractMPPHandler): + """Dummy implementation of :class:`AbstractMPPHandler`.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`NullHandler`. + + :param postgresql: a reference to :class:`Postgresql` object. + :param config: configuration of MPP section. + """ + AbstractMPPHandler.__init__(self, postgresql, config) + + def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None: + """Handle an event sent from a worker node. + + :param cluster: the currently known cluster state from DCS. + :param event: the event to be handled. + """ + + def sync_meta_data(self, cluster: Cluster) -> None: + """Sync meta data on the coordinator. + + :param cluster: the currently known cluster state from DCS. + """ + + def on_demote(self) -> None: + """On demote handler. + + Is called when the primary was demoted. + """ + + def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ + + def bootstrap(self) -> None: + """Bootstrap handler. + + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ + + def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: + """Adjust GUCs in the current PostgreSQL configuration. + + :param parameters: dictionary of GUCs, with key as GUC name and corresponding value as current GUC value. + """ + + def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: always ``False``. + """ + return False + + +def iter_mpp_classes( + config: Optional[Union['Config', Dict[str, Any]]] = None +) -> Iterator[Tuple[str, Type[AbstractMPP]]]: + """Attempt to import MPP modules that are present in the given configuration. + + :param config: configuration information with possible MPP names as keys. If given, only attempt to import MPP + modules defined in the configuration. Else, if ``None``, attempt to import any supported MPP module. + + :yields: tuples, each containing the module ``name`` and the imported MPP class object. + """ + yield from iter_classes(__package__, AbstractMPP, config) + + +def get_mpp(config: Union['Config', Dict[str, Any]]) -> AbstractMPP: + """Attempt to load and instantiate a MPP module from known available implementations. + + :param config: object or dictionary with Patroni configuration. + + :returns: The successfully loaded MPP or fallback to :class:`Null`. + """ + for name, mpp_class in iter_mpp_classes(config): + if mpp_class.validate_config(config[name]): + return mpp_class(config[name]) + return Null() diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/mpp/citus.py similarity index 89% rename from patroni/postgresql/citus.py rename to patroni/postgresql/mpp/citus.py index 7499c5e2..1719723a 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/mpp/citus.py @@ -6,12 +6,15 @@ from threading import Condition, Event, Thread from urllib.parse import urlparse from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Set, Tuple, TYPE_CHECKING -from ..dcs import CITUS_COORDINATOR_GROUP_ID, Cluster -from ..psycopg import connect, quote_ident +from . import AbstractMPP, AbstractMPPHandler +from ...dcs import Cluster +from ...psycopg import connect, quote_ident, quote_literal +from ...utils import parse_int if TYPE_CHECKING: # pragma: no cover - from . import Postgresql + from .. import Postgresql +CITUS_COORDINATOR_GROUP_ID = 0 CITUS_SLOT_NAME_RE = re.compile(r'^citus_shard_(move|split)_slot(_[1-9][0-9]*){2,3}$') logger = logging.getLogger(__name__) @@ -351,13 +354,45 @@ class PgDistTask(PgDistGroup): return not self == other -class CitusHandler(Thread): +class Citus(AbstractMPP): - def __init__(self, postgresql: 'Postgresql', config: Optional[Dict[str, Union[str, int]]]) -> None: - super(CitusHandler, self).__init__() + group_re = re.compile('^(0|[1-9][0-9]*)$') + + @staticmethod + def validate_config(config: Union[Any, Dict[str, Union[str, int]]]) -> bool: + """Check whether provided config is good for a given MPP. + + :param config: configuration of ``citus`` MPP section. + + :returns: ``True`` is config passes validation, otherwise ``False``. + """ + return isinstance(config, dict) \ + and isinstance(config.get('database'), str) \ + and parse_int(config.get('group')) is not None + + @property + def group(self) -> int: + """The group of this Citus node.""" + return int(self._config['group']) + + @property + def coordinator_group_id(self) -> int: + """The group id of the Citus coordinator PostgreSQL cluster.""" + return CITUS_COORDINATOR_GROUP_ID + + +class CitusHandler(Citus, AbstractMPPHandler, Thread): + """Define the interfaces for handling an underlying Citus cluster.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """"Initialize a new instance of :class:`CitusHandler`. + + :param postgresql: the Postgres node. + :param config: the ``citus`` MPP config section. + """ + Thread.__init__(self) + AbstractMPPHandler.__init__(self, postgresql, config) self.daemon = True - self._postgresql = postgresql - self._config = config if config: self._connection = postgresql.connection_pool.get( 'citus', {'dbname': config['database'], @@ -369,19 +404,11 @@ class CitusHandler(Thread): self._condition = Condition() # protects _pg_dist_group, _tasks, _in_flight, and _schedule_load_pg_dist_group self.schedule_cache_rebuild() - def is_enabled(self) -> bool: - return isinstance(self._config, dict) - - def group(self) -> Optional[int]: - return int(self._config['group']) if isinstance(self._config, dict) else None - - def is_coordinator(self) -> bool: - return self.is_enabled() and self.group() == CITUS_COORDINATOR_GROUP_ID - - def is_worker(self) -> bool: - return self.is_enabled() and not self.is_coordinator() - def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ with self._condition: self._schedule_load_pg_dist_group = True @@ -428,8 +455,8 @@ class CitusHandler(Thread): self._pg_dist_group = pg_dist_group return True - def sync_pg_dist_node(self, cluster: Cluster) -> None: - """Maintain the `pg_dist_node` from the coordinator leader every heartbeat loop. + def sync_meta_data(self, cluster: Cluster) -> None: + """Maintain the ``pg_dist_node`` from the coordinator leader every heartbeat loop. We can't always rely on REST API calls from worker nodes in order to maintain `pg_dist_node`, therefore at least once per heartbeat @@ -599,16 +626,16 @@ class CitusHandler(Thread): with self._condition: i = self.find_task_by_groupid(task.groupid) - # The `PgDistTask.timeout` == None is an indicator that it was scheduled from the sync_pg_dist_group(). + # The `PgDistNode.timeout` == None is an indicator that it was scheduled from the sync_meta_data(). if task.timeout is None: # We don't want to override the already existing task created from REST API. if i is not None and self._tasks[i].timeout is not None: return False # There is a little race condition with tasks created from REST API - the call made "before" the member - # key is updated in DCS. Therefore it is possible that :func:`sync_pg_dist_group` will try to create a - # task based on the outdated values of "state"/"role". To solve it we introduce an artificial timeout. - # Only when the timeout is reached new tasks could be scheduled from sync_pg_dist_group() + # key is updated in DCS. Therefore it is possible that :func:`sync_meta_data` will try to create a task + # based on the outdated values of "state"/"role". To solve it we introduce an artificial timeout. + # Only when the timeout is reached new tasks could be scheduled from sync_meta_data() if self._in_flight and self._in_flight.groupid == task.groupid and self._in_flight.timeout is not None\ and self._in_flight.deadline > time.time(): return False @@ -667,17 +694,25 @@ class CitusHandler(Thread): task.wait() def bootstrap(self) -> None: - if not isinstance(self._config, dict): # self.is_enabled() - return + """Bootstrap handler. + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ conn_kwargs = {**self._postgresql.connection_pool.conn_kwargs, 'options': '-c synchronous_commit=local -c statement_timeout=0'} if self._config['database'] != self._postgresql.database: conn = connect(**conn_kwargs) try: + database = self._config['database'] + sql = """DO $$ +BEGIN + PERFORM * FROM pg_catalog.pg_database WHERE datname = {0}; + IF NOT FOUND THEN + CREATE DATABASE {1}; + END IF; +END;$$""".format(quote_literal(database), quote_ident(database, conn)) with conn.cursor() as cur: - cur.execute('CREATE DATABASE {0}'.format( - quote_ident(self._config['database'], conn)).encode('utf-8')) + cur.execute(sql.encode('utf-8')) finally: conn.close() @@ -685,7 +720,7 @@ class CitusHandler(Thread): conn = connect(**conn_kwargs) try: with conn.cursor() as cur: - cur.execute('CREATE EXTENSION citus') + cur.execute('CREATE EXTENSION IF NOT EXISTS citus') superuser = self._postgresql.config.superuser params = {k: superuser[k] for k in ('password', 'sslcert', 'sslkey') if k in superuser} @@ -702,9 +737,10 @@ class CitusHandler(Thread): conn.close() def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: - if not self.is_enabled(): - return + """Adjust GUCs in the current PostgreSQL configuration. + :param parameters: dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value. + """ # citus extension must be on the first place in shared_preload_libraries shared_preload_libraries = list(filter( lambda el: el and el != 'citus', @@ -722,8 +758,18 @@ class CitusHandler(Thread): parameters['citus.local_hostname'] = self._postgresql.connection_pool.conn_kwargs.get('host', 'localhost') def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: - if isinstance(self._config, dict) and self._postgresql.is_primary() and\ - slot['type'] == 'logical' and slot['database'] == self._config['database']: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: ``True`` if the replication slots should not be removed, otherwise ``False``. + """ + if self._postgresql.is_primary() and slot['type'] == 'logical' and slot['database'] == self._config['database']: m = CITUS_SLOT_NAME_RE.match(slot['name']) return bool(m and {'move': 'pgoutput', 'split': 'citus'}.get(m.group(1)) == slot['plugin']) return False diff --git a/patroni/postgresql/postmaster.py b/patroni/postgresql/postmaster.py index 4505e7f7..97eb10e4 100644 --- a/patroni/postgresql/postmaster.py +++ b/patroni/postgresql/postmaster.py @@ -176,7 +176,7 @@ class PostmasterProcess(psutil.Process): return not self.is_running() def wait_for_user_backends_to_close(self, stop_timeout: Optional[float]) -> None: - # These regexps are cross checked against versions PostgreSQL 9.1 .. 15 + # These regexps are cross checked against versions PostgreSQL 9.1 .. 16 aux_proc_re = re.compile("(?:postgres:)( .*:)? (?:(?:archiver|startup|autovacuum launcher|autovacuum worker|" "checkpointer|logger|stats collector|wal receiver|wal writer|writer)(?: process )?|" "walreceiver|wal sender process|walsender|walwriter|background writer|" diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index fb9448cd..7f7fd294 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -291,7 +291,7 @@ class SlotsHandler: :param name: name of the slot to ignore :returns: ``True`` if slot *name* matches any slot specified in ``ignore_slots`` configuration, - otherwise will pass through and return result of :meth:`CitusHandler.ignore_replication_slot`. + otherwise will pass through and return result of :meth:`AbstractMPPHandler.ignore_replication_slot`. """ slot = self._replication_slots[name] if cluster.config: diff --git a/patroni/utils.py b/patroni/utils.py index 23f419e5..2b1ded5b 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -400,22 +400,23 @@ def parse_real(value: Any, base_unit: Optional[str] = None) -> Optional[float]: return convert_to_base_unit(val, unit, base_unit) -def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value: Any) -> bool: - """Check if *old_value* and *new_value* are equivalent after parsing them as *vartype*. +def compare_values(vartype: str, unit: Optional[str], settings_value: Any, config_value: Any) -> bool: + """Check if the value from ``pg_settings`` and from Patroni config are equivalent after parsing them as *vartype*. - :param vartpe: the target type to parse *old_value* and *new_value* before comparing them. Accepts any among of the - following (case sensitive): + :param vartype: the target type to parse *settings_value* and *config_value* before comparing them. + Accepts any among of the following (case sensitive): * ``bool``: parse values using :func:`parse_bool`; or * ``integer``: parse values using :func:`parse_int`; or * ``real``: parse values using :func:`parse_real`; or * ``enum``: parse values as lowercase strings; or * ``string``: parse values as strings. This one is used by default if no valid value is passed as *vartype*. - :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real` for *new_value*. - :param old_value: value to be compared with *new_value*. - :param new_value: value to be compared with *old_value*. + :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real` + for *config_value*. + :param settings_value: value to be compared with *config_value*. + :param config_value: value to be compared with *settings_value*. - :returns: ``True`` if *old_value* is equivalent to *new_value* when both are parsed as *vartype*. + :returns: ``True`` if *settings_value* is equivalent to *config_value* when both are parsed as *vartype*. :Example: @@ -455,8 +456,8 @@ def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value: } converter = converters.get(vartype) or converters['string'] - old_converted = converter(old_value, None) - new_converted = converter(new_value, unit) + old_converted = converter(settings_value, None) + new_converted = converter(config_value, unit) return old_converted is not None and new_converted is not None and old_converted == new_converted diff --git a/patroni/validator.py b/patroni/validator.py index d0b168be..bd69cc0e 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -937,6 +937,18 @@ validate_etcd = { schema = Schema({ "name": str, "scope": str, + Optional("log"): { + Optional("level"): EnumValidator(('DEBUG', 'INFO', 'WARN', 'WARNING', 'ERROR', 'FATAL', 'CRITICAL'), + case_sensitive=True, raise_assert=True), + Optional("traceback_level"): EnumValidator(('DEBUG', 'ERROR'), raise_assert=True), + Optional("format"): str, + Optional("dateformat"): str, + Optional("max_queue_size"): int, + Optional("dir"): str, + Optional("file_num"): int, + Optional("file_size"): int, + Optional("loggers"): dict + }, Optional("ctl"): { Optional("insecure"): bool, Optional("cacert"): str, diff --git a/tests/__init__.py b/tests/__init__.py index aca90bee..0537c764 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -12,6 +12,7 @@ import patroni.psycopg as psycopg from patroni.dcs import Leader, Member from patroni.postgresql import Postgresql from patroni.postgresql.config import ConfigHandler +from patroni.postgresql.mpp import get_mpp from patroni.utils import RetryFailedError, tzutc @@ -25,8 +26,41 @@ mock_available_gucs = PropertyMock(return_value={ 'max_wal_senders', 'max_worker_processes', 'port', 'search_path', 'shared_preload_libraries', 'stats_temp_directory', 'synchronous_standby_names', 'track_commit_timestamp', 'unix_socket_directories', 'vacuum_cost_delay', 'vacuum_cost_limit', 'wal_keep_size', 'wal_level', 'wal_log_hints', 'zero_damaged_pages', + 'autovacuum', 'wal_segment_size', 'wal_block_size', 'shared_buffers', 'wal_buffers', }) +GET_PG_SETTINGS_RESULT = [ + ('wal_segment_size', '2048', '8kB', 'integer', 'internal'), + ('wal_block_size', '8192', None, 'integer', 'internal'), + ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'), + ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'), + ('max_connections', '100', None, 'integer', 'postmaster'), + ('max_prepared_transactions', '200', None, 'integer', 'postmaster'), + ('max_worker_processes', '8', None, 'integer', 'postmaster'), + ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'), + ('max_wal_senders', '5', None, 'integer', 'postmaster'), + ('search_path', 'public', None, 'string', 'user'), + ('port', '5432', None, 'integer', 'postmaster'), + ('listen_addresses', '127.0.0.2, 127.0.0.3', None, 'string', 'postmaster'), + ('autovacuum', 'on', None, 'bool', 'sighup'), + ('unix_socket_directories', '/tmp', None, 'string', 'postmaster'), + ('shared_preload_libraries', 'citus', None, 'string', 'postmaster'), + ('wal_keep_size', '128', 'MB', 'integer', 'sighup'), + ('cluster_name', 'batman', None, 'string', 'postmaster'), + ('vacuum_cost_delay', '200', 'ms', 'real', 'user'), + ('vacuum_cost_limit', '-1', None, 'integer', 'user'), + ('max_stack_depth', '2048', 'kB', 'integer', 'superuser'), + ('constraint_exclusion', '', None, 'enum', 'user'), + ('force_parallel_mode', '1', None, 'enum', 'user'), + ('zero_damaged_pages', 'off', None, 'bool', 'superuser'), + ('stats_temp_directory', '/tmp', None, 'string', 'sighup'), + ('track_commit_timestamp', 'off', None, 'bool', 'postmaster'), + ('wal_log_hints', 'on', None, 'bool', 'superuser'), + ('hot_standby', 'on', None, 'bool', 'superuser'), + ('max_replication_slots', '5', None, 'integer', 'superuser'), + ('wal_level', 'logical', None, 'enum', 'superuser'), +] + class MockResponse(object): @@ -133,22 +167,9 @@ class MockCursor(object): ('archive_command', 'my archive command'), ('cluster_name', 'my_cluster')] elif sql.startswith('SELECT name, setting'): - self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'), - ('wal_block_size', '8192', None, 'integer', 'internal'), - ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'), - ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'), - ('max_connections', '100', None, 'integer', 'postmaster'), - ('max_prepared_transactions', '0', None, 'integer', 'postmaster'), - ('max_worker_processes', '8', None, 'integer', 'postmaster'), - ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'), - ('max_wal_senders', '5', None, 'integer', 'postmaster'), - ('search_path', 'public', None, 'string', 'user'), - ('port', '5433', None, 'integer', 'postmaster'), - ('listen_addresses', '*', None, 'string', 'postmaster'), - ('autovacuum', 'on', None, 'bool', 'sighup'), - ('unix_socket_directories', '/tmp', None, 'string', 'postmaster')] + self.results = GET_PG_SETTINGS_RESULT elif sql.startswith('SELECT COUNT(*) FROM pg_catalog.pg_settings'): - self.results = [(1,)] + self.results = [(0,)] elif sql.startswith('IDENTIFY_SYSTEM'): self.results = [('1', 3, '0/402EEC0', '')] elif sql.startswith('TIMELINE_HISTORY '): @@ -222,11 +243,11 @@ class PostgresInit(unittest.TestCase): _PARAMETERS = {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'f.oo': 'bar', 'search_path': 'public', 'hot_standby': 'on', 'max_wal_senders': 5, 'wal_keep_segments': 8, 'wal_log_hints': 'on', 'max_locks_per_transaction': 64, - 'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 0, + 'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 200, 'track_commit_timestamp': 'off', 'unix_socket_directories': '/tmp', - 'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': '', + 'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': 'off', 'force_parallel_mode': '1', 'constraint_exclusion': '', - 'max_stack_depth': 'Z', 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200} + 'max_stack_depth': 2048, 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200} @patch('patroni.psycopg._connect', psycopg_connect) @patch('patroni.postgresql.CallbackExecutor', Mock()) @@ -236,23 +257,24 @@ class PostgresInit(unittest.TestCase): @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='primary')) def setUp(self): data_dir = os.path.join('data', 'test0') - self.p = Postgresql({'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir, - 'config_dir': data_dir, 'retry_timeout': 10, - 'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'), - 'listen': '127.0.0.2, 127.0.0.3:5432', - 'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433', - 'authentication': {'superuser': {'username': 'foo', 'password': 'test'}, - 'replication': {'username': '', 'password': 'rep-pass'}, - 'rewind': {'username': 'rewind', 'password': 'test'}}, - 'remove_data_directory_on_rewind_failure': True, - 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True, - 'parameters': self._PARAMETERS, - 'recovery_conf': {'foo': 'bar'}, - 'pg_hba': ['host all all 0.0.0.0/0 md5'], - 'pg_ident': ['krb realm postgres'], - 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true', - 'on_restart': 'true', 'on_role_change': 'true'}, - 'citus': {'group': 0, 'database': 'citus'}}) + config = {'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir, + 'config_dir': data_dir, 'retry_timeout': 10, + 'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'), + 'listen': '127.0.0.2, 127.0.0.3:5432', + 'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433', + 'authentication': {'superuser': {'username': 'foo', 'password': 'test'}, + 'replication': {'username': '', 'password': 'rep-pass'}, + 'rewind': {'username': 'rewind', 'password': 'test'}}, + 'remove_data_directory_on_rewind_failure': True, + 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True, + 'parameters': self._PARAMETERS, + 'recovery_conf': {'foo': 'bar'}, + 'pg_hba': ['host all all 0.0.0.0/0 md5'], + 'pg_ident': ['krb realm postgres'], + 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true', + 'on_restart': 'true', 'on_role_change': 'true'}, + 'citus': {'group': 0, 'database': 'citus'}} + self.p = Postgresql(config, get_mpp(config)) class BaseTestPostgresql(PostgresInit): diff --git a/tests/test_citus.py b/tests/test_citus.py index 1eddded1..b0e04548 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -4,13 +4,13 @@ import unittest from copy import deepcopy from mock import Mock, patch from typing import List -from patroni.postgresql.citus import CitusHandler, PgDistGroup, PgDistNode +from patroni.postgresql.mpp.citus import CitusHandler, PgDistGroup, PgDistNode from . import BaseTestPostgresql, MockCursor, psycopg_connect, SleepException from .test_ha import get_cluster_initialized_with_leader -@patch('patroni.postgresql.citus.Thread', Mock()) +@patch('patroni.postgresql.mpp.citus.Thread', Mock()) @patch('patroni.psycopg.connect', psycopg_connect) class TestCitus(BaseTestPostgresql): @@ -21,9 +21,9 @@ class TestCitus(BaseTestPostgresql): self.cluster.workers[1] = self.cluster @patch('time.time', Mock(side_effect=[100, 130, 160, 190, 220, 250, 280, 310, 340, 370])) - @patch('patroni.postgresql.citus.logger.exception', Mock(side_effect=SleepException)) - @patch('patroni.postgresql.citus.logger.warning') - @patch('patroni.postgresql.citus.PgDistTask.wait', Mock()) + @patch('patroni.postgresql.mpp.citus.logger.exception', Mock(side_effect=SleepException)) + @patch('patroni.postgresql.mpp.citus.logger.warning') + @patch('patroni.postgresql.mpp.citus.PgDistTask.wait', Mock()) @patch.object(CitusHandler, 'is_alive', Mock(return_value=True)) def test_run(self, mock_logger_warning): # `before_demote` or `before_promote` REST API calls starting a @@ -43,10 +43,10 @@ class TestCitus(BaseTestPostgresql): @patch.object(CitusHandler, 'is_alive', Mock(return_value=False)) @patch.object(CitusHandler, 'start', Mock()) - def test_sync_pg_dist_node(self): + def test_sync_meta_data(self): with patch.object(CitusHandler, 'is_enabled', Mock(return_value=False)): - self.c.sync_pg_dist_node(self.cluster) - self.c.sync_pg_dist_node(self.cluster) + self.c.sync_meta_data(self.cluster) + self.c.sync_meta_data(self.cluster) def test_handle_event(self): self.c.handle_event(self.cluster, {}) @@ -55,18 +55,18 @@ class TestCitus(BaseTestPostgresql): 'leader': 'leader', 'timeout': 30, 'cooldown': 10}) def test_add_task(self): - with patch('patroni.postgresql.citus.logger.error') as mock_logger, \ - patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)): + with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \ + patch('patroni.postgresql.mpp.citus.urlparse', Mock(side_effect=Exception)): self.c.add_task('', 1, self.cluster, '', None) mock_logger.assert_called_once() - with patch('patroni.postgresql.citus.logger.debug') as mock_logger: + with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger: self.c.add_task('before_demote', 1, self.cluster, self.cluster.leader_name, 'postgres://host:5432/postgres', 30) mock_logger.assert_called_once() self.assertTrue(mock_logger.call_args[0][0].startswith('Adding the new task:')) - with patch('patroni.postgresql.citus.logger.debug') as mock_logger: + with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger: self.c.add_task('before_promote', 1, self.cluster, self.cluster.leader_name, 'postgres://host:5432/postgres', 30) mock_logger.assert_called_once() @@ -119,7 +119,7 @@ class TestCitus(BaseTestPostgresql): self.c.process_tasks() self.c.add_task('after_promote', 0, self.cluster, self.cluster.leader_name, 'postgres://host3:5432/postgres') - with patch('patroni.postgresql.citus.logger.error') as mock_logger, \ + with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \ patch.object(CitusHandler, 'query', Mock(side_effect=Exception)): self.c.process_tasks() mock_logger.assert_called_once() @@ -128,7 +128,7 @@ class TestCitus(BaseTestPostgresql): def test_on_demote(self): self.c.on_demote() - @patch('patroni.postgresql.citus.logger.error') + @patch('patroni.postgresql.mpp.citus.logger.error') @patch.object(MockCursor, 'execute', Mock(side_effect=Exception)) def test_load_pg_dist_group(self, mock_logger): # load_pg_dist_group) triggers, query fails and exception is property handled @@ -154,10 +154,6 @@ class TestCitus(BaseTestPostgresql): self.assertEqual(parameters['wal_level'], 'logical') self.assertEqual(parameters['citus.local_hostname'], '/tmp') - def test_bootstrap(self): - self.c._config = None - self.c.bootstrap() - def test_ignore_replication_slot(self): self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'physical', 'database': 'bar', 'plugin': 'wal2json'})) diff --git a/tests/test_consul.py b/tests/test_consul.py index 83ee67d8..494d1126 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -3,8 +3,10 @@ import unittest from consul import ConsulException, NotFound from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \ ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError +from patroni.postgresql.mpp import get_mpp from . import SleepException @@ -91,13 +93,17 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.KV, 'get', kv_get) @patch.object(consul.Consul.KV, 'delete', Mock()) def setUp(self): - Consul({'ttl': 30, 'scope': 't', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10, - 'verify': 'on', 'key': 'foo', 'cert': 'bar', 'cacert': 'buz', 'token': 'asd', 'dc': 'dc1', - 'register_service': True}) - Consul({'ttl': 30, 'scope': 't_', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10, - 'verify': 'on', 'cert': 'bar', 'cacert': 'buz', 'register_service': True}) - self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10, - 'register_service': True, 'service_check_tls_server_name': True}) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't', 'name': 'p', 'retry_timeout': 10, + 'consul': {'url': 'https://l:1', 'verify': 'on', + 'key': 'foo', 'cert': 'bar', 'cacert': 'buz', + 'token': 'asd', 'dc': 'dc1', 'register_service': True}}), Consul) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't_', 'name': 'p', 'retry_timeout': 10, + 'consul': {'url': 'https://l:1', 'verify': 'on', + 'cert': 'bar', 'cacert': 'buz', 'register_service': True}}), Consul) + self.c = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'retry_timeout': 10, + 'consul': {'host': 'localhost:1', 'register_service': True, + 'service_check_tls_server_name': True}}) + self.assertIsInstance(self.c, Consul) self.c._base_path = 'service/good' self.c.get_cluster() @@ -130,7 +136,7 @@ class TestConsul(unittest.TestCase): self.assertIsInstance(self.c.get_cluster(), Cluster) def test__get_citus_cluster(self): - self.c._citus_group = '0' + self.c._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.c.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index a174b03d..f5341f88 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -12,6 +12,7 @@ from patroni.ctl import ctl, load_config, output_members, get_dcs, parse_dcs, \ get_all_members, get_any_member, get_cursor, query_member, PatroniCtlException, apply_config_changes, \ format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH, PatronictlPrettyTable from patroni.dcs import Cluster, Failover +from patroni.postgresql.mpp import get_mpp from patroni.psycopg import OperationalError from patroni.utils import tzutc from prettytable import PrettyTable, ALL @@ -69,7 +70,7 @@ class TestCtl(unittest.TestCase): @patch('patroni.psycopg.connect', psycopg_connect) def test_get_cursor(self): with click.Context(click.Command('query')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), None, {}, role=role)) self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role=role)) @@ -107,7 +108,7 @@ class TestCtl(unittest.TestCase): def test_output_members(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at)) del cluster.members[1].data['conn_url'] @@ -157,7 +158,8 @@ class TestCtl(unittest.TestCase): # Target and source are equal result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny') self.assertEqual(result.exit_code, 1) - self.assertIn('Switchover target and source are the same', result.output) + self.assertIn("Candidate ['other']", result.output) + self.assertIn('Member leader is already the leader of cluster dummy', result.output) # Candidate is not a member of the cluster result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny') @@ -220,29 +222,36 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n') self.assertIn('Failover could be performed only to a specific candidate', result.output) + # Candidate is the same as the leader + result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0'], input='leader\n') + self.assertIn("Candidate ['other']", result.output) + self.assertIn('Member leader is already the leader of cluster dummy', result.output) + # Temp test to check a fallback to switchover if leader is specified with patch('patroni.ctl._do_failover_or_switchover') as failover_func_mock: result = self.runner.invoke(ctl, ['failover', '--leader', 'leader', 'dummy'], input='0\n') self.assertIn('Supplying a leader name using this command is deprecated', result.output) failover_func_mock.assert_called_once_with('switchover', 'dummy', None, 'leader', None, False) - # Failover to an async member in sync mode (confirm) cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) cluster.members.append(Member(0, 'async', 28, {'api_url': 'http://127.0.0.1:8012/patroni'})) cluster.config.data['synchronous_mode'] = True with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=cluster)): + # Failover to an async member in sync mode (confirm) result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny') self.assertIn('Are you sure you want to failover to the asynchronous node async', result.output) + self.assertEqual(result.exit_code, 0) - # Failover to an async member in sync mode (abort) - result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N') - self.assertEqual(result.exit_code, 1) + # Failover to an async member in sync mode (abort) + result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N') + self.assertEqual(result.exit_code, 1) + self.assertIn('Aborting failover', result.output) @patch('patroni.dynamic_loader.iter_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd'])) def test_get_dcs(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {'dummy': {}}} + ctx.obj = {'__config': {'dummy': {}}, '__mpp': get_mpp({})} self.assertRaises(PatroniCtlException, get_dcs, 'dummy', 0) @patch('patroni.psycopg.connect', psycopg_connect) @@ -431,7 +440,7 @@ class TestCtl(unittest.TestCase): def test_get_any_member(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertIsNone(get_any_member(get_cluster_initialized_without_leader(), None, role=role)) @@ -440,7 +449,7 @@ class TestCtl(unittest.TestCase): def test_get_all_members(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertEqual(list(get_all_members(get_cluster_initialized_without_leader(), None, role=role)), []) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 874aac5c..d7a423be 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -5,8 +5,10 @@ import unittest from dns.exception import DNSException from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.etcd import AbstractDCS, EtcdClient, Cluster, Etcd, EtcdError, DnsCachingResolver from patroni.exceptions import DCSError +from patroni.postgresql.mpp import get_mpp from patroni.utils import Retry from urllib3.exceptions import ReadTimeoutError @@ -138,8 +140,9 @@ class TestClient(unittest.TestCase): @patch.object(EtcdClient, '_get_machines_list', Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])) def setUp(self): - self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3, - 'srv': 'test', 'scope': 'test', 'name': 'foo'}) + self.etcd = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3, + 'etcd': {'srv': 'test'}, 'scope': 'test', 'name': 'foo'}) + self.assertIsInstance(self.etcd, Etcd) self.client = self.etcd._client self.client.http.request = http_request self.client.http.request_encode_body = http_request @@ -235,7 +238,7 @@ class TestEtcd(unittest.TestCase): Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])) def setUp(self): self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, - 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}) + 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}, get_mpp({})) def test_base_path(self): self.assertEqual(self.etcd._base_path, '/patroni/test') @@ -270,7 +273,7 @@ class TestEtcd(unittest.TestCase): self.assertRaises(EtcdError, self.etcd.get_cluster) def test__get_citus_cluster(self): - self.etcd._citus_group = '0' + self.etcd._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.etcd.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 10ab1ea5..fcfd4e4b 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -4,10 +4,12 @@ import unittest import urllib3 from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.etcd import DnsCachingResolver from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \ Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \ Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode +from patroni.postgresql.mpp import get_mpp from threading import Thread from . import SleepException, MockResponse @@ -80,9 +82,9 @@ class BaseTestEtcd3(unittest.TestCase): @patch.object(Thread, 'start', Mock()) @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen) def setUp(self): - self.etcd3 = Etcd3({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, - 'host': 'localhost:2378', 'scope': 'test', 'name': 'foo', - 'username': 'etcduser', 'password': 'etcdpassword'}) + self.etcd3 = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, 'name': 'foo', 'scope': 'test', + 'etcd3': {'host': 'localhost:2378', 'username': 'etcduser', 'password': 'etcdpassword'}}) + self.assertIsInstance(self.etcd3, Etcd3) self.client = self.etcd3._client self.kv_cache = self.client._kv_cache @@ -236,7 +238,7 @@ class TestEtcd3(BaseTestEtcd3): self.assertRaises(Etcd3Error, self.etcd3.get_cluster) def test__get_citus_cluster(self): - self.etcd3._citus_group = '0' + self.etcd3._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.etcd3.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py index a908e1fc..5a72eb21 100644 --- a/tests/test_exhibitor.py +++ b/tests/test_exhibitor.py @@ -2,6 +2,7 @@ import unittest import urllib3 from mock import Mock, patch +from patroni.dcs import get_dcs from patroni.dcs.exhibitor import ExhibitorEnsembleProvider, Exhibitor from patroni.dcs.zookeeper import ZooKeeperError @@ -26,8 +27,9 @@ class TestExhibitor(unittest.TestCase): status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}'))) @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient) def setUp(self): - self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test', - 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + self.e = get_dcs({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, + 'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + self.assertIsInstance(self.e, Exhibitor) @patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True)) @patch.object(MockKazooClient, 'get_children', Mock(side_effect=Exception)) diff --git a/tests/test_ha.py b/tests/test_ha.py index 40063018..45f64164 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -197,7 +197,7 @@ def run_async(self, func, args=()): @patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False)) @patch('patroni.async_executor.AsyncExecutor.run_async', run_async) @patch('patroni.postgresql.rewind.Thread', Mock()) -@patch('patroni.postgresql.citus.CitusHandler.start', Mock()) +@patch('patroni.postgresql.mpp.citus.CitusHandler.start', Mock()) @patch('subprocess.call', Mock(return_value=0)) @patch('time.sleep', Mock()) class TestHa(PostgresInit): @@ -593,8 +593,8 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.bootstrap(), 'failed to acquire initialize lock') @patch('patroni.psycopg.connect', psycopg_connect) - @patch('patroni.postgresql.citus.connect', psycopg_connect) - @patch('patroni.postgresql.citus.quote_ident', Mock()) + @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect) + @patch('patroni.postgresql.mpp.citus.quote_ident', Mock()) @patch.object(Postgresql, 'connection', Mock(return_value=None)) def test_bootstrap_initialized_new_cluster(self): self.ha.cluster = get_cluster_not_initialized_without_leader() @@ -615,8 +615,8 @@ class TestHa(PostgresInit): self.assertRaises(PatroniFatalException, self.ha.post_bootstrap) @patch('patroni.psycopg.connect', psycopg_connect) - @patch('patroni.postgresql.citus.connect', psycopg_connect) - @patch('patroni.postgresql.citus.quote_ident', Mock()) + @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect) + @patch('patroni.postgresql.mpp.citus.quote_ident', Mock()) @patch.object(Postgresql, 'connection', Mock(return_value=None)) def test_bootstrap_release_initialize_key_on_watchdog_failure(self): self.ha.cluster = get_cluster_not_initialized_without_leader() @@ -659,7 +659,7 @@ class TestHa(PostgresInit): @patch.object(ConfigHandler, 'replace_pg_hba', Mock()) @patch.object(ConfigHandler, 'replace_pg_ident', Mock()) @patch.object(PostmasterProcess, 'start', Mock(return_value=MockPostmaster())) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_worker_restart(self): self.ha.has_lock = true self.ha.patroni.request = Mock() @@ -694,7 +694,7 @@ class TestHa(PostgresInit): self.ha.is_paused = true self.assertEqual(self.ha.run_cycle(), 'PAUSE: restart in progress') - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_manual_failover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -733,7 +733,7 @@ class TestHa(PostgresInit): ('Member %s exceeds maximum replication lag', 'b')) self.ha.cluster.members.pop() - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_manual_switchover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -774,7 +774,7 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock') self.assertEqual(mock_info.call_args_list[0][0], ('Member %s exceeds maximum replication lag', 'leader')) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_scheduled_switchover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -1544,7 +1544,7 @@ class TestHa(PostgresInit): self.ha.is_failover_possible = true self.ha.shutdown() - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_shutdown_citus_worker(self): self.ha.is_leader = true self.p.is_running = Mock(side_effect=[Mock(), False]) @@ -1585,6 +1585,11 @@ class TestHa(PostgresInit): self.p.is_primary = false self.ha.run_cycle() exit_mock.assert_called_once_with(1) + self.p.set_role('replica') + self.ha.dcs.initialize = Mock() + with patch.object(Postgresql, 'cb_called', PropertyMock(return_value=True)): + self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock') + self.ha.dcs.initialize.assert_not_called() @patch.object(Cluster, 'is_unlocked', Mock(return_value=False)) def test_after_pause(self): @@ -1651,7 +1656,7 @@ class TestHa(PostgresInit): self.assertRaises(DCSError, self.ha.acquire_lock) self.assertFalse(self.ha.acquire_lock()) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_notify_citus_coordinator(self): self.ha.patroni.request = Mock() self.ha.notify_citus_coordinator('before_demote') diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index b6db7fb3..c493d799 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -8,9 +8,11 @@ import unittest import urllib3 from mock import Mock, PropertyMock, mock_open, patch +from patroni.dcs import get_dcs from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \ K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \ Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME +from patroni.postgresql.mpp import get_mpp from threading import Thread from . import MockResponse, SleepException @@ -225,11 +227,12 @@ class BaseTestKubernetes(unittest.TestCase): @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod, create=True) @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True) def setUp(self, config=None): - config = config or {} - config.update(ttl=30, scope='test', name='p-0', loop_wait=10, group=0, - retry_timeout=10, labels={'f': 'b'}, bypass_api_service=True) - self.k = Kubernetes(config) - self.k._citus_group = None + config = {'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10, + 'kubernetes': {'labels': {'f': 'b'}, 'bypass_api_service': True, **(config or {})}, + 'citus': {'group': 0, 'database': 'postgres'}} + self.k = get_dcs(config) + self.assertIsInstance(self.k, Kubernetes) + self.k._mpp = get_mpp({}) self.assertRaises(AttributeError, self.k._pods._build_cache) self.k._pods._is_ready = True self.assertRaises(TypeError, self.k._kinds._build_cache) @@ -254,7 +257,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.assertRaises(KubernetesError, self.k.get_cluster) def test__get_citus_cluster(self): - self.k._citus_group = '0' + self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.k.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) @@ -466,7 +469,7 @@ class TestCacheBuilder(BaseTestKubernetes): @patch('patroni.dcs.kubernetes.ObjectCache._watch', mock_watch) @patch.object(urllib3.HTTPResponse, 'read_chunked') def test__build_cache(self, mock_read_chunked): - self.k._citus_group = '0' + self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) mock_read_chunked.return_value = [json.dumps( {'type': 'MODIFIED', 'object': {'metadata': { 'name': self.k.config_path, 'resourceVersion': '2', 'annotations': {self.k._CONFIG: 'foo'}}}} diff --git a/tests/test_mpp.py b/tests/test_mpp.py new file mode 100644 index 00000000..9eb87633 --- /dev/null +++ b/tests/test_mpp.py @@ -0,0 +1,52 @@ +from typing import Any +from patroni.exceptions import PatroniException +from patroni.postgresql.mpp import AbstractMPP, get_mpp, Null + +from . import BaseTestPostgresql +from .test_ha import get_cluster_initialized_with_leader + + +class TestMPP(BaseTestPostgresql): + + def setUp(self): + super(TestMPP, self).setUp() + self.cluster = get_cluster_initialized_with_leader() + + def test_get_handler_impl_exception(self): + class DummyMPP(AbstractMPP): + def __init__(self) -> None: + super().__init__({}) + + @staticmethod + def validate_config(config: Any) -> bool: + return True + + @property + def group(self) -> None: + return None + + @property + def coordinator_group_id(self) -> None: + return None + + @property + def type(self) -> str: + return "dummy" + + mpp = DummyMPP() + self.assertRaises(PatroniException, mpp.get_handler_impl, self.p) + + def test_null_handler(self): + config = {} + mpp = get_mpp(config) + self.assertIsInstance(mpp, Null) + self.assertIsNone(mpp.group) + self.assertTrue(mpp.validate_config(config)) + nullHandler = mpp.get_handler_impl(self.p) + self.assertIsNone(nullHandler.handle_event(self.cluster, {})) + self.assertIsNone(nullHandler.sync_meta_data(self.cluster)) + self.assertIsNone(nullHandler.on_demote()) + self.assertIsNone(nullHandler.schedule_cache_rebuild()) + self.assertIsNone(nullHandler.bootstrap()) + self.assertIsNone(nullHandler.adjust_postgres_gucs({})) + self.assertFalse(nullHandler.ignore_replication_slot({})) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 31454479..8a5c491b 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -5,6 +5,7 @@ import re import subprocess import time +from copy import deepcopy from mock import Mock, MagicMock, PropertyMock, patch, mock_open import patroni.psycopg as psycopg @@ -25,7 +26,8 @@ from patroni.postgresql.validator import (ValidatorFactoryNoType, ValidatorFacto from patroni.utils import RetryFailedError from threading import Thread, current_thread -from . import BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs +from . import (BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs, + GET_PG_SETTINGS_RESULT) mtime_ret = {} @@ -559,31 +561,103 @@ class TestPostgresql(BaseTestPostgresql): @patch('time.sleep', Mock()) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) - def test_reload_config(self): - parameters = self._PARAMETERS.copy() - parameters.pop('f.oo') - parameters['wal_buffers'] = '512' - config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'use_unix_socket_repl': True, - 'authentication': {}, - 'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters} + @patch('patroni.postgresql.config.logger.info') + @patch('patroni.postgresql.config.logger.warning') + def test_reload_config(self, mock_warning, mock_info): + config = deepcopy(self.p.config._config) + + # Nothing changed self.p.reload_config(config) - parameters['b.ar'] = 'bar' - with patch.object(MockCursor, 'fetchall', - Mock(side_effect=[[('wal_block_size', '8191', None, 'integer', 'internal'), - ('wal_segment_size', '2048', '8kB', 'integer', 'internal'), - ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'), - ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'), - ('port', '5433', None, 'integer', 'postmaster')], Exception])): + mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.') + mock_warning.assert_not_called() + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # Handle wal_buffers + self.p.config._config['parameters']['wal_buffers'] = '512' + self.p.reload_config(config) + mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.') + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + config = deepcopy(self.p.config._config) + + # hba/ident_changed + config['pg_hba'] = [''] + config['pg_ident'] = [''] + self.p.reload_config(config) + mock_info.assert_called_once_with('Reloading PostgreSQL configuration.') + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # Postmaster parameter change (pending_restart) + init_max_worker_processes = config['parameters']['max_worker_processes'] + config['parameters']['max_worker_processes'] *= 2 + with patch('patroni.postgresql.Postgresql._query', Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)]])): self.p.reload_config(config) - parameters['autovacuum'] = 'on' + self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s (restart might be required)', + 'max_worker_processes', str(init_max_worker_processes), + config['parameters']['max_worker_processes'])) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, True) + + mock_info.reset_mock() + + # Reset to the initial value without restart + config['parameters']['max_worker_processes'] = init_max_worker_processes self.p.reload_config(config) - parameters['autovacuum'] = 'off' - parameters.pop('search_path') - config['listen'] = '*:5433' + self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'max_worker_processes', + init_max_worker_processes * 2, + str(config['parameters']['max_worker_processes']))) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # User-defined parameter changed (removed) + config['parameters'].pop('f.oo') self.p.reload_config(config) - parameters['unix_socket_directories'] = '.' + self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'f.oo', 'bar', None)) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, False) + + mock_info.reset_mock() + + # Non-postmaster parameter change + config['parameters']['autovacuum'] = 'off' self.p.reload_config(config) - self.p.config.resolve_connection_addresses() + self.assertEqual(mock_info.call_args_list[0][0], ("Changed %s from %s to %s", 'autovacuum', 'on', 'off')) + self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, False) + + config['parameters']['autovacuum'] = 'on' + mock_info.reset_mock() + + # Remove invalid parameter + config['parameters']['invalid'] = 'value' + self.p.reload_config(config) + self.assertEqual(mock_warning.call_args_list[0][0], + ('Removing invalid parameter `%s` from postgresql.parameters', 'invalid')) + config['parameters'].pop('invalid') + + mock_warning.reset_mock() + mock_info.reset_mock() + + # Non-empty result (outside changes) and exception while querying pending_restart parameters + with patch('patroni.postgresql.Postgresql._query', + Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)], GET_PG_SETTINGS_RESULT, Exception])): + self.p.reload_config(config, True) + self.assertEqual(mock_info.call_args_list[0][0], ('Reloading PostgreSQL configuration.',)) + self.assertEqual(self.p.pending_restart, True) + + # Invalid values, just to increase silly coverage in postgresql.validator. + # One day we will have proper tests there. + config['parameters']['autovacuum'] = 'of' # Bool.transform() + config['parameters']['vacuum_cost_limit'] = 'smth' # Number.transform() + self.p.reload_config(config, True) + self.assertEqual(mock_warning.call_args_list[-1][0][0], 'Exception %r when running query') def test_resolve_connection_addresses(self): self.p.config._config['use_unix_socket'] = self.p.config._config['use_unix_socket_repl'] = True diff --git a/tests/test_raft.py b/tests/test_raft.py index 9bb109e9..387a5a56 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -4,8 +4,10 @@ import tempfile import time from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \ Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport +from patroni.postgresql.mpp import get_mpp from pysyncobj import SyncObjConf, FAIL_REASON @@ -128,9 +130,10 @@ class TestRaft(unittest.TestCase): _TMP = tempfile.gettempdir() def test_raft(self): - raft = Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'self_addr': '127.0.0.1:1234', - 'retry_timeout': 10, 'data_dir': self._TMP, - 'database': 'citus', 'group': 0}) + raft = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'retry_timeout': 10, + 'raft': {'self_addr': '127.0.0.1:1234', 'data_dir': self._TMP}, + 'citus': {'group': 0, 'database': 'postgres'}}) + self.assertIsInstance(raft, Raft) raft.reload_config({'retry_timeout': 20, 'ttl': 60, 'loop_wait': 10}) self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}')) self.assertTrue(raft.touch_member('')) @@ -139,9 +142,9 @@ class TestRaft(unittest.TestCase): self.assertTrue(raft.set_config_value('{}')) self.assertTrue(raft.write_sync_state('foo', 'bar')) self.assertFalse(raft.write_sync_state('foo', 'bar', 1)) - raft._citus_group = '1' + raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}}) self.assertTrue(raft.manual_failover('foo', 'bar')) - raft._citus_group = '0' + raft._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) self.assertTrue(raft.take_leader()) cluster = raft.get_cluster() self.assertIsInstance(cluster, Cluster) @@ -157,9 +160,9 @@ class TestRaft(unittest.TestCase): self.assertTrue(raft.delete_sync_state()) self.assertTrue(raft.set_history_value('')) self.assertTrue(raft.delete_cluster()) - raft._citus_group = '1' + raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}}) self.assertTrue(raft.delete_cluster()) - raft._citus_group = None + raft._mpp = get_mpp({}) raft.get_cluster() raft.watch(None, 0.001) raft._sync_obj.destroy() @@ -175,5 +178,5 @@ class TestRaft(unittest.TestCase): def test_init(self, mock_event, mock_kvstore): mock_kvstore.return_value.applied_local_log = False mock_event.return_value.is_set.side_effect = [False, True] - self.assertIsNotNone(Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True, - 'self_addr': '1', 'data_dir': self._TMP})) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True, + 'raft': {'self_addr': '1', 'data_dir': self._TMP}}), Raft) diff --git a/tests/test_validator.py b/tests/test_validator.py index 1f647dbe..d5e82992 100644 --- a/tests/test_validator.py +++ b/tests/test_validator.py @@ -13,6 +13,20 @@ available_dcs = [m.split(".")[-1] for m in dcs_modules()] config = { "name": "string", "scope": "string", + "log": { + "level": "DEBUG", + "traceback_level": "DEBUG", + "format": "%(asctime)s %(levelname)s: %(message)s", + "dateformat": "%Y-%m-%d %H:%M:%S", + "max_queue_size": 100, + "dir": "/tmp", + "file_num": 10, + "file_size": 1000000, + "loggers": { + "patroni.postmaster": "WARNING", + "urllib3": "DEBUG" + } + }, "restapi": { "listen": "127.0.0.2:800", "connect_address": "127.0.0.2:800", diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 3ce3ea75..3cd03467 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -7,8 +7,10 @@ from kazoo.handlers.threading import SequentialThreadingHandler from kazoo.protocol.states import KeeperState, WatchedEvent, ZnodeStat from kazoo.retry import RetryFailedError from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.zookeeper import Cluster, PatroniKazooClient, \ PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError +from patroni.postgresql.mpp import get_mpp class MockKazooClient(Mock): @@ -148,9 +150,9 @@ class TestZooKeeper(unittest.TestCase): @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient) def setUp(self): - self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test', - 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10, - 'set_acls': {'CN=principal2': ['ALL']}}) + self.zk = get_dcs({'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10, + 'zookeeper': {'hosts': ['localhost:2181'], 'set_acls': {'CN=principal2': ['ALL']}}}) + self.assertIsInstance(self.zk, ZooKeeper) def test_reload_config(self): self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10}) @@ -177,7 +179,7 @@ class TestZooKeeper(unittest.TestCase): self.assertEqual(cluster.last_lsn, 500) def test__get_citus_cluster(self): - self.zk._citus_group = '0' + self.zk._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) for _ in range(0, 2): cluster = self.zk.get_cluster() self.assertIsInstance(cluster, Cluster)