[DPE-9325]: Network HA tests

pyproject.toml

@@ -54,6 +54,7 @@ python-dateutil = "*"
 tenacity = "^9.1.2"
 # https://github.com/valkey-io/valkey-glide/pull/5124 not yet released
 valkey-glide = { git = "https://github.com/skourta/valkey-glide", subdirectory = "python/glide-async", branch = "add-build-rs-to-async-client" }
+kubernetes = "^35.0.0"
 
 [tool.coverage.run]
 branch = true

src/common/locks.py

@@ -4,6 +4,7 @@
 """Collection of locks for cluster operations."""
 
 import logging
+import time
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Protocol, override
 
@@ -52,10 +53,16 @@ class DataBagLock(Lockable):
 
     unit_request_lock_atr_name: str
     member_with_lock_atr_name: str
+    lock_timestamp: str = "databaglock_timestamp"
 
     def __init__(self, state: "ClusterState") -> None:
         self.state = state
 
+    def __init_subclass__(cls) -> None:
+        """Initialize subclass attributes."""
+        super().__init_subclass__()
+        cls.lock_timestamp = cls.__name__.lower() + "_timestamp"
+
     @property
     def units_requesting_lock(self) -> list[str]:
         """Get the list of units requesting the start lock."""
@@ -68,6 +75,8 @@ def units_requesting_lock(self) -> list[str]:
     @property
     def next_unit_to_give_lock(self) -> str | None:
         """Get the next unit to give the start lock to."""
+        if self.state.unit_server.model[self.unit_request_lock_atr_name]:
+            return self.state.unit_server.unit_name
         return self.units_requesting_lock[0] if self.units_requesting_lock else None
 
     @property
@@ -98,11 +107,13 @@ def is_held_by_this_unit(self) -> bool:
 
     def request_lock(self) -> bool:
         """Request the lock for the local unit."""
-        self.state.unit_server.update(
-            {
-                self.unit_request_lock_atr_name: True,
-            }
-        )
+        if not self.state.unit_server.model[self.unit_request_lock_atr_name]:
+            self.state.unit_server.update(
+                {
+                    self.unit_request_lock_atr_name: True,
+                    self.lock_timestamp: time.time(),
+                }
+            )
         if self.state.unit_server.unit.is_leader():
             logger.info(
                 "Leader unit requesting %s lock. Triggering lock request processing.",
@@ -114,11 +125,13 @@ def request_lock(self) -> bool:
 
     def release_lock(self) -> bool:
         """Release the lock from the local unit."""
-        self.state.unit_server.update(
-            {
-                self.unit_request_lock_atr_name: False,
-            }
-        )
+        if self.state.unit_server.model[self.unit_request_lock_atr_name]:
+            self.state.unit_server.update(
+                {
+                    self.unit_request_lock_atr_name: False,
+                    self.lock_timestamp: time.time(),
+                }
+            )
         if self.state.unit_server.unit.is_leader():
             logger.info(
                 "Leader unit releasing %s lock. Triggering lock request processing.",
@@ -157,6 +170,24 @@ def is_lock_free_to_give(self) -> bool:
             not self.state.cluster.model.start_member
             or not starting_unit
             or starting_unit.is_started
+            or not starting_unit.model.request_start_lock
+        )
+
+
+class RestartLock(DataBagLock):
+    """Lock for restart operations."""
+
+    unit_request_lock_atr_name = "request_restart_lock"
+    member_with_lock_atr_name = "restart_member"
+
+    @property
+    def is_lock_free_to_give(self) -> bool:
+        """Check if the unit with the restart lock has completed its operation."""
+        restarting_unit = self.unit_with_lock
+        return (
+            not self.state.cluster.model.restart_member
+            or not restarting_unit
+            or not restarting_unit.model.request_restart_lock
         )
 
 

src/core/models.py

@@ -53,6 +53,7 @@ class PeerAppModel(PeerModel):
     charmed_sentinel_peers_password: InternalUsersSecret = Field(default="")
     charmed_sentinel_operator_password: InternalUsersSecret = Field(default="")
     start_member: str = Field(default="")
+    restart_member: str = Field(default="")
     internal_ca_certificate: InternalCertificatesSecret = Field(default="")
     internal_ca_private_key: InternalCertificatesSecret = Field(default="")
 
@@ -65,6 +66,7 @@ class PeerUnitModel(PeerModel):
     hostname: str = Field(default="")
     private_ip: str = Field(default="")
     request_start_lock: bool = Field(default=False)
+    request_restart_lock: bool = Field(default=False)
     scale_down_state: str = Field(default="")
     tls_client_state: str = Field(default="")
     client_cert_ready: bool = Field(default=False)

src/events/base_events.py

@@ -19,7 +19,7 @@
     ValkeyServicesFailedToStartError,
     ValkeyWorkloadCommandError,
 )
