charm.py

#!/usr/bin/env -S LD_LIBRARY_PATH=lib python3
# Copyright 2021 Canonical Ltd.
# See LICENSE file for licensing details.

"""Charmed Kubernetes Operator for the PostgreSQL database."""

import itertools
import json
import logging
import os
import re
import shutil
import sys
import time
from pathlib import Path
from typing import Dict, List, Literal, Optional, Tuple, get_args

# First platform-specific import, will fail on wrong architecture
try:
    import psycopg2
except ModuleNotFoundError:
    from ops.main import main

    from arch_utils import WrongArchitectureWarningCharm, is_wrong_architecture

    # If the charm was deployed inside a host with different architecture
    # (possibly due to user specifying an incompatible revision)
    # then deploy an empty blocked charm with a warning.
    if is_wrong_architecture() and __name__ == "__main__":
        main(WrongArchitectureWarningCharm, use_juju_for_storage=True)
    raise

from charms.data_platform_libs.v0.data_interfaces import DataPeerData, DataPeerUnitData
from charms.data_platform_libs.v0.data_models import TypedCharmBase
from charms.grafana_k8s.v0.grafana_dashboard import GrafanaDashboardProvider
from charms.loki_k8s.v1.loki_push_api import LogForwarder, LogProxyConsumer
from charms.observability_libs.v1.kubernetes_service_patch import KubernetesServicePatch
from charms.postgresql_k8s.v0.postgresql import (
    REQUIRED_PLUGINS,
    PostgreSQL,
    PostgreSQLEnableDisableExtensionError,
    PostgreSQLGetCurrentTimelineError,
    PostgreSQLUpdateUserPasswordError,
)
from charms.postgresql_k8s.v0.postgresql_tls import PostgreSQLTLS
from charms.prometheus_k8s.v0.prometheus_scrape import MetricsEndpointProvider
from charms.rolling_ops.v0.rollingops import RollingOpsManager, RunWithLock
from charms.tempo_coordinator_k8s.v0.charm_tracing import trace_charm
from charms.tempo_coordinator_k8s.v0.tracing import TracingEndpointRequirer
from lightkube import ApiError, Client
from lightkube.models.core_v1 import ServicePort, ServiceSpec
from lightkube.models.meta_v1 import ObjectMeta
from lightkube.resources.core_v1 import Endpoints, Node, Pod, Service
from ops import main
from ops.charm import (
    ActionEvent,
    HookEvent,
    LeaderElectedEvent,
    RelationDepartedEvent,
    WorkloadEvent,
)
from ops.jujuversion import JujuVersion
from ops.model import (
    ActiveStatus,
    BlockedStatus,
    Container,
    MaintenanceStatus,
    ModelError,
    Relation,
    Unit,
    UnknownStatus,
    WaitingStatus,
)
from ops.pebble import (
    ChangeError,
    ExecError,
    Layer,
    PathError,
    ProtocolError,
    ServiceInfo,
    ServiceStatus,
)
from requests import ConnectionError
from tenacity import RetryError, Retrying, stop_after_attempt, stop_after_delay, wait_fixed

from backups import CANNOT_RESTORE_PITR, S3_BLOCK_MESSAGES, PostgreSQLBackups
from config import CharmConfig
from constants import (
    APP_SCOPE,
    BACKUP_USER,
    METRICS_PORT,
    MONITORING_PASSWORD_KEY,
    MONITORING_USER,
    PATRONI_PASSWORD_KEY,
    PEER,
    PLUGIN_OVERRIDES,
    POSTGRES_LOG_FILES,
    REPLICATION_PASSWORD_KEY,
    REPLICATION_USER,
    REWIND_PASSWORD_KEY,
    SECRET_DELETED_LABEL,
    SECRET_INTERNAL_LABEL,
    SECRET_KEY_OVERRIDES,
    SPI_MODULE,
    SYSTEM_USERS,
    TLS_CA_FILE,
    TLS_CERT_FILE,
    TLS_KEY_FILE,
    TRACING_PROTOCOL,
    TRACING_RELATION_NAME,
    UNIT_SCOPE,
    USER,
    USER_PASSWORD_KEY,
    WORKLOAD_OS_GROUP,
    WORKLOAD_OS_USER,
)
from patroni import NotReadyError, Patroni, SwitchoverFailedError
from relations.async_replication import (
    REPLICATION_CONSUMER_RELATION,
    REPLICATION_OFFER_RELATION,
    PostgreSQLAsyncReplication,
)
from relations.db import EXTENSIONS_BLOCKING_MESSAGE, DbProvides
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

logger = logging.getLogger(__name__)

EXTENSIONS_DEPENDENCY_MESSAGE = "Unsatisfied plugin dependencies. Please check the logs"
EXTENSION_OBJECT_MESSAGE = "Cannot disable plugins: Existing objects depend on it. See logs"
INSUFFICIENT_SIZE_WARNING = "<10% free space on pgdata volume."

ORIGINAL_PATRONI_ON_FAILURE_CONDITION = "restart"

# http{x,core} clutter the logs with debug messages
logging.getLogger("httpcore").setLevel(logging.ERROR)
logging.getLogger("httpx").setLevel(logging.ERROR)

Scopes = Literal[APP_SCOPE, UNIT_SCOPE]
PASSWORD_USERS = [*SYSTEM_USERS, "patroni"]


