[DPE-9751] add K8s services for primary and replica endpoints

charmcraft.yaml

@@ -77,8 +77,8 @@ parts:
       # rpds-py (Python package) >=0.19.0 requires rustc >=1.76, which is not available in the
       # Ubuntu 22.04 archive. Install rustc and cargo using rustup instead of the Ubuntu archive
       rustup set profile minimal
-      rustup default 1.90.0  # renovate: charmcraft-rust-latest
-
+      rustup default 1.94.0  # renovate: charmcraft-rust-latest
+      
       craftctl default
       # Include requirements.txt in *.charm artifact for easier debugging
       cp requirements.txt "$CRAFT_PART_INSTALL/requirements.txt"

pyproject.toml

@@ -16,6 +16,7 @@ tenacity = "*"
 data-platform-helpers = ">=0.1.7"
 validators = ">=0.35.0"
 dpcharmlibs-interfaces = ">=1.0.2"
+lightkube = ">=0.19.0"
 
 [tool.poetry.requires-plugins]
 poetry-plugin-export = ">=1.8"

src/common/exceptions.py

@@ -66,3 +66,7 @@ class RequestingLockTimedOutError(Exception):
 
 class ValkeyCertificatesNotReadyError(Exception):
     """Custom Exception if not all units have stored the TLS certificates."""
+
+
+class KubernetesClientError(Exception):
+    """Custom Exception if a connection to the Kubernetes Cluster API fails."""

src/common/k8s_client.py

@@ -0,0 +1,106 @@
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""K8sClient utility class to connect to the Kubernetes API server."""
+
+import logging
+
+from lightkube.core.client import Client
+from lightkube.core.exceptions import ApiError
+from lightkube.models.core_v1 import ServicePort, ServiceSpec
+from lightkube.models.meta_v1 import ObjectMeta
+from lightkube.resources.core_v1 import Pod, Service
+from lightkube.types import PatchType
+
+from common.exceptions import KubernetesClientError
+
+logger = logging.getLogger(__name__)
+
+
+class K8sClient:
+    """Expose Kubernetes API commands to the charm."""
+
+    def __init__(self, namespace: str, app_name: str):
+        self.namespace = namespace
+        self.app_name = app_name
+        self.client = Client()
+
+    def ensure_endpoint_service(self, role: str, port: int) -> None:
+        """Create or update a K8s service.
+
+        Args:
+            role(str): name of the role to create the service for, e.g "primary" or "replicas"
+            port(int): the port number to set for the service
+        """
+        service_name = f"{self.app_name}-{role}"
+        service_port = ServicePort(port=port, targetPort=port)
+
+        try:
+            service = self.client.get(res=Service, name=service_name, namespace=self.namespace)
+            if service.spec.ports != [service_port]:
+                service.spec.ports = [service_port]
+                self.client.patch(Service, service_name, service, patch_type=PatchType.MERGE)
+                logger.info("Updated Kubernetes service %s to port %s", service_name, port)
+            return
+        except ApiError as e:
+            # 404 will be raised if service does not exist yet
+            if e.status.code != 404:
+                raise KubernetesClientError from e
+
+        try:
+            pod0 = self.client.get(
+                res=Pod,
+                name=self.app_name + "-0",
+                namespace=self.namespace,
+            )
+        except ApiError as e:
+            raise KubernetesClientError from e
+
+        service = Service(
+            apiVersion="v1",
+            kind="Service",
+            metadata=ObjectMeta(
+                namespace=self.namespace,
+                name=service_name,
+                ownerReferences=pod0.metadata.ownerReferences,
+            ),
+            spec=ServiceSpec(
+                selector={"application-name": self.app_name, "role": role},
+                ports=[service_port],
+                type="ClusterIP",
+            ),
+        )
+
+        try:
+            self.client.create(service)
+            logger.info("Created Kubernetes service %s for port %s", service_name, port)
+        except ApiError as e:
+            logger.error("Kubernetes service creation failed: %s", e)
+            raise KubernetesClientError from e
+
+    def update_pod_label(self, pod_name: str, role: str) -> None:
+        """Create or update a label for a pod.
+
+        Args:
+            pod_name(str): the name of the pod
+            role(str): name of the role to create the label for, e.g "primary" or "replicas"
+        """
+        try:
+            pod = self.client.get(Pod, pod_name, namespace=self.namespace)
+        except ApiError as e:
+            raise KubernetesClientError from e
+
+        if not pod.metadata.labels:
+            pod.metadata.labels = {}
+
+        if pod.metadata.labels.get("role") == role:
+            return
+
+        logger.info("Updating pod %s to role %s", pod_name, role)
+        pod.metadata.labels["application-name"] = self.app_name
+        pod.metadata.labels["role"] = role
+        try:
+            self.client.patch(Pod, pod_name, pod)
+        except ApiError as e:
+            logger.error("Failed to update Kubernetes pod labels: %s", e)
+            raise KubernetesClientError from e

src/events/external_clients.py

@@ -23,13 +23,19 @@
 )
 
 from common.exceptions import (
+    KubernetesClientError,
     ValkeyACLLoadError,
     ValkeyCannotGetPrimaryIPError,
     ValkeyServicesFailedToStartError,
     ValkeyTLSLoadError,
     ValkeyWorkloadCommandError,
 )
-from literals import CERTIFICATE_TRANSFER_RELATION, EXTERNAL_CLIENTS_RELATION, PEER_RELATION
+from literals import (
+    CERTIFICATE_TRANSFER_RELATION,
+    EXTERNAL_CLIENTS_RELATION,
+    PEER_RELATION,
+    Substrate,
+)
 
 if TYPE_CHECKING:
     from charm import ValkeyCharm
@@ -178,15 +184,26 @@ def _on_bulk_resources_requested(
         self.charm.state.cluster.update({"client_user_epoch": time.time()})
 
     def _on_peer_relation_changed(self, event: ops.RelationChangedEvent) -> None:
-        """Handle peer relation changes in regard to external client relations."""
-        if (
-            not self.charm.state.unit_server.is_started
-            or not self.charm.state.external_client_relations
-        ):
+        """Handle peer relation changes in regard to external client relations.
+
+        This handler catches all changes from scaling operations, TLS switchover, TLS CA rotation,
+        IP changes, etc.
+        """
+        if not self.charm.state.unit_server.is_started:
+            return
+
+        if self.charm.unit.is_leader() and self.charm.state.substrate == Substrate.K8S:
+            try:
+                self.charm.sentinel_manager.reconcile_k8s_services()
+            except (KubernetesClientError, ValkeyCannotGetPrimaryIPError) as e:
+                logger.error("Error updating Kubernetes services: %s", e)
+                event.defer()
+                return
+
+        if not self.charm.state.external_client_relations:
             return
 
         if self.charm.unit.is_leader():
-            # this catches all changes from scaling operations, TLS switchover, IP changes, etc.
             try:
                 self._update_client_relations()
             except (ValkeyCannotGetPrimaryIPError, ValkeyWorkloadCommandError) as e:

src/literals.py

@@ -122,3 +122,10 @@ class TLSCARotationState(StrEnum):
     NEW_CA_DETECTED = "new-ca-detected"
     NEW_CA_ADDED = "new-ca-added"
     CA_UPDATED = "ca-updated"
+
+
+class K8sService(StrEnum):
+    """Services managed by the charm in Kubernetes."""
+
+    PRIMARY = "primary"
+    REPLICAS = "replicas"