Kf 739 distribute secrets (#6)

* Create relation "secrets" which allows to distribute secrets manifests to resource dispatcher. Secrets are distributed across namespaces.
* On relation broken remove the secrets manifests of the relation from resource dispatcher container.
* Create manifest tester charm for integration tests to test the relation.

.gitignore

@@ -11,3 +11,5 @@ __pycache__/
 kubeconfig.tmp
 new_config
 .mypy_cache
+
+namespace-example.yaml

lib/charms/observability_libs/v1/kubernetes_service_patch.py

@@ -146,7 +146,7 @@ def setUp(self, *unused):
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 5
+LIBPATCH = 6
 
 ServiceType = Literal["ClusterIP", "LoadBalancer"]
 
@@ -201,6 +201,7 @@ def __init__(
         # Ensure this patch is applied during the 'install' and 'upgrade-charm' events
         self.framework.observe(charm.on.install, self._patch)
         self.framework.observe(charm.on.upgrade_charm, self._patch)
+        self.framework.observe(charm.on.update_status, self._patch)
 
         # apply user defined events
         if refresh_event:

metadata.yaml

@@ -14,4 +14,17 @@ resources:
   oci-image:
     type: oci-image
     description: Backing OCI image
-    upstream-source: charmedkubeflow/resource-dispatcher:1.0-22.04
+    upstream-source: charmedkubeflow/resource-dispatcher:1.0_beta-22.04
+requires:
+  secrets:
+    interface: secrets
+    schema:
+      v1:
+        provides:
+          type: object
+          properties:
+            secrets:
+              type: string
+        required:
+        - secrets
+    versions: [v1]

src/charm.py

@@ -3,21 +3,25 @@
 # See LICENSE file for licensing details.
 #
 
+import json
 import logging
 
+import yaml
 from charmed_kubeflow_chisme.exceptions import ErrorWithStatus, GenericCharmRuntimeError
 from charmed_kubeflow_chisme.kubernetes import KubernetesResourceHandler
 from charmed_kubeflow_chisme.lightkube.batch import delete_many
 from charms.observability_libs.v1.kubernetes_service_patch import KubernetesServicePatch
 from lightkube import ApiError
 from lightkube.generic_resource import load_in_cluster_generic_resources
 from lightkube.models.core_v1 import ServicePort
-from ops.charm import CharmBase
+from ops.charm import CharmBase, RelationBrokenEvent
 from ops.main import main
-from ops.model import ActiveStatus, MaintenanceStatus, WaitingStatus
+from ops.model import ActiveStatus, BlockedStatus, MaintenanceStatus, WaitingStatus
 from ops.pebble import ChangeError, Layer
+from serialized_data_interface import NoCompatibleVersions, NoVersionsListed, get_interfaces
 
 K8S_RESOURCE_FILES = ["src/templates/composite-controller.yaml.j2"]
+DISPATCHER_SECRETS_PATH = "/app/resources"
 
 
 class ResourceDispatcherOperator(CharmBase):
@@ -44,8 +48,9 @@ def __init__(self, *args):
         self.framework.observe(self.on.config_changed, self._on_event)
         self.framework.observe(self.on.remove, self._on_remove)
 
-        # for rel in self.model.relations.keys():
-        #     self.framework.observe(self.on[rel].relation_changed, self._on_event)
+        for rel in self.model.relations.keys():
+            self.framework.observe(self.on[rel].relation_changed, self._on_event)
+            self.framework.observe(self.on[rel].relation_broken, self._on_event)
 
         port = ServicePort(int(self._port), name=f"{self.app.name}")
         self.service_patcher = KubernetesServicePatch(
@@ -107,6 +112,11 @@ def _check_leader(self):
             self.logger.info("Not a leader, skipping setup")
             raise ErrorWithStatus("Waiting for leadership", WaitingStatus)
 
+    def _check_container(self):
+        """Check if we can connect the container."""
+        if not self.container.can_connect():
+            raise ErrorWithStatus("Container is not ready", WaitingStatus)
+
     def _deploy_k8s_resources(self) -> None:
         """Deploys K8S resources."""
         try:
@@ -123,6 +133,16 @@ def _on_install(self, _):
         # deploy K8S resources to speed up deployment
         self._deploy_k8s_resources()
 
+    def _get_interfaces(self):
+        """Retrieve interface object."""
+        try:
+            interfaces = get_interfaces(self)
+        except NoVersionsListed as err:
+            raise ErrorWithStatus(err, WaitingStatus)
+        except NoCompatibleVersions as err:
+            raise ErrorWithStatus(err, BlockedStatus)
+        return interfaces
+
     def _update_layer(self) -> None:
         """Update the Pebble configuration layer (if changed)."""
         current_layer = self.container.get_plan()
@@ -136,13 +156,90 @@ def _update_layer(self) -> None:
             except ChangeError as err:
                 raise GenericCharmRuntimeError(f"Failed to replan with error: {str(err)}") from err
 
+    def _get_manifests(self, interfaces, relation, event):
+        """Unpacks and returns the manifests relation data."""
+        if not ((relation_interface := interfaces[relation]) and relation_interface.get_data()):
+            self.logger.info(f"No {relation} data presented in relation")
+            return None
+        try:
+            relations_data = {
+                (rel, app): route
+                for (rel, app), route in sorted(
+                    relation_interface.get_data().items(), key=lambda tup: tup[0][0].id
+                )
+                if app != self.app
+            }
+        except Exception as e:
+            raise ErrorWithStatus(
+                f"Unexpected error unpacking {relation} data - data format not "
+                f"as expected. Caught exception: '{str(e)}'",
+                BlockedStatus,
+            )
+        if isinstance(event, (RelationBrokenEvent)):
+            del relations_data[(event.relation, event.app)]
+
+        manifests = []
+        for relation_data in relations_data.values():
+            manifests += json.loads(relation_data[relation])
+        self.logger.debug(f"manifests are {manifests}")
+        return manifests
+
+    def _manifests_valid(self, manifests):
+        """Checks if manifests are unique."""
+        if manifests:
+            for manifest in manifests:
+                if (
+                    sum([m["metadata"]["name"] == manifest["metadata"]["name"] for m in manifests])
+                    > 1
+                ):
+                    return False
+        return True
+
+    def _sync_manifests(self, manifests, push_location):
+        """Push list of manifests into layer.
+
+        Args:
+            manifests: List of kubernetes manifests to be pushed to pebble layer.
+            push_location: Container location where the manifests should be pushed to.
+        """
+        all_files = self.container.list_files(push_location)
+        if manifests:
+            manifests_locations = [
+                f"{push_location}/{m['metadata']['name']}.yaml" for m in manifests
+            ]
+        else:
+            manifests_locations = []
+        if all_files:
+            for file in all_files:
+                if file.path not in manifests_locations:
+                    self.container.remove_path(file.path)
+        if manifests:
+            for manifest in manifests:
+                filename = manifest["metadata"]["name"]
+                self.container.push(f"{push_location}/{filename}.yaml", yaml.dump(manifest))
+
+    def _update_manifests(self, interfaces, dispatch_folder, relation, event):
+        """Get manifests from relation and update them in dispatcher folder."""
+        manifests = self._get_manifests(interfaces, relation, event)
+        if not self._manifests_valid(manifests):
+            self.logger.debug(
+                f"Manifests names in all relations must be unique {','.join(manifests)}"
+            )
+            raise ErrorWithStatus(
+                "Failed to process invalid manifest. See debug logs.",
+                BlockedStatus,
+            )
+        self.logger.debug(f"received {relation} are {manifests}")
+        self._sync_manifests(manifests, dispatch_folder)
+
     def _on_event(self, event) -> None:
         """Perform all required actions for the Charm."""
         try:
             self._check_leader()
-            self._deploy_k8s_resources()
-            # interfaces = self._get_interfaces()
+            self._check_container()
+            interfaces = self._get_interfaces()
             self._update_layer()
+            self._update_manifests(interfaces, DISPATCHER_SECRETS_PATH, "secrets", event)
         except ErrorWithStatus as err:
             self.model.unit.status = err.status
             self.logger.info(f"Event {event} stopped early with message: {str(err)}")

src/templates/composite-controller.yaml.j2

@@ -12,7 +12,7 @@ spec:
     - apiVersion: v1
       resource: secrets
       updateStrategy:
-        method: OnDelete
+        method: InPlace
   hooks:
     sync:
       webhook:

tests/integration/manifests-tester/.gitignore

@@ -0,0 +1,9 @@
+venv/
+build/
+*.charm
+.tox/
+.coverage
+__pycache__/
+*.py[cod]
+.idea
+.vscode/

tests/integration/manifests-tester/charmcraft.yaml

@@ -0,0 +1,14 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+type: "charm"
+bases:
+  - build-on:
+    - name: "ubuntu"
+      channel: "20.04"
+    run-on:
+    - name: "ubuntu"
+      channel: "20.04"
+parts:
+  charm:
+    charm-python-packages: [setuptools, pip]

tests/integration/manifests-tester/config.yaml

@@ -0,0 +1,9 @@
+# This file defines charm config options, and populates the Configure tab on Charmhub.
+# If your charm does not require configuration options, delete this file entirely.
+#
+# See https://juju.is/docs/config for guidance.
+
+options:
+  test_data:
+    default: src/secrets
+    type: string

tests/integration/manifests-tester/metadata.yaml

@@ -0,0 +1,20 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+name: manifests-tester
+summary: |
+  Charm for sending manifests to ResourceDispatcher relations.
+description: |
+  Charm for sending manifests to ResourceDispatcher relations.
+provides:
+  secrets:
+    interface: secrets
+    schema:
+      v1:
+        provides:
+          type: object
+          properties:
+            secrets:
+              type: string
+        required:
+        - secrets
+    versions: [v1]

tests/integration/manifests-tester/requirements.txt

@@ -0,0 +1,3 @@
+ops
+pyyaml
+serialized-data-interface

tests/integration/manifests-tester/src/charm.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+#
+
+"""Mock relation provider charms."""
+
+import glob
+import json
+import logging
+from pathlib import Path
+
+import yaml
+from ops.charm import CharmBase
+from ops.main import main
+from ops.model import ActiveStatus, BlockedStatus, WaitingStatus
+from serialized_data_interface import NoCompatibleVersions, NoVersionsListed, get_interfaces
+
+logger = logging.getLogger(__name__)
+
+
+class ManifestsTesterCharm(CharmBase):
+    """Charm for sending manifests to ResourceDispatcher relations."""
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self._name = "manifests-tester"
+        self._secrets_folder = self.model.config["test_data"]
+
+        self.framework.observe(self.on.start, self._on_start)
+        self.framework.observe(self.on.config_changed, self._on_event)
+
+        for rel in self.model.relations.keys():
+            self.framework.observe(self.on[rel].relation_changed, self._on_event)
+
+    def _on_start(self, _):
+        """Set active on start."""
+        self.model.unit.status = ActiveStatus()
+
+    def _get_interfaces(self):
+        """Retrieve interface object."""
+        try:
+            interfaces = get_interfaces(self)
+        except NoVersionsListed:
+            self.model.unit.status = WaitingStatus()
+            return {"secrets": None}
+        except NoCompatibleVersions:
+            self.model.unit.status = BlockedStatus()
+            return {"secrets": None}
+        return interfaces
+
+    def _send_manifests(self, interfaces, folder, relation):
+        """Send manifests from folder to desired relation."""
+        if relation in interfaces and interfaces[relation]:
+            manifests = []
+            logger.info(f"Scanning folder {folder}")
+            manifest_files = glob.glob(f"{folder}/*.yaml")
+            for file in manifest_files:
+                manifest = yaml.safe_load(Path(file).read_text())
+                manifests.append(manifest)
+            data = {relation: json.dumps(manifests)}
+            interfaces[relation].send_data(data)
+
+    def _on_event(self, _) -> None:
+        """Perform all required actions for the Charm."""
+        interfaces = self._get_interfaces()
+        self._send_manifests(interfaces, self._secrets_folder, "secrets")
+        self.model.unit.status = ActiveStatus()
+
+
+if __name__ == "__main__":  # pragma: nocover
+    main(ManifestsTesterCharm)

tests/integration/manifests-tester/src/secrets/secret1.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: mlpipeline-minio-artifact
+stringData:
+  AWS_ACCESS_KEY_ID: access_key
+  AWS_SECRET_ACCESS_KEY: secret_access_key

tests/integration/manifests-tester/src/secrets/secret2.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: seldon-rclone-secret
+stringData:
+  RCLONE_CONFIG_MYS3_TYPE: test
+  RCLONE_CONFIG_MYS3_PROVIDER: test
+  RCLONE_CONFIG_MYS3_ACCESS_KEY_ID: test
+  RCLONE_CONFIG_MYS3_SECRET_ACCESS_KEY: test
+  RCLONE_CONFIG_MYS3_ENDPOINT: test

tests/integration/manifests-tester/src/secrets2/secret1.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: mlpipeline-minio-artifact2
+stringData:
+  AWS_ACCESS_KEY_ID: access_key
+  AWS_SECRET_ACCESS_KEY: secret_access_key

tests/integration/manifests-tester/src/secrets2/secret2.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: seldon-rclone-secret2
+stringData:
+  RCLONE_CONFIG_MYS3_TYPE: test
+  RCLONE_CONFIG_MYS3_PROVIDER: test
+  RCLONE_CONFIG_MYS3_ACCESS_KEY_ID: test
+  RCLONE_CONFIG_MYS3_SECRET_ACCESS_KEY: test
+  RCLONE_CONFIG_MYS3_ENDPOINT: test