@trace_charm(
    tracing_endpoint="tracing_endpoint",
    extra_types=(
        DbProvides,
        GrafanaDashboardProvider,
        LogProxyConsumer,
        MetricsEndpointProvider,
        Patroni,
        PostgreSQL,
        PostgreSQLAsyncReplication,
        PostgreSQLBackups,
        PostgreSQLProvider,
        PostgreSQLTLS,
        PostgreSQLUpgrade,
        RollingOpsManager,
    ),
)
class PostgresqlOperatorCharm(TypedCharmBase[CharmConfig]):
    """Charmed Operator for the PostgreSQL database."""

    config_type = CharmConfig

    def __init__(self, *args):
        super().__init__(*args)

        # Support for disabling the operator.
        disable_file = Path(f"{os.environ.get('CHARM_DIR')}/disable")
        if disable_file.exists():
            logger.warning(
                f"\n\tDisable file `{disable_file.resolve()}` found, the charm will skip all events."
                "\n\tTo resume normal operations, please remove the file."
            )
            self.unit.status = BlockedStatus("Disabled")
            sys.exit(0)

        self.peer_relation_app = DataPeerData(
            self.model,
            relation_name=PEER,
            secret_field_name=SECRET_INTERNAL_LABEL,
            deleted_label=SECRET_DELETED_LABEL,
        )
        self.peer_relation_unit = DataPeerUnitData(
            self.model,
            relation_name=PEER,
            secret_field_name=SECRET_INTERNAL_LABEL,
            deleted_label=SECRET_DELETED_LABEL,
        )

        self._postgresql_service = "postgresql"
        self.rotate_logs_service = "rotate-logs"
        self.pgbackrest_server_service = "pgbackrest server"
        self._metrics_service = "metrics_server"
        self._unit = self.model.unit.name
        self._name = self.model.app.name
        self._namespace = self.model.name
        self._context = {"namespace": self._namespace, "app_name": self._name}
        self.cluster_name = f"patroni-{self._name}"

        self.framework.observe(self.on.config_changed, self._on_config_changed)
        self.framework.observe(self.on.leader_elected, self._on_leader_elected)
        self.framework.observe(self.on[PEER].relation_changed, self._on_peer_relation_changed)
        self.framework.observe(self.on.secret_changed, self._on_peer_relation_changed)
        self.framework.observe(self.on[PEER].relation_departed, self._on_peer_relation_departed)
        self.framework.observe(self.on.postgresql_pebble_ready, self._on_postgresql_pebble_ready)
        self.framework.observe(self.on.pgdata_storage_detaching, self._on_pgdata_storage_detaching)
        self.framework.observe(self.on.stop, self._on_stop)
        self.framework.observe(self.on.get_password_action, self._on_get_password)
        self.framework.observe(self.on.set_password_action, self._on_set_password)
        self.framework.observe(self.on.get_primary_action, self._on_get_primary)
        self.framework.observe(self.on.update_status, self._on_update_status)
        self._storage_path = self.meta.storages["pgdata"].location
        self.pgdata_path = f"{self._storage_path}/pgdata"

        self.upgrade = PostgreSQLUpgrade(
            self,
            model=get_postgresql_k8s_dependencies_model(),
            relation_name="upgrade",
            substrate="k8s",
        )
        self.framework.observe(self.on.upgrade_charm, self._on_upgrade_charm)
        self.postgresql_client_relation = PostgreSQLProvider(self)
        self.legacy_db_relation = DbProvides(self, admin=False)
        self.legacy_db_admin_relation = DbProvides(self, admin=True)
        self.backup = PostgreSQLBackups(self, "s3-parameters")
        self.tls = PostgreSQLTLS(self, PEER, [self.primary_endpoint, self.replicas_endpoint])
        self.async_replication = PostgreSQLAsyncReplication(self)
        self.restart_manager = RollingOpsManager(
            charm=self, relation="restart", callback=self._restart
        )
        self.grafana_dashboards = GrafanaDashboardProvider(self)
        self.metrics_endpoint = MetricsEndpointProvider(
            self,
            refresh_event=[self.on.start],
            jobs=self._generate_metrics_jobs(self.is_tls_enabled),
        )
        self.loki_push = (
            LogForwarder(
                self,
                relation_name="logging",
            )
            if self._pebble_log_forwarding_supported
            else LogProxyConsumer(
                self,
                logs_scheme={"postgresql": {"log-files": POSTGRES_LOG_FILES}},
                relation_name="logging",
            )
        )

        postgresql_db_port = ServicePort(5432, name="database")
        patroni_api_port = ServicePort(8008, name="api")
        self.service_patcher = KubernetesServicePatch(self, [postgresql_db_port, patroni_api_port])
        self.tracing = TracingEndpointRequirer(
            self, relation_name=TRACING_RELATION_NAME, protocols=[TRACING_PROTOCOL]
        )

    @property
    def tracing_endpoint(self) -> Optional[str]:
        """Otlp http endpoint for charm instrumentation."""
        if self.tracing.is_ready():
            return self.tracing.get_endpoint(TRACING_PROTOCOL)

    @property
    def _pebble_log_forwarding_supported(self) -> bool:
        # https://github.com/canonical/operator/issues/1230
        from ops.jujuversion import JujuVersion

        juju_version = JujuVersion.from_environ()
        return juju_version > JujuVersion(version=str("3.3"))

    def _generate_metrics_jobs(self, enable_tls: bool) -> Dict:
        """Generate spec for Prometheus scraping."""
        return [
            {"static_configs": [{"targets": [f"*:{METRICS_PORT}"]}]},
            {
                "static_configs": [{"targets": ["*:8008"]}],
                "scheme": "https" if enable_tls else "http",
                "tls_config": {"insecure_skip_verify": True},
            },
        ]

    @property
    def app_units(self) -> set[Unit]:
        """The peer-related units in the application."""
        if not self._peers:
            return set()

        return {self.unit, *self._peers.units}

    @property
    def app_peer_data(self) -> Dict:
        """Application peer relation data object."""
        relation = self.model.get_relation(PEER)
        if relation is None:
            return {}

        return relation.data[self.app]

    @property
    def unit_peer_data(self) -> Dict:
        """Unit peer relation data object."""
        relation = self.model.get_relation(PEER)
        if relation is None:
            return {}

        return relation.data[self.unit]

    def _peer_data(self, scope: Scopes) -> Dict:
        """Return corresponding databag for app/unit."""
        relation = self.model.get_relation(PEER)
        if relation is None:
            return {}

        return relation.data[self._scope_obj(scope)]

    def _scope_obj(self, scope: Scopes):
        if scope == APP_SCOPE:
            return self.app
        if scope == UNIT_SCOPE:
            return self.unit

    def peer_relation_data(self, scope: Scopes) -> DataPeerData:
        """Returns the peer relation data per scope."""
        if scope == APP_SCOPE:
            return self.peer_relation_app
        elif scope == UNIT_SCOPE:
            return self.peer_relation_unit

    def _translate_field_to_secret_key(self, key: str) -> str:
        """Change 'key' to secrets-compatible key field."""
        if not JujuVersion.from_environ().has_secrets:
            return key
        key = SECRET_KEY_OVERRIDES.get(key, key)
        new_key = key.replace("_", "-")
        return new_key.strip("-")

    def get_secret(self, scope: Scopes, key: str) -> Optional[str]:
        """Get secret from the secret storage."""
        if scope not in get_args(Scopes):
            raise RuntimeError("Unknown secret scope.")

        if not (peers := self.model.get_relation(PEER)):
            return None

        secret_key = self._translate_field_to_secret_key(key)
        # Old translation in databag is to be taken
        if result := self.peer_relation_data(scope).fetch_my_relation_field(peers.id, key):
            return result

        return self.peer_relation_data(scope).get_secret(peers.id, secret_key)

    def set_secret(self, scope: Scopes, key: str, value: Optional[str]) -> Optional[str]:
        """Set secret from the secret storage."""
        if scope not in get_args(Scopes):
            raise RuntimeError("Unknown secret scope.")

        if not value:
            return self.remove_secret(scope, key)

        if not (peers := self.model.get_relation(PEER)):
            return None

        secret_key = self._translate_field_to_secret_key(key)
        # Old translation in databag is to be deleted
        self.peer_relation_data(scope).delete_relation_data(peers.id, [key])
        self.peer_relation_data(scope).set_secret(peers.id, secret_key, value)

    def remove_secret(self, scope: Scopes, key: str) -> None:
        """Removing a secret."""
        if scope not in get_args(Scopes):
            raise RuntimeError("Unknown secret scope.")

        if not (peers := self.model.get_relation(PEER)):
            return None

        secret_key = self._translate_field_to_secret_key(key)
        if scope == APP_SCOPE:
            self.peer_relation_app.delete_relation_data(peers.id, [secret_key])
        else:
            self.peer_relation_unit.delete_relation_data(peers.id, [secret_key])

    @property
    def is_cluster_initialised(self) -> bool:
        """Returns whether the cluster is already initialised."""
        return "cluster_initialised" in self.app_peer_data

    @property
    def postgresql(self) -> PostgreSQL:
        """Returns an instance of the object used to interact with the database."""
        return PostgreSQL(
            primary_host=self.primary_endpoint,
            current_host=self.endpoint,
            user=USER,
            password=self.get_secret(APP_SCOPE, f"{USER}-password"),
            database="postgres",
            system_users=SYSTEM_USERS,
        )

    @property
    def endpoint(self) -> str:
        """Returns the endpoint of this instance's pod."""
        return f"{self._unit.replace('/', '-')}.{self._build_service_name('endpoints')}"

    @property
    def primary_endpoint(self) -> str:
        """Returns the endpoint of the primary instance's service."""
        return self._build_service_name("primary")

    @property
    def replicas_endpoint(self) -> str:
        """Returns the endpoint of the replicas instances' service."""
        return self._build_service_name("replicas")

    def _build_service_name(self, service: str) -> str:
        """Build a full k8s service name based on the service name."""
        return f"{self._name}-{service}.{self._namespace}.svc.cluster.local"

    def get_hostname_by_unit(self, unit_name: str) -> str:
        """Create a DNS name for a PostgreSQL unit.

        Args:
            unit_name: the juju unit name, e.g. "postgre-sql/1".

        Returns:
            A string representing the hostname of the PostgreSQL unit.
        """
        unit_id = unit_name.split("/")[1]
        return f"{self.app.name}-{unit_id}.{self.app.name}-endpoints"

    def _get_endpoints_to_remove(self) -> List[str]:
        """List the endpoints that were part of the cluster but departed."""
        old = self._endpoints
        current = [self._get_hostname_from_unit(member) for member in self._hosts]
        endpoints_to_remove = list(set(old) - set(current))
        return endpoints_to_remove

    def get_unit_ip(self, unit: Unit) -> Optional[str]:
        """Get the IP address of a specific unit."""
        # Check if host is current host.
        if unit == self.unit:
            return str(self.model.get_binding(PEER).network.bind_address)
        # Check if host is a peer.
        elif unit in self._peers.data:
            return str(self._peers.data[unit].get("private-address"))
        # Return None if the unit is not a peer neither the current unit.
        else:
            return None

    def _on_peer_relation_departed(self, event: RelationDepartedEvent) -> None:
        """The leader removes the departing units from the list of cluster members."""
        # Allow leader to update endpoints if it isn't leaving.
        if not self.unit.is_leader() or event.departing_unit == self.unit:
            return

        if "cluster_initialised" not in self._peers.data[self.app]:
            logger.debug(
                "Deferring on_peer_relation_departed: Cluster must be initialized before members can leave"
            )
            event.defer()
            return

        endpoints_to_remove = self._get_endpoints_to_remove()
        self.postgresql_client_relation.update_read_only_endpoint()
        self._remove_from_endpoints(endpoints_to_remove)

        # Update the sync-standby endpoint in the async replication data.
        self.async_replication.update_async_replication_data()

    def _on_pgdata_storage_detaching(self, _) -> None:
        # Change the primary if it's the unit that is being removed.
        try:
            primary = self._patroni.get_primary(unit_name_pattern=True)
        except RetryError:
            # Ignore the event if the primary couldn't be retrieved.
            # If a switchover is needed, an automatic failover will be triggered
            # when the unit is removed.
            logger.debug("Early exit on_pgdata_storage_detaching: primary cannot be retrieved")
            return

        if self.unit.name != primary:
            return

        if not self._patroni.are_all_members_ready():
            logger.warning(
                "could not switchover because not all members are ready"
                " - an automatic failover will be triggered"
            )
            return

        # Try to switchover to another member and raise an exception if it doesn't succeed.
        # If it doesn't happen on time, Patroni will automatically run a fail-over.
        try:
            # Get the current primary to check if it has changed later.
            current_primary = self._patroni.get_primary()

            # Trigger the switchover.
            self._patroni.switchover()

            # Wait for the switchover to complete.
            self._patroni.primary_changed(current_primary)

            logger.info("successful switchover")
        except (RetryError, SwitchoverFailedError) as e:
            logger.warning(
                f"switchover failed with reason: {e} - an automatic failover will be triggered"
            )
            return

        # Only update the connection endpoints if there is a primary.
        # A cluster can have all members as replicas for some time after
        # a failed switchover, so wait until the primary is elected.
        endpoints_to_remove = self._get_endpoints_to_remove()
        self.postgresql_client_relation.update_read_only_endpoint()
        self._remove_from_endpoints(endpoints_to_remove)

    def _on_peer_relation_changed(self, event: HookEvent) -> None:  # noqa: C901
        """Reconfigure cluster members."""
        # The cluster must be initialized first in the leader unit
        # before any other member joins the cluster.
        if "cluster_initialised" not in self._peers.data[self.app]:
            if self.unit.is_leader():
                if self._initialize_cluster(event):
                    logger.debug("Deferring on_peer_relation_changed: Leader initialized cluster")
                else:
                    logger.debug("_initialized_cluster failed on _peer_relation_changed")
                    return
            else:
                logger.debug(
                    "Deferring on_peer_relation_changed: Cluster must be initialized before members can join"
                )
            event.defer()
            return

        # If the leader is the one receiving the event, it adds the new members,
        # one at a time.
        if self.unit.is_leader():
            self._add_members(event)

        # Don't update this member before it's part of the members list.
        if self._endpoint not in self._endpoints:
            return

        # Update the list of the cluster members in the replicas to make them know each other.
        # Update the cluster members in this unit (updating patroni configuration).
        container = self.unit.get_container("postgresql")
        if not container.can_connect():
            logger.debug(
                "Early exit on_peer_relation_changed: Waiting for container to become available"
            )
            return
        try:
            self.update_config()
        except ValueError as e:
            self.unit.status = BlockedStatus("Configuration Error. Please check the logs")
            logger.error("Invalid configuration: %s", str(e))
            return

        # Should not override a blocked status
        if isinstance(self.unit.status, BlockedStatus):
            logger.debug("on_peer_relation_changed early exit: Unit in blocked status")
            return

        services = container.pebble.get_services(names=[self._postgresql_service])
        if (
            ("restoring-backup" in self.app_peer_data or "restore-to-time" in self.app_peer_data)
            and len(services) > 0
            and not self._was_restore_successful(container, services[0])
        ):
            logger.debug("on_peer_relation_changed early exit: Backup restore check failed")
            return

        # Validate the status of the member before setting an ActiveStatus.
        if not self._patroni.member_started:
            logger.debug("Deferring on_peer_relation_changed: Waiting for member to start")
            self.unit.status = WaitingStatus("awaiting for member to start")
            event.defer()
            return

        # Restart the workload if it's stuck on the starting state after a timeline divergence
        # due to a backup that was restored.
        if (
            not self.is_primary
            and not self.is_standby_leader
            and (
                self._patroni.member_replication_lag == "unknown"
                or int(self._patroni.member_replication_lag) > 1000
            )
        ):
            logger.warning("Workload failure detected. Reinitialising unit.")
            self.unit.status = MaintenanceStatus("reinitialising replica")
            self._patroni.reinitialize_postgresql()
            logger.debug("Deferring on_peer_relation_changed: reinitialising replica")
            event.defer()
            return

        try:
            self.postgresql_client_relation.update_read_only_endpoint()
        except ModelError as e:
            logger.warning("Cannot update read_only endpoints: %s", str(e))

        # Start or stop the pgBackRest TLS server service when TLS certificate change.
        if not self.backup.start_stop_pgbackrest_service():
            # Ping primary to start its TLS server.
            self.unit_peer_data.update({"start-tls-server": "True"})
            logger.debug(
                "Deferring on_peer_relation_changed: awaiting for TLS server service to start on primary"
            )
            event.defer()
            return
        else:
            self.unit_peer_data.pop("start-tls-server", None)

        self.backup.coordinate_stanza_fields()

        # This is intended to be executed only when leader is reinitializing S3 connection due to the leader change.
        if (
            "s3-initialization-start" in self.app_peer_data
            and "s3-initialization-done" not in self.unit_peer_data
            and self.is_primary
            and not self.backup._on_s3_credential_changed_primary(event)
        ):
            return

        # Clean-up unit initialization data after successful sync to the leader.
        if "s3-initialization-done" in self.app_peer_data and not self.unit.is_leader():
            self.unit_peer_data.update({
                "stanza": "",
                "s3-initialization-block-message": "",
                "s3-initialization-done": "",
                "s3-initialization-start": "",
            })

        self.async_replication.handle_read_only_mode()

    def _on_config_changed(self, event) -> None:
        """Handle configuration changes, like enabling plugins."""
        if not self.is_cluster_initialised:
            logger.debug("Defer on_config_changed: cluster not initialised yet")
            event.defer()
            return

        if not self.upgrade.idle:
            logger.debug("Defer on_config_changed: upgrade in progress")
            event.defer()
            return

        try:
            self._validate_config_options()
            # update config on every run
            self.update_config()
        except psycopg2.OperationalError:
            logger.debug("Defer on_config_changed: Cannot connect to database")
            event.defer()
            return
        except ValueError as e:
            self.unit.status = BlockedStatus("Configuration Error. Please check the logs")
            logger.error("Invalid configuration: %s", str(e))
            return

        if self.is_blocked and "Configuration Error" in self.unit.status.message:
            self._set_active_status()

        # Update the sync-standby endpoint in the async replication data.
        self.async_replication.update_async_replication_data()

        if not self.unit.is_leader():
            return

        # Enable and/or disable the extensions.
        self.enable_disable_extensions()

        # Unblock the charm after extensions are enabled (only if it's blocked due to application
        # charms requesting extensions).
        if self.unit.status.message != EXTENSIONS_BLOCKING_MESSAGE:
            return

        for relation in [
            *self.model.relations.get("db", []),
            *self.model.relations.get("db-admin", []),
        ]:
            if not self.legacy_db_relation.set_up_relation(relation):
                logger.debug(
                    "Early exit on_config_changed: legacy relation requested extensions that are still disabled"
                )
                return

    def enable_disable_extensions(self, database: str = None) -> None:
        """Enable/disable PostgreSQL extensions set through config options.

        Args:
            database: optional database where to enable/disable the extension.
        """
        if self._patroni.get_primary() is None:
            logger.debug("Early exit enable_disable_extensions: standby cluster")
            return
        original_status = self.unit.status
        extensions = {}
        # collect extensions
        for plugin in self.config.plugin_keys():
            enable = self.config[plugin]

            # Enable or disable the plugin/extension.
            extension = "_".join(plugin.split("_")[1:-1])
            if extension == "spi":
                for ext in SPI_MODULE:
                    extensions[ext] = enable
                continue
            extension = PLUGIN_OVERRIDES.get(extension, extension)
            if self._check_extension_dependencies(extension, enable):
                self.unit.status = BlockedStatus(EXTENSIONS_DEPENDENCY_MESSAGE)
                return
            extensions[extension] = enable
        if self.is_blocked and self.unit.status.message == EXTENSIONS_DEPENDENCY_MESSAGE:
            self._set_active_status()
            original_status = self.unit.status

        self._handle_enable_disable_extensions(original_status, extensions, database)

    def _handle_enable_disable_extensions(self, original_status, extensions, database) -> None:
        """Try enablind/disabling Postgresql extensions and handle exceptions appropriately."""
        if not isinstance(original_status, UnknownStatus):
            self.unit.status = WaitingStatus("Updating extensions")
        try:
            self.postgresql.enable_disable_extensions(extensions, database)
        except psycopg2.errors.DependentObjectsStillExist as e:
            logger.error(
                "Failed to disable plugin: %s\nWas the plugin enabled manually? If so, update charm config with `juju config postgresql-k8s plugin_<plugin_name>_enable=True`",
                str(e),
            )
            self.unit.status = BlockedStatus(EXTENSION_OBJECT_MESSAGE)
            return
        except PostgreSQLEnableDisableExtensionError as e:
            logger.exception("failed to change plugins: %s", str(e))
        if original_status.message == EXTENSION_OBJECT_MESSAGE:
            self._set_active_status()
            return
        if not isinstance(original_status, UnknownStatus):
            self.unit.status = original_status

    def _check_extension_dependencies(self, extension: str, enable: bool) -> bool:
        skip = False
        if enable and extension in REQUIRED_PLUGINS:
            for ext in REQUIRED_PLUGINS[extension]:
                if not self.config[f"plugin_{ext}_enable"]:
                    skip = True
                    logger.exception(
                        "cannot enable %s, extension required %s to be enabled before",
                        extension,
                        ext,
                    )
        return skip

    def _add_members(self, event) -> None:
        """Add new cluster members.

        This method is responsible for adding new members to the cluster
        when new units are added to the application. This event is deferred if
        one of the current units is copying data from the primary, to avoid
        multiple units copying data at the same time, which can cause slow
        transfer rates in these processes and overload the primary instance.
        """
        # Only the leader can reconfigure.
        if not self.unit.is_leader():
            return

        # Reconfiguration can be successful only if the cluster is initialised
        # (i.e. first unit has bootstrap the cluster).
        if "cluster_initialised" not in self._peers.data[self.app]:
            return

        try:
            # Compare set of Patroni cluster members and Juju hosts
            # to avoid the unnecessary reconfiguration.
            if self._patroni.cluster_members == self._hosts:
                return

            logger.info("Reconfiguring cluster")
            self.unit.status = MaintenanceStatus("reconfiguring cluster")
            for member in self._hosts - self._patroni.cluster_members:
                logger.debug("Adding %s to cluster", member)
                self.add_cluster_member(member)
        except NotReadyError:
            logger.info("Deferring reconfigure: another member doing sync right now")
            event.defer()
        except RetryError:
            logger.info("Deferring reconfigure: failed to obtain cluster members from Patroni")
            event.defer()

    def add_cluster_member(self, member: str) -> None:
        """Add member to the cluster if all members are already up and running.

        Raises:
            NotReadyError if either the new member or the current members are not ready.
        """
        hostname = self._get_hostname_from_unit(member)

        if not self._patroni.are_all_members_ready():
            logger.info("not all members are ready")
            raise NotReadyError("not all members are ready")

        # Add the member to the list that should be updated in each other member.
        self._add_to_endpoints(hostname)

        # Add the labels needed for replication in this pod.
        # This also enables the member as part of the cluster.
        try:
            self._patch_pod_labels(member)
        except ApiError as e:
            logger.error("failed to patch pod")
            self.unit.status = BlockedStatus(f"failed to patch pod with error {e}")
            return

    @property
    def _hosts(self) -> set:
        """List of the current Juju hosts.

        Returns:
            a set containing the current Juju hosts
                with the names in the k8s pod name format
        """
        peers = self.model.get_relation(PEER)
        hosts = [self._unit_name_to_pod_name(self.unit.name)] + [
            self._unit_name_to_pod_name(unit.name) for unit in peers.units
        ]
        return set(hosts)

    def _get_hostname_from_unit(self, member: str) -> str:
        """Create a DNS name for a PostgreSQL/Patroni cluster member.

        Args:
            member: the Patroni member name, e.g. "postgresql-k8s-0".

        Returns:
            A string representing the hostname of the PostgreSQL unit.
        """
        unit_id = member.split("-")[-1]
        return f"{self.app.name}-{unit_id}.{self.app.name}-endpoints"

    def _on_leader_elected(self, event: LeaderElectedEvent) -> None:
        """Handle the leader-elected event."""
        for password in {
            USER_PASSWORD_KEY,
            REPLICATION_PASSWORD_KEY,
            REWIND_PASSWORD_KEY,
            MONITORING_PASSWORD_KEY,
            PATRONI_PASSWORD_KEY,
        }:
            if self.get_secret(APP_SCOPE, password) is None:
                self.set_secret(APP_SCOPE, password, new_password())

        # Add this unit to the list of cluster members
        # (the cluster should start with only this member).
        if self._endpoint not in self._endpoints:
            self._add_to_endpoints(self._endpoint)

        self._cleanup_old_cluster_resources()

        if not self.fix_leader_annotation():
            return

        # Create resources and add labels needed for replication.
        if self.upgrade.idle:
            try:
                self._create_services()
            except ApiError:
                logger.exception("failed to create k8s services")
                self.unit.status = BlockedStatus("failed to create k8s services")
                return

        # Remove departing units when the leader changes.
        self._remove_from_endpoints(self._get_endpoints_to_remove())

        self._add_members(event)

    def fix_leader_annotation(self) -> bool:
        """Fix the leader annotation if it's missing."""
        client = Client()
        try:
            endpoint = client.get(Endpoints, name=self.cluster_name, namespace=self._namespace)
            if "leader" not in endpoint.metadata.annotations:
                patch = {
                    "metadata": {
                        "annotations": {"leader": self._unit_name_to_pod_name(self._unit)}
                    }
                }
                client.patch(
                    Endpoints, name=self.cluster_name, namespace=self._namespace, obj=patch
                )
                self.app_peer_data.pop("cluster_initialised", None)
                logger.info("Fixed missing leader annotation")
        except ApiError as e:
            if e.status.code == 403:
                self.on_deployed_without_trust()
                return False
            # Ignore the error only when the resource doesn't exist.
            if e.status.code != 404:
                raise e
        return True

    def _create_pgdata(self, container: Container):
        """Create the PostgreSQL data directory."""
        if not container.exists(self.pgdata_path):
            container.make_dir(
                self.pgdata_path, permissions=0o770, user=WORKLOAD_OS_USER, group=WORKLOAD_OS_GROUP
            )
        # Also, fix the permissions from the parent directory.
        container.exec([
            "chown",
            f"{WORKLOAD_OS_USER}:{WORKLOAD_OS_GROUP}",
            self._storage_path,
        ]).wait()

    def _on_postgresql_pebble_ready(self, event: WorkloadEvent) -> None:
        """Event handler for PostgreSQL container on PebbleReadyEvent."""
        if self._endpoint in self._endpoints:
            self._fix_pod()

        # TODO: move this code to an "_update_layer" method in order to also utilize it in
        # config-changed hook.
        # Get the postgresql container so we can configure/manipulate it.
        container = event.workload
        if not container.can_connect():
            logger.debug(
                "Defer on_postgresql_pebble_ready: Waiting for container to become available"
            )
            event.defer()
            return

        # Create the PostgreSQL data directory. This is needed on cloud environments
        # where the volume is mounted with more restrictive permissions.
        self._create_pgdata(container)

        self.unit.set_workload_version(self._patroni.rock_postgresql_version)

        # Defer the initialization of the workload in the replicas
        # if the cluster hasn't been bootstrap on the primary yet.
        # Otherwise, each unit will create a different cluster and
        # any update in the members list on the units won't have effect
        # on fixing that.
        if not self.unit.is_leader() and "cluster_initialised" not in self._peers.data[self.app]:
            logger.debug(
                "Deferring on_postgresql_pebble_ready: Not leader and cluster not initialized"
            )
            event.defer()
            return

        try:
            self.push_tls_files_to_workload(container)
        except (PathError, ProtocolError) as e:
            logger.error(
                "Deferring on_postgresql_pebble_ready: Cannot push TLS certificates: %r", e
            )
            event.defer()
            return

        # Start the database service.
        self._update_pebble_layers()

        # Ensure the member is up and running before marking the cluster as initialised.
        if not self._patroni.member_started:
            logger.debug("Deferring on_postgresql_pebble_ready: Waiting for cluster to start")
            self.unit.status = WaitingStatus("awaiting for cluster to start")
            event.defer()
            return

        if self.unit.is_leader() and not self._initialize_cluster(event):
            return

        # Update the archive command and replication configurations.
        self.update_config()

        # Enable/disable PostgreSQL extensions if they were set before the cluster
        # was fully initialised.
        self.enable_disable_extensions()

        # Enable pgbackrest service
        self.backup.start_stop_pgbackrest_service()

        # All is well, set an ActiveStatus.
        self._set_active_status()

    def _set_active_status(self):
        # The charm should not override this status outside of the function checking disk space.
        if self.unit.status.message == INSUFFICIENT_SIZE_WARNING:
            return
        try:
            if self.unit.is_leader() and "s3-initialization-block-message" in self.app_peer_data:
                self.unit.status = BlockedStatus(
                    self.app_peer_data["s3-initialization-block-message"]
                )
                return
            if self._patroni.get_primary(unit_name_pattern=True) == self.unit.name:
                self.unit.status = ActiveStatus("Primary")
            elif self.is_standby_leader:
                self.unit.status = ActiveStatus("Standby")
            elif self._patroni.member_started:
                self.unit.status = ActiveStatus()
        except (RetryError, ConnectionError) as e:
            logger.error(f"failed to get primary with error {e}")

    def _initialize_cluster(self, event: WorkloadEvent) -> bool:
        # Add the labels needed for replication in this pod.
        # This also enables the member as part of the cluster.
        try:
            self._patch_pod_labels(self._unit)
        except ApiError as e:
            logger.error("failed to patch pod")
            self.unit.status = BlockedStatus(f"failed to patch pod with error {e}")
            return False

        # Create resources and add labels needed for replication
        if self.upgrade.idle:
            try:
                self._create_services()
            except ApiError:
                logger.exception("failed to create k8s services")
                self.unit.status = BlockedStatus("failed to create k8s services")
                return False

        async_replication_primary_cluster = self.async_replication.get_primary_cluster()
        if (
            async_replication_primary_cluster is not None
            and async_replication_primary_cluster != self.app
        ):
            logger.debug(
                "Early exit _initialize_cluster: not the primary cluster in async replication"
            )
            return True

        if not self._patroni.primary_endpoint_ready:
            logger.debug(
                "Deferring on_postgresql_pebble_ready: Waiting for primary endpoint to be ready"
            )
            self.unit.status = WaitingStatus("awaiting for primary endpoint to be ready")
            event.defer()
            return False

        pg_users = self.postgresql.list_users()
        # Create the backup user.
        if BACKUP_USER not in pg_users:
            self.postgresql.create_user(BACKUP_USER, new_password(), admin=True)
        # Create the monitoring user.
        if MONITORING_USER not in pg_users:
            self.postgresql.create_user(
                MONITORING_USER,
                self.get_secret(APP_SCOPE, MONITORING_PASSWORD_KEY),
                extra_user_roles="pg_monitor",
            )

        self.postgresql.set_up_database()

        # Mark the cluster as initialised.
        self._peers.data[self.app]["cluster_initialised"] = "True"

        return True

    @property
    def is_blocked(self) -> bool:
        """Returns whether the unit is in a blocked state."""
        return isinstance(self.unit.status, BlockedStatus)

    def _on_upgrade_charm(self, _) -> None:
        self._fix_pod()

    def _patch_pod_labels(self, member: str) -> None:
        """Add labels required for replication to the current pod.

        Args:
            member: name of the unit that needs the labels
        Raises:
            ApiError when there is any problem communicating
                to K8s API
        """
        client = Client()
        patch = {
            "metadata": {"labels": {"application": "patroni", "cluster-name": self.cluster_name}}
        }
        client.patch(
            Pod,
            name=self._unit_name_to_pod_name(member),
            namespace=self._namespace,
            obj=patch,
        )

    def _create_services(self) -> None:
        """Create kubernetes services for primary and replicas endpoints."""
        client = Client()

        pod0 = client.get(
            res=Pod,
            name=f"{self.app.name}-0",
            namespace=self.model.name,
        )

        services = {
            "primary": "primary",
            "replicas": "replica",
        }
        for service_name_suffix, role_selector in services.items():
            service = Service(
                metadata=ObjectMeta(
                    name=f"{self._name}-{service_name_suffix}",
                    namespace=self.model.name,
                    ownerReferences=pod0.metadata.ownerReferences,
                    labels={
                        "app.kubernetes.io/name": self.app.name,
                    },
                ),
                spec=ServiceSpec(
                    ports=[
                        ServicePort(
                            name="api",
                            port=8008,
                            targetPort=8008,
                        ),
                        ServicePort(
                            name="database",
                            port=5432,
                            targetPort=5432,
                        ),
                    ],
                    selector={
                        "app.kubernetes.io/name": self.app.name,
                        "cluster-name": f"patroni-{self.app.name}",
                        "role": role_selector,
                    },
                ),
            )
            client.apply(
                obj=service,
                name=service.metadata.name,
                namespace=service.metadata.namespace,
                force=True,
                field_manager=self.model.app.name,
            )

    def _cleanup_old_cluster_resources(self) -> None:
        """Delete kubernetes services and endpoints from previous deployment."""
        if self.is_cluster_initialised:
            logger.debug("Early exit _cleanup_old_cluster_resources: cluster already initialised")
            return

        client = Client()
        for kind, suffix in itertools.product([Service, Endpoints], ["", "-config", "-sync"]):
            try:
                client.delete(
                    res=kind,
                    name=f"{self.cluster_name}{suffix}",
                    namespace=self._namespace,
                )
                logger.info(f"deleted {kind.__name__}/{self.cluster_name}{suffix}")
            except ApiError as e:
                if e.status.code == 403:
                    self.on_deployed_without_trust()
                    return
                # Ignore the error only when the resource doesn't exist.
                if e.status.code != 404:
                    raise e

    @property
    def _has_blocked_status(self) -> bool:
        """Returns whether the unit is in a blocked state."""
        return isinstance(self.unit.status, BlockedStatus)

    @property
    def _has_non_restore_waiting_status(self) -> bool:
        """Returns whether the unit is in a waiting state and there is no restore process ongoing."""
        return (
            isinstance(self.unit.status, WaitingStatus)
            and "restoring-backup" not in self.app_peer_data
            and "restore-to-time" not in self.app_peer_data
        )

    def _on_get_password(self, event: ActionEvent) -> None:
        """Returns the password for a user as an action response.

        If no user is provided, the password of the operator user is returned.
        """
        username = event.params.get("username", USER)
        if username not in PASSWORD_USERS:
            event.fail(
                f"The action can be run only for users used by the charm or Patroni:"
                f" {', '.join(PASSWORD_USERS)} not {username}"
            )
            return
        event.set_results({"password": self.get_secret(APP_SCOPE, f"{username}-password")})

    def _on_set_password(self, event: ActionEvent) -> None:
        """Set the password for the specified user."""
        # Only leader can write the new password into peer relation.
        if not self.unit.is_leader():
            event.fail("The action can be run only on leader unit")
            return

        username = event.params.get("username", USER)
        if username not in SYSTEM_USERS:
            event.fail(
                f"The action can be run only for users used by the charm:"
                f" {', '.join(SYSTEM_USERS)} not {username}"
            )
            return

        password = new_password()
        if "password" in event.params:
            password = event.params["password"]

        if password == self.get_secret(APP_SCOPE, f"{username}-password"):
            event.log("The old and new passwords are equal.")
            event.set_results({"password": password})
            return

        # Ensure all members are ready before trying to reload Patroni
        # configuration to avoid errors (like the API not responding in
        # one instance because PostgreSQL and/or Patroni are not ready).
        if not self._patroni.are_all_members_ready():
            event.fail(
                "Failed changing the password: Not all members healthy or finished initial sync."
            )
            return

        replication_offer_relation = self.model.get_relation(REPLICATION_OFFER_RELATION)
        if (
            replication_offer_relation is not None
            and not self.async_replication.is_primary_cluster()
        ):
            # Update the password in the other cluster PostgreSQL primary instance.
            other_cluster_endpoints = self.async_replication.get_all_primary_cluster_endpoints()
            other_cluster_primary = self._patroni.get_primary(
                alternative_endpoints=other_cluster_endpoints
            )
            other_cluster_primary_ip = [
                replication_offer_relation.data[unit].get("private-address")
                for unit in replication_offer_relation.units
                if unit.name.replace("/", "-") == other_cluster_primary
            ][0]
            try:
                self.postgresql.update_user_password(
                    username, password, database_host=other_cluster_primary_ip
                )
            except PostgreSQLUpdateUserPasswordError as e:
                logger.exception(e)
                event.fail("Failed changing the password.")
                return
        elif self.model.get_relation(REPLICATION_CONSUMER_RELATION) is not None:
            event.fail(
                "Failed changing the password: This action can be ran only in the cluster from the offer side."
            )
            return
        else:
            # Update the password in this cluster PostgreSQL primary instance.
            try:
                self.postgresql.update_user_password(username, password)
            except PostgreSQLUpdateUserPasswordError as e:
                logger.exception(e)
                event.fail("Failed changing the password.")
                return

        # Update the password in the secret store.
        self.set_secret(APP_SCOPE, f"{username}-password", password)

        # Update and reload Patroni configuration in this unit to use the new password.
        # Other units Patroni configuration will be reloaded in the peer relation changed event.
        self.update_config()

        event.set_results({"password": password})

    def _on_get_primary(self, event: ActionEvent) -> None:
        """Get primary instance."""
        try:
            primary = self._patroni.get_primary(unit_name_pattern=True)
            event.set_results({"primary": primary})
        except RetryError as e:
            logger.error(f"failed to get primary with error {e}")

    def _fix_pod(self) -> None:
        # Recreate k8s resources and add labels required for replication
        # when the pod loses them (like when it's deleted).
        if self.upgrade.idle:
            try:
                self._create_services()
            except ApiError:
                logger.exception("failed to create k8s services")
                self.unit.status = BlockedStatus("failed to create k8s services")
                return

        try:
            self._patch_pod_labels(self.unit.name)
        except ApiError as e:
            logger.error("failed to patch pod")
            self.unit.status = BlockedStatus(f"failed to patch pod with error {e}")
            return

        # Update the sync-standby endpoint in the async replication data.
        self.async_replication.update_async_replication_data()

    def _on_stop(self, _):
        # Remove data from the drive when scaling down to zero to prevent
        # the cluster from getting stuck when scaling back up.
        if self.app.planned_units() == 0:
            self.unit_peer_data.clear()

        # Patch the services to remove them when the StatefulSet is deleted
        # (i.e. application is removed).
        try:
            client = Client(field_manager=self.model.app.name)

            pod0 = client.get(
                res=Pod,
                name=f"{self.app.name}-0",
                namespace=self.model.name,
            )
        except ApiError:
            # Only log the exception.
            logger.exception("failed to get first pod info")
            return

        try:
            # Get the k8s resources created by the charm and Patroni.
            resources_to_patch = []
            for kind in [Endpoints, Service]:
                resources_to_patch.extend(
                    client.list(
                        kind,
                        namespace=self._namespace,
                        labels={"app.juju.is/created-by": f"{self._name}"},
                    )
                )
        except ApiError:
            # Only log the exception.
            logger.exception("failed to get the k8s resources created by the charm and Patroni")
            return

        for resource in resources_to_patch:
            # Ignore resources created by Juju or the charm
            # (which are already patched).
            if (
                type(resource) is Service
                and resource.metadata.name
                in [
                    self._name,
                    f"{self._name}-endpoints",
                    f"{self._name}-primary",
                    f"{self._name}-replicas",
                ]
            ) or resource.metadata.ownerReferences == pod0.metadata.ownerReferences:
                continue
            # Patch the resource.
            try:
                resource.metadata.ownerReferences = pod0.metadata.ownerReferences
                resource.metadata.managedFields = None
                client.apply(
                    obj=resource,
                    name=resource.metadata.name,
                    namespace=resource.metadata.namespace,
                    force=True,
                )
            except ApiError:
                # Only log the exception.
                logger.exception(
                    f"failed to patch k8s {type(resource).__name__} {resource.metadata.name}"
                )

    def _on_update_status_early_exit_checks(self, container) -> bool:
        if not self.upgrade.idle:
            logger.debug("Early exit on_update_status: upgrade in progress")
            return False

        if not container.can_connect():
            logger.debug("on_update_status early exit: Cannot connect to container")
            return False

        self._check_pgdata_storage_size()

        if (
            self._has_blocked_status and self.unit.status not in S3_BLOCK_MESSAGES
        ) 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:
                self.enable_disable_extensions()
                return True

            logger.debug("on_update_status early exit: Unit is in Blocked/Waiting status")
            return False
        return True

    def _check_pgdata_storage_size(self) -> None:
        """Asserts that pgdata volume has at least 10% free space and blocks charm if not."""
        try:
            total_size, _, free_size = shutil.disk_usage(self.pgdata_path)
        except FileNotFoundError:
            logger.error("pgdata folder not found in %s", self.pgdata_path)
            return

        logger.debug(
            "pgdata free disk space: %s out of %s, ratio of %s",
            free_size,
            total_size,
            free_size / total_size,
        )
        if free_size / total_size < 0.1:
            self.unit.status = BlockedStatus(INSUFFICIENT_SIZE_WARNING)
        elif self.unit.status.message == INSUFFICIENT_SIZE_WARNING:
            self.unit.status = ActiveStatus()
            self._set_active_status()

    def _on_update_status(self, _) -> None:
        """Update the unit status message."""
        container = self.unit.get_container("postgresql")
        if not self._on_update_status_early_exit_checks(container):
            return

        services = container.pebble.get_services(names=[self._postgresql_service])
        if len(services) == 0:
            # Service has not been added nor started yet, so don't try to check Patroni API.
            logger.debug("on_update_status early exit: Service has not been added nor started yet")
            return

        if (
            "restoring-backup" not in self.app_peer_data
            and "restore-to-time" not in self.app_peer_data
            and "stopped" not in self.unit_peer_data
            and services[0].current != ServiceStatus.ACTIVE
        ):
            logger.warning(
                "%s pebble service inactive, restarting service" % self._postgresql_service
            )
            try:
                container.restart(self._postgresql_service)
            except ChangeError:
                logger.exception("Failed to restart patroni")
            # If service doesn't recover fast, exit and wait for next hook run to re-check
            if not self._patroni.member_started:
                self.unit.status = MaintenanceStatus("Database service inactive, restarting")
                return

        if (
            "restoring-backup" in self.app_peer_data or "restore-to-time" in self.app_peer_data
        ) and not self._was_restore_successful(container, services[0]):
            return

        if self._handle_processes_failures():
            return

        # Update the sync-standby endpoint in the async replication data.
        self.async_replication.update_async_replication_data()

        self.backup.coordinate_stanza_fields()

        self._set_active_status()

    def _was_restore_successful(self, container: Container, service: ServiceInfo) -> bool:
        """Checks if restore operation succeeded and S3 is properly configured."""
        if "restore-to-time" in self.app_peer_data and all(self.is_pitr_failed(container)):
            logger.error(
                "Restore failed: database service failed to reach point-in-time-recovery target. "
                "You can launch another restore with different parameters"
            )
            self.log_pitr_last_transaction_time()
            self.unit.status = BlockedStatus(CANNOT_RESTORE_PITR)
            return False

        if (
            service.current != ServiceStatus.ACTIVE
            and self.unit.status.message != CANNOT_RESTORE_PITR
        ):
            logger.error("Restore failed: database service failed to start")
            self.unit.status = BlockedStatus("Failed to restore backup")
            return False

        if not self._patroni.member_started:
            logger.debug("Restore check early exit: Patroni has not started yet")
            return False

        restoring_backup = self.app_peer_data.get("restoring-backup")
        restore_timeline = self.app_peer_data.get("restore-timeline")
        restore_to_time = self.app_peer_data.get("restore-to-time")
        try:
            current_timeline = self.postgresql.get_current_timeline()
        except PostgreSQLGetCurrentTimelineError:
            logger.debug("Restore check early exit: can't get current wal timeline")
            return False

        # Remove the restoring backup flag and the restore stanza name.
        self.app_peer_data.update({
            "restoring-backup": "",
            "restore-stanza": "",
            "restore-to-time": "",
            "restore-timeline": "",
        })
        self.update_config()
        self.restore_patroni_on_failure_condition()

        logger.info(
            "Restored"
            f"{f' to {restore_to_time}' if restore_to_time else ''}"
            f"{f' from timeline {restore_timeline}' if restore_timeline and not restoring_backup else ''}"
            f"{f' from backup {self.backup._parse_backup_id(restoring_backup)[0]}' if restoring_backup else ''}"
            f". Currently tracking the newly created timeline {current_timeline}."
        )

        can_use_s3_repository, validation_message = self.backup.can_use_s3_repository()
        if not can_use_s3_repository:
            self.app_peer_data.update({
                "stanza": "",
                "s3-initialization-start": "",
                "s3-initialization-done": "",
                "s3-initialization-block-message": validation_message,
            })

        return True

    def _handle_processes_failures(self) -> bool:
        """Handle Patroni and PostgreSQL OS processes failures.

        Returns:
            a bool indicating whether the charm performed any action.
        """
        container = self.unit.get_container("postgresql")

        # Restart the Patroni process if it was killed (in that case, the PostgreSQL
        # process is still running). This is needed until
        # https://github.com/canonical/pebble/issues/149 is resolved.
        if not self._patroni.member_started and self._patroni.is_database_running:
            try:
                container.restart(self._postgresql_service)
                logger.info("restarted Patroni because it was not running")
            except ChangeError:
                logger.error("failed to restart Patroni after checking that it was not running")
                return False
            return True

        try:
            is_primary = self.is_primary
            is_standby_leader = self.is_standby_leader
        except RetryError:
            return False

        if (
            not is_primary
            and not is_standby_leader
            and self._patroni.member_started
            and not self._patroni.member_streaming
        ):
            try:
                logger.warning("Degraded member detected: reinitialising unit")
                self.unit.status = MaintenanceStatus("reinitialising replica")
                self._patroni.reinitialize_postgresql()
            except RetryError:
                logger.error(
                    "failed to reinitialise replica after checking that it was not streaming from primary"
                )
                return False
            return True

        return False

    @property
    def _patroni(self):
        """Returns an instance of the Patroni object."""
        return Patroni(
            self,
            self._endpoint,
            self._endpoints,
            self.primary_endpoint,
            self._namespace,
            self._storage_path,
            self.get_secret(APP_SCOPE, USER_PASSWORD_KEY),
            self.get_secret(APP_SCOPE, REPLICATION_PASSWORD_KEY),
            self.get_secret(APP_SCOPE, REWIND_PASSWORD_KEY),
            bool(self.unit_peer_data.get("tls")),
            self.get_secret(APP_SCOPE, PATRONI_PASSWORD_KEY),
        )

    @property
    def is_primary(self) -> bool:
        """Return whether this unit is the primary instance."""
        return self._unit == self._patroni.get_primary(unit_name_pattern=True)

    @property
    def is_standby_leader(self) -> bool:
        """Return whether this unit is the standby leader instance."""
        return self._unit == self._patroni.get_standby_leader(unit_name_pattern=True)

    @property
    def is_tls_enabled(self) -> bool:
        """Return whether TLS is enabled."""
        if not self.model.get_relation(PEER):
            return False
        return all(self.tls.get_tls_files())

    @property
    def _endpoint(self) -> str:
        """Current unit hostname."""
        return self._get_hostname_from_unit(self._unit_name_to_pod_name(self.unit.name))

    @property
    def _endpoints(self) -> List[str]:
        """Cluster members hostnames."""
        if self._peers:
            return json.loads(self._peers.data[self.app].get("endpoints", "[]"))
        else:
            # If the peer relations was not created yet, return only the current member hostname.
            return [self._endpoint]

    @property
    def peer_members_endpoints(self) -> List[str]:
        """Fetch current list of peer members endpoints.

        Returns:
            A list of peer members addresses (strings).
        """
        # Get all members endpoints and remove the current unit endpoint from the list.
        endpoints = self._endpoints
        current_unit_endpoint = self._get_hostname_from_unit(
            self._unit_name_to_pod_name(self._unit)
        )
        if current_unit_endpoint in endpoints:
            endpoints.remove(current_unit_endpoint)
        return endpoints

    def _add_to_endpoints(self, endpoint) -> None:
        """Add one endpoint to the members list."""
        self._update_endpoints(endpoint_to_add=endpoint)

    def _remove_from_endpoints(self, endpoints: List[str]) -> None:
        """Remove endpoints from the members list."""
        self._update_endpoints(endpoints_to_remove=endpoints)

    def _update_endpoints(
        self, endpoint_to_add: str = None, endpoints_to_remove: List[str] = None
    ) -> None:
        """Update members IPs."""
        # Allow leader to reset which members are part of the cluster.
        if not self.unit.is_leader():
            return

        endpoints = json.loads(self._peers.data[self.app].get("endpoints", "[]"))
        if endpoint_to_add:
            endpoints.append(endpoint_to_add)
        elif endpoints_to_remove:
            for endpoint in endpoints_to_remove:
                endpoints.remove(endpoint)
        self._peers.data[self.app]["endpoints"] = json.dumps(endpoints)

    def _generate_metrics_service(self) -> Dict:
        """Generate the metrics service definition."""
        return {
            "override": "replace",
            "summary": "postgresql metrics exporter",
            "command": "/start-exporter.sh",
            "startup": (
                "enabled"
                if self.get_secret("app", MONITORING_PASSWORD_KEY) is not None
                else "disabled"
            ),
            "after": [self._postgresql_service],
            "user": WORKLOAD_OS_USER,
            "group": WORKLOAD_OS_GROUP,
            "environment": {
                "DATA_SOURCE_NAME": (
                    f"user={MONITORING_USER} "
                    f"password={self.get_secret('app', MONITORING_PASSWORD_KEY)} "
                    "host=/var/run/postgresql port=5432 database=postgres"
                ),
            },
        }

    def _postgresql_layer(self) -> Layer:
        """Returns a Pebble configuration layer for PostgreSQL."""
        pod_name = self._unit_name_to_pod_name(self._unit)
        layer_config = {
            "summary": "postgresql + patroni layer",
            "description": "pebble config layer for postgresql + patroni",
            "services": {
                self._postgresql_service: {
                    "override": "replace",
                    "summary": "entrypoint of the postgresql + patroni image",
                    "command": f"patroni {self._storage_path}/patroni.yml",
                    "startup": "enabled",
                    "on-failure": self.unit_peer_data.get(
                        "patroni-on-failure-condition-override", None
                    )
                    or ORIGINAL_PATRONI_ON_FAILURE_CONDITION,
                    "user": WORKLOAD_OS_USER,
                    "group": WORKLOAD_OS_GROUP,
                    "environment": {
                        "PATRONI_KUBERNETES_LABELS": f"{{application: patroni, cluster-name: {self.cluster_name}}}",
                        "PATRONI_KUBERNETES_LEADER_LABEL_VALUE": "primary",
                        "PATRONI_KUBERNETES_NAMESPACE": self._namespace,
                        "PATRONI_KUBERNETES_USE_ENDPOINTS": "true",
                        "PATRONI_NAME": pod_name,
                        "PATRONI_SCOPE": self.cluster_name,
                        "PATRONI_REPLICATION_USERNAME": REPLICATION_USER,
                        "PATRONI_SUPERUSER_USERNAME": USER,
                    },
                },
                self.pgbackrest_server_service: {
                    "override": "replace",
                    "summary": "pgBackRest server",
                    "command": self.pgbackrest_server_service,
                    "startup": "disabled",
                    "user": WORKLOAD_OS_USER,
                    "group": WORKLOAD_OS_GROUP,
                },
                self._metrics_service: self._generate_metrics_service(),
                self.rotate_logs_service: {
                    "override": "replace",
                    "summary": "rotate logs",
                    "command": "python3 /home/postgres/rotate_logs.py",
                    "startup": "disabled",
                },
            },
            "checks": {
                self._postgresql_service: {
                    "override": "replace",
                    "level": "ready",
                    "http": {
                        "url": f"{self._patroni._patroni_url}/health",
                    },
                }
            },
        }
        return Layer(layer_config)

    @property
    def _peers(self) -> Relation:
        """Fetch the peer relation.

        Returns:
             A :class:`ops.model.Relation` object representing
             the peer relation.
        """
        return self.model.get_relation(PEER)

    def push_tls_files_to_workload(self, container: Container = None) -> bool:
        """Uploads TLS files to the workload container."""
        if container is None:
            container = self.unit.get_container("postgresql")

        key, ca, cert = self.tls.get_tls_files()

        if key is not None:
            container.push(
                f"{self._storage_path}/{TLS_KEY_FILE}",
                key,
                make_dirs=True,
                permissions=0o400,
                user=WORKLOAD_OS_USER,
                group=WORKLOAD_OS_GROUP,
            )
        if ca is not None:
            container.push(
                f"{self._storage_path}/{TLS_CA_FILE}",
                ca,
                make_dirs=True,
                permissions=0o400,
                user=WORKLOAD_OS_USER,
                group=WORKLOAD_OS_GROUP,
            )
            container.push(
                "/usr/local/share/ca-certificates/ca.crt",
                ca,
                make_dirs=True,
                permissions=0o400,
                user=WORKLOAD_OS_USER,
                group=WORKLOAD_OS_GROUP,
            )
            container.exec(["update-ca-certificates"]).wait()
        if cert is not None:
            container.push(
                f"{self._storage_path}/{TLS_CERT_FILE}",
                cert,
                make_dirs=True,
                permissions=0o400,
                user=WORKLOAD_OS_USER,
                group=WORKLOAD_OS_GROUP,
            )

        return self.update_config()

    def _restart(self, event: RunWithLock) -> None:
        """Restart PostgreSQL."""
        if not self._patroni.are_all_members_ready():
            logger.debug("Early exit _restart: not all members ready yet")
            event.defer()
            return

        try:
            logger.debug("Restarting PostgreSQL")
            self._patroni.restart_postgresql()
        except RetryError:
            error_message = "failed to restart PostgreSQL"
            logger.exception(error_message)
            self.unit.status = BlockedStatus(error_message)
            return

        # Update health check URL.
        self._update_pebble_layers()

        try:
            for attempt in Retrying(wait=wait_fixed(3), stop=stop_after_delay(300)):
                with attempt:
                    if not self._can_connect_to_postgresql:
                        assert False
        except Exception:
            logger.exception("Unable to reconnect to postgresql")

        # Start or stop the pgBackRest TLS server service when TLS certificate change.
        self.backup.start_stop_pgbackrest_service()

    @property
    def _is_workload_running(self) -> bool:
        """Returns whether the workload is running (in an active state)."""
        container = self.unit.get_container("postgresql")
        if not container.can_connect():
            return False

        services = container.pebble.get_services(names=[self._postgresql_service])
        if len(services) == 0:
            return False

        return services[0].current == ServiceStatus.ACTIVE

    @property
    def _can_connect_to_postgresql(self) -> bool:
        try:
            for attempt in Retrying(stop=stop_after_delay(30), wait=wait_fixed(3)):
                with attempt:
                    assert self.postgresql.get_postgresql_timezones()
        except RetryError:
            logger.debug("Cannot connect to database")
            return False
        return True

    def update_config(self, is_creating_backup: bool = False) -> bool:
        """Updates Patroni config file based on the existence of the TLS files."""
        # Retrieve PostgreSQL parameters.
        if self.config.profile_limit_memory:
            limit_memory = self.config.profile_limit_memory * 10**6
        else:
            limit_memory = None
        try:
            available_cpu_cores, available_memory = self.get_available_resources()
        except ApiError as e:
            if e.status.code == 403:
                self.on_deployed_without_trust()
                return
            raise e

        postgresql_parameters = self.postgresql.build_postgresql_parameters(
            self.model.config, available_memory, limit_memory
        )

        logger.info("Updating Patroni config file")
        # Update and reload configuration based on TLS files availability.
        self._patroni.render_patroni_yml_file(
            connectivity=self.unit_peer_data.get("connectivity", "on") == "on",
            is_creating_backup=is_creating_backup,
            enable_tls=self.is_tls_enabled,
            is_no_sync_member=self.upgrade.is_no_sync_member,
            backup_id=self.app_peer_data.get("restoring-backup"),
            pitr_target=self.app_peer_data.get("restore-to-time"),
            restore_timeline=self.app_peer_data.get("restore-timeline"),
            restore_to_latest=self.app_peer_data.get("restore-to-time", None) == "latest",
            stanza=self.app_peer_data.get("stanza", self.unit_peer_data.get("stanza")),
            restore_stanza=self.app_peer_data.get("restore-stanza"),
            parameters=postgresql_parameters,
        )

        if not self._is_workload_running:
            # If Patroni/PostgreSQL has not started yet and TLS relations was initialised,
            # then mark TLS as enabled. This commonly happens when the charm is deployed
            # in a bundle together with the TLS certificates operator. This flag is used to
            # know when to call the Patroni API using HTTP or HTTPS.
            self.unit_peer_data.update({"tls": "enabled" if self.is_tls_enabled else ""})
            self.postgresql_client_relation.update_tls_flag(
                "True" if self.is_tls_enabled else "False"
            )
            logger.debug("Early exit update_config: Workload not started yet")
            return True

        if not self._patroni.member_started:
            logger.debug("Early exit update_config: Patroni not started yet")
            return False

        # Use config value if set, calculate otherwise
        if self.config.experimental_max_connections:
            max_connections = self.config.experimental_max_connections
        else:
            max_connections = max(4 * available_cpu_cores, 100)

        self._patroni.bulk_update_parameters_controller_by_patroni({
            "max_connections": max_connections,
            "max_prepared_transactions": self.config.memory_max_prepared_transactions,
            "shared_buffers": self.config.memory_shared_buffers,
        })

        self._handle_postgresql_restart_need()

        # Restart the monitoring service if the password was rotated
        container = self.unit.get_container("postgresql")
        current_layer = container.get_plan()
        if metrics_service := current_layer.services[self._metrics_service]:
            if not metrics_service.environment.get("DATA_SOURCE_NAME", "").startswith(
                f"user={MONITORING_USER} password={self.get_secret('app', MONITORING_PASSWORD_KEY)} "
            ):
                container.add_layer(
                    self._metrics_service,
                    Layer({"services": {self._metrics_service: self._generate_metrics_service()}}),
                    combine=True,
                )
                container.restart(self._metrics_service)

        return True

    def _validate_config_options(self) -> None:
        """Validates specific config options that need access to the database or to the TLS status."""
        if (
            self.config.instance_default_text_search_config
            not in self.postgresql.get_postgresql_text_search_configs()
        ):
            raise ValueError(
                "instance_default_text_search_config config option has an invalid value"
            )

        if not self.postgresql.validate_date_style(self.config.request_date_style):
            raise ValueError("request_date_style config option has an invalid value")

        if self.config.request_time_zone not in self.postgresql.get_postgresql_timezones():
            raise ValueError("request_time_zone config option has an invalid value")

        container = self.unit.get_container("postgresql")
        output, _ = container.exec(["locale", "-a"]).wait_output()
        locales = list(output.splitlines())
        for parameter in ["response_lc_monetary", "response_lc_numeric", "response_lc_time"]:
            value = self.model.config.get(parameter)
            if value is not None and value not in locales:
                raise ValueError(
                    f"Value for {parameter} not one of the locales available in the system"
                )

    def _handle_postgresql_restart_need(self):
        """Handle PostgreSQL restart need based on the TLS configuration and configuration changes."""
        restart_postgresql = self.is_tls_enabled != self.postgresql.is_tls_enabled()
        self._patroni.reload_patroni_configuration()
        # Wait for some more time than the Patroni's loop_wait default value (10 seconds),
        # which tells how much time Patroni will wait before checking the configuration
        # file again to reload it.
        try:
            for attempt in Retrying(stop=stop_after_attempt(5), wait=wait_fixed(3)):
                with attempt:
                    restart_postgresql = restart_postgresql or self.postgresql.is_restart_pending()
                    if not restart_postgresql:
                        raise Exception
        except RetryError:
            # Ignore the error, as it happens only to indicate that the configuration has not changed.
            pass
        self.unit_peer_data.update({"tls": "enabled" if self.is_tls_enabled else ""})
        self.postgresql_client_relation.update_tls_flag("True" if self.is_tls_enabled else "False")

        # Restart PostgreSQL if TLS configuration has changed
        # (so the both old and new connections use the configuration).
        if restart_postgresql:
            logger.info("PostgreSQL restart required")
            self.metrics_endpoint.update_scrape_job_spec(
                self._generate_metrics_jobs(self.is_tls_enabled)
            )
            self.on[self.restart_manager.name].acquire_lock.emit()

    def _update_pebble_layers(self, replan: bool = True) -> None:
        """Update the pebble layers to keep the health check URL up-to-date."""
        container = self.unit.get_container("postgresql")

        # Get the current layer.
        current_layer = container.get_plan()

        # Create a new config layer.
        new_layer = self._postgresql_layer()

        # Check if there are any changes to layer services.
        if current_layer.services != new_layer.services:
            # Changes were made, add the new layer.
            container.add_layer(self._postgresql_service, new_layer, combine=True)
            logging.info("Added updated layer 'postgresql' to Pebble plan")
            if replan:
                container.replan()
                logging.info("Restarted postgresql service")
        if current_layer.checks != new_layer.checks:
            # Changes were made, add the new layer.
            container.add_layer(self._postgresql_service, new_layer, combine=True)
            logging.info("Updated health checks")

    def _unit_name_to_pod_name(self, unit_name: str) -> str:
        """Converts unit name to pod name.

        Args:
            unit_name: name in "postgresql-k8s/0" format.

        Returns:
            pod name in "postgresql-k8s-0" format.
        """
        return unit_name.replace("/", "-")

    def _get_node_name_for_pod(self) -> str:
        """Return the node name for a given pod."""
        client = Client()
        pod = client.get(
            Pod, name=self._unit_name_to_pod_name(self.unit.name), namespace=self._namespace
        )
        return pod.spec.nodeName

    def get_resources_limits(self, container_name: str) -> Dict:
        """Return resources limits for a given container.

        Args:
            container_name: name of the container to get resources limits for
        """
        client = Client()
        pod = client.get(
            Pod, self._unit_name_to_pod_name(self.unit.name), namespace=self._namespace
        )

        for container in pod.spec.containers:
            if container.name == container_name:
                return container.resources.limits or {}
        return {}

    def get_node_allocable_memory(self) -> int:
        """Return the allocable memory in bytes for the current K8S node."""
        client = Client()
        node = client.get(Node, name=self._get_node_name_for_pod(), namespace=self._namespace)
        return any_memory_to_bytes(node.status.allocatable["memory"])

    def get_node_cpu_cores(self) -> int:
        """Return the number of CPU cores for the current K8S node."""
        client = Client()
        node = client.get(Node, name=self._get_node_name_for_pod(), namespace=self._namespace)
        return any_cpu_to_cores(node.status.allocatable["cpu"])

    def get_available_resources(self) -> Tuple[int, int]:
        """Get available CPU cores and memory (in bytes) for the container."""
        cpu_cores = self.get_node_cpu_cores()
        allocable_memory = self.get_node_allocable_memory()
        container_limits = self.get_resources_limits(container_name="postgresql")
        if "cpu" in container_limits:
            cpu_str = container_limits["cpu"]
            constrained_cpu = int(cpu_str)
            if constrained_cpu < cpu_cores:
                logger.debug(f"CPU constrained to {cpu_str} cores from resource limit")
                cpu_cores = constrained_cpu
        if "memory" in container_limits:
            memory_str = container_limits["memory"]
            constrained_memory = any_memory_to_bytes(memory_str)
            if constrained_memory < allocable_memory:
                logger.debug(f"Memory constrained to {memory_str} from resource limit")
                allocable_memory = constrained_memory

        return cpu_cores, allocable_memory

    def on_deployed_without_trust(self) -> None:
        """Blocks the application and returns a specific error message for deployments made without --trust."""
        self.unit.status = BlockedStatus(
            f"Insufficient permissions, try: `juju trust {self._name} --scope=cluster`"
        )
        logger.error(
            f"""
            Access to k8s cluster resources is not authorized. This happens when RBAC is enabled and the deployed application was not trusted by the juju admin.
            To fix this issue, run `juju trust {self._name} --scope=cluster` (or remove & re-deploy {self._name} with `--trust`)
            """
        )

    @property
    def client_relations(self) -> List[Relation]:
        """Return the list of established client relations."""
        relations = []
        for relation_name in ["database", "db", "db-admin"]:
            for relation in self.model.relations.get(relation_name, []):
                relations.append(relation)
        return relations

    def override_patroni_on_failure_condition(
        self, new_condition: str, repeat_cause: str | None
    ) -> bool:
        """Temporary override Patroni pebble service on-failure condition.

        Executes only on current unit.

        Args:
            new_condition: new Patroni pebble service on-failure condition.
            repeat_cause: whether this field is equal to the last success override operation repeat cause, Patroni
                on-failure condition will be overridden (keeping the original restart condition reference untouched) and
                success code will be returned. But if this field is distinct from previous repeat cause or None,
                repeated operation will cause failure code will be returned.
        """
        if "patroni-on-failure-condition-override" in self.unit_peer_data:
            current_condition = self.unit_peer_data["patroni-on-failure-condition-override"]
            if repeat_cause is None:
                logger.error(
                    f"failure trying to override patroni on-failure condition to {new_condition}"
                    f"as it already overridden from {ORIGINAL_PATRONI_ON_FAILURE_CONDITION} to {current_condition}"
                )
                return False
            previous_repeat_cause = self.unit_peer_data.get(
                "overridden-patroni-on-failure-condition-repeat-cause", None
            )
            if previous_repeat_cause != repeat_cause:
                logger.error(
                    f"failure trying to override patroni on-failure condition to {new_condition}"
                    f"as it already overridden from {ORIGINAL_PATRONI_ON_FAILURE_CONDITION} to {current_condition}"
                    f"and repeat cause is not equal: {previous_repeat_cause} != {repeat_cause}"
                )
                return False
            self.unit_peer_data["patroni-on-failure-condition-override"] = new_condition
            self._update_pebble_layers(False)
            logger.debug(
                f"Patroni on-failure condition re-overridden to {new_condition} within repeat cause {repeat_cause}"
                f"(original on-failure condition reference is untouched and is {ORIGINAL_PATRONI_ON_FAILURE_CONDITION})"
            )
            return True

        self.unit_peer_data["patroni-on-failure-condition-override"] = new_condition
        if repeat_cause:
            self.unit_peer_data["overridden-patroni-on-failure-condition-repeat-cause"] = (
                repeat_cause
            )
        self._update_pebble_layers(False)
        logger.debug(
            f"Patroni on-failure condition overridden from {ORIGINAL_PATRONI_ON_FAILURE_CONDITION} to {new_condition}"
            f"{' with repeat cause ' + repeat_cause if repeat_cause is not None else ''}"
        )
        return True

    def restore_patroni_on_failure_condition(self) -> None:
        """Restore Patroni pebble service original on-failure condition.

        Will do nothing if not overridden. Executes only on current unit.
        """
        if "patroni-on-failure-condition-override" in self.unit_peer_data:
            self.unit_peer_data.update({
                "patroni-on-failure-condition-override": "",
                "overridden-patroni-on-failure-condition-repeat-cause": "",
            })
            self._update_pebble_layers(False)
            logger.debug(
                f"restored Patroni on-failure condition to {ORIGINAL_PATRONI_ON_FAILURE_CONDITION}"
            )
        else:
            logger.warning("not restoring patroni on-failure condition as it's not overridden")

    def is_pitr_failed(self, container: Container) -> Tuple[bool, bool]:
        """Check if Patroni service failed to bootstrap cluster during point-in-time-recovery.

        Typically, this means that database service failed to reach point-in-time-recovery target or has been
        supplied with bad PITR parameter. Also, remembers last state and can provide info is it new event, or
        it belongs to previous action. Executes only on current unit.

        Returns:
            Tuple[bool, bool]:
                - Is patroni service failed to bootstrap cluster.
                - Is it new fail, that wasn't observed previously.
        """
        patroni_exceptions = []
        count = 0
        while len(patroni_exceptions) == 0 and count < 10:
            if count > 0:
                time.sleep(3)
            try:
                log_exec = container.pebble.exec(
                    ["pebble", "logs", "postgresql", "-n", "all"], combine_stderr=True
                )
                patroni_logs = log_exec.wait_output()[0]
                patroni_exceptions = re.findall(
                    r"^([0-9-:TZ.]+) \[postgresql] patroni\.exceptions\.PatroniFatalException: Failed to bootstrap cluster$",
                    patroni_logs,
                    re.MULTILINE,
                )
            except ExecError:  # For Juju 2.
                log_exec = container.pebble.exec(["cat", "/var/log/postgresql/patroni.log"])
                patroni_logs = log_exec.wait_output()[0]
                patroni_exceptions = re.findall(
                    r"^([0-9- :]+) UTC \[[0-9]+\]: INFO: removing initialize key after failed attempt to bootstrap the cluster",
                    patroni_logs,
                    re.MULTILINE,
                )
                if len(patroni_exceptions) != 0:
                    break
                # If no match, look at older logs
                log_exec = container.pebble.exec([
                    "find",
                    "/var/log/postgresql/",
                    "-name",
                    "'patroni.log.*'",
                    "-exec",
                    "cat",
                    "{}",
                    "+",
                ])
                patroni_logs = log_exec.wait_output()[0]
                patroni_exceptions = re.findall(
                    r"^([0-9- :]+) UTC \[[0-9]+\]: INFO: removing initialize key after failed attempt to bootstrap the cluster",
                    patroni_logs,
                    re.MULTILINE,
                )
            count += 1

        if len(patroni_exceptions) > 0:
            logger.debug("Failures to bootstrap cluster detected on Patroni service logs")
            old_pitr_fail_id = self.unit_peer_data.get("last_pitr_fail_id", None)
            self.unit_peer_data["last_pitr_fail_id"] = patroni_exceptions[-1]
            return True, patroni_exceptions[-1] != old_pitr_fail_id

        logger.debug("No failures detected on Patroni service logs")
        return False, False

    def log_pitr_last_transaction_time(self) -> None:
        """Log to user last completed transaction time acquired from postgresql logs."""
        postgresql_logs = self._patroni.last_postgresql_logs()
        log_time = re.findall(
            r"last completed transaction was at log time (.*)$",
            postgresql_logs,
            re.MULTILINE,
        )
        if len(log_time) > 0:
            logger.info(f"Last completed transaction was at {log_time[-1]}")
        else:
            logger.error("Can't tell last completed transaction time")

    def get_plugins(self) -> List[str]:
        """Return a list of installed plugins."""
        plugins = [
            "_".join(plugin.split("_")[1:-1])
            for plugin in self.config.plugin_keys()
            if self.config[plugin]
        ]
        plugins = [PLUGIN_OVERRIDES.get(plugin, plugin) for plugin in plugins]
        if "spi" in plugins:
            plugins.remove("spi")
            for ext in SPI_MODULE:
                plugins.append(ext)
        return plugins


if __name__ == "__main__":
    main(PostgresqlOperatorCharm, use_juju_for_storage=True)