[DPE-7625] Add Logical Replication (#982)

* Migrated logical replication from VM.

* Add logical replication integration test.

* Lint fix.

* Add logical replication integration test.

* Improved saved resources info.

* Review suggestions.

* Fix pg_hba rules.

* Fix unit tests.

* Properly close psycopg connection.

* Review suggestions.

* Fix integration test. Fix bug with duplicating data.

* Include fix from @marceloneppel.

* Fix test_smoke.py

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

* Fix linting

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

* Try to make faster the check for all DBs down

Signed-off-by: Marcelo Henrique Neppel <[email protected]>

* Logical replication config description.

---------

Signed-off-by: Marcelo Henrique Neppel <[email protected]>
Co-authored-by: Alex Lutay <[email protected]>
Co-authored-by: Marcelo Henrique Neppel <[email protected]>

Author: 3 people

config.yaml

@@ -117,6 +117,13 @@ options:
       Enables tracking of function call counts and time used. Specify pl to track only procedural-language functions
     type: string
     default: "none"
+  logical_replication_subscription_request:
+    description: |
+      Set of databases corresponding to list of tables with schema notation in JSON format, which will be requested from
+      publiblisher cluster to subscribe on via logical replication.
+      Example: {"<database>": ["<schema>.<table>", ...], ...}
+    type: string
+    default: "{}"
   memory_maintenance_work_mem:
     description: |
       Sets the maximum memory (KB) to be used for maintenance operations.

lib/charms/postgresql_k8s/v0/postgresql.py

@@ -49,6 +49,9 @@ provides:
     interface: postgresql_async
     limit: 1
     optional: true
+  logical-replication-offer:
+    interface: postgresql_logical_replication
+    optional: true
   database:
     interface: postgresql_client
   metrics-endpoint:
@@ -61,6 +64,10 @@ requires:
     interface: postgresql_async
     limit: 1
     optional: true
+  logical-replication:
+    interface: postgresql_logical_replication
+    limit: 1
+    optional: true
   certificates:
     interface: tls-certificates
     limit: 1

metadata.yaml

@@ -35,6 +35,10 @@
     WORKLOAD_OS_USER,
 )
 from relations.async_replication import REPLICATION_CONSUMER_RELATION, REPLICATION_OFFER_RELATION
