[DPE-5097] - test: add mtls int-tests (#133)

.github/workflows/sync_docs.yaml

@@ -26,4 +26,4 @@ jobs:
       - name: Show migrate output
         run: echo '${{ steps.docs-pr.outputs.migrate }}'
       - name: Show reconcile output
-        run: echo '${{ steps.docs-pr.outputs.reconcile }}'
+        run: echo '${{ steps.docs-pr.outputs.reconcile }}'

src/core/models.py

@@ -547,23 +547,23 @@ def pod(self) -> Pod:
 
         K8s-only.
         """
-        return self.k8s.get_pod(pod_name=self.pod_name)
+        return self.k8s.get_pod(self.pod_name)
 
     @cached_property
     def node(self) -> Node:
         """The Node the unit is scheduled on.
 
         K8s-only.
         """
-        return self.k8s.get_node(pod=self.pod)
+        return self.k8s.get_node(self.pod_name)
 
     @cached_property
     def node_ip(self) -> str:
         """The IPV4/IPV6 IP address the Node the unit is on.
 
         K8s-only.
         """
-        return self.k8s.get_node_ip(node=self.node)
+        return self.k8s.get_node_ip(self.pod_name)
 
 
 class ZooKeeper(RelationState):
@@ -700,7 +700,7 @@ def zookeeper_version(self) -> str:
     # retry to give ZK time to update its broker zNodes before failing
     @retry(
         wait=wait_fixed(5),
-        stop=stop_after_attempt(10),
+        stop=stop_after_attempt(3),
         retry=retry_if_result(lambda result: result is False),
         retry_error_callback=lambda _: False,
     )

src/events/broker.py

@@ -163,13 +163,13 @@ def _on_start(self, event: StartEvent | PebbleReadyEvent) -> None:
         if not self.upgrade.idle:
             return
 
+        self.update_external_services()
+
         self.charm._set_status(self.charm.state.ready_to_start)
         if not isinstance(self.charm.unit.status, ActiveStatus):
             event.defer()
             return
 
-        self.update_external_services()
-
         # required settings given zookeeper connection config has been created
         self.config_manager.set_server_properties()
         self.config_manager.set_zk_jaas_config()

src/events/zookeeper.py

@@ -78,8 +78,12 @@ def _on_zookeeper_changed(self, event: RelationChangedEvent) -> None:
 
             try:
                 internal_user_credentials = self._create_internal_credentials()
-            except (KeyError, RuntimeError, subprocess.CalledProcessError, ExecError) as e:
-                logger.warning(str(e))
+            except (KeyError, RuntimeError) as e:
+                logger.warning(e)
+                event.defer()
+                return
+            except (subprocess.CalledProcessError, ExecError) as e:
+                logger.warning(f"{e.stdout}, {e.stderr}")
                 event.defer()
                 return
 

src/managers/k8s.py

@@ -4,8 +4,11 @@
 
 """Manager for handling Kafka Kubernetes resources for a single Kafka pod."""
 
+import json
 import logging
-from functools import cached_property
+import math
+import time
+from functools import cache
 
 from lightkube.core.client import Client
 from lightkube.core.exceptions import ApiError
@@ -15,13 +18,12 @@
 
 from literals import SECURITY_PROTOCOL_PORTS, AuthMap, AuthMechanism
 
-logger = logging.getLogger(__name__)
-
 # default logging from lightkube httpx requests is very noisy
-logging.getLogger("lightkube").disabled = True
-logging.getLogger("lightkube.core.client").disabled = True
-logging.getLogger("httpx").disabled = True
-logging.getLogger("httpcore").disabled = True
+logging.getLogger("lightkube").setLevel(logging.CRITICAL)
+logging.getLogger("httpx").setLevel(logging.CRITICAL)
+logging.getLogger("httpcore").setLevel(logging.CRITICAL)
+
+logger = logging.getLogger(__name__)
 
 
 class K8sManager:
@@ -42,54 +44,57 @@ def __init__(
             "SSL": "ssl",
         }
 
-    @cached_property
+    def __eq__(self, other: object) -> bool:
+        """__eq__ dunder.
+
+        Needed to get an cache hit on calls on the same method from different instances of K8sManager
+        as `self` is passed to methods.
+        """
+        return isinstance(other, K8sManager) and self.__dict__ == other.__dict__
+
+    def __hash__(self) -> int:
+        """__hash__ dunder.
+
+        K8sManager needs to be hashable so that `self` can be passed to the 'dict-like' cache.
+        """
+        return hash(json.dumps(self.__dict__, sort_keys=True))
+
+    @property
     def client(self) -> Client:
         """The Lightkube client."""
         return Client(  # pyright: ignore[reportArgumentType]
             field_manager=self.pod_name,
             namespace=self.namespace,
         )
 
+    @staticmethod
+    def get_ttl_hash(seconds=60 * 2) -> int:
+        """Gets a unique time hash for the cache, expiring after 2 minutes.
+
+        When 2m has passed, a new value will be created, ensuring an cache miss
+        and a re-loading of that K8s API call.
+        """
+        return math.floor(time.time() / seconds)
+
     # --- GETTERS ---
 
     def get_pod(self, pod_name: str = "") -> Pod:
         """Gets the Pod via the K8s API."""
-        # Allows us to get pods from other peer units
-        pod_name = pod_name or self.pod_name
-
-        return self.client.get(
-            res=Pod,
-            name=self.pod_name,
-        )
+        return self._get_pod(pod_name, self.get_ttl_hash())
 
-    def get_node(self, pod: Pod) -> Node:
+    def get_node(self, pod_name: str) -> Node:
         """Gets the Node the Pod is running on via the K8s API."""
-        if not pod.spec or not pod.spec.nodeName:
-            raise Exception("Could not find podSpec or nodeName")
-
-        return self.client.get(
-            Node,
-            name=pod.spec.nodeName,
-        )
-
-    def get_node_ip(self, node: Node) -> str:
-        """Gets the IP Address of the Node via the K8s API."""
-        # all these redundant checks are because Lightkube's typing is awful
-        if not node.status or not node.status.addresses:
-            raise Exception(f"No status found for {node}")
+        return self._get_node(pod_name, self.get_ttl_hash())
 
-        for addresses in node.status.addresses:
-            if addresses.type in ["ExternalIP", "InternalIP", "Hostname"]:
-                return addresses.address
-
-        return ""
+    def get_node_ip(self, pod_name: str) -> str:
+        """Gets the IP Address of the Node of a given Pod via the K8s API."""
+        return self._get_node_ip(pod_name, self.get_ttl_hash())
 
     def get_service(self, service_name: str) -> Service | None:
         """Gets the Service via the K8s API."""
-        return self.client.get(
-            res=Service,
-            name=service_name,
-        )
+        return self._get_service(service_name, self.get_ttl_hash())
+
+    # SERVICE BUILDERS
 
     def get_node_port(
         self,
@@ -139,7 +144,7 @@ def get_bootstrap_nodeport(self, auth_map: AuthMap) -> int:
 
     def build_bootstrap_services(self) -> Service:
         """Builds a ClusterIP service for initial client connection."""
-        pod = self.get_pod(pod_name=self.pod_name)
+        pod = self.get_pod(self.pod_name)
         if not pod.metadata:
             raise Exception(f"Could not find metadata for {pod}")
 
@@ -231,3 +236,46 @@ def apply_service(self, service: Service) -> None:
                 return
             else:
                 raise
+
+    # PRIVATE METHODS
+
+    @cache
+    def _get_pod(self, pod_name: str = "", *_) -> Pod:
+        # Allows us to get pods from other peer units
+        pod_name = pod_name or self.pod_name
+
+        return self.client.get(
+            res=Pod,
+            name=pod_name,
+        )
+
+    @cache
+    def _get_node(self, pod_name: str, *_) -> Node:
+        pod = self.get_pod(pod_name)
+        if not pod.spec or not pod.spec.nodeName:
+            raise Exception("Could not find podSpec or nodeName")
+
+        return self.client.get(
+            Node,
+            name=pod.spec.nodeName,
+        )
+
+    @cache
+    def _get_node_ip(self, pod_name: str, *_) -> str:
+        # all these redundant checks are because Lightkube's typing is awful
+        node = self.get_node(pod_name)
+        if not node.status or not node.status.addresses:
+            raise Exception(f"No status found for {node}")
+
+        for addresses in node.status.addresses:
+            if addresses.type in ["ExternalIP", "InternalIP", "Hostname"]:
+                return addresses.address
+
+        return ""
+
+    @cache
+    def _get_service(self, service_name: str, *_) -> Service | None:
+        return self.client.get(
+            res=Service,
+            name=service_name,
+        )

tests/integration/app-charm/actions.yaml

@@ -3,5 +3,36 @@
 
 produce:
   description: Produces messages to a test-topic
+
 consume:
   description: Consumes messages from a test-topic
+
+create-certificate:
+  description: Creates JKS keystore and signed certificate on unit
+
+run-mtls-producer:
+  description: Runs producer
+  params:
+    mtls-nodeport:
+      type: integer
+      description: The NodePort for the mTLS bootstrap service
+    broker-ca:
+      type: string
+      description: The CA used for broker identity from certificates relation
+    num-messages:
+      type: integer
+      description: The number of messages to be sent for testing
+
+get-offsets:
+  description: Retrieve offset for test topic
+  params:
+    mtls-nodeport:
+      type: integer
+      description: The NodePort for the mTLS bootstrap service
+
+create-topic:
+  description: Attempts the configured topic
+  params:
+    bootstrap-server:
+      type: string
+      description: The address for SASL_PLAINTEXT Kafka

tests/integration/app-charm/config.yaml

@@ -0,0 +1,6 @@
+options:
+  topic-name:
+    description: |
+      The topic-name to request when relating to the Kafka application
+    type: string
+    default: test-topic