[MISC] Conditionally create the legacy interface to support Juju 4 (#368)

* Conditionally create the legacy interface to support Juju 4

* Update libs

* Unhandled exception

* Longer idle period

* Increse timeout

* Retry restart check

* Disable lxd restart test

Author: dragomirp

lib/charms/data_platform_libs/v0/data_interfaces.py

@@ -331,7 +331,7 @@ def _on_topic_requested(self, event: TopicRequestedEvent):
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 46
+LIBPATCH = 49
 
 PYDEPS = ["ops>=2.0.0"]
 
@@ -2569,7 +2569,7 @@ def __init__(
 
 
 ################################################################################
-# Cross-charm Relatoins Data Handling and Evenets
+# Cross-charm Relations Data Handling and Events
 ################################################################################
 
 # Generic events
@@ -3268,7 +3268,7 @@ def __init__(
 # Kafka Events
 
 
-class KafkaProvidesEvent(RelationEvent):
+class KafkaProvidesEvent(RelationEventWithSecret):
     """Base class for Kafka events."""
 
     @property
@@ -3287,6 +3287,40 @@ def consumer_group_prefix(self) -> Optional[str]:
 
         return self.relation.data[self.relation.app].get("consumer-group-prefix")
 
+    @property
+    def mtls_cert(self) -> Optional[str]:
+        """Returns TLS cert of the client."""
+        if not self.relation.app:
+            return None
+
+        if not self.secrets_enabled:
+            raise SecretsUnavailableError("Secrets unavailable on current Juju version")
+
+        secret_field = f"{PROV_SECRET_PREFIX}{SECRET_GROUPS.MTLS}"
+        if secret_uri := self.relation.data[self.app].get(secret_field):
+            secret = self.framework.model.get_secret(id=secret_uri)
+            content = secret.get_content(refresh=True)
+            if content:
+                return content.get("mtls-cert")
+
+
+class KafkaClientMtlsCertUpdatedEvent(KafkaProvidesEvent):
+    """Event emitted when the mtls relation is updated."""
+
+    def __init__(self, handle, relation, old_mtls_cert: Optional[str] = None, app=None, unit=None):
+        super().__init__(handle, relation, app, unit)
+
+        self.old_mtls_cert = old_mtls_cert
+
+    def snapshot(self):
+        """Return a snapshot of the event."""
+        return super().snapshot() | {"old_mtls_cert": self.old_mtls_cert}
+
+    def restore(self, snapshot):
+        """Restore the event from a snapshot."""
+        super().restore(snapshot)
+        self.old_mtls_cert = snapshot["old_mtls_cert"]
+
 
 class TopicRequestedEvent(KafkaProvidesEvent, ExtraRoleEvent):
     """Event emitted when a new topic is requested for use on this relation."""
@@ -3299,6 +3333,7 @@ class KafkaProvidesEvents(CharmEvents):
     """
 
     topic_requested = EventSource(TopicRequestedEvent)
+    mtls_cert_updated = EventSource(KafkaClientMtlsCertUpdatedEvent)
 
 
 class KafkaRequiresEvent(RelationEvent):
@@ -3416,6 +3451,13 @@ def __init__(self, charm: CharmBase, relation_data: KafkaProviderData) -> None:
     def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         """Event emitted when the relation has changed."""
         super()._on_relation_changed_event(event)
+
+        new_data_keys = list(event.relation.data[event.app].keys())
+        if any(newval for newval in new_data_keys if self.relation_data._is_secret_field(newval)):
+            self.relation_data._register_secrets_to_relation(event.relation, new_data_keys)
+
+        getattr(self.on, "mtls_cert_updated").emit(event.relation, app=event.app, unit=event.unit)
+
         # Leader only
         if not self.relation_data.local_unit.is_leader():
             return
@@ -3430,6 +3472,33 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
                 event.relation, app=event.app, unit=event.unit
             )
 
+    def _on_secret_changed_event(self, event: SecretChangedEvent):
+        """Event notifying about a new value of a secret."""
+        if not event.secret.label:
+            return
+
+        relation = self.relation_data._relation_from_secret_label(event.secret.label)
+        if not relation:
+            logging.info(
+                f"Received secret {event.secret.label} but couldn't parse, seems irrelevant"
+            )
+            return
+
+        if relation.app == self.charm.app:
+            logging.info("Secret changed event ignored for Secret Owner")
+
+        remote_unit = None
+        for unit in relation.units:
+            if unit.app != self.charm.app:
+                remote_unit = unit
+
+        old_mtls_cert = event.secret.get_content().get("mtls-cert")
+        # mtls-cert is the only secret that can be updated
+        logger.info("mtls-cert updated")
+        getattr(self.on, "mtls_cert_updated").emit(
+            relation, app=relation.app, unit=remote_unit, old_mtls_cert=old_mtls_cert
+        )
+
 
 class KafkaProvides(KafkaProviderData, KafkaProviderEventHandlers):
     """Provider-side of the Kafka relation."""
@@ -3450,11 +3519,18 @@ def __init__(
         extra_user_roles: Optional[str] = None,
         consumer_group_prefix: Optional[str] = None,
         additional_secret_fields: Optional[List[str]] = [],
+        mtls_cert: Optional[str] = None,
     ):
         """Manager of Kafka client relations."""
         super().__init__(model, relation_name, extra_user_roles, additional_secret_fields)
         self.topic = topic
         self.consumer_group_prefix = consumer_group_prefix or ""
+        self.mtls_cert = mtls_cert
+
+    @staticmethod
+    def is_topic_value_acceptable(topic_value: str) -> bool:
+        """Check whether the given Kafka topic value is acceptable."""
+        return "*" not in topic_value[:3]
 
     @property
     def topic(self):
@@ -3463,11 +3539,19 @@ def topic(self):
 
     @topic.setter
     def topic(self, value):
-        # Avoid wildcards
-        if value == "*":
-            raise ValueError(f"Error on topic '{value}', cannot be a wildcard.")
+        if not self.is_topic_value_acceptable(value):
+            raise ValueError(f"Error on topic '{value}', unacceptable value.")
         self._topic = value
 
+    def set_mtls_cert(self, relation_id: int, mtls_cert: str) -> None:
+        """Set the mtls cert in the application relation databag / secret.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            mtls_cert: mtls cert.
+        """
+        self.update_relation_data(relation_id, {"mtls-cert": mtls_cert})
+
 
 class KafkaRequirerEventHandlers(RequirerEventHandlers):
     """Requires-side of the Kafka relation."""
@@ -3489,6 +3573,9 @@ def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
         # Sets topic, extra user roles, and "consumer-group-prefix" in the relation
         relation_data = {"topic": self.relation_data.topic}
 
+        if self.relation_data.mtls_cert:
+            relation_data["mtls-cert"] = self.relation_data.mtls_cert
+
         if self.relation_data.extra_user_roles:
             relation_data["extra-user-roles"] = self.relation_data.extra_user_roles
 
@@ -3547,15 +3634,17 @@ def __init__(
         extra_user_roles: Optional[str] = None,
         consumer_group_prefix: Optional[str] = None,
         additional_secret_fields: Optional[List[str]] = [],
+        mtls_cert: Optional[str] = None,
     ) -> None:
         KafkaRequirerData.__init__(
             self,
             charm.model,
             relation_name,
             topic,
-            extra_user_roles,
-            consumer_group_prefix,
-            additional_secret_fields,
+            extra_user_roles=extra_user_roles,
+            consumer_group_prefix=consumer_group_prefix,
+            additional_secret_fields=additional_secret_fields,
+            mtls_cert=mtls_cert,
         )
         KafkaRequirerEventHandlers.__init__(self, charm, self)
 
@@ -3675,6 +3764,10 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
                 event.relation, app=event.app, unit=event.unit
             )
 
+    def _on_secret_changed_event(self, event: SecretChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        pass
+
 
 class OpenSearchProvides(OpenSearchProvidesData, OpenSearchProvidesEventHandlers):
     """Provider-side of the OpenSearch relation."""

src/charm.py

@@ -27,8 +27,6 @@
 
 logger = logging.getLogger(__name__)
 
-pgsql = ops.lib.use("pgsql", 1, "postgresql-charmers@lists.launchpad.net")
-
 PEER = "postgresql-test-peers"
 # Expected tmp access
 LAST_WRITTEN_FILE = "/tmp/last_written_value"  # noqa: S108
@@ -151,10 +149,12 @@ def __init__(self, *args):
         self.no_database = DatabaseRequires(self, "no-database", database_name="")
 
         # Legacy interface
-        self.db = pgsql.PostgreSQLClient(self, "db")
-        self.framework.observe(
-            self.db.on.database_relation_joined, self._on_database_relation_joined
-        )
+        if self.model.juju_version.major < 4:
+            pgsql = ops.lib.use("pgsql", 1, "postgresql-charmers@lists.launchpad.net")
+            self.db = pgsql.PostgreSQLClient(self, "db")
+            self.framework.observe(
+                self.db.on.database_relation_joined, self._on_database_relation_joined
+            )
 
         self.framework.observe(self.on.run_sql_action, self._on_run_sql_action)
         self.framework.observe(self.on.test_tls_action, self._on_test_tls_action)
@@ -338,6 +338,7 @@ def _on_start_continuous_writes_action(self, event: ActionEvent) -> None:
             logger.exception("Unable to stop writes to create table", exc_info=e)
             return
 
+        connection = None
         try:
             # Create the table to write records on and also a unique index to prevent duplicate
             # writes.
@@ -355,12 +356,14 @@ def _on_start_continuous_writes_action(self, event: ActionEvent) -> None:
             logger.exception("Unable to create table", exc_info=e)
             return
         finally:
-            connection.close()
+            if connection:
+                connection.close()
 
         self._start_continuous_writes(1)
         event.set_results({"result": "True"})
 
     def _get_db_writes(self) -> int:
+        connection = None
         try:
             with (
                 psycopg2.connect(self._connection_string) as connection,
@@ -373,7 +376,8 @@ def _get_db_writes(self) -> int:
             writes = -1
             logger.exception("Unable to count writes")
         finally:
-            connection.close()
+            if connection:
+                connection.close()
         return writes
 
     def _on_show_continuous_writes_action(self, event: ActionEvent) -> None:
@@ -436,10 +440,7 @@ def _stop_continuous_writes(self) -> int | None:
         return last_written_value
 
     # Legacy event handlers
-    def _on_database_relation_joined(
-        self,
-        event: pgsql.DatabaseRelationJoinedEvent,  # type: ignore
-    ) -> None:
+    def _on_database_relation_joined(self, event) -> None:
         """Handle db-relation-joined.
 
         Args:

tests/integration/test_smoke.py

@@ -6,6 +6,7 @@
 import logging
 import time
 
+import pytest
 from juju.relation import Relation
 from lightkube.core.client import Client
 from lightkube.resources.core_v1 import Pod
@@ -63,7 +64,7 @@ async def test_smoke(ops_test: OpsTest, charm) -> None:
     await integrate(ops_test, postgresql, pgbouncer)
     await integrate(ops_test, f"{TEST_APP_NAME}:database", pgbouncer)
     await ops_test.model.wait_for_idle(
-        apps=[postgresql, pgbouncer, TEST_APP_NAME], status="active", timeout=1000
+        apps=[postgresql, pgbouncer, TEST_APP_NAME], status="active", timeout=1000, idle_period=30
     )
 
     logger.info("Test continuous writes")
@@ -142,11 +143,14 @@ async def test_restart(ops_test: OpsTest) -> None:
         client = Client(namespace=ops_test.model.info.name)
         client.delete(Pod, name=f"{TEST_APP_NAME}-0")
     else:
+        pytest.skip("Unstable LXC restart test")
         logger.info("Restarting lxc")
         await restart_machine(ops_test, ops_test.model.applications[TEST_APP_NAME].units[0].name)
 
     logger.info("Wait for idle")
-    await ops_test.model.wait_for_idle(apps=[TEST_APP_NAME], status="active", timeout=600)
+    await ops_test.model.wait_for_idle(
+        apps=[TEST_APP_NAME], status="active", timeout=600, idle_period=30
+    )
 
     logger.info("Check that writes are increasing")
     results = await (