diff --git a/patroni/api.py b/patroni/api.py index d738fb34..d9136d24 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -1156,6 +1156,14 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_POST_citus(self) -> None: """Handle a ``POST`` request to ``/citus`` path. + .. note:: + We keep this entrypoint for backward compatibility and simply dispatch the request to :meth:`do_POST_mpp`. + """ + self.do_POST_mpp() + + def do_POST_mpp(self) -> None: + """Handle a ``POST`` request to ``/mpp`` path. + Call :func:`~patroni.postgresql.mpp.AbstractMPPHandler.handle_event` to handle the request, then write a response with HTTP status code ``200``. @@ -1167,9 +1175,9 @@ class RestApiHandler(BaseHTTPRequestHandler): return patroni = self.server.patroni - if patroni.postgresql.citus_handler.is_coordinator() and patroni.ha.is_leader(): + if patroni.postgresql.mpp_handler.is_coordinator() and patroni.ha.is_leader(): cluster = patroni.dcs.get_cluster() - patroni.postgresql.citus_handler.handle_event(cluster, request) + patroni.postgresql.mpp_handler.handle_event(cluster, request) self.write_response(200, 'OK') def parse_request(self) -> bool: diff --git a/patroni/ctl.py b/patroni/ctl.py index 274c2250..87204a64 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -346,7 +346,7 @@ def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: try: dcs = _get_dcs(config) if is_citus_cluster() and group is None: - dcs.is_citus_coordinator = lambda: True + dcs.is_mpp_coordinator = lambda: True click.get_current_context().obj['__mpp'] = dcs.mpp return dcs except PatroniException as e: diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index a210615b..481a8e6c 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -782,7 +782,7 @@ class Cluster(NamedTuple('Cluster', ('history', Optional[TimelineHistory]), ('failsafe', Optional[Dict[str, str]]), ('workers', Dict[int, 'Cluster'])])): - """Immutable object (namedtuple) which represents PostgreSQL or Citus cluster. + """Immutable object (namedtuple) which represents PostgreSQL or MPP cluster. .. note:: We are using an old-style attribute declaration here because otherwise it is not possible to override `__new__` @@ -799,8 +799,8 @@ class Cluster(NamedTuple('Cluster', :ivar sync: reference to :class:`SyncState` object, last observed synchronous replication state. :ivar history: reference to `TimelineHistory` object. :ivar failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list. - :ivar workers: dictionary of workers of the Citus cluster, optional. Each key is an :class:`int` representing - the group, and the corresponding value is a :class:`Cluster` instance. + :ivar workers: dictionary of workers of the MPP cluster, optional. Each key representing the group and the + corresponding value is a :class:`Cluster` instance. """ def __new__(cls, *args: Any, **kwargs: Any): @@ -1263,11 +1263,11 @@ class AbstractDCS(abc.ABC): Functional methods that are critical in their timing, required to complete within ``retry_timeout`` period in order to prevent the DCS considered inaccessible, each perform construction of complex data objects: - * :meth:`~AbstractDCS._cluster_loader`: + * :meth:`~AbstractDCS._postgresql_cluster_loader`: method which processes the structure of data stored in the DCS used to build the :class:`Cluster` object with all relevant associated data. - * :meth:`~AbstractDCS._citus_cluster_loader`: - Similar to above but specifically representing Citus group and workers information. + * :meth:`~AbstractDCS._mpp_cluster_loader`: + Similar to above but specifically representing MPP group and workers information. * :meth:`~AbstractDCS._load_cluster`: main method for calling specific ``loader`` method to build the :class:`Cluster` object representing the state and topology of the cluster. @@ -1337,7 +1337,7 @@ class AbstractDCS(abc.ABC): _FAILSAFE = 'failsafe' def __init__(self, config: Dict[str, Any], mpp: 'AbstractMPP') -> None: - """Prepare DCS paths, Citus group ID, initial values for state information and processing dependencies. + """Prepare DCS paths, MPP object, 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... @@ -1472,22 +1472,21 @@ class AbstractDCS(abc.ABC): return self._last_seen @abc.abstractmethod - def _cluster_loader(self, path: Any) -> Cluster: - """Load and build the :class:`Cluster` object from DCS, which represents a single Patroni or Citus cluster. + def _postgresql_cluster_loader(self, path: Any) -> Cluster: + """Load and build the :class:`Cluster` object from DCS, which represents a single PostgreSQL cluster. - :param path: the path in DCS where to load Cluster(s) from. + :param path: the path in DCS where to load :class:`Cluster` from. :returns: :class:`Cluster` instance. """ @abc.abstractmethod - def _citus_cluster_loader(self, path: Any) -> Dict[int, Cluster]: - """Load and build all Patroni clusters from a single Citus cluster. + def _mpp_cluster_loader(self, path: Any) -> Dict[int, Cluster]: + """Load and build all PostgreSQL clusters from a single MPP cluster. :param path: the path in DCS where to load Cluster(s) from. - :returns: all Citus groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values or a - :class:`Cluster` object representing the coordinator with filled `Cluster.workers` attribute. + :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values. """ @abc.abstractmethod @@ -1502,13 +1501,14 @@ class AbstractDCS(abc.ABC): the :meth:`~AbstractDCS.get_cluster` method. :param path: the path in DCS where to load Cluster(s) from. - :param loader: one of :meth:`~AbstractDCS._cluster_loader` or :meth:`~AbstractDCS._citus_cluster_loader`. + :param loader: one of :meth:`~AbstractDCS._postgresql_cluster_loader` or + :meth:`~AbstractDCS._mpp_cluster_loader`. :raise: :exc:`~DCSError` in case of communication problems with DCS. If the current node was running as a primary and exception raised, instance would be demoted. """ - def __get_patroni_cluster(self, path: Optional[str] = None) -> Cluster: + def __get_postgresql_cluster(self, path: Optional[str] = None) -> Cluster: """Low level method to load a :class:`Cluster` object from DCS. :param path: optional client path in DCS backend to load from. @@ -1517,39 +1517,40 @@ class AbstractDCS(abc.ABC): """ if path is None: path = self.client_path('') - cluster = self._load_cluster(path, self._cluster_loader) + cluster = self._load_cluster(path, self._postgresql_cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(cluster, Cluster) return cluster - def is_citus_coordinator(self) -> bool: - """:class:`Cluster` instance has a Citus Coordinator group ID. + def is_mpp_coordinator(self) -> bool: + """:class:`Cluster` instance has a Coordinator group ID. :returns: ``True`` if the given node is running as the MPP Coordinator. """ return self._mpp.is_coordinator() - def get_citus_coordinator(self) -> Optional[Cluster]: - """Load the Patroni cluster for the Citus Coordinator. + def get_mpp_coordinator(self) -> Optional[Cluster]: + """Load the PostgreSQL cluster for the MPP Coordinator. - .. note:: - This method is only executed on the worker nodes (``group!=0``) to find the coordinator. + .. note:: + This method is only executed on the worker nodes to find the coordinator. :returns: Select :class:`Cluster` instance associated with the MPP Coordinator group ID. """ try: - return self.__get_patroni_cluster(f'{self._base_path}/{self._mpp.coordinator_group_id}/') + return self.__get_postgresql_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) + logger.error('Failed to load %s coordinator cluster from %s: %r', + self._mpp.type, self.__class__.__name__, e) return None - def _get_citus_cluster(self) -> Cluster: - """Load Citus cluster from DCS. + def _get_mpp_cluster(self) -> Cluster: + """Load MPP cluster from DCS. - :returns: A Citus :class:`Cluster` instance for the coordinator with workers clusters in the `Cluster.workers` + :returns: A MPP :class:`Cluster` instance for the coordinator with workers clusters in the `Cluster.workers` dict. """ - groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader) + groups = self._load_cluster(self._base_path + '/', self._mpp_cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(groups, dict) cluster = groups.pop(self._mpp.coordinator_group_id, Cluster.empty()) @@ -1563,12 +1564,12 @@ class AbstractDCS(abc.ABC): Stores copy of time, status and failsafe values for comparison in DCS update decisions. Caching is required to avoid overhead placed upon the REST API. - Returns either a Citus or Patroni implementation of :class:`Cluster` depending on availability. + Returns either a PostgreSQL or MPP implementation of :class:`Cluster` depending on availability. :returns: """ try: - cluster = self._get_citus_cluster() if self.is_citus_coordinator() else self.__get_patroni_cluster() + cluster = self._get_mpp_cluster() if self.is_mpp_coordinator() else self.__get_postgresql_cluster() except Exception: self.reset_cluster() raise diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 19d65306..bbfd2453 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -420,7 +420,13 @@ class Consul(AbstractDCS): def _consistency(self) -> str: return 'consistent' if self._ctl else self._client.consistency - def _cluster_loader(self, path: str) -> Cluster: + def _postgresql_cluster_loader(self, path: str) -> Cluster: + """Load and build the :class:`Cluster` object from DCS, which represents a single PostgreSQL cluster. + + :param path: the path in DCS where to load :class:`Cluster` from. + + :returns: :class:`Cluster` instance. + """ _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency) if results is None: return Cluster.empty() @@ -431,7 +437,13 @@ class Consul(AbstractDCS): return self._cluster_from_nodes(nodes) - def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: + def _mpp_cluster_loader(self, path: str) -> Dict[int, Cluster]: + """Load and build all PostgreSQL clusters from a single MPP cluster. + + :param path: the path in DCS where to load Cluster(s) from. + + :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values. + """ _, results = self.retry(self._client.kv.get, path, recurse=True, consistency=self._consistency) clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict) for node in results or []: diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index b9d3e0ab..0b6c6dd3 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -710,7 +710,13 @@ class Etcd(AbstractEtcd): return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe) - def _cluster_loader(self, path: str) -> Cluster: + def _postgresql_cluster_loader(self, path: str) -> Cluster: + """Load and build the :class:`Cluster` object from DCS, which represents a single PostgreSQL cluster. + + :param path: the path in DCS where to load :class:`Cluster` from. + + :returns: :class:`Cluster` instance. + """ try: result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl) except etcd.EtcdKeyNotFound: @@ -718,7 +724,13 @@ class Etcd(AbstractEtcd): nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves} return self._cluster_from_nodes(result.etcd_index, nodes) - def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: + def _mpp_cluster_loader(self, path: str) -> Dict[int, Cluster]: + """Load and build all PostgreSQL clusters from a single MPP cluster. + + :param path: the path in DCS where to load Cluster(s) from. + + :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values. + """ try: result = self.retry(self._client.read, path, recursive=True, quorum=self._ctl) except etcd.EtcdKeyNotFound: diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 7cc2a115..946f9318 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -733,7 +733,11 @@ class Etcd3(AbstractEtcd): @property def cluster_prefix(self) -> str: - return self._base_path + '/' if self.is_citus_coordinator() else self.client_path('') + """Construct the cluster prefix for the cluster. + + :returns: path in the DCS under which we store information about this Patroni cluster. + """ + return self._base_path + '/' if self.is_mpp_coordinator() else self.client_path('') @staticmethod def member(node: Dict[str, str]) -> Member: @@ -787,13 +791,25 @@ class Etcd3(AbstractEtcd): return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe) - def _cluster_loader(self, path: str) -> Cluster: + def _postgresql_cluster_loader(self, path: str) -> Cluster: + """Load and build the :class:`Cluster` object from DCS, which represents a single PostgreSQL cluster. + + :param path: the path in DCS where to load :class:`Cluster` from. + + :returns: :class:`Cluster` instance. + """ nodes = {node['key'][len(path):]: node for node in self._client.get_cluster(path) if node['key'].startswith(path)} return self._cluster_from_nodes(nodes) - def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: + def _mpp_cluster_loader(self, path: str) -> Dict[int, Cluster]: + """Load and build all PostgreSQL clusters from a single MPP cluster. + + :param path: the path in DCS where to load Cluster(s) from. + + :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values. + """ clusters: Dict[int, Dict[str, Dict[str, Any]]] = defaultdict(dict) path = self._base_path + '/' for node in self._client.get_cluster(path): diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 343b496c..4b32130d 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -746,8 +746,6 @@ class ObjectCache(Thread): class Kubernetes(AbstractDCS): - _CITUS_LABEL = 'citus-group' - 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'] @@ -761,7 +759,7 @@ class Kubernetes(AbstractDCS): self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME super(Kubernetes, self).__init__({**config, 'namespace': ''}, mpp) if self._mpp.is_enabled(): - self._labels[self._CITUS_LABEL] = str(self._mpp.group) + self._labels[self._mpp.k8s_group_label] = str(self._mpp.group) self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1, retry_exceptions=KubernetesRetriableException) @@ -936,19 +934,31 @@ class Kubernetes(AbstractDCS): return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe) - def _cluster_loader(self, path: Dict[str, Any]) -> Cluster: + def _postgresql_cluster_loader(self, path: Dict[str, Any]) -> Cluster: + """Load and build the :class:`Cluster` object from DCS, which represents a single PostgreSQL cluster. + + :param path: the path in DCS where to load :class:`Cluster` from. + + :returns: :class:`Cluster` instance. + """ return self._cluster_from_nodes(path['group'], path['nodes'], path['pods'].values()) - def _citus_cluster_loader(self, path: Dict[str, Any]) -> Dict[int, Cluster]: + def _mpp_cluster_loader(self, path: Dict[str, Any]) -> Dict[int, Cluster]: + """Load and build all PostgreSQL clusters from a single MPP cluster. + + :param path: the path in DCS where to load Cluster(s) from. + + :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values. + """ clusters: Dict[str, Dict[str, Dict[str, K8sObject]]] = defaultdict(lambda: defaultdict(dict)) for name, pod in path['pods'].items(): - group = pod.metadata.labels.get(self._CITUS_LABEL) + group = pod.metadata.labels.get(self._mpp.k8s_group_label) 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) + group = kind.metadata.labels.get(self._mpp.k8s_group_label) 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()) @@ -965,9 +975,9 @@ class Kubernetes(AbstractDCS): with self._condition: self._wait_caches(stop_time) pods = {name: pod for name, pod in self._pods.copy().items() - if not group or pod.metadata.labels.get(self._CITUS_LABEL) == group} + if not group or pod.metadata.labels.get(self._mpp.k8s_group_label) == group} nodes = {name: kind for name, kind in self._kinds.copy().items() - if not group or kind.metadata.labels.get(self._CITUS_LABEL) == group} + if not group or kind.metadata.labels.get(self._mpp.k8s_group_label) == group} return loader({'group': group, 'pods': pods, 'nodes': nodes}) except Exception: logger.exception('get_cluster') @@ -979,14 +989,21 @@ class Kubernetes(AbstractDCS): 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]: + def get_mpp_coordinator(self) -> Optional[Cluster]: + """Load the PostgreSQL cluster for the MPP Coordinator. + + .. note:: + This method is only executed on the worker nodes to find the coordinator. + + :returns: Select :class:`Cluster` instance associated with the MPP Coordinator group ID. + """ try: - ret = self.__load_cluster(str(self._mpp.coordinator_group_id), self._cluster_loader) + ret = self.__load_cluster(str(self._mpp.coordinator_group_id), self._postgresql_cluster_loader) if TYPE_CHECKING: # pragma: no cover assert isinstance(ret, Cluster) return ret except Exception as e: - logger.error('Failed to load Citus coordinator cluster from Kubernetes: %r', e) + logger.error('Failed to load %s coordinator cluster from Kubernetes: %r', self._mpp.type, e) @staticmethod def compare_ports(p1: K8sObject, p2: K8sObject) -> bool: diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index 0528cfb0..9965611e 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -375,14 +375,26 @@ class Raft(AbstractDCS): return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe) - def _cluster_loader(self, path: str) -> Cluster: + def _postgresql_cluster_loader(self, path: str) -> Cluster: + """Load and build the :class:`Cluster` object from DCS, which represents a single PostgreSQL cluster. + + :param path: the path in DCS where to load :class:`Cluster` from. + + :returns: :class:`Cluster` instance. + """ response = self._sync_obj.get(path, recursive=True) if not response: return Cluster.empty() nodes = {key[len(path):]: value for key, value in response.items()} return self._cluster_from_nodes(nodes) - def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: + def _mpp_cluster_loader(self, path: str) -> Dict[int, Cluster]: + """Load and build all PostgreSQL clusters from a single MPP cluster. + + :param path: the path in DCS where to load Cluster(s) from. + + :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values. + """ clusters: Dict[int, Dict[str, Any]] = defaultdict(dict) response = self._sync_obj.get(path, recursive=True) for key, value in (response or {}).items(): diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 6bf77ae4..00c8e4bd 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -214,7 +214,13 @@ class ZooKeeper(AbstractDCS): members.append(self.member(member, *data)) return members - def _cluster_loader(self, path: str) -> Cluster: + def _postgresql_cluster_loader(self, path: str) -> Cluster: + """Load and build the :class:`Cluster` object from DCS, which represents a single PostgreSQL cluster. + + :param path: the path in DCS where to load :class:`Cluster` from. + + :returns: :class:`Cluster` instance. + """ nodes = set(self.get_children(path)) # get initialize flag @@ -258,11 +264,17 @@ class ZooKeeper(AbstractDCS): return Cluster(initialize, config, leader, status, members, failover, sync, history, failsafe) - def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]: + def _mpp_cluster_loader(self, path: str) -> Dict[int, Cluster]: + """Load and build all PostgreSQL clusters from a single MPP cluster. + + :param path: the path in DCS where to load Cluster(s) from. + + :returns: all MPP groups as :class:`dict`, with group IDs as keys and :class:`Cluster` objects as values. + """ ret: Dict[int, Cluster] = {} for node in self.get_children(path): if self._mpp.group_re.match(node): - ret[int(node)] = self._cluster_loader(path + node + '/') + ret[int(node)] = self._postgresql_cluster_loader(path + node + '/') return ret def _load_cluster( diff --git a/patroni/ha.py b/patroni/ha.py index 80e877c8..e763905c 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -175,7 +175,7 @@ class Ha(object): # Count of concurrent sync disabling requests. Value above zero means that we don't want to be synchronous # standby. Changes protected by _member_state_lock. self._disable_sync = 0 - # Remember the last known member role and state written to the DCS in order to notify Citus coordinator + # Remember the last known member role and state written to the DCS in order to notify MPP coordinator self._last_state = None # We need following property to avoid shutdown of postgres when join of Patroni to the postgres @@ -326,20 +326,26 @@ class Ha(object): tags['nosync'] = True return tags - def notify_citus_coordinator(self, event: str) -> None: - if self.state_handler.citus_handler.is_worker(): - coordinator = self.dcs.get_citus_coordinator() + def notify_mpp_coordinator(self, event: str) -> None: + """Send an event to the MPP coordinator. + + :param event: the type of event for coordinator to parse. + """ + mpp_handler = self.state_handler.mpp_handler + if mpp_handler.is_worker(): + coordinator = self.dcs.get_mpp_coordinator() if coordinator and coordinator.leader and coordinator.leader.conn_url: try: data = {'type': event, - 'group': self.state_handler.citus_handler.group, + 'group': mpp_handler.group, 'leader': self.state_handler.name, 'timeout': self.dcs.ttl, 'cooldown': self.patroni.config['retry_timeout']} timeout = self.dcs.ttl if event == 'before_demote' else 2 - self.patroni.request(coordinator.leader.member, 'post', 'citus', data, timeout=timeout, retries=0) + endpoint = 'citus' if mpp_handler.type == 'Citus' else 'mpp' + self.patroni.request(coordinator.leader.member, 'post', endpoint, data, timeout=timeout, retries=0) except Exception as e: - logger.warning('Request to Citus coordinator leader %s %s failed: %r', + logger.warning('Request to %s coordinator leader %s %s failed: %r', mpp_handler.type, coordinator.leader.name, coordinator.leader.member.api_url, e) def touch_member(self) -> bool: @@ -403,7 +409,7 @@ class Ha(object): if ret: new_state = (data['state'], {'master': 'primary'}.get(data['role'], data['role'])) if self._last_state != new_state and new_state == ('running', 'primary'): - self.notify_citus_coordinator('after_promote') + self.notify_mpp_coordinator('after_promote') self._last_state = new_state return ret @@ -848,7 +854,7 @@ class Ha(object): self.state_handler.set_role('master') self.process_sync_replication() self.update_cluster_history() - self.state_handler.citus_handler.sync_meta_data(self.cluster) + self.state_handler.mpp_handler.sync_meta_data(self.cluster) return message elif self.state_handler.role in ('master', 'promoted', 'primary'): self.process_sync_replication() @@ -868,7 +874,7 @@ class Ha(object): self._failsafe.set_is_active(0) def before_promote(): - self.notify_citus_coordinator('before_promote') + self.notify_mpp_coordinator('before_promote') with self._async_response: self._async_response.reset() @@ -1239,10 +1245,10 @@ class Ha(object): status['released'] = True def before_shutdown() -> None: - if self.state_handler.citus_handler.is_coordinator(): - self.state_handler.citus_handler.on_demote() + if self.state_handler.mpp_handler.is_coordinator(): + self.state_handler.mpp_handler.on_demote() else: - self.notify_citus_coordinator('before_demote') + self.notify_mpp_coordinator('before_demote') self.state_handler.stop(str(mode_control['stop']), checkpoint=bool(mode_control['checkpoint']), on_safepoint=self.watchdog.disable if self.watchdog.is_running else None, @@ -1544,10 +1550,10 @@ class Ha(object): self.set_start_timeout(timeout) def before_shutdown() -> None: - self.notify_citus_coordinator('before_demote') + self.notify_mpp_coordinator('before_demote') def after_start() -> None: - self.notify_citus_coordinator('after_promote') + self.notify_mpp_coordinator('after_promote') # For non async cases we want to wait for restart to complete or timeout before returning. do_restart = functools.partial(self.state_handler.restart, timeout, self._async_executor.critical_task, @@ -2000,7 +2006,7 @@ class Ha(object): self.dcs.write_leader_optime(prev_location) def _before_shutdown() -> None: - self.notify_citus_coordinator('before_demote') + self.notify_mpp_coordinator('before_demote') on_shutdown = _on_shutdown if self.is_leader() else None before_shutdown = _before_shutdown if self.is_leader() else None diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 8fc8c8fb..059c8034 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -80,7 +80,7 @@ class Postgresql(object): self._pending_restart_reason = CaseInsensitiveDict() self.connection_pool = ConnectionPool() self._connection = self.connection_pool.get('heartbeat') - self.citus_handler = mpp.get_handler_impl(self) + self.mpp_handler = mpp.get_handler_impl(self) self.config = ConfigHandler(self, config) self.config.check_directories() @@ -1208,7 +1208,7 @@ class Postgresql(object): before_promote() self.slots_handler.on_promote() - self.citus_handler.schedule_cache_rebuild() + self.mpp_handler.schedule_cache_rebuild() ret = self.pg_ctl('promote', '-W') if ret: @@ -1355,7 +1355,7 @@ class Postgresql(object): """ self.ensure_major_version_is_known() self.slots_handler.schedule() - self.citus_handler.schedule_cache_rebuild() + self.mpp_handler.schedule_cache_rebuild() self._sysid = '' def _get_gucs(self) -> CaseInsensitiveSet: diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py index b9b913d6..d112dbe6 100644 --- a/patroni/postgresql/bootstrap.py +++ b/patroni/postgresql/bootstrap.py @@ -471,8 +471,8 @@ END;$$""".format(f, quote_ident(rewind['username'], postgresql.connection())) time.sleep(1) # give a time to postgres to "reload" configuration files postgresql.connection().close() # close connection to reconnect with a new password else: # initdb - # We may want create database and extension for citus - self._postgresql.citus_handler.bootstrap() + # We may want create database and extension for some MPP clusters + self._postgresql.mpp_handler.bootstrap() except Exception: logger.exception('post_bootstrap') task.complete(False) diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index e5187b3d..8e08b5e7 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -992,7 +992,7 @@ class ConfigHandler(object): wal_keep_size = parse_int(parameters.pop('wal_keep_size', self.CMDLINE_OPTIONS['wal_keep_size'][0]), 'MB') parameters.setdefault('wal_keep_segments', int(((wal_keep_size or 0) + 8) / 16)) - self._postgresql.citus_handler.adjust_postgres_gucs(parameters) + self._postgresql.mpp_handler.adjust_postgres_gucs(parameters) ret = CaseInsensitiveDict({k: v for k, v in parameters.items() if not self._postgresql.major_version or self._postgresql.major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 90100))[2]}) diff --git a/patroni/postgresql/mpp/__init__.py b/patroni/postgresql/mpp/__init__.py index 3494793b..bb180d8a 100644 --- a/patroni/postgresql/mpp/__init__.py +++ b/patroni/postgresql/mpp/__init__.py @@ -65,6 +65,25 @@ class AbstractMPP(abc.ABC): def coordinator_group_id(self) -> Any: """The group id of the coordinator PostgreSQL cluster.""" + @property + def type(self) -> str: + """The type of the MPP cluster. + + :returns: A string representation of the type of a given MPP implementation. + """ + for base in self.__class__.__bases__: + if not base.__name__.startswith('Abstract'): + return base.__name__ + return self.__class__.__name__ + + @property + def k8s_group_label(self): + """Group label used for kubernetes DCS of the MPP cluster. + + :returns: A string representation of the k8s group label of a given MPP implementation. + """ + return self.type.lower() + '-group' + def is_coordinator(self) -> bool: """Check whether this node is running in the coordinator PostgreSQL cluster. diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index 7f7fd294..6e00d6d0 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -302,7 +302,7 @@ class SlotsHandler: for a in ('database', 'plugin', 'type')) ): return True - return self._postgresql.citus_handler.ignore_replication_slot(slot) + return self._postgresql.mpp_handler.ignore_replication_slot(slot) def drop_replication_slot(self, name: str) -> Tuple[bool, bool]: """Drop a named slot from Postgres. diff --git a/tests/test_api.py b/tests/test_api.py index ae47f51e..678d896c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -61,7 +61,7 @@ class MockPostgresql: wal_flush = '_flush' POSTMASTER_START_TIME = 'pg_catalog.pg_postmaster_start_time()' TL_LSN = 'CASE WHEN pg_catalog.pg_is_in_recovery()' - citus_handler = Mock() + mpp_handler = Mock() @staticmethod def postmaster_start_time(): @@ -675,6 +675,12 @@ class TestRestApiHandler(unittest.TestCase): MockRestApiServer(RestApiHandler, post + '0\n\n') MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}') + @patch.object(MockHa, 'is_leader', Mock(return_value=True)) + def test_do_POST_mpp(self): + post = 'POST /mpp HTTP/1.0' + self._authorization + '\nContent-Length: ' + MockRestApiServer(RestApiHandler, post + '0\n\n') + MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}') + class TestRestApiServer(unittest.TestCase): diff --git a/tests/test_citus.py b/tests/test_citus.py index 3d78668f..7cf26e2c 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -13,7 +13,7 @@ class TestCitus(BaseTestPostgresql): def setUp(self): super(TestCitus, self).setUp() - self.c = self.p.citus_handler + self.c = self.p.mpp_handler self.cluster = get_cluster_initialized_with_leader() self.cluster.workers[1] = self.cluster diff --git a/tests/test_ha.py b/tests/test_ha.py index 45f64164..ef315685 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -1659,12 +1659,13 @@ class TestHa(PostgresInit): @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') + self.ha.notify_mpp_coordinator('before_demote') self.ha.patroni.request.assert_called_once() self.assertEqual(self.ha.patroni.request.call_args[1]['timeout'], 30) self.ha.patroni.request = Mock(side_effect=Exception) with patch('patroni.ha.logger.warning') as mock_logger: - self.ha.notify_citus_coordinator('before_promote') + self.ha.notify_mpp_coordinator('before_promote') self.assertEqual(self.ha.patroni.request.call_args[1]['timeout'], 2) mock_logger.assert_called() - self.assertTrue(mock_logger.call_args[0][0].startswith('Request to Citus coordinator')) + self.assertTrue(mock_logger.call_args[0][0].startswith('Request to %s coordinator leader')) + self.assertEqual(mock_logger.call_args[0][1], 'Citus') diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index c493d799..235d24b2 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -18,6 +18,7 @@ from . import MockResponse, SleepException def mock_list_namespaced_config_map(*args, **kwargs): + k8s_group_label = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}).k8s_group_label metadata = {'resource_version': '1', 'labels': {'f': 'b'}, 'name': 'test-config', 'annotations': {'initialize': '123', 'config': '{}'}} items = [k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))] @@ -28,16 +29,16 @@ def mock_list_namespaced_config_map(*args, **kwargs): items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) metadata.update({'name': 'test-sync', 'annotations': {'leader': 'p-0'}}) items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) - metadata.update({'name': 'test-0-leader', 'labels': {Kubernetes._CITUS_LABEL: '0'}, + metadata.update({'name': 'test-0-leader', 'labels': {k8s_group_label: '0'}, 'annotations': {'optime': '1234x', 'leader': 'p-0', 'ttl': '30s', 'slots': '{', 'failsafe': '{'}}) items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) - metadata.update({'name': 'test-0-config', 'labels': {Kubernetes._CITUS_LABEL: '0'}, + metadata.update({'name': 'test-0-config', 'labels': {k8s_group_label: '0'}, 'annotations': {'initialize': '123', 'config': '{}'}}) items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) - metadata.update({'name': 'test-1-leader', 'labels': {Kubernetes._CITUS_LABEL: '1'}, + metadata.update({'name': 'test-1-leader', 'labels': {k8s_group_label: '1'}, 'annotations': {'leader': 'p-3', 'ttl': '30s'}}) items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) - metadata.update({'name': 'test-2-config', 'labels': {Kubernetes._CITUS_LABEL: '2'}, 'annotations': {}}) + metadata.update({'name': 'test-2-config', 'labels': {k8s_group_label: '2'}, 'annotations': {}}) items.append(k8s_client.V1ConfigMap(metadata=k8s_client.V1ObjectMeta(**metadata))) metadata = k8s_client.V1ObjectMeta(resource_version='1') @@ -62,7 +63,8 @@ def mock_list_namespaced_endpoints(*args, **kwargs): def mock_list_namespaced_pod(*args, **kwargs): - metadata = k8s_client.V1ObjectMeta(resource_version='1', labels={'f': 'b', Kubernetes._CITUS_LABEL: '1'}, + k8s_group_label = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}).k8s_group_label + metadata = k8s_client.V1ObjectMeta(resource_version='1', labels={'f': 'b', k8s_group_label: '1'}, name='p-0', annotations={'status': '{}'}, uid='964dfeae-e79b-4476-8a5a-1920b5c2a69d') status = k8s_client.V1PodStatus(pod_ip='10.0.0.1') @@ -263,12 +265,25 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.assertIsInstance(cluster.workers[1], Cluster) @patch('patroni.dcs.kubernetes.logger.error') - def test_get_citus_coordinator(self, mock_logger): - self.assertIsInstance(self.k.get_citus_coordinator(), Cluster) - with patch.object(Kubernetes, '_cluster_loader', Mock(side_effect=Exception)): - self.assertIsNone(self.k.get_citus_coordinator()) + def test_get_mpp_coordinator(self, mock_logger): + self.assertIsInstance(self.k.get_mpp_coordinator(), Cluster) + with patch.object(Kubernetes, '_postgresql_cluster_loader', Mock(side_effect=Exception)): + self.assertIsNone(self.k.get_mpp_coordinator()) mock_logger.assert_called() - self.assertTrue(mock_logger.call_args[0][0].startswith('Failed to load Citus coordinator')) + self.assertEqual(mock_logger.call_args[0][0], 'Failed to load %s coordinator cluster from Kubernetes: %r') + self.assertEqual(mock_logger.call_args[0][1], 'Null') + self.assertIsInstance(mock_logger.call_args[0][2], KubernetesError) + + @patch('patroni.dcs.kubernetes.logger.error') + def test_get_citus_coordinator(self, mock_logger): + self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) + self.assertIsInstance(self.k.get_mpp_coordinator(), Cluster) + with patch.object(Kubernetes, '_postgresql_cluster_loader', Mock(side_effect=Exception)): + self.assertIsNone(self.k.get_mpp_coordinator()) + mock_logger.assert_called() + self.assertEqual(mock_logger.call_args[0][0], 'Failed to load %s coordinator cluster from Kubernetes: %r') + self.assertEqual(mock_logger.call_args[0][1], 'Citus') + self.assertIsInstance(mock_logger.call_args[0][2], KubernetesError) def test_attempt_to_acquire_leader(self): with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', create=True) as mock_patch: diff --git a/tests/test_raft.py b/tests/test_raft.py index 387a5a56..b25196f6 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -156,7 +156,7 @@ class TestRaft(unittest.TestCase): self.assertTrue(raft.update_leader(leader, '1', failsafe={'foo': 'bat'})) self.assertTrue(raft._sync_obj.set(raft.failsafe_path, '{"foo"}')) self.assertTrue(raft._sync_obj.set(raft.status_path, '{')) - raft.get_citus_coordinator() + raft.get_mpp_coordinator() self.assertTrue(raft.delete_sync_state()) self.assertTrue(raft.set_history_value('')) self.assertTrue(raft.delete_cluster()) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 3cd03467..1bb6a23b 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -166,13 +166,13 @@ class TestZooKeeper(unittest.TestCase): def test__cluster_loader(self): self.zk._base_path = self.zk._base_path.replace('test', 'bla') - self.zk._cluster_loader(self.zk.client_path('')) + self.zk._postgresql_cluster_loader(self.zk.client_path('')) self.zk._base_path = self.zk._base_path = '/broken' - self.zk._cluster_loader(self.zk.client_path('')) + self.zk._postgresql_cluster_loader(self.zk.client_path('')) self.zk._base_path = self.zk._base_path = '/legacy' - self.zk._cluster_loader(self.zk.client_path('')) + self.zk._postgresql_cluster_loader(self.zk.client_path('')) self.zk._base_path = self.zk._base_path = '/no_node' - self.zk._cluster_loader(self.zk.client_path('')) + self.zk._postgresql_cluster_loader(self.zk.client_path('')) def test_get_cluster(self): cluster = self.zk.get_cluster() @@ -185,11 +185,28 @@ class TestZooKeeper(unittest.TestCase): self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) - @patch('patroni.dcs.zookeeper.logger.error') - @patch.object(ZooKeeper, '_cluster_loader', Mock(side_effect=Exception)) + @patch('patroni.dcs.logger.error') + def test_get_mpp_coordinator(self, mock_logger): + self.assertIsInstance(self.zk.get_mpp_coordinator(), Cluster) + with patch.object(ZooKeeper, '_postgresql_cluster_loader', Mock(side_effect=Exception)): + self.assertIsNone(self.zk.get_mpp_coordinator()) + mock_logger.assert_called_once() + self.assertEqual(mock_logger.call_args[0][0], 'Failed to load %s coordinator cluster from %s: %r') + self.assertEqual(mock_logger.call_args[0][1], 'Null') + self.assertEqual(mock_logger.call_args[0][2], 'ZooKeeper') + self.assertIsInstance(mock_logger.call_args[0][3], ZooKeeperError) + + @patch('patroni.dcs.logger.error') def test_get_citus_coordinator(self, mock_logger): - self.assertIsNone(self.zk.get_citus_coordinator()) - mock_logger.assert_called_once() + self.zk._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}}) + self.assertIsInstance(self.zk.get_mpp_coordinator(), Cluster) + with patch.object(ZooKeeper, '_postgresql_cluster_loader', Mock(side_effect=Exception)): + self.assertIsNone(self.zk.get_mpp_coordinator()) + mock_logger.assert_called_once() + self.assertEqual(mock_logger.call_args[0][0], 'Failed to load %s coordinator cluster from %s: %r') + self.assertEqual(mock_logger.call_args[0][1], 'Citus') + self.assertEqual(mock_logger.call_args[0][2], 'ZooKeeper') + self.assertIsInstance(mock_logger.call_args[0][3], ZooKeeperError) def test_delete_leader(self): self.assertTrue(self.zk.delete_leader(self.zk.get_cluster().leader))