Add support for pebble log forwarding

lib/charms/loki_k8s/v1/loki_push_api.py

@@ -20,6 +20,9 @@
 send telemetry, such as logs, to Loki through a Log Proxy by implementing the consumer side of the
 `loki_push_api` relation interface.
 
+- 'LogForwarder': This object can be used by any Charmed Operator which needs to send its
+standard output (stdout) to Loki, through Pebble's log forwarding mechanism.
+
 Filtering logs in Loki is largely performed on the basis of labels. In the Juju ecosystem, Juju
 topology labels are used to uniquely identify the workload which generates telemetry like logs.
 
@@ -349,6 +352,72 @@ def _promtail_error(self, event):
     )
 ```
 
+## LogForwarder Library Usage
+
+Let's say that we have a workload charm that produces logs to the standard output (stdout),
+and we need to send those logs to a workload implementing the `loki_push_api` interface,
+such as `Loki` or `Grafana Agent`.
+
+Use the `LogForwarder` class by instantiating it in the `__init__` method of the charmed
+operator. There are two ways to provide Loki endpoint(s) to the forwarder. You can let the
+object extract the endpoint(s) from relation data, or you can explicitly pass the Loki
+endpoint(s) to forward your logs to.
+
+1. If you want your charm to relate to another implementing the `loki_push_api` interface,
+   you only need to instantiate the object; for example:
+
+   ```python
+   from charms.loki_k8s.v1.loki_push_api import LogForwarder
+
+   ...
+
+      def __init__(self, *args):
+          ...
+          self._log_forwarder = LogForwarder(self)
+   ```
+
+   The `LogForwarder` will listen to relation events out-of-the-box and enable or disable
+   the log forwarding accordingly. Next, modify the `metadata.yaml` file to add:
+
+   - The `log-forwarding` relation in the `requires` section:
+     ```yaml
+     requires:
+       log-forwarding:
+         interface: loki_push_api
+         optional: true
+     ```
+
+2. If you don't want to relate your charm to another implementing the `loki_push_api` interface,
+   you need to explicitly provide the endpoint(s) to forward your logs to. If your charm receives
+   the endpoint(s) from another different relation, this is the approach to follow.
+   However, you also need to manually enable and disable the log forwarding. For example,
+   let's say the charm gets the endpoint(s) from the `foo` relation:
+
+   ```python
+   from charms.loki_k8s.v1.loki_push_api import LogForwarder
+
+   ...
+
+     def __init__(self, *args):
+         ...
+         self._log_forwarder = LogForwarder(
+             self,
+             loki_endpoints=self.get_loki_endpoints,
+         )
+         self.framework.observe(self.on["foo"].relation_joined, self._log_forwarder.enable())
+         self.framework.observe(self.on["foo"].relation_changed, self._log_forwarder.enable())
+         self.framework.observe(self.on["foo"].relation_departed, self._log_forwarder.disable())
+         self.framework.observe(self.on["foo"].relation_broken, self._log_forwarder.disable())
+
+
+     def get_loki_endpoints(self) -> List[str]:
+         return self.model.relations["foo"].get("loki_endpoints", None)
+
+   ```
+
+Once the library is implemented in a Charmed Operator, and log forwarding is enabled, the library
+will inject a Pebble layer in the workload container to send logs.
+
 ## Alerting Rules
 
 This charm library also supports gathering alerting rules from all related Loki client
@@ -474,7 +543,7 @@ def _alert_rules_error(self, event):
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 1
+LIBPATCH = 2
 
 logger = logging.getLogger(__name__)
 
@@ -2313,6 +2382,123 @@ def _containers(self) -> Dict[str, Container]:
         return {cont: self._charm.unit.get_container(cont) for cont in self._logs_scheme.keys()}
 
 
+class LogForwarder(Object):
+    """Forward the StdOut output to one or multiple Loki endpoints."""
+
+    def __init__(
+        self,
+        charm: CharmBase,
+        *,
+        relation_name: str = DEFAULT_RELATION_NAME,
+        loki_endpoints: Optional[List[str]] = None,
+    ):
+        super().__init__(charm, relation_name)
+        self._charm = charm
+        self._relation_name = relation_name
+        self.topology = JujuTopology.from_charm(charm)
+        self.loki_endpoints = loki_endpoints
+
+        if not loki_endpoints:
+            on_relation = self._charm.on[self._relation_name]
+            self.framework.observe(on_relation.relation_joined, self._on_logging_relation_joined)
+            self.framework.observe(on_relation.relation_changed, self._on_logging_relation_changed)
+            self.framework.observe(
+                on_relation.relation_departed, self._on_logging_relation_departed
+            )
+            self.framework.observe(on_relation.relation_broken, self._on_logging_relation_broken)
+
+    def _on_logging_relation_joined(self, _):
+        self.loki_endpoints = self._fetch_endpoints
+        self.enable()
+
+    def _on_logging_relation_changed(self, _):
+        self.loki_endpoints = self._fetch_endpoints
+        self.enable()
+
+    def _on_logging_relation_departed(self, _):
+        self.disable()
+
+    def _on_logging_relation_broken(self, _):
+        self.disable()
+
+    def enable(self):
+        """Enable log forwarding."""
+        self.handle_logging(enabled=True)
+
+    def disable(self):
+        """Disable log forwarding."""
+        self.handle_logging(enabled=False)
+
+    def _build_log_target(self, endpoint, index, enabled=False):
+        """Build a log target for the log forwarding Pebble layer."""
+        dest_name = f"loki{index}"
+        services_value = ["all"] if enabled else ["-all"]
+
+        return {
+            dest_name: {
+                "override": "merge",
+                "type": "loki",
+                "location": endpoint,
+                "services": services_value,
+                "labels": {
+                    "product": "Juju",
+                    "charm": self.topology._charm_name,
+                    "juju_model": self.topology._model,
+                    "juju_model_uuid": self.topology._model_uuid,
+                    "juju_application": self.topology._application,
+                    "juju_unit": self.topology._unit,
+                },
+            }
+        }
+
+    def _build_log_targets(self, endpoints: Optional[List[str]], enable=False):
+        """Build the targets for the log forwarding Pebble layer."""
+        targets = {}
+        if endpoints:
+            for i, endpoint in enumerate(endpoints):
+                targets.update(self._build_log_target(endpoint, i, enable))
+        return targets
+
+    def handle_logging(self, enabled=False):
+        """Enable or disable the log forwarding."""
+        if self.loki_endpoints:
+            layer_config = {"log-targets": self._build_log_targets(self.loki_endpoints, enabled)}
+            layer = Layer(layer_config)  # pyright: ignore
+
+            for container_name, container in self._charm.unit.containers.items():
+                container.add_layer(f"{container_name}-log-forwarding", layer, combine=True)
+        else:
+            logger.warning("No loki endpoints available")
+
+    @property
+    def _fetch_endpoints(self) -> List[str]:
+        """Fetch Loki Push API endpoints through relation data.
+
+        Returns:
+            A list of Loki Push API endpoint URLs, for instance:
+            [
+                "http://loki1:3100/loki/api/v1/push",
+                "http://loki2:3100/loki/api/v1/push",
+            ]
+        """
+        endpoints = []  # type: list
+
+        for relation in self._charm.model.relations[self._relation_name]:
+            for unit in relation.units:
+                if unit.app == self._charm.app:
+                    # This is a peer unit
+                    continue
+
+                endpoint = relation.data[unit].get("endpoint")
+                if endpoint:
+                    deserialized_endpoint = json.loads(endpoint)
+                    url = deserialized_endpoint.get("url")
+                    if url:
+                        endpoints.append(url)
+        endpoints.sort()
+        return endpoints
+
+
 class CosTool:
     """Uses cos-tool to inject label matchers into alert rule expressions and validate rules."""
 

tests/unit/test_log_forwarder.py

@@ -0,0 +1,151 @@
+# Copyright 2020 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import json
+import textwrap
+import unittest
+from unittest.mock import MagicMock
+
+from charms.loki_k8s.v1.loki_push_api import LogForwarder
+from ops.charm import CharmBase
+from ops.testing import Harness
+
+
+class FakeCharm(CharmBase):
+    """Container charm for forwarding logs using the logforwarder class."""
+
+    metadata_yaml = textwrap.dedent(
+        """
+        containers:
+          consumer:
+            resource: consumer-image
+
+        requires:
+          logging:
+            interface: loki_push_api
+        """
+    )
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.log_forwarder = LogForwarder(self)
+
+
+class TestTransform(unittest.TestCase):
+    """Test that the cos-tool implementation works."""
+
+    def setUp(self):
+        self.harness = Harness(FakeCharm, meta=FakeCharm.metadata_yaml)
+        self.addCleanup(self.harness.cleanup)
+        self.harness.begin_with_initial_hooks()
+
+    def test_handle_logging_with_endpoints_and_relation_lifecycle(self):
+        rel_id = self.harness.add_relation("logging", "loki")
+
+        for i in range(2):
+            loki_unit = f"loki/{i}"
+            endpoint = f"http://loki-{i}:3100/loki/api/v1/push"
+            data = json.dumps({"url": f"{endpoint}"})
+            self.harness.add_relation_unit(rel_id, loki_unit)
+            self.harness.set_planned_units(1)
+            self.harness.update_relation_data(
+                rel_id,
+                loki_unit,
+                {"endpoint": data},
+            )
+
+        expected_endpoints = [
+            "http://loki-0:3100/loki/api/v1/push",
+            "http://loki-1:3100/loki/api/v1/push",
+        ]
+        self.assertEqual(self.harness.charm.log_forwarder._fetch_endpoints, expected_endpoints)
+
+        expected_layer_config = {
+            "loki0": {
+                "override": "merge",
+                "type": "loki",
+                "location": "http://loki-0:3100/loki/api/v1/push",
+                "services": ["all"],
+                "labels": {
+                    "product": "Juju",
+                    "charm": self.harness.charm.log_forwarder.topology._charm_name,
+                    "juju_model": self.harness.charm.log_forwarder.topology._model,
+                    "juju_model_uuid": self.harness.charm.log_forwarder.topology._model_uuid,
+                    "juju_application": self.harness.charm.log_forwarder.topology._application,
+                    "juju_unit": self.harness.charm.log_forwarder.topology._unit,
+                },
+            },
+            "loki1": {
+                "override": "merge",
+                "type": "loki",
+                "location": "http://loki-1:3100/loki/api/v1/push",
+                "services": ["all"],
+                "labels": {
+                    "product": "Juju",
+                    "charm": self.harness.charm.log_forwarder.topology._charm_name,
+                    "juju_model": self.harness.charm.log_forwarder.topology._model,
+                    "juju_model_uuid": self.harness.charm.log_forwarder.topology._model_uuid,
+                    "juju_application": self.harness.charm.log_forwarder.topology._application,
+                    "juju_unit": self.harness.charm.log_forwarder.topology._unit,
+                },
+            },
+        }
+        actual_layer_config = self.harness.charm.log_forwarder._build_log_targets(
+            self.harness.charm.log_forwarder._fetch_endpoints, True
+        )
+        self.assertDictEqual(expected_layer_config, actual_layer_config)
+
+        self.harness.remove_relation(rel_id)
+        self.assertEqual(self.harness.charm.log_forwarder.loki_endpoints, expected_endpoints)
+        expected_layer_config = {
+            "loki0": {
+                "override": "merge",
+                "type": "loki",
+                "location": "http://loki-0:3100/loki/api/v1/push",
+                "services": ["-all"],
+                "labels": {
+                    "product": "Juju",
+                    "charm": self.harness.charm.log_forwarder.topology._charm_name,
+                    "juju_model": self.harness.charm.log_forwarder.topology._model,
+                    "juju_model_uuid": self.harness.charm.log_forwarder.topology._model_uuid,
+                    "juju_application": self.harness.charm.log_forwarder.topology._application,
+                    "juju_unit": self.harness.charm.log_forwarder.topology._unit,
+                },
+            },
+            "loki1": {
+                "override": "merge",
+                "type": "loki",
+                "location": "http://loki-1:3100/loki/api/v1/push",
+                "services": ["-all"],
+                "labels": {
+                    "product": "Juju",
+                    "charm": self.harness.charm.log_forwarder.topology._charm_name,
+                    "juju_model": self.harness.charm.log_forwarder.topology._model,
+                    "juju_model_uuid": self.harness.charm.log_forwarder.topology._model_uuid,
+                    "juju_application": self.harness.charm.log_forwarder.topology._application,
+                    "juju_unit": self.harness.charm.log_forwarder.topology._unit,
+                },
+            },
+        }
+        actual_layer_config = self.harness.charm.log_forwarder._build_log_targets(
+            self.harness.charm.log_forwarder.loki_endpoints, False
+        )
+        self.assertDictEqual(expected_layer_config, actual_layer_config)
+
+    def test_handle_logging_called_on_relation_lifecycle(self):
+        rel_id = self.harness.add_relation("logging", "loki")
+        self.harness.add_relation_unit(rel_id, "loki/0")
+        self.harness.charm.log_forwarder.handle_logging = MagicMock()
+        self.harness.update_relation_data(
+            rel_id,
+            "loki/0",
+            {"endpoint": json.dumps({"url": "http://loki-0:3100/loki/api/v1/push"})},
+        )
+
+        self.harness.charm.log_forwarder.handle_logging.assert_called_with(
+            ["http://loki-0:3100/loki/api/v1/push"], enable=True
+        )
+        self.harness.remove_relation(rel_id)
+        self.harness.charm.log_forwarder.handle_logging.assert_called_with(
+            ["http://loki-0:3100/loki/api/v1/push"], enable=False
+        )