From bcfd8438a50a1f36b65be76ee2d17d1da3b903e3 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 21 Dec 2023 08:58:26 +0100 Subject: [PATCH] Abstract CitusHandler and decouple it from configuration (#2950) the main issue was that the configuration for Citus handler and for DCS existed in two places, while ideally AbstractDCS should not know many details about what kind of MPP is in use. To solve the problem we first dynamically create an object implementing AbstractMPP interfaces, which is a configuration for DCS. Later this object is used to instantiate the class implementing AbstractMPPHandler interface. This is just a starting point, which does some heavy lifting. As a next steps all kind of variables named after Citus in files different from patroni/postgres/mpp/citus.py should be renamed. In other words this commit takes over the most complex part of #2940, which was never implemented. Co-authored-by: zhjwpku --- patroni/__main__.py | 2 +- patroni/api.py | 4 +- patroni/ctl.py | 8 +- patroni/dcs/__init__.py | 33 +-- patroni/dcs/consul.py | 9 +- patroni/dcs/etcd.py | 13 +- patroni/dcs/etcd3.py | 10 +- patroni/dcs/exhibitor.py | 5 +- patroni/dcs/kubernetes.py | 20 +- patroni/dcs/raft.py | 10 +- patroni/dcs/zookeeper.py | 10 +- patroni/ha.py | 4 +- patroni/postgresql/__init__.py | 6 +- patroni/postgresql/mpp/__init__.py | 296 ++++++++++++++++++++++++++ patroni/postgresql/{ => mpp}/citus.py | 103 ++++++--- patroni/postgresql/slots.py | 2 +- tests/__init__.py | 36 ++-- tests/test_citus.py | 34 ++- tests/test_consul.py | 22 +- tests/test_ctl.py | 11 +- tests/test_etcd.py | 11 +- tests/test_etcd3.py | 10 +- tests/test_exhibitor.py | 6 +- tests/test_ha.py | 22 +- tests/test_kubernetes.py | 17 +- tests/test_mpp.py | 52 +++++ tests/test_raft.py | 21 +- tests/test_zookeeper.py | 10 +- 28 files changed, 602 insertions(+), 185 deletions(-) create mode 100644 patroni/postgresql/mpp/__init__.py rename patroni/postgresql/{ => mpp}/citus.py (84%) create mode 100644 tests/test_mpp.py diff --git a/patroni/__main__.py b/patroni/__main__.py index 229ccfb9..2c253da4 100644 --- a/patroni/__main__.py +++ b/patroni/__main__.py @@ -68,7 +68,7 @@ class Patroni(AbstractPatroniDaemon, Tags): self.watchdog = Watchdog(self.config) self.load_dynamic_configuration() - self.postgresql = Postgresql(self.config['postgresql']) + self.postgresql = Postgresql(self.config['postgresql'], self.dcs.mpp) self.api = RestApiServer(self, self.config['restapi']) self.ha = Ha(self) diff --git a/patroni/api.py b/patroni/api.py index 5761d359..5c9d6603 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -1153,8 +1153,8 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_POST_citus(self) -> None: """Handle a ``POST`` request to ``/citus`` path. - Call :func:`~patroni.postgresql.CitusHandler.handle_event` to handle the request, then write a response with - HTTP status code ``200``. + Call :func:`~patroni.postgresql.mpp.AbstractMPPHandler.handle_event` to handle the request, + then write a response with HTTP status code ``200``. .. note:: If unable to parse the request body, then the request is silently discarded. diff --git a/patroni/ctl.py b/patroni/ctl.py index de1fe892..6a5fe0e6 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -51,6 +51,7 @@ from .config import Config from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member from .exceptions import PatroniException from .postgresql.misc import postgres_version_to_int +from .postgresql.mpp import get_mpp from .utils import cluster_as_json, patch_config, polling_loop from .request import PatroniRequest from .version import __version__ @@ -313,7 +314,7 @@ def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: config = load_config(config_file, dcs_url) # backward compatibility for configuration file where ctl section is not defined config.setdefault('ctl', {})['insecure'] = config.get('ctl', {}).get('insecure') or insecure - ctx.obj = {'__config': config} + ctx.obj = {'__config': config, '__mpp': get_mpp(config)} def is_citus_cluster() -> bool: @@ -321,7 +322,7 @@ def is_citus_cluster() -> bool: :returns: ``True`` if configuration has ``citus`` section, otherwise ``False``. """ - return bool(_get_configuration().get('citus')) + return click.get_current_context().obj['__mpp'].is_enabled() def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: @@ -340,12 +341,13 @@ def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: config = _get_configuration() config.update({'scope': scope, 'patronictl': True}) if group is not None: - config['citus'] = {'group': group} + config['citus'] = {'group': group, 'database': 'postgres'} config.setdefault('name', scope) try: dcs = _get_dcs(config) if is_citus_cluster() and group is None: dcs.is_citus_coordinator = lambda: True + click.get_current_context().obj['__mpp'] = dcs.mpp return dcs except PatroniException as e: raise PatroniCtlException(str(e)) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 28c3734f..a210615b 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -25,10 +25,9 @@ from ..utils import parse_int if TYPE_CHECKING: # pragma: no cover from ..config import Config from ..postgresql import Postgresql + from ..postgresql.mpp import AbstractMPP SLOT_ADVANCE_AVAILABLE_VERSION = 110000 -CITUS_COORDINATOR_GROUP_ID = 0 -citus_group_re = re.compile('^(0|[1-9][0-9]*)$') slot_name_re = re.compile('^[a-z0-9_]{1,63}$') logger = logging.getLogger(__name__) @@ -130,10 +129,9 @@ def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS': p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait', 'patronictl', 'ttl', 'retry_timeout') if p in config}) - # From citus section we only need "group" parameter, but will propagate everything just in case. - if isinstance(config.get('citus'), dict): - config[name].update(config['citus']) - return dcs_class(config[name]) + + from patroni.postgresql.mpp import get_mpp + return dcs_class(config[name], get_mpp(config)) available_implementations = ', '.join(sorted([n for n, _ in iter_dcs_classes()])) raise PatroniFatalException("Can not find suitable configuration of distributed configuration store\n" @@ -1338,15 +1336,15 @@ class AbstractDCS(abc.ABC): _SYNC = 'sync' _FAILSAFE = 'failsafe' - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: 'AbstractMPP') -> None: """Prepare DCS paths, Citus group ID, initial values for state information and processing dependencies. :ivar config: :class:`dict`, reference to config section of selected DCS. i.e.: ``zookeeper`` for zookeeper, ``etcd`` for etcd, etc... """ + self._mpp = mpp self._name = config['name'] self._base_path = re.sub('/+', '/', '/'.join(['', config.get('namespace', 'service'), config['scope']])) - self._citus_group = str(config['group']) if isinstance(config.get('group'), int) else None self._set_loop_wait(config.get('loop_wait', 10)) self._ctl = bool(config.get('patronictl', False)) @@ -1359,6 +1357,11 @@ class AbstractDCS(abc.ABC): self._last_failsafe: Optional[Dict[str, str]] = {} self.event = Event() + @property + def mpp(self) -> 'AbstractMPP': + """Get the effective underlying MPP, if any has been configured.""" + return self._mpp + def client_path(self, path: str) -> str: """Construct the absolute key name from appropriate parts for the DCS type. @@ -1367,8 +1370,8 @@ class AbstractDCS(abc.ABC): :returns: absolute key name for the current Patroni cluster. """ components = [self._base_path] - if self._citus_group: - components.append(self._citus_group) + if self._mpp.is_enabled(): + components.append(str(self._mpp.group)) components.append(path.lstrip('/')) return '/'.join(components) @@ -1522,9 +1525,9 @@ class AbstractDCS(abc.ABC): def is_citus_coordinator(self) -> bool: """:class:`Cluster` instance has a Citus Coordinator group ID. - :returns: ``True`` if the given node is running as Citus Coordinator (``group=0``). + :returns: ``True`` if the given node is running as the MPP Coordinator. """ - return self._citus_group == str(CITUS_COORDINATOR_GROUP_ID) + return self._mpp.is_coordinator() def get_citus_coordinator(self) -> Optional[Cluster]: """Load the Patroni cluster for the Citus Coordinator. @@ -1532,10 +1535,10 @@ class AbstractDCS(abc.ABC): .. note:: This method is only executed on the worker nodes (``group!=0``) to find the coordinator. - :returns: Select :class:`Cluster` instance associated with the Citus Coordinator group ID. + :returns: Select :class:`Cluster` instance associated with the MPP Coordinator group ID. """ try: - return self.__get_patroni_cluster(f'{self._base_path}/{CITUS_COORDINATOR_GROUP_ID}/') + return self.__get_patroni_cluster(f'{self._base_path}/{self._mpp.coordinator_group_id}/') except Exception as e: logger.error('Failed to load Citus coordinator cluster from %s: %r', self.__class__.__name__, e) return None @@ -1549,7 +1552,7 @@ class AbstractDCS(abc.ABC): groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(groups, dict) - cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty()) + cluster = groups.pop(self._mpp.coordinator_group_id, Cluster.empty()) cluster.workers.update(groups) return cluster diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index fe66c6d8..19d65306 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -16,8 +16,9 @@ from urllib.parse import urlencode, urlparse, quote from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -232,8 +233,8 @@ def service_name_from_scope_name(scope_name: str) -> str: class Consul(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(Consul, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Consul, self).__init__(config, mpp) self._base_path = self._base_path[1:] self._scope = config['scope'] self._session = None @@ -435,7 +436,7 @@ class Consul(AbstractDCS): clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict) for node in results or []: key = node['Key'][len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): node['Value'] = (node['Value'] or b'').decode('utf-8') clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 3be699a6..b9d3e0ab 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -22,8 +22,9 @@ from urllib3 import Timeout from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..request import get as requests_get from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover @@ -470,9 +471,9 @@ class EtcdClient(AbstractEtcdClientWithFailover): class AbstractEtcd(AbstractDCS): - def __init__(self, config: Dict[str, Any], client_cls: Type[AbstractEtcdClientWithFailover], + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP, client_cls: Type[AbstractEtcdClientWithFailover], retry_errors_cls: Union[Type[Exception], Tuple[Type[Exception], ...]]) -> None: - super(AbstractEtcd, self).__init__(config) + super(AbstractEtcd, self).__init__(config, mpp) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, retry_exceptions=retry_errors_cls) self._ttl = int(config.get('ttl') or 30) @@ -645,8 +646,8 @@ def catch_etcd_errors(func: Callable[..., Any]) -> Any: class Etcd(AbstractEtcd): - def __init__(self, config: Dict[str, Any]) -> None: - super(Etcd, self).__init__(config, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Etcd, self).__init__(config, mpp, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal)) self.__do_not_watch = False @property @@ -726,7 +727,7 @@ class Etcd(AbstractEtcd): clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict) for node in result.leaves: key = node.key[len(result.key):].lstrip('/').split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(result.etcd_index, nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index ea7e52f2..7cc2a115 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -16,9 +16,10 @@ from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union from . import ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, catch_return_false_exception, citus_group_re + TimelineHistory, catch_return_false_exception from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry from ..exceptions import DCSError, PatroniException +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT logger = logging.getLogger(__name__) @@ -671,8 +672,9 @@ class PatroniEtcd3Client(Etcd3Client): class Etcd3(AbstractEtcd): - def __init__(self, config: Dict[str, Any]) -> None: - super(Etcd3, self).__init__(config, PatroniEtcd3Client, (DeadlineExceeded, Unavailable, FailedPrecondition)) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Etcd3, self).__init__(config, mpp, PatroniEtcd3Client, + (DeadlineExceeded, Unavailable, FailedPrecondition)) self.__do_not_watch = False self._lease = None self._last_lease_refresh = 0 @@ -796,7 +798,7 @@ class Etcd3(AbstractEtcd): path = self._base_path + '/' for node in self._client.get_cluster(path): key = node['key'][len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = node return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py index 2b06073b..03d23575 100644 --- a/patroni/dcs/exhibitor.py +++ b/patroni/dcs/exhibitor.py @@ -7,6 +7,7 @@ from typing import Any, Callable, Dict, List, Union from . import Cluster from .zookeeper import ZooKeeper +from ..postgresql.mpp import AbstractMPP from ..request import get as requests_get from ..utils import uri @@ -66,10 +67,10 @@ class ExhibitorEnsembleProvider(object): class Exhibitor(ZooKeeper): - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: interval = config.get('poll_interval', 300) self._ensemble_provider = ExhibitorEnsembleProvider(config['hosts'], config['port'], poll_interval=interval) - super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts}) + super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts}, mpp) def _load_cluster( self, path: str, loader: Callable[[str], Union[Cluster, Dict[int, Cluster]]] diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index aee87bd3..343b496c 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -19,9 +19,9 @@ from urllib3.exceptions import HTTPError from threading import Condition, Lock, Thread from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING -from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re +from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \ Retry, RetryFailedError, tzutc, uri, USER_AGENT if TYPE_CHECKING: # pragma: no cover @@ -748,7 +748,7 @@ class Kubernetes(AbstractDCS): _CITUS_LABEL = 'citus-group' - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: self._labels = deepcopy(config['labels']) self._labels[config.get('scope_label', 'cluster-name')] = config['scope'] self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items()) @@ -759,9 +759,9 @@ class Kubernetes(AbstractDCS): self._standby_leader_label_value = config.get('standby_leader_label_value', 'master') self._tmp_role_label = config.get('tmp_role_label') self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME - super(Kubernetes, self).__init__({**config, 'namespace': ''}) - if self._citus_group: - self._labels[self._CITUS_LABEL] = self._citus_group + super(Kubernetes, self).__init__({**config, 'namespace': ''}, mpp) + if self._mpp.is_enabled(): + self._labels[self._CITUS_LABEL] = str(self._mpp.group) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, retry_exceptions=KubernetesRetriableException) @@ -944,12 +944,12 @@ class Kubernetes(AbstractDCS): for name, pod in path['pods'].items(): group = pod.metadata.labels.get(self._CITUS_LABEL) - if group and citus_group_re.match(group): + if group and self._mpp.group_re.match(group): clusters[group]['pods'][name] = pod for name, kind in path['nodes'].items(): group = kind.metadata.labels.get(self._CITUS_LABEL) - if group and citus_group_re.match(group): + if group and self._mpp.group_re.match(group): clusters[group]['nodes'][name] = kind return {int(group): self._cluster_from_nodes(group, value['nodes'], value['pods'].values()) for group, value in clusters.items()} @@ -976,12 +976,12 @@ class Kubernetes(AbstractDCS): def _load_cluster( self, path: str, loader: Callable[[Any], Union[Cluster, Dict[int, Cluster]]] ) -> Union[Cluster, Dict[int, Cluster]]: - group = self._citus_group if path == self.client_path('') else None + group = str(self._mpp.group) if self._mpp.is_enabled() and path == self.client_path('') else None return self.__load_cluster(group, loader) def get_citus_coordinator(self) -> Optional[Cluster]: try: - ret = self.__load_cluster(str(CITUS_COORDINATOR_GROUP_ID), self._cluster_loader) + ret = self.__load_cluster(str(self._mpp.coordinator_group_id), self._cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(ret, Cluster) return ret diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index 98c48f44..0528cfb0 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -12,9 +12,9 @@ from pysyncobj.transport import TCPTransport, CONNECTION_STATE from pysyncobj.utility import TcpUtility from typing import Any, Callable, Collection, Dict, List, Optional, Set, Union, TYPE_CHECKING -from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, citus_group_re +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import validate_directory if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -285,8 +285,8 @@ class KVStoreTTL(DynMemberSyncObj): class Raft(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(Raft, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(Raft, self).__init__(config, mpp) self._ttl = int(config.get('ttl') or 30) ready_event = threading.Event() @@ -387,7 +387,7 @@ class Raft(AbstractDCS): response = self._sync_obj.get(path, recursive=True) for key, value in (response or {}).items(): key = key[len(path):].split('/', 1) - if len(key) == 2 and citus_group_re.match(key[0]): + if len(key) == 2 and self._mpp.group_re.match(key[0]): clusters[int(key[0])][key[1]] = value return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()} diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 3704b579..6bf77ae4 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -12,9 +12,9 @@ from kazoo.retry import RetryFailedError from kazoo.security import ACL, make_acl from typing import Any, Callable, Dict, List, Optional, Union, Tuple, TYPE_CHECKING -from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \ - TimelineHistory, citus_group_re +from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory from ..exceptions import DCSError +from ..postgresql.mpp import AbstractMPP from ..utils import deep_compare if TYPE_CHECKING: # pragma: no cover from ..config import Config @@ -87,8 +87,8 @@ class PatroniKazooClient(KazooClient): class ZooKeeper(AbstractDCS): - def __init__(self, config: Dict[str, Any]) -> None: - super(ZooKeeper, self).__init__(config) + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: + super(ZooKeeper, self).__init__(config, mpp) hosts: Union[str, List[str]] = config.get('hosts', []) if isinstance(hosts, list): @@ -261,7 +261,7 @@ class ZooKeeper(AbstractDCS): def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: ret: Dict[int, Cluster] = {} for node in self.get_children(path): - if citus_group_re.match(node): + if self._mpp.group_re.match(node): ret[int(node)] = self._cluster_loader(path + node + '/') return ret diff --git a/patroni/ha.py b/patroni/ha.py index 0d3e05a4..dea78163 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -332,7 +332,7 @@ class Ha(object): if coordinator and coordinator.leader and coordinator.leader.conn_url: try: data = {'type': event, - 'group': self.state_handler.citus_handler.group(), + 'group': self.state_handler.citus_handler.group, 'leader': self.state_handler.name, 'timeout': self.dcs.ttl, 'cooldown': self.patroni.config['retry_timeout']} @@ -847,7 +847,7 @@ class Ha(object): self.state_handler.set_role('master') self.process_sync_replication() self.update_cluster_history() - self.state_handler.citus_handler.sync_pg_dist_node(self.cluster) + self.state_handler.citus_handler.sync_meta_data(self.cluster) return message elif self.state_handler.role in ('master', 'promoted', 'primary'): self.process_sync_replication() diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index c87c29fa..f6e3ab54 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -19,8 +19,8 @@ from .callback_executor import CallbackAction, CallbackExecutor from .cancellable import CancellableSubprocess from .config import ConfigHandler, mtime from .connection import ConnectionPool, get_connection_cursor -from .citus import CitusHandler from .misc import parse_history, parse_lsn, postgres_major_version_to_int +from .mpp import AbstractMPP from .postmaster import PostmasterProcess from .slots import SlotsHandler from .sync import SyncHandler @@ -63,7 +63,7 @@ class Postgresql(object): "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, " "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()") - def __init__(self, config: Dict[str, Any]) -> None: + def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None: self.name: str = config['name'] self.scope: str = config['scope'] self._data_dir: str = config['data_dir'] @@ -80,7 +80,7 @@ class Postgresql(object): self._pending_restart = False self.connection_pool = ConnectionPool() self._connection = self.connection_pool.get('heartbeat') - self.citus_handler = CitusHandler(self, config.get('citus')) + self.citus_handler = mpp.get_handler_impl(self) self.config = ConfigHandler(self, config) self.config.check_directories() diff --git a/patroni/postgresql/mpp/__init__.py b/patroni/postgresql/mpp/__init__.py new file mode 100644 index 00000000..3494793b --- /dev/null +++ b/patroni/postgresql/mpp/__init__.py @@ -0,0 +1,296 @@ +"""Abstract classes for MPP handler. + +MPP stands for Massively Parallel Processing, and Citus belongs to this architecture. Currently, Citus is the only +supported MPP cluster. However, we may consider adapting other databases such as TimescaleDB, GPDB, etc. into Patroni. +""" +import abc + +from typing import Any, Dict, Iterator, Optional, Union, Tuple, Type, TYPE_CHECKING + +from ...dcs import Cluster +from ...dynamic_loader import iter_classes +from ...exceptions import PatroniException + +if TYPE_CHECKING: # pragma: no cover + from .. import Postgresql + from ...config import Config + + +class AbstractMPP(abc.ABC): + """An abstract class which should be passed to :class:`AbstractDCS`. + + .. note:: + We create :class:`AbstractMPP` and :class:`AbstractMPPHandler` to solve the chicken-egg initialization problem. + When initializing DCS, we dynamically create an object implementing :class:`AbstractMPP`, later this object is + used to instantiate an object implementing :class:`AbstractMPPHandler`. + """ + + group_re: Any # re.Pattern[str] + + def __init__(self, config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`AbstractMPP`. + + :param config: configuration of MPP section. + """ + self._config = config + + def is_enabled(self) -> bool: + """Check if MPP is enabled for a given MPP. + + .. note:: + We just check that the :attr:`_config` object isn't empty and expect + it to be empty only in case of :class:`Null`. + + :returns: ``True`` if MPP is enabled, otherwise ``False``. + """ + return bool(self._config) + + @staticmethod + @abc.abstractmethod + def validate_config(config: Any) -> bool: + """Check whether provided config is good for a given MPP. + + :param config: configuration of MPP section. + + :returns: ``True`` is config passes validation, otherwise ``False``. + """ + + @property + @abc.abstractmethod + def group(self) -> Any: + """The group for a given MPP implementation.""" + + @property + @abc.abstractmethod + def coordinator_group_id(self) -> Any: + """The group id of the coordinator PostgreSQL cluster.""" + + def is_coordinator(self) -> bool: + """Check whether this node is running in the coordinator PostgreSQL cluster. + + :returns: ``True`` if MPP is enabled and the group id of this node + matches with the :attr:`coordinator_group_id`, otherwise ``False``. + """ + return self.is_enabled() and self.group == self.coordinator_group_id + + def is_worker(self) -> bool: + """Check whether this node is running as a MPP worker PostgreSQL cluster. + + :returns: ``True`` if MPP is enabled and this node is known to be not running + as the coordinator PostgreSQL cluster, otherwise ``False``. + """ + return self.is_enabled() and not self.is_coordinator() + + def _get_handler_cls(self) -> Iterator[Type['AbstractMPPHandler']]: + """Find Handler classes inherited from a class type of this object. + + :yields: handler classes for this object. + """ + for cls in self.__class__.__subclasses__(): + if issubclass(cls, AbstractMPPHandler) and cls.__name__.startswith(self.__class__.__name__): + yield cls + + def get_handler_impl(self, postgresql: 'Postgresql') -> 'AbstractMPPHandler': + """Find and instantiate Handler implementation of this object. + + :param postgresql: a reference to :class:`Postgresql` object. + + :raises: + :exc:`PatroniException`: if the Handler class haven't been found. + + :returns: an instantiated class that implements Handler for this object. + """ + for cls in self._get_handler_cls(): + return cls(postgresql, self._config) + raise PatroniException(f'Failed to initialize {self.__class__.__name__}Handler object') + + +class AbstractMPPHandler(AbstractMPP): + """An abstract class which defines interfaces that should be implemented by real handlers.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`AbstractMPPHandler`. + + :param postgresql: a reference to :class:`Postgresql` object. + :param config: configuration of MPP section. + """ + super().__init__(config) + self._postgresql = postgresql + + @abc.abstractmethod + def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None: + """Handle an event sent from a worker node. + + :param cluster: the currently known cluster state from DCS. + :param event: the event to be handled. + """ + + @abc.abstractmethod + def sync_meta_data(self, cluster: Cluster) -> None: + """Sync meta data on the coordinator. + + :param cluster: the currently known cluster state from DCS. + """ + + @abc.abstractmethod + def on_demote(self) -> None: + """On demote handler. + + Is called when the primary was demoted. + """ + + @abc.abstractmethod + def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ + + @abc.abstractmethod + def bootstrap(self) -> None: + """Bootstrap handler. + + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ + + @abc.abstractmethod + def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: + """Adjust GUCs in the current PostgreSQL configuration. + + :param parameters: dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value. + """ + + @abc.abstractmethod + def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: ``True`` if the replication slots should not be removed, otherwise ``False``. + """ + + +class Null(AbstractMPP): + """Dummy implementation of :class:`AbstractMPP`.""" + + def __init__(self) -> None: + """Init method for :class:`Null`.""" + super().__init__({}) + + @staticmethod + def validate_config(config: Any) -> bool: + """Check whether provided config is good for :class:`Null`. + + :returns: always ``True``. + """ + return True + + @property + def group(self) -> None: + """The group for :class:`Null`. + + :returns: always ``None``. + """ + return None + + @property + def coordinator_group_id(self) -> None: + """The group id of the coordinator PostgreSQL cluster. + + :returns: always ``None``. + """ + return None + + +class NullHandler(Null, AbstractMPPHandler): + """Dummy implementation of :class:`AbstractMPPHandler`.""" + + def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None: + """Init method for :class:`NullHandler`. + + :param postgresql: a reference to :class:`Postgresql` object. + :param config: configuration of MPP section. + """ + AbstractMPPHandler.__init__(self, postgresql, config) + + def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None: + """Handle an event sent from a worker node. + + :param cluster: the currently known cluster state from DCS. + :param event: the event to be handled. + """ + + def sync_meta_data(self, cluster: Cluster) -> None: + """Sync meta data on the coordinator. + + :param cluster: the currently known cluster state from DCS. + """ + + def on_demote(self) -> None: + """On demote handler. + + Is called when the primary was demoted. + """ + + def schedule_cache_rebuild(self) -> None: + """Cache rebuild handler. + + Is called to notify handler that it has to refresh its metadata cache from the database. + """ + + def bootstrap(self) -> None: + """Bootstrap handler. + + Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method). + """ + + def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None: + """Adjust GUCs in the current PostgreSQL configuration. + + :param parameters: dictionary of GUCs, with key as GUC name and corresponding value as current GUC value. + """ + + def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: always ``False``. + """ + return False + + +def iter_mpp_classes( + config: Optional[Union['Config', Dict[str, Any]]] = None +) -> Iterator[Tuple[str, Type[AbstractMPP]]]: + """Attempt to import MPP modules that are present in the given configuration. + + :param config: configuration information with possible MPP names as keys. If given, only attempt to import MPP + modules defined in the configuration. Else, if ``None``, attempt to import any supported MPP module. + + :yields: tuples, each containing the module ``name`` and the imported MPP class object. + """ + yield from iter_classes(__package__, AbstractMPP, config) + + +def get_mpp(config: Union['Config', Dict[str, Any]]) -> AbstractMPP: + """Attempt to load and instantiate a MPP module from known available implementations. + + :param config: object or dictionary with Patroni configuration. + + :returns: The successfully loaded MPP or fallback to :class:`Null`. + """ + for name, mpp_class in iter_mpp_classes(config): + if mpp_class.validate_config(config[name]): + return mpp_class(config[name]) + return Null() diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/mpp/citus.py similarity index 84% rename from patroni/postgresql/citus.py rename to patroni/postgresql/mpp/citus.py index b50dc1d0..a17d1435 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/mpp/citus.py @@ -6,12 +6,15 @@ from threading import Condition, Event, Thread from urllib.parse import urlparse from typing import Any, Dict, List, Optional, Union, 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 +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__) @@ -63,13 +66,45 @@ class PgDistNode(object): return str(self) -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'], @@ -81,19 +116,11 @@ class CitusHandler(Thread): self._condition = Condition() # protects _pg_dist_node, _tasks, _in_flight, and _schedule_load_pg_dist_node 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_node = True @@ -134,8 +161,8 @@ class CitusHandler(Thread): self._pg_dist_node = {r[1]: PgDistNode(r[1], r[2], r[3], 'after_promote', r[0]) for r in rows} 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 @@ -296,16 +323,16 @@ class CitusHandler(Thread): with self._condition: i = self.find_task_by_group(task.group) - # The `PgDistNode.timeout` == None is an indicator that it was scheduled from the sync_pg_dist_node(). + # 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_node` 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_node() + # 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.group == task.group and self._in_flight.timeout is not None\ and self._in_flight.deadline > time.time(): return False @@ -353,9 +380,10 @@ 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: @@ -388,9 +416,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', @@ -408,8 +437,18 @@ class CitusHandler(Thread): parameters['citus.local_hostname'] = self._postgresql.connection_pool.conn_kwargs.get('host', 'localhost') def ignore_replication_slot(self, slot: Dict[str, str]) -> bool: - if isinstance(self._config, dict) and self._postgresql.is_primary() and\ - slot['type'] == 'logical' and slot['database'] == self._config['database']: + """Check whether provided replication *slot* existing in the database should not be removed. + + .. note:: + MPP database may create replication slots for its own use, for example to migrate data between workers + using logical replication, and we don't want to suddenly drop them. + + :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and + ``plugin``. + + :returns: ``True`` if the replication slots should not be removed, otherwise ``False``. + """ + if self._postgresql.is_primary() and slot['type'] == 'logical' and slot['database'] == self._config['database']: m = CITUS_SLOT_NAME_RE.match(slot['name']) return bool(m and {'move': 'pgoutput', 'split': 'citus'}.get(m.group(1)) == slot['plugin']) return False diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index fb9448cd..7f7fd294 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -291,7 +291,7 @@ class SlotsHandler: :param name: name of the slot to ignore :returns: ``True`` if slot *name* matches any slot specified in ``ignore_slots`` configuration, - otherwise will pass through and return result of :meth:`CitusHandler.ignore_replication_slot`. + otherwise will pass through and return result of :meth:`AbstractMPPHandler.ignore_replication_slot`. """ slot = self._replication_slots[name] if cluster.config: diff --git a/tests/__init__.py b/tests/__init__.py index b013e4e1..986bd88b 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -12,6 +12,7 @@ import patroni.psycopg as psycopg from patroni.dcs import Leader, Member from patroni.postgresql import Postgresql from patroni.postgresql.config import ConfigHandler +from patroni.postgresql.mpp import get_mpp from patroni.utils import RetryFailedError, tzutc @@ -252,23 +253,24 @@ class PostgresInit(unittest.TestCase): @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='primary')) def setUp(self): data_dir = os.path.join('data', 'test0') - self.p = Postgresql({'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir, - 'config_dir': data_dir, 'retry_timeout': 10, - 'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'), - 'listen': '127.0.0.2, 127.0.0.3:5432', - 'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433', - 'authentication': {'superuser': {'username': 'foo', 'password': 'test'}, - 'replication': {'username': '', 'password': 'rep-pass'}, - 'rewind': {'username': 'rewind', 'password': 'test'}}, - 'remove_data_directory_on_rewind_failure': True, - 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True, - 'parameters': self._PARAMETERS, - 'recovery_conf': {'foo': 'bar'}, - 'pg_hba': ['host all all 0.0.0.0/0 md5'], - 'pg_ident': ['krb realm postgres'], - 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true', - 'on_restart': 'true', 'on_role_change': 'true'}, - 'citus': {'group': 0, 'database': 'citus'}}) + config = {'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir, + 'config_dir': data_dir, 'retry_timeout': 10, + 'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'), + 'listen': '127.0.0.2, 127.0.0.3:5432', + 'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433', + 'authentication': {'superuser': {'username': 'foo', 'password': 'test'}, + 'replication': {'username': '', 'password': 'rep-pass'}, + 'rewind': {'username': 'rewind', 'password': 'test'}}, + 'remove_data_directory_on_rewind_failure': True, + 'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True, + 'parameters': self._PARAMETERS, + 'recovery_conf': {'foo': 'bar'}, + 'pg_hba': ['host all all 0.0.0.0/0 md5'], + 'pg_ident': ['krb realm postgres'], + 'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true', + 'on_restart': 'true', 'on_role_change': 'true'}, + 'citus': {'group': 0, 'database': 'citus'}} + self.p = Postgresql(config, get_mpp(config)) class BaseTestPostgresql(PostgresInit): diff --git a/tests/test_citus.py b/tests/test_citus.py index 7279893e..dbf6d9cf 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -1,12 +1,12 @@ import time from mock import Mock, patch -from patroni.postgresql.citus import CitusHandler +from patroni.postgresql.mpp.citus import CitusHandler 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): @@ -17,9 +17,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.PgDistNode.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.PgDistNode.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 @@ -39,10 +39,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, {}) @@ -51,22 +51,22 @@ 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, 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, '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, 'postgres://host:5432/postgres', 30) mock_logger.assert_called_once() self.assertTrue(mock_logger.call_args[0][0].startswith('Overriding existing task:')) - # add_task called from sync_pg_dist_node should not override already scheduled or in flight task until deadline + # add_task called from sync_meta_data should not override already scheduled or in flight task until deadline self.assertIsNotNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres', 30)) self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres')) self.c._in_flight = self.c._tasks.pop() @@ -106,7 +106,7 @@ class TestCitus(BaseTestPostgresql): self.c.process_tasks() self.c.add_task('after_promote', 0, '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() @@ -115,7 +115,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_node(self, mock_logger): # load_pg_dist_node() triggers, query fails and exception is property handled @@ -140,10 +140,6 @@ class TestCitus(BaseTestPostgresql): self.assertEqual(parameters['wal_level'], 'logical') self.assertEqual(parameters['citus.local_hostname'], '/tmp') - def test_bootstrap(self): - self.c._config = None - self.c.bootstrap() - def test_ignore_replication_slot(self): self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'physical', 'database': 'bar', 'plugin': 'wal2json'})) diff --git a/tests/test_consul.py b/tests/test_consul.py index 83ee67d8..494d1126 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -3,8 +3,10 @@ import unittest from consul import ConsulException, NotFound from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \ ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError +from patroni.postgresql.mpp import get_mpp from . import SleepException @@ -91,13 +93,17 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.KV, 'get', kv_get) @patch.object(consul.Consul.KV, 'delete', Mock()) def setUp(self): - Consul({'ttl': 30, 'scope': 't', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10, - 'verify': 'on', 'key': 'foo', 'cert': 'bar', 'cacert': 'buz', 'token': 'asd', 'dc': 'dc1', - 'register_service': True}) - Consul({'ttl': 30, 'scope': 't_', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10, - 'verify': 'on', 'cert': 'bar', 'cacert': 'buz', 'register_service': True}) - self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10, - 'register_service': True, 'service_check_tls_server_name': True}) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't', 'name': 'p', 'retry_timeout': 10, + 'consul': {'url': 'https://l:1', 'verify': 'on', + 'key': 'foo', 'cert': 'bar', 'cacert': 'buz', + 'token': 'asd', 'dc': 'dc1', 'register_service': True}}), Consul) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't_', 'name': 'p', 'retry_timeout': 10, + 'consul': {'url': 'https://l:1', 'verify': 'on', + 'cert': 'bar', 'cacert': 'buz', 'register_service': True}}), Consul) + self.c = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'retry_timeout': 10, + 'consul': {'host': 'localhost:1', 'register_service': True, + 'service_check_tls_server_name': True}}) + self.assertIsInstance(self.c, Consul) self.c._base_path = 'service/good' self.c.get_cluster() @@ -130,7 +136,7 @@ class TestConsul(unittest.TestCase): self.assertIsInstance(self.c.get_cluster(), Cluster) def test__get_citus_cluster(self): - self.c._citus_group = '0' + self.c._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.c.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index f9ee62ce..f5341f88 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -12,6 +12,7 @@ from patroni.ctl import ctl, load_config, output_members, get_dcs, parse_dcs, \ get_all_members, get_any_member, get_cursor, query_member, PatroniCtlException, apply_config_changes, \ format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH, PatronictlPrettyTable from patroni.dcs import Cluster, Failover +from patroni.postgresql.mpp import get_mpp from patroni.psycopg import OperationalError from patroni.utils import tzutc from prettytable import PrettyTable, ALL @@ -69,7 +70,7 @@ class TestCtl(unittest.TestCase): @patch('patroni.psycopg.connect', psycopg_connect) def test_get_cursor(self): with click.Context(click.Command('query')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), None, {}, role=role)) self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role=role)) @@ -107,7 +108,7 @@ class TestCtl(unittest.TestCase): def test_output_members(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at)) del cluster.members[1].data['conn_url'] @@ -250,7 +251,7 @@ class TestCtl(unittest.TestCase): @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) @@ -439,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)) @@ -448,7 +449,7 @@ class TestCtl(unittest.TestCase): def test_get_all_members(self): with click.Context(click.Command('list')) as ctx: - ctx.obj = {'__config': {}} + ctx.obj = {'__config': {}, '__mpp': get_mpp({})} for role in self.TEST_ROLES: self.assertEqual(list(get_all_members(get_cluster_initialized_without_leader(), None, role=role)), []) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 874aac5c..d7a423be 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -5,8 +5,10 @@ import unittest from dns.exception import DNSException from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.etcd import AbstractDCS, EtcdClient, Cluster, Etcd, EtcdError, DnsCachingResolver from patroni.exceptions import DCSError +from patroni.postgresql.mpp import get_mpp from patroni.utils import Retry from urllib3.exceptions import ReadTimeoutError @@ -138,8 +140,9 @@ class TestClient(unittest.TestCase): @patch.object(EtcdClient, '_get_machines_list', Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])) def setUp(self): - self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3, - 'srv': 'test', 'scope': 'test', 'name': 'foo'}) + self.etcd = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3, + 'etcd': {'srv': 'test'}, 'scope': 'test', 'name': 'foo'}) + self.assertIsInstance(self.etcd, Etcd) self.client = self.etcd._client self.client.http.request = http_request self.client.http.request_encode_body = http_request @@ -235,7 +238,7 @@ class TestEtcd(unittest.TestCase): Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])) def setUp(self): self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, - 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}) + 'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}, get_mpp({})) def test_base_path(self): self.assertEqual(self.etcd._base_path, '/patroni/test') @@ -270,7 +273,7 @@ class TestEtcd(unittest.TestCase): self.assertRaises(EtcdError, self.etcd.get_cluster) def test__get_citus_cluster(self): - self.etcd._citus_group = '0' + self.etcd._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.etcd.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 10ab1ea5..fcfd4e4b 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -4,10 +4,12 @@ import unittest import urllib3 from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.etcd import DnsCachingResolver from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \ Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \ Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode +from patroni.postgresql.mpp import get_mpp from threading import Thread from . import SleepException, MockResponse @@ -80,9 +82,9 @@ class BaseTestEtcd3(unittest.TestCase): @patch.object(Thread, 'start', Mock()) @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen) def setUp(self): - self.etcd3 = Etcd3({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, - 'host': 'localhost:2378', 'scope': 'test', 'name': 'foo', - 'username': 'etcduser', 'password': 'etcdpassword'}) + self.etcd3 = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, 'name': 'foo', 'scope': 'test', + 'etcd3': {'host': 'localhost:2378', 'username': 'etcduser', 'password': 'etcdpassword'}}) + self.assertIsInstance(self.etcd3, Etcd3) self.client = self.etcd3._client self.kv_cache = self.client._kv_cache @@ -236,7 +238,7 @@ class TestEtcd3(BaseTestEtcd3): self.assertRaises(Etcd3Error, self.etcd3.get_cluster) def test__get_citus_cluster(self): - self.etcd3._citus_group = '0' + self.etcd3._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.etcd3.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py index a908e1fc..5a72eb21 100644 --- a/tests/test_exhibitor.py +++ b/tests/test_exhibitor.py @@ -2,6 +2,7 @@ import unittest import urllib3 from mock import Mock, patch +from patroni.dcs import get_dcs from patroni.dcs.exhibitor import ExhibitorEnsembleProvider, Exhibitor from patroni.dcs.zookeeper import ZooKeeperError @@ -26,8 +27,9 @@ class TestExhibitor(unittest.TestCase): status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}'))) @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient) def setUp(self): - self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test', - 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + self.e = get_dcs({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, + 'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + self.assertIsInstance(self.e, Exhibitor) @patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True)) @patch.object(MockKazooClient, 'get_children', Mock(side_effect=Exception)) diff --git a/tests/test_ha.py b/tests/test_ha.py index 5b1d4562..45f64164 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -197,7 +197,7 @@ def run_async(self, func, args=()): @patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False)) @patch('patroni.async_executor.AsyncExecutor.run_async', run_async) @patch('patroni.postgresql.rewind.Thread', Mock()) -@patch('patroni.postgresql.citus.CitusHandler.start', Mock()) +@patch('patroni.postgresql.mpp.citus.CitusHandler.start', Mock()) @patch('subprocess.call', Mock(return_value=0)) @patch('time.sleep', Mock()) class TestHa(PostgresInit): @@ -593,8 +593,8 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.bootstrap(), 'failed to acquire initialize lock') @patch('patroni.psycopg.connect', psycopg_connect) - @patch('patroni.postgresql.citus.connect', psycopg_connect) - @patch('patroni.postgresql.citus.quote_ident', Mock()) + @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect) + @patch('patroni.postgresql.mpp.citus.quote_ident', Mock()) @patch.object(Postgresql, 'connection', Mock(return_value=None)) def test_bootstrap_initialized_new_cluster(self): self.ha.cluster = get_cluster_not_initialized_without_leader() @@ -615,8 +615,8 @@ class TestHa(PostgresInit): self.assertRaises(PatroniFatalException, self.ha.post_bootstrap) @patch('patroni.psycopg.connect', psycopg_connect) - @patch('patroni.postgresql.citus.connect', psycopg_connect) - @patch('patroni.postgresql.citus.quote_ident', Mock()) + @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect) + @patch('patroni.postgresql.mpp.citus.quote_ident', Mock()) @patch.object(Postgresql, 'connection', Mock(return_value=None)) def test_bootstrap_release_initialize_key_on_watchdog_failure(self): self.ha.cluster = get_cluster_not_initialized_without_leader() @@ -659,7 +659,7 @@ class TestHa(PostgresInit): @patch.object(ConfigHandler, 'replace_pg_hba', Mock()) @patch.object(ConfigHandler, 'replace_pg_ident', Mock()) @patch.object(PostmasterProcess, 'start', Mock(return_value=MockPostmaster())) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_worker_restart(self): self.ha.has_lock = true self.ha.patroni.request = Mock() @@ -694,7 +694,7 @@ class TestHa(PostgresInit): self.ha.is_paused = true self.assertEqual(self.ha.run_cycle(), 'PAUSE: restart in progress') - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_manual_failover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -733,7 +733,7 @@ class TestHa(PostgresInit): ('Member %s exceeds maximum replication lag', 'b')) self.ha.cluster.members.pop() - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_manual_switchover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -774,7 +774,7 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock') self.assertEqual(mock_info.call_args_list[0][0], ('Member %s exceeds maximum replication lag', 'leader')) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_scheduled_switchover_from_leader(self): self.ha.has_lock = true # I am the leader @@ -1544,7 +1544,7 @@ class TestHa(PostgresInit): self.ha.is_failover_possible = true self.ha.shutdown() - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_shutdown_citus_worker(self): self.ha.is_leader = true self.p.is_running = Mock(side_effect=[Mock(), False]) @@ -1656,7 +1656,7 @@ class TestHa(PostgresInit): self.assertRaises(DCSError, self.ha.acquire_lock) self.assertFalse(self.ha.acquire_lock()) - @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False)) + @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False)) def test_notify_citus_coordinator(self): self.ha.patroni.request = Mock() self.ha.notify_citus_coordinator('before_demote') diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index b6db7fb3..c493d799 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -8,9 +8,11 @@ import unittest import urllib3 from mock import Mock, PropertyMock, mock_open, patch +from patroni.dcs import get_dcs from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \ K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \ Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME +from patroni.postgresql.mpp import get_mpp from threading import Thread from . import MockResponse, SleepException @@ -225,11 +227,12 @@ class BaseTestKubernetes(unittest.TestCase): @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod, create=True) @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True) def setUp(self, config=None): - config = config or {} - config.update(ttl=30, scope='test', name='p-0', loop_wait=10, group=0, - retry_timeout=10, labels={'f': 'b'}, bypass_api_service=True) - self.k = Kubernetes(config) - self.k._citus_group = None + config = {'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10, + 'kubernetes': {'labels': {'f': 'b'}, 'bypass_api_service': True, **(config or {})}, + 'citus': {'group': 0, 'database': 'postgres'}} + self.k = get_dcs(config) + self.assertIsInstance(self.k, Kubernetes) + self.k._mpp = get_mpp({}) self.assertRaises(AttributeError, self.k._pods._build_cache) self.k._pods._is_ready = True self.assertRaises(TypeError, self.k._kinds._build_cache) @@ -254,7 +257,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.assertRaises(KubernetesError, self.k.get_cluster) def test__get_citus_cluster(self): - self.k._citus_group = '0' + self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) cluster = self.k.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) @@ -466,7 +469,7 @@ class TestCacheBuilder(BaseTestKubernetes): @patch('patroni.dcs.kubernetes.ObjectCache._watch', mock_watch) @patch.object(urllib3.HTTPResponse, 'read_chunked') def test__build_cache(self, mock_read_chunked): - self.k._citus_group = '0' + self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) mock_read_chunked.return_value = [json.dumps( {'type': 'MODIFIED', 'object': {'metadata': { 'name': self.k.config_path, 'resourceVersion': '2', 'annotations': {self.k._CONFIG: 'foo'}}}} diff --git a/tests/test_mpp.py b/tests/test_mpp.py new file mode 100644 index 00000000..9eb87633 --- /dev/null +++ b/tests/test_mpp.py @@ -0,0 +1,52 @@ +from typing import Any +from patroni.exceptions import PatroniException +from patroni.postgresql.mpp import AbstractMPP, get_mpp, Null + +from . import BaseTestPostgresql +from .test_ha import get_cluster_initialized_with_leader + + +class TestMPP(BaseTestPostgresql): + + def setUp(self): + super(TestMPP, self).setUp() + self.cluster = get_cluster_initialized_with_leader() + + def test_get_handler_impl_exception(self): + class DummyMPP(AbstractMPP): + def __init__(self) -> None: + super().__init__({}) + + @staticmethod + def validate_config(config: Any) -> bool: + return True + + @property + def group(self) -> None: + return None + + @property + def coordinator_group_id(self) -> None: + return None + + @property + def type(self) -> str: + return "dummy" + + mpp = DummyMPP() + self.assertRaises(PatroniException, mpp.get_handler_impl, self.p) + + def test_null_handler(self): + config = {} + mpp = get_mpp(config) + self.assertIsInstance(mpp, Null) + self.assertIsNone(mpp.group) + self.assertTrue(mpp.validate_config(config)) + nullHandler = mpp.get_handler_impl(self.p) + self.assertIsNone(nullHandler.handle_event(self.cluster, {})) + self.assertIsNone(nullHandler.sync_meta_data(self.cluster)) + self.assertIsNone(nullHandler.on_demote()) + self.assertIsNone(nullHandler.schedule_cache_rebuild()) + self.assertIsNone(nullHandler.bootstrap()) + self.assertIsNone(nullHandler.adjust_postgres_gucs({})) + self.assertFalse(nullHandler.ignore_replication_slot({})) diff --git a/tests/test_raft.py b/tests/test_raft.py index 9bb109e9..387a5a56 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -4,8 +4,10 @@ import tempfile import time from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \ Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport +from patroni.postgresql.mpp import get_mpp from pysyncobj import SyncObjConf, FAIL_REASON @@ -128,9 +130,10 @@ class TestRaft(unittest.TestCase): _TMP = tempfile.gettempdir() def test_raft(self): - raft = Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'self_addr': '127.0.0.1:1234', - 'retry_timeout': 10, 'data_dir': self._TMP, - 'database': 'citus', 'group': 0}) + raft = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'retry_timeout': 10, + 'raft': {'self_addr': '127.0.0.1:1234', 'data_dir': self._TMP}, + 'citus': {'group': 0, 'database': 'postgres'}}) + self.assertIsInstance(raft, Raft) raft.reload_config({'retry_timeout': 20, 'ttl': 60, 'loop_wait': 10}) self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}')) self.assertTrue(raft.touch_member('')) @@ -139,9 +142,9 @@ class TestRaft(unittest.TestCase): self.assertTrue(raft.set_config_value('{}')) self.assertTrue(raft.write_sync_state('foo', 'bar')) self.assertFalse(raft.write_sync_state('foo', 'bar', 1)) - raft._citus_group = '1' + raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}}) self.assertTrue(raft.manual_failover('foo', 'bar')) - raft._citus_group = '0' + raft._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) self.assertTrue(raft.take_leader()) cluster = raft.get_cluster() self.assertIsInstance(cluster, Cluster) @@ -157,9 +160,9 @@ class TestRaft(unittest.TestCase): self.assertTrue(raft.delete_sync_state()) self.assertTrue(raft.set_history_value('')) self.assertTrue(raft.delete_cluster()) - raft._citus_group = '1' + raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}}) self.assertTrue(raft.delete_cluster()) - raft._citus_group = None + raft._mpp = get_mpp({}) raft.get_cluster() raft.watch(None, 0.001) raft._sync_obj.destroy() @@ -175,5 +178,5 @@ class TestRaft(unittest.TestCase): def test_init(self, mock_event, mock_kvstore): mock_kvstore.return_value.applied_local_log = False mock_event.return_value.is_set.side_effect = [False, True] - self.assertIsNotNone(Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True, - 'self_addr': '1', 'data_dir': self._TMP})) + self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True, + 'raft': {'self_addr': '1', 'data_dir': self._TMP}}), Raft) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 3ce3ea75..3cd03467 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -7,8 +7,10 @@ from kazoo.handlers.threading import SequentialThreadingHandler from kazoo.protocol.states import KeeperState, WatchedEvent, ZnodeStat from kazoo.retry import RetryFailedError from mock import Mock, PropertyMock, patch +from patroni.dcs import get_dcs from patroni.dcs.zookeeper import Cluster, PatroniKazooClient, \ PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError +from patroni.postgresql.mpp import get_mpp class MockKazooClient(Mock): @@ -148,9 +150,9 @@ class TestZooKeeper(unittest.TestCase): @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient) def setUp(self): - self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test', - 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10, - 'set_acls': {'CN=principal2': ['ALL']}}) + self.zk = get_dcs({'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10, + 'zookeeper': {'hosts': ['localhost:2181'], 'set_acls': {'CN=principal2': ['ALL']}}}) + self.assertIsInstance(self.zk, ZooKeeper) def test_reload_config(self): self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10}) @@ -177,7 +179,7 @@ class TestZooKeeper(unittest.TestCase): self.assertEqual(cluster.last_lsn, 500) def test__get_citus_cluster(self): - self.zk._citus_group = '0' + self.zk._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) for _ in range(0, 2): cluster = self.zk.get_cluster() self.assertIsInstance(cluster, Cluster)