-from common.locks import ScaleDownLock, StartLock
+from common.locks import RestartLock, ScaleDownLock, StartLock
 from literals import (
     CLIENT_PORT,
     DATA_STORAGE,
@@ -41,6 +41,29 @@
 logger = logging.getLogger(__name__)
 
 
+class RestartWorkloadEvent(ops.EventBase):
+    """Event for restarting the workload when certain events happen, e.g. IP change."""
+
+    def __init__(
+        self, handle: ops.Handle, restart_valkey: bool = True, restart_sentinel: bool = True
+    ):
+        super().__init__(handle)
+        self.restart_valkey = restart_valkey
+        self.restart_sentinel = restart_sentinel
+
+    def snapshot(self) -> dict[str, str]:
+        """Save the state of the event."""
+        return {
+            "restart_valkey": str(self.restart_valkey),
+            "restart_sentinel": str(self.restart_sentinel),
+        }
+
+    def restore(self, snapshot: dict[str, str]) -> None:
+        """Restore the state of the event."""
+        self.restart_valkey = snapshot.get("restart_valkey", "True") == "True"
+        self.restart_sentinel = snapshot.get("restart_sentinel", "True") == "True"
+
+
 class UnitFullyStarted(ops.EventBase):
     """Event that signals that the unit's has fully started.
 
@@ -66,6 +89,7 @@ class BaseEvents(ops.Object):
     """Handle all base events."""
 
     unit_fully_started = ops.EventSource(UnitFullyStarted)
+    restart_workload = ops.EventSource(RestartWorkloadEvent)
 
     def __init__(self, charm: "ValkeyCharm"):
         super().__init__(charm, key="base_events")
@@ -81,6 +105,7 @@ def __init__(self, charm: "ValkeyCharm"):
         self.framework.observe(self.charm.on.config_changed, self._on_config_changed)
         self.framework.observe(self.charm.on.secret_changed, self._on_secret_changed)
         self.framework.observe(self.unit_fully_started, self._on_unit_fully_started)
+        self.framework.observe(self.restart_workload, self._on_restart_workload)
         self.framework.observe(
             self.charm.on[DATA_STORAGE].storage_detaching, self._on_storage_detaching
         )
@@ -230,7 +255,7 @@ def _on_peer_relation_changed(self, event: ops.RelationChangedEvent) -> None:
         if not self.charm.unit.is_leader():
             return
 
-        for lock in [StartLock(self.charm.state)]:
+        for lock in [StartLock(self.charm.state), RestartLock(self.charm.state)]:
             lock.process()
 
     def _on_update_status(self, event: ops.UpdateStatusEvent) -> None:
@@ -287,12 +312,15 @@ def _on_leader_elected(self, event: ops.LeaderElectedEvent) -> None:
 
     def _on_config_changed(self, event: ops.ConfigChangedEvent) -> None:
         """Handle the config_changed event."""
-        self.charm.state.unit_server.update(
-            {
-                "hostname": self.charm.state.hostname,
-                "private_ip": self.charm.state.bind_address,
-            }
-        )
+        # on k8s we use hostnames so we do not have to reconfigure on ip change
+        if (
+            self.charm.state.unit_server.model.private_ip
+            and self.charm.state.bind_address != self.charm.state.unit_server.model.private_ip
+            and self.charm.state.substrate == Substrate.VM
+        ):
+            self.charm.config_manager.configure_services(
+                self.charm.sentinel_manager.get_primary_ip()
+            )
 
         if not self.charm.unit.is_leader():
             return
@@ -524,3 +552,54 @@ def _set_state_for_going_away(self) -> None:
             )
 
         self.charm.state.unit_server.update({"scale_down_state": ScaleDownState.GOING_AWAY})
+
+    def _on_restart_workload(self, event: RestartWorkloadEvent) -> None:
+        """Handle the restart_workload event."""
+        logger.info(
+            "Restarting workload Event. Restart Valkey: %s, Restart Sentinel: %s",
+            event.restart_valkey,
+            event.restart_sentinel,
+        )
+        restart_lock = RestartLock(self.charm.state)
+        restart_lock.request_lock()
+        if not restart_lock.is_held_by_this_unit:
+            logger.info("Waiting for lock to restart workload")
+            event.defer()
+            return
+
+        if event.restart_valkey:
+            self.charm.workload.restart(self.charm.workload.valkey_service)
+        if event.restart_sentinel:
+            self.charm.sentinel_manager.restart_service()
+
+        if event.restart_valkey and not self.charm.cluster_manager.is_healthy(
+            check_replica_sync=False
+        ):
+            self.charm.status.set_running_status(
+                ClusterStatuses.VALKEY_UNHEALTHY_RESTART.value,
+                scope="unit",
+                component_name=self.charm.cluster_manager.name,
+                statuses_state=self.charm.state.statuses,
+            )
+
+        self.charm.state.statuses.delete(
+            ClusterStatuses.VALKEY_UNHEALTHY_RESTART.value,
+            scope="unit",
+            component=self.charm.cluster_manager.name,
+        )
+
+        if event.restart_sentinel and not self.charm.sentinel_manager.is_healthy():
+            self.charm.status.set_running_status(
+                ClusterStatuses.SENTINEL_UNHEALTHY_RESTART.value,
+                scope="unit",
+                component_name=self.charm.cluster_manager.name,
+                statuses_state=self.charm.state.statuses,
+            )
+
+        self.charm.state.statuses.delete(
+            ClusterStatuses.SENTINEL_UNHEALTHY_RESTART.value,
+            scope="unit",
+            component=self.charm.cluster_manager.name,
+        )
+
+        restart_lock.release_lock()

src/events/tls.py

@@ -24,6 +24,7 @@
     CLIENT_PORT,
     CLIENT_TLS_RELATION_NAME,
     PEER_RELATION,
+    Substrate,
     TLSCARotationState,
     TLSState,
 )
@@ -78,6 +79,7 @@ def __init__(self, charm: "ValkeyCharm"):
             self.charm.on[PEER_RELATION].relation_changed, self._on_peer_relation_changed
         )
         self.framework.observe(self.charm.on.update_status, self._on_update_status)
+        self.framework.observe(self.charm.on.config_changed, self._on_config_changed)
 
     def _on_peer_relation_created(self, event: ops.RelationCreatedEvent) -> None:
         """Set up self-signed certificates for peer TLS by default."""
@@ -299,6 +301,30 @@ def _enable_client_tls(self) -> None:
         self.charm.cluster_manager.reload_tls_settings(tls_config)
         self.charm.sentinel_manager.restart_service()
 
+    def _on_config_changed(self, event: ops.ConfigChangedEvent) -> None:
+        """Handle the `config-changed` event."""
+        if (
+            self.charm.state.unit_server.model.private_ip
+            and self.charm.state.bind_address != self.charm.state.unit_server.model.private_ip
+        ):
+            if self.charm.tls_manager.certificate_sans_require_update():
+                if not self.charm.state.client_tls_relation:
+                    self.charm.tls_manager.create_and_store_self_signed_certificate()
+                else:
+                    self.charm.tls_events.refresh_tls_certificates_event.emit()
+                    event.defer()
+                    return
+
+            self.charm.state.unit_server.update(
+                {
+                    "hostname": self.charm.state.hostname,
+                    "private_ip": self.charm.state.bind_address,
+                }
+            )
+            # only restart on VM because on k8s the hostname is stable and does not change with IP changes
+            if self.charm.state.substrate == Substrate.VM:
+                self.charm.base_events.restart_workload.emit()
+
     def _orchestrate_ca_rotation(self) -> None:
         """Orchestrate the workflow when a TLS CA rotation has been initiated."""
         match self.charm.state.unit_server.tls_ca_rotation_state:

src/managers/config.py

@@ -78,7 +78,6 @@ def get_config_properties(self, primary_endpoint: str) -> dict[str, str]:
         config_properties["aclfile"] = self.workload.acl_file.as_posix()
         config_properties["dir"] = self.workload.working_dir.as_posix()
 
-        # bind to all interfaces
         config_properties["bind"] = self.state.endpoint
 
         # replica related config
@@ -93,7 +92,7 @@ def get_config_properties(self, primary_endpoint: str) -> dict[str, str]:
 
     def _generate_replica_config(self, primary_endpoint: str) -> dict[str, str]:
         """Generate the config properties related to replica configuration based on the current cluster state."""
-        local_unit_endpoint = self.state.unit_server.get_endpoint(self.state.substrate)
+        local_unit_endpoint = self.state.endpoint
         replica_config = {
             "primaryuser": CharmUsers.VALKEY_REPLICA.value,
             "primaryauth": self.state.cluster.internal_users_credentials.get(

src/managers/tls.py

@@ -246,6 +246,53 @@ def start_ca_rotation_if_required(
         )
         return True
 
+    def get_current_sans(self) -> dict[str, set[str]]:
+        """Get the current SANs for a unit's cert."""
+        cert_file = self.workload.tls_paths.client_cert
+
+        sans_ip = set()
+        sans_dns = set()
+        if not (
+            san_lines := self.workload.exec(
+                [
+                    "openssl",
+                    "x509",
+                    "-ext",
+                    "subjectAltName",
+                    "-noout",
+                    "-in",
+                    cert_file.as_posix(),
+                ]
+            )[0].splitlines()
+        ):
+            return {"sans_ip": sans_ip, "sans_dns": sans_dns}
+
+        for line in san_lines:
+            for sans in line.split(", "):
+                san_type, san_value = sans.split(":")
+
+                if san_type.strip() == "DNS":
+                    sans_dns.add(san_value)
+                if san_type.strip() == "IP Address":
+                    sans_ip.add(san_value)
+
+        return {"sans_ip": sans_ip, "sans_dns": sans_dns}
+
+    def certificate_sans_require_update(self) -> bool:
+        """Check current certificate sans and determine if certificate requires update.
+
+        Returns:
+            bool: True if certificate sans have changed, False if they are still the same.
+        """
+        current_sans = self.get_current_sans()
+        new_sans_ip = self.build_sans_ip()
+        new_sans_dns = self.build_sans_dns()
+
+        if new_sans_ip ^ current_sans["sans_ip"] or new_sans_dns ^ current_sans["sans_dns"]:
+            return True
+
+        return False
+
     def get_statuses(self, scope: Scope, recompute: bool = False) -> list[StatusObject]:
         """Compute the TLS statuses."""
         status_list: list[StatusObject] = []

src/statuses.py

@@ -34,6 +34,18 @@ class ClusterStatuses(Enum):
         running="async",
     )
 
+    VALKEY_UNHEALTHY_RESTART = StatusObject(
+        status="maintenance",
+        message="Valkey unhealthy after restart",
+        running="async",
+    )
+
+    SENTINEL_UNHEALTHY_RESTART = StatusObject(
+        status="maintenance",
+        message="Sentinel unhealthy after restart",
+        running="async",
+    )
+
 
 class StartStatuses(Enum):
     """Collection of possible statuses related to the service start."""