[DPE-10062] Single kernel changes (#168)

Author: dragomirp

pyproject.toml

@@ -23,19 +23,25 @@ dependencies = [
 
 [project.optional-dependencies]
 postgresql = [
-    "psycopg2>=2.9.10",
     "httpx; python_version >= '3.12'",
     "data-platform-helpers>=0.1.7; python_version >= '3.12'",
     "charmlibs-pathops>=1.0.1; python_version >= '3.12'",
-    "charmlibs-snap>=1.0.1; python_version >= '3.12'",
     "charmlibs-interfaces-tls-certificates>=1.8.3; python_version >= '3.12'",
     "charm-refresh; python_version >= '3.12'",
     "requests; python_version >= '3.12'",
     "tomli; python_version >= '3.12'",
     "pydantic>=2.0; python_version >= '3.12'",
     "jinja2 >=3.1.6; python_version >= '3.12'",
     "charmlibs-rollingops>=1.1.1; python_version >= '3.12'",
- ]
+]
+vm = [
+    "pysyncobj>=0.3.15; python_version >= '3.12'",
+    "psutil>=7.2.2; python_version >= '3.12'",
+    "charmlibs-snap>=1.0.1; python_version >= '3.12'",
+]
+db-driver = [
+    "psycopg2>=2.9.10",
+]
 
 [build-system]
 requires = ["uv_build>=0.11.0,<0.12.0"]

single_kernel_postgresql/charms/abstract_charm.py

@@ -29,18 +29,10 @@ def __init__(self, *args):
         self.state = CharmState(charm=self, substrate=self.substrate)
 
         # Managers
-        self.cluster_manager = ClusterManager(
-            state=self.state, workload=self.workload, client=self.postgresql
-        )
-        self.tls_manager = TLSManager(
-            state=self.state, workload=self.workload, client=self.postgresql
-        )
-        self.config_manager = ConfigManager(
-            state=self.state, workload=self.workload, client=self.postgresql
-        )
-        self.patroni_manager = PatroniManager(
-            state=self.state, workload=self.workload, client=self.postgresql
-        )
+        self.tls_manager = TLSManager(state=self.state, workload=self.workload)
+        self.patroni_manager = PatroniManager(state=self.state, workload=self.workload)
+        self.cluster_manager = ClusterManager(state=self.state, workload=self.workload)
+        self.config_manager = ConfigManager(state=self.state, workload=self.workload)
 
         # Events Handler
         self.postgresql_events_handler = PostgreSQLEventsHandler(

single_kernel_postgresql/charms/k8s_charm.py

@@ -25,7 +25,7 @@ def __init__(self, *args):
         assert isinstance(self.workload, K8sWorkload), (  # noqa: S101
             "Workload must be an instance of K8sWorkload"
         )
-        self.k8s_manager = K8sManager(self.state, self.workload, self.postgresql)
+        self.k8s_manager = K8sManager(self.state, self.workload)
 
     @property
     def postgresql(self) -> PostgreSQL:

single_kernel_postgresql/config/exceptions.py

@@ -25,3 +25,51 @@ class PostgreSQLCannotConnectError(Exception):
 
 class TlsError(Exception):
     """TLS implementation internal exception."""
+
+
+class RaftPostgresqlNotUpError(Exception):
+    """Postgresql not yet started."""
+
+
+class RaftPostgresqlStillUpError(Exception):
+    """Postgresql not yet down."""
+
+
+class RaftNotPromotedError(Exception):
+    """Leader not yet set when reinitialising raft."""
+
+
+class ClusterNotPromotedError(Exception):
+    """Raised when a cluster is not promoted."""
+
+
+class NotReadyError(Exception):
+    """Raised when not all cluster members healthy or finished initial sync."""
+
+
+class EndpointNotReadyError(Exception):
+    """Raised when an endpoint is not ready."""
+
+
+class StandbyClusterAlreadyPromotedError(Exception):
+    """Raised when a standby cluster is already promoted."""
+
+
+class RemoveRaftMemberFailedError(Exception):
+    """Raised when a remove raft member failed for some reason."""
+
+
+class AddRaftMemberFailedError(Exception):
+    """Raised when adding raft member failed for some reason."""
+
+
+class SwitchoverFailedError(Exception):
+    """Raised when a switchover failed for some reason."""
+
+
+class SwitchoverNotSyncError(SwitchoverFailedError):
+    """Raised when a switchover failed because node is not sync."""
+
+
+class UpdateSyncNodeCountError(Exception):
+    """Raised when updating synchronous_node_count failed for some reason."""

single_kernel_postgresql/config/literals.py

@@ -31,6 +31,9 @@
 SNAP = "/snap/charmed-postgresql/current"
 VM_LOGS_PATH = "var/log/postgresql"
 VM_DATA_PATH = "var/lib/postgresql"
+VM_ARCHIVE_PATH = "data/archive"
+VM_DATA_LOGS_PATH = "data/logs"
+VM_TEMP_PATH = "data/temp"
 
 ## K8s Paths
 K8S_DATA_PATH = "var/lib/pg/data"
@@ -42,7 +45,9 @@
 # e.g. snap_current / "/etc/postgresql" will result in "/etc/postgresql" instead of "/var/snap/postgresql/current/etc/postgresql"
 POSTGRESQL_CONF_PATH = "etc/postgresql"
 POSTGRESQL_CONF_FILE = "postgresql.conf"
-
+## Pgbackreest Paths
+PGBACKREST_CONF_PATH = "etc/pgbackrest"
+PGBACKREST_CONF_FILE = "pgbackrest.conf"
 
 ## TLS Paths
 TLS_CA_BUNDLE_FILE = "peer_ca_bundle.pem"
@@ -161,3 +166,11 @@
 WATCHER_USER = "watcher"
 WATCHER_PASSWORD_KEY = "watcher-password"  # noqa: S105
 WATCHER_SECRET_LABEL = "watcher-secret"  # noqa: S105
+
+# Raft
+RAFT_PORT = 2222
+RAFT_PARTNER_PREFIX = "partner_node_status_server_"
+
+# VM services
+VM_PATRONI_SERVICE_NAME = "snap.charmed-postgresql.patroni.service"
+VM_PATRONI_SERVICE_DEFAULT_PATH = f"/etc/systemd/system/{VM_PATRONI_SERVICE_NAME}"

single_kernel_postgresql/core/peer_relation.py

@@ -6,6 +6,8 @@
 """State objects for database-peers relation."""
 
 import json
+from collections.abc import MutableMapping
+from functools import cached_property
 
 from ops import Application, BlockedStatus, Relation, Unit
 
@@ -177,6 +179,34 @@ def peer_addresses(self) -> set[str]:
             peer_addrs.add(addr)
         return peer_addrs
 
+    @property
+    def is_unit_departing(self) -> bool:
+        """Returns whether the unit is departing."""
+        if not self.relation:
+            return False
+        return "departing" in self.relation.data[self.unit]
+
+    @property
+    def is_unit_stopped(self) -> bool:
+        """Returns whether the unit is stopped."""
+        if not self.relation:
+            return False
+        return "stopped" in self.relation.data[self.unit]
+
+    @property
+    def is_connectivity_enabled(self) -> bool:
+        """Return whether this unit can be connected externally."""
+        if not self.relation:
+            return True
+        return self.relation.data[self.unit].get("connectivity", "on") == "on"
+
+    @cached_property
+    def data(self) -> MutableMapping[str, str]:
+        """Escape hatch method to access the peer data directly."""
+        if not self.relation:
+            return {}
+        return self.relation.data[self.unit]
+
 
 class PostgreSQLApplication(RelationState):
     """An PostgreSQL Application is the peer application state.
@@ -290,7 +320,7 @@ def cluster_name(self) -> str:
             return f"patroni-{self.app.name}"
         return self.app.name
 
-    @property
+    @cached_property
     def planned_units(self) -> int:
         """Get the number of planned units for the application."""
         return self.app.planned_units()
@@ -309,6 +339,39 @@ def endpoints(self) -> set[str]:
             return set()
         return set(json.loads(self.relation.data[self.app].get("endpoints", "[]")))
 
+    @property
+    def is_cluster_initialised(self) -> bool:
+        """Returns whether the cluster is already initialised."""
+        if not self.relation:
+            return False
+        return "cluster_initialised" in self.relation.data[self.app]
+
+    @property
+    def is_cluster_restoring_backup(self) -> bool:
+        """Returns whether the cluster is restoring a backup."""
+        if not self.relation:
+            return False
+        return "restoring-backup" in self.relation.data[self.app]
+
+    @property
+    def is_cluster_restoring_to_time(self) -> bool:
+        """Returns whether the cluster is restoring a backup to a specific time."""
+        if not self.relation:
+            return False
+        return "restore-to-time" in self.relation.data[self.app]
+
+    @property
+    def is_ldap_charm_related(self) -> bool:
+        """Return whether this unit has an LDAP charm related."""
+        if not self.relation:
+            return False
+        return self.relation.data[self.app].get("ldap_enabled", "False") == "True"
+
+    @property
+    def is_ldap_enabled(self) -> bool:
+        """Return whether this unit has LDAP enabled."""
+        return self.is_ldap_charm_related and self.is_cluster_initialised
+
     def get_secret(self, key: str) -> str | None:
         """Get the secret value for 'key' from the peer relation data."""
         if not self.relation:
@@ -326,3 +389,10 @@ def remove_secret(self, key: str) -> None:
         if not self.relation:
             return
         self.data_interface.delete_relation_data(self.relation.id, [key])
+
+    @cached_property
+    def data(self) -> MutableMapping[str, str]:
+        """Escape hatch method to access the peer data directly."""
+        if not self.relation:
+            return {}
+        return self.relation.data[self.app]

single_kernel_postgresql/core/state.py

@@ -6,17 +6,30 @@
 
 import re
 import socket
+from contextlib import suppress
 from functools import cached_property
 from typing import TYPE_CHECKING, Any, get_args
 
 from data_platform_helpers.advanced_statuses import StatusesState, StatusObject
 from data_platform_helpers.advanced_statuses.types import Scope as AdvancedStatusesScope
-from ops import ConfigData, JujuVersion, ModelError, Object, Relation, SecretNotFoundError, Unit
+from ops import (
+    ConfigData,
+    JujuVersion,
+    ModelError,
+    Object,
+    Relation,
+    RelationNotFoundError,
+    SecretNotFoundError,
+    Unit,
+)
 
 from single_kernel_postgresql.config.enums import Substrates
 from single_kernel_postgresql.config.literals import (
     APP_SCOPE,
+    DATABASE,
     PEER_RELATION,
+    REPLICATION_CONSUMER_RELATION,
+    REPLICATION_OFFER_RELATION,
     SCOPES,
     STATUS_PEERS_RELATION,
 )
@@ -153,6 +166,27 @@ def unit_ip(self) -> str | None:
         if binding := self.model.get_binding(PEER_RELATION):
             return str(binding.network.bind_address)
 
+    @property
+    def database_ip(self) -> str | None:
+        """Database endpoint address."""
+        with suppress(RelationNotFoundError):
+            if binding := self.model.get_binding(DATABASE):
+                return str(binding.network.bind_address)
+
+    @property
+    def replication_offer_ip(self) -> str | None:
+        """Async replication offer endpoint address."""
+        with suppress(RelationNotFoundError):
+            if binding := self.model.get_binding(REPLICATION_OFFER_RELATION):
+                return str(binding.network.bind_address)
+
+    @property
+    def replication_consumer_ip(self) -> str | None:
+        """Async replication consumer endpoint address."""
+        with suppress(RelationNotFoundError):
+            if binding := self.model.get_binding(REPLICATION_CONSUMER_RELATION):
+                return str(binding.network.bind_address)
+
     @property
     def fqdn(self) -> str | None:
         """Current unit fqdn."""
@@ -165,17 +199,19 @@ def fqdn(self) -> str | None:
     def endpoint(self) -> str | None:
         """Current unit endpoint."""
         if self.substrate == Substrates.K8S:
-            return self.fqdn
+            return f"{self.peer.unit.name.replace('/', '-')}.{self.application.app.name}-endpoints"
         else:
             return self.unit_ip
 
     @property
     def endpoints(self) -> set[str]:
         """Returns the list of endpoints of the current members of the cluster."""
         if self.peer_relation:
-            return self.application.endpoints
-        else:
-            return {self.endpoint} if self.endpoint else set()
+            if self.substrate == Substrates.K8S:
+                return self.application.endpoints
+            else:
+                return self.peer_members_ips
+        return {self.endpoint} if self.endpoint else set()
 
     @property
     def model_name(self) -> str:
@@ -185,7 +221,7 @@ def model_name(self) -> str:
     @cached_property
     def patroni_url(self) -> str:
         """Patroni REST API URL."""
-        return f"https://{self.unit_ip}:8007"
+        return f"https://{self.endpoint}:8008"
 
     @property
     def peer_members_ips(self) -> set[str]:
@@ -227,7 +263,12 @@ def listen_ips(self) -> list[str]:
         ips = []
         if self.unit_ip:
             ips.append(self.unit_ip)
-        # TODO: Add other ips
+        if self.database_ip and self.database_ip not in ips:
+            ips.append(self.database_ip)
+        if self.replication_offer_ip and self.replication_offer_ip not in ips:
+            ips.append(self.replication_offer_ip)
+        if self.replication_consumer_ip and self.replication_consumer_ip not in ips:
+            ips.append(self.replication_consumer_ip)
         return ips
 
     # -- Secrets
@@ -403,3 +444,38 @@ def _search_interpolated_status(
             if re.fullmatch(regex_pattern, present_status.message) is not None:
                 return present_status
         return None
+
+    @cached_property
+    def synchronous_node_count(self) -> int:
+        """Number of expected sync standbys."""
+        planned_units = self.application.planned_units
+        if self.config.synchronous_node_count == "all":
+            return planned_units - 1
+        elif self.config.synchronous_node_count == "majority":
+            return planned_units // 2
+        # -1 for leader
+        return (
+            self.config.synchronous_node_count
+            if self.config.synchronous_node_count < planned_units - 1
+            else planned_units - 1
+        )
+
+    @cached_property
+    def synchronous_configuration(self) -> dict[str, Any]:
+        """Synchronous mode configuration."""
+        # Try to update synchronous_node_count.
+        return {
+            "synchronous_node_count": self.synchronous_node_count,
+            "synchronous_mode_strict": len(self.application.members_ips) > 1
+            and self.config.synchronous_mode_strict
+            and self.synchronous_node_count > 0,
+        }
+
+    def _build_service_name(self, service: str) -> str:
+        """Build a full k8s service name based on the service name."""
+        return f"{self.application.app.name}-{service}.{self.model_name}.svc.cluster.local"
+
+    @property
+    def primary_endpoint(self) -> str:
+        """Returns the endpoint of the primary instance's service."""
+        return self._build_service_name("primary")