+from relations.logical_replication import (
+    LOGICAL_REPLICATION_OFFER_RELATION,
+    LOGICAL_REPLICATION_RELATION,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -1088,7 +1092,7 @@ def _fetch_backup_from_id(self, backup_id: str) -> str:
 
         return None
 
-    def _pre_restore_checks(self, event: ActionEvent) -> bool:
+    def _pre_restore_checks(self, event: ActionEvent) -> bool:  # noqa: C901
         """Run some checks before starting the restore.
 
         Returns:
@@ -1143,14 +1147,25 @@ def _pre_restore_checks(self, event: ActionEvent) -> bool:
             event.fail(error_message)
             return False
 
-        logger.info("Checking that the cluster is not replicating data to a standby cluster")
+        logger.info("Checking that cluster does not have an active async replication relation")
         for relation in [
             self.model.get_relation(REPLICATION_CONSUMER_RELATION),
             self.model.get_relation(REPLICATION_OFFER_RELATION),
         ]:
             if not relation:
                 continue
-            error_message = "Unit cannot restore backup as the cluster is replicating data to a standby cluster"
+            error_message = "Unit cannot restore backup with an active async replication relation"
+            logger.error(f"Restore failed: {error_message}")
+            event.fail(error_message)
+            return False
+
+        logger.info("Checking that cluster does not have an active logical replication relation")
+        if self.model.get_relation(LOGICAL_REPLICATION_RELATION) or len(
+            self.model.relations.get(LOGICAL_REPLICATION_OFFER_RELATION, ())
+        ):
+            error_message = (
+                "Unit cannot restore backup with an active logical replication connection"
+            )
             logger.error(f"Restore failed: {error_message}")
             event.fail(error_message)
             return False

src/backups.py

@@ -49,6 +49,7 @@
     PostgreSQL,
     PostgreSQLEnableDisableExtensionError,
     PostgreSQLGetCurrentTimelineError,
+    PostgreSQLListGroupsError,
     PostgreSQLUpdateUserPasswordError,
 )
 from charms.postgresql_k8s.v0.postgresql_tls import PostgreSQLTLS
@@ -134,6 +135,10 @@
     REPLICATION_OFFER_RELATION,
     PostgreSQLAsyncReplication,
 )
+from relations.logical_replication import (
+    LOGICAL_REPLICATION_VALIDATION_ERROR_STATUS,
+    PostgreSQLLogicalReplication,
+)
 from relations.postgresql_provider import PostgreSQLProvider
 from upgrade import PostgreSQLUpgrade, get_postgresql_k8s_dependencies_model
 from utils import any_cpu_to_cores, any_memory_to_bytes, new_password
@@ -250,6 +255,7 @@ def __init__(self, *args):
         self.ldap = PostgreSQLLDAP(self, "ldap")
         self.tls = PostgreSQLTLS(self, PEER, [self.primary_endpoint, self.replicas_endpoint])
         self.async_replication = PostgreSQLAsyncReplication(self)
+        self.logical_replication = PostgreSQLLogicalReplication(self)
         self.restart_manager = RollingOpsManager(
             charm=self, relation="restart", callback=self._restart
         )
@@ -709,7 +715,7 @@ def _on_secret_changed(self, event: SecretChangedEvent) -> None:
             except PostgreSQLUpdateUserPasswordError:
                 event.defer()
 
-    def _on_config_changed(self, event) -> None:
+    def _on_config_changed(self, event) -> None:  # noqa: C901
         """Handle configuration changes, like enabling plugins."""
         if not self.is_cluster_initialised:
             logger.debug("Defer on_config_changed: cluster not initialised yet")
@@ -744,6 +750,9 @@ def _on_config_changed(self, event) -> None:
         # Update the sync-standby endpoint in the async replication data.
         self.async_replication.update_async_replication_data()
 
+        if not self.logical_replication.apply_changed_config(event):
+            return
+
         if not self.unit.is_leader():
             return
 
@@ -1097,6 +1106,12 @@ def _set_active_status(self):
                     self.app_peer_data["s3-initialization-block-message"]
                 )
                 return
+            if self.unit.is_leader() and (
+                self.app_peer_data.get("logical-replication-validation") == "error"
+                or self.logical_replication.has_remote_publisher_errors()
+            ):
+                self.unit.status = BlockedStatus(LOGICAL_REPLICATION_VALIDATION_ERROR_STATUS)
+                return
             if (
                 self._patroni.get_primary(unit_name_pattern=True) == self.unit.name
                 or self.is_standby_leader
@@ -1488,7 +1503,9 @@ def _on_update_status_early_exit_checks(self, container) -> bool:
         self._check_pgdata_storage_size()
 
         if (
-            self._has_blocked_status and self.unit.status not in S3_BLOCK_MESSAGES
+            self._has_blocked_status
+            and self.unit.status not in S3_BLOCK_MESSAGES
+            and self.unit.status.message != LOGICAL_REPLICATION_VALIDATION_ERROR_STATUS
         ) or self._has_non_restore_waiting_status:
             # If charm was failing to disable plugin, try again and continue (user may have removed the objects)
             if self.unit.status.message == EXTENSION_OBJECT_MESSAGE:
@@ -1565,6 +1582,8 @@ def _on_update_status(self, _) -> None:
 
         self.backup.coordinate_stanza_fields()
 
+        self.logical_replication.retry_validations()
+
         self._set_active_status()
 
     def _was_restore_successful(self, container: Container, service: ServiceInfo) -> bool:
@@ -2078,6 +2097,8 @@ def update_config(self, is_creating_backup: bool = False) -> bool:
             self.model.config, available_memory, limit_memory
         )
 
+        replication_slots = self.logical_replication.replication_slots()
+
         logger.info("Updating Patroni config file")
         # Update and reload configuration based on TLS files availability.
         self._patroni.render_patroni_yml_file(
@@ -2094,6 +2115,7 @@ def update_config(self, is_creating_backup: bool = False) -> bool:
             restore_stanza=self.app_peer_data.get("restore-stanza"),
             parameters=postgresql_parameters,
             user_databases_map=self.relations_user_databases_map,
+            slots=replication_slots or None,
         )
 
         if not self._is_workload_running:
@@ -2133,6 +2155,8 @@ def update_config(self, is_creating_backup: bool = False) -> bool:
             "wal_keep_size": self.config.durability_wal_keep_size,
         })
 
+        self._patroni.ensure_slots_controller_by_patroni(replication_slots)
+
         self._handle_postgresql_restart_need()
         self._restart_metrics_service()
         self._restart_ldap_sync_service()
@@ -2323,22 +2347,30 @@ def client_relations(self) -> list[Relation]:
     @property
     def relations_user_databases_map(self) -> dict:
         """Returns a user->databases map for all relations."""
-        if (
-            not self.is_cluster_initialised
-            or not self._patroni.member_started
-            or self.postgresql.list_access_groups(current_host=self.is_connectivity_enabled)
-            != set(ACCESS_GROUPS)
-        ):
+        try:
+            if (
+                not self.is_cluster_initialised
+                or not self._patroni.member_started
+                or self.postgresql.list_access_groups(current_host=self.is_connectivity_enabled)
+                != set(ACCESS_GROUPS)
+            ):
+                return {USER: "all", REPLICATION_USER: "all", REWIND_USER: "all"}
+        except PostgreSQLListGroupsError as e:
+            logger.warning(f"Failed to list access groups: {e}")
             return {USER: "all", REPLICATION_USER: "all", REWIND_USER: "all"}
         user_database_map = {}
         for user in self.postgresql.list_users_from_relation(
             current_host=self.is_connectivity_enabled
         ):
-            user_database_map[user] = ",".join(
+            databases = ",".join(
                 self.postgresql.list_accessible_databases_for_user(
                     user, current_host=self.is_connectivity_enabled
                 )
             )
+            if databases:
+                user_database_map[user] = databases
+            else:
+                logger.debug(f"User {user} has no databases to connect to")
 
         # Copy relations users directly instead of waiting for them to be created
         for relation in self.model.relations[self.postgresql_client_relation.relation_name]:

src/charm.py

@@ -35,6 +35,7 @@ class CharmConfig(BaseConfigModel):
     logging_log_lock_waits: bool | None
     logging_log_min_duration_statement: int | None
     logging_track_functions: str | None
+    logical_replication_subscription_request: str | None
     memory_maintenance_work_mem: int | None
     memory_max_prepared_transactions: int | None
     memory_shared_buffers: int | None

src/config.py

@@ -456,6 +456,35 @@ def bulk_update_parameters_controller_by_patroni(self, parameters: dict[str, Any
             timeout=PATRONI_TIMEOUT,
         )
 
+    def ensure_slots_controller_by_patroni(self, slots: dict[str, str]) -> None:
+        """Synchronises slots controlled by Patroni with the provided state by removing unneeded slots and creating new ones.
+
+        Args:
+            slots: dictionary of slots in the {slot: database} format.
+        """
+        current_config = requests.get(
+            f"{self._patroni_url}/config",
+            verify=self._verify,
+            timeout=PATRONI_TIMEOUT,
+            auth=self._patroni_auth,
+        )
+        slots_patch: dict[str, dict[str, str] | None] = dict.fromkeys(
+            current_config.json().get("slots", ())
+        )
+        for slot, database in slots.items():
+            slots_patch[slot] = {
+                "database": database,
+                "plugin": "pgoutput",
+                "type": "logical",
+            }
+        requests.patch(
+            f"{self._patroni_url}/config",
+            verify=self._verify,
+            json={"slots": slots_patch},
+            auth=self._patroni_auth,
+            timeout=PATRONI_TIMEOUT,
+        )
+
     def promote_standby_cluster(self) -> None:
         """Promote a standby cluster to be a regular cluster."""
         config_response = requests.get(
@@ -526,6 +555,7 @@ def render_patroni_yml_file(
         restore_to_latest: bool = False,
         parameters: dict[str, str] | None = None,
         user_databases_map: dict[str, str] | None = None,
+        slots: dict[str, str] | None = None,
     ) -> None:
         """Render the Patroni configuration file.
 
@@ -545,6 +575,7 @@ def render_patroni_yml_file(
             restore_to_latest: restore all the WAL transaction logs from the stanza.
             parameters: PostgreSQL parameters to be added to the postgresql.conf file.
             user_databases_map: map of databases to be accessible by each user.
+            slots: replication slots (keys) with assigned database name (values).
         """
         # Open the template patroni.yml file.
         with open("templates/patroni.yml.j2") as file:
@@ -584,6 +615,7 @@ def render_patroni_yml_file(
             ldap_parameters=self._dict_to_hba_string(ldap_params),
             patroni_password=self._patroni_password,
             user_databases_map=user_databases_map,
+            slots=slots,
         )
         self._render_file(f"{self._storage_path}/patroni.yml", rendered, 0o644)