Merge branch 'master' of github.com:zalando/patroni into feature/citus-secondaries

.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])])

.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'}

.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:

.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:

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

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

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,

kubernetes/Dockerfile

@@ -1,4 +1,4 @@
-FROM postgres:15
+FROM postgres:16
 LABEL maintainer="Alexander Kukushkin <[email protected]>"
 
 RUN export DEBIAN_FRONTEND=noninteractive \

kubernetes/Dockerfile.citus

@@ -1,4 +1,4 @@
-FROM postgres:15
+FROM postgres:16
 LABEL maintainer="Alexander Kukushkin <[email protected]>"
 
 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 \

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)
 

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.

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)
 

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
 

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

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()}
 

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()}
 

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]]]

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

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()}
 

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
 

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')

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()
 

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)

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()

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

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|"

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:

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
 

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,

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):

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'}))

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)

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)), [])
 

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)

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)

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))

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')

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'}}}}

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({}))

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

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)

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",

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)