feat: send/receive relations for forwarding OTLP metrics (#131)

Author: MichaelThamm

charmcraft.yaml

@@ -81,6 +81,17 @@ provides:
     interface: profiling
     optional: true
     description: Receive profiles from other charms.
+  receive-otlp:
+    interface: otlp
+    optional: true
+    description: |
+      Receive OTLP data with an OTLP server by publishing all OTLP endpoints for the supported
+      protocols and telemetries of this server. OTLP data can contain multiple telemetries, making
+      this endpoint a modern alternative to having one relation per telemetry-type, e.g.
+      `receive-loki-logs` and/or `receive-remote-write`.
+
+      This endpoint currently supports:
+        - HTTP for ["metrics"]
 
 requires:
   send-profiles:
@@ -152,6 +163,16 @@ requires:
     description: |
       Ingress integration for the Otelcol server, so that cross-model workloads can send their data
       through the ingress. Uses `traefik_route` to open ports on Traefik host for tracing ingesters.
+  send-otlp:
+    interface: otlp
+    optional: true
+    description: |
+      Send OTLP data to an OTLP server by agregating all supported OTLP endpoints offered by an
+      OTLP server. If multiple endpoints are supported, then the first supported one is chosen.
+      
+      This endpoint currently supports:
+        - gRPC for ["logs", "metrics"]
+        - HTTP for ["logs", "metrics"]
 
 peers:
   peers:

src/charm.py

@@ -36,7 +36,10 @@
 
 logger = logging.getLogger(__name__)
 
-def charm_address(container: Container, ingress: integrations.TraefikRouteRequirer) -> integrations.Address:
+
+def charm_address(
+    container: Container, ingress: integrations.TraefikRouteRequirer
+) -> integrations.Address:
     """Return the Address dataclass from charm context.
 
     Args:
@@ -50,18 +53,22 @@ def charm_address(container: Container, ingress: integrations.TraefikRouteRequir
     internal_scheme = "https" if tls else "http"
     internal_url = f"{internal_scheme}://{socket.getfqdn()}"
     external_url = (
-        f"{ingress.scheme}://{ingress.external_host}" if integrations.ingress_ready(ingress) else None
+        f"{ingress.scheme}://{ingress.external_host}"
+        if integrations.ingress_ready(ingress)
+        else None
     )
     resolved_url = external_url if external_url else internal_url
     resolved_scheme = urlparse(external_url).scheme if external_url else "https" if tls else "http"
 
     return integrations.Address(
-        internal_scheme,
-        internal_url,
-        resolved_scheme,
-        resolved_url,
+        ingress=integrations.ingress_ready(ingress),
+        internal_scheme=internal_scheme,
+        internal_url=internal_url,
+        resolved_scheme=resolved_scheme,
+        resolved_url=resolved_url,
     )
 
+
 def refresh_certs(container: Container):
     """Run `update-ca-certificates` to refresh the trusted system certs."""
     container.exec(["update-ca-certificates", "--fresh"]).wait()
@@ -83,6 +90,7 @@ def _get_missing_mandatory_relations(charm: CharmBase) -> Optional[str]:
         pairs={
             "metrics-endpoint": [  # must be paired with:
                 {"send-remote-write"},  # or
+                {"send-otlp"},
                 {"cloud-config"},
             ],
             "receive-loki-logs": [  # must be paired with:
@@ -97,6 +105,15 @@ def _get_missing_mandatory_relations(charm: CharmBase) -> Optional[str]:
                 {"grafana-dashboards-provider"},  # or
                 {"cloud-config"},
             ],
+            "receive-otlp": [  # must be paired with:
+                {"send-otlp"},  # or
+                # Technically, this would depend on databag contents for enabled pipelines,
+                # but we keep it simple for now.
+                {"send-traces"},
+                {"send-loki-logs"},
+                {"send-remote-write"},
+                {"cloud-config"},
+            ],
         }
     )
     active_relations = {name for name, relation in charm.model.relations.items() if relation}
@@ -200,6 +217,11 @@ def _reconcile(self):
         #  cf: https://github.com/canonical/opentelemetry-collector-k8s-operator/issues/17
         feature_gates: Optional[str] = None
 
+        # OTLP setup
+        integrations.receive_otlp(self, otelcol_address.resolved_url)
+        otlp_endpoints = integrations.send_otlp(self)
+        config_manager.add_otlp_forwarding(otlp_endpoints)
+
         # Logs setup
         integrations.receive_loki_logs(self, otelcol_address)
         loki_endpoints = integrations.send_loki_logs(self)
@@ -319,6 +341,21 @@ def _reconcile(self):
         if missing_relations:
             self.unit.status = BlockedStatus(missing_relations)
 
+        # Cyclic OTLP relations
+        if integrations.cyclic_otlp_relations_exist(self):
+            self.unit.status = BlockedStatus("cyclic OTLP relations exist")
+
+        # Ingress and scaling status
+        if self.model.unit.is_leader():
+            if self.app.planned_units() > 1 and not otelcol_address.ingress:
+                self.unit.status = BlockedStatus(
+                    "Ingress missing - routing only to leader; see debug-log"
+                )
+                logger.warning(
+                    "without ingress and planned_units > 1, all data is forwarded to the leader "
+                    "unit, with nothing sent to non-leader units."
+                )
+
         # Workload version
         self.unit.set_workload_version(self._otelcol_version or "")
 

src/config_builder.py

@@ -156,7 +156,7 @@ def add_default_config(self):
         #       fighting for port bindings. This is only for relevant for the vm charm
         self.add_component(
             Component.receiver,
-            "otlp",
+            f"otlp/{self._unit_name}",
             {
                 "protocols": {
                     "http": {"endpoint": f"0.0.0.0:{Port.otlp_http.value}"},

src/config_manager.py

@@ -2,12 +2,14 @@
 
 import logging
 from typing import Any, Dict, List, Literal, Optional, Set
+from urllib.parse import urlparse
 
 import yaml
 
 from config_builder import Component, ConfigBuilder, Port
 from constants import FILE_STORAGE_DIRECTORY
 from integrations import ProfilingEndpoint
+from otlp import OtlpEndpoint
 
 logger = logging.getLogger(__name__)
 
@@ -251,14 +253,14 @@ def add_profile_ingestion(self):
         """Configure ingesting profiles."""
         self.config.add_component(
             Component.receiver,
-            "otlp",
+            f"otlp/{self._unit_name}",
             {
                 "protocols": {
                     "http": {"endpoint": f"0.0.0.0:{Port.otlp_http.value}"},
                     "grpc": {"endpoint": f"0.0.0.0:{Port.otlp_grpc.value}"},
                 },
             },
-            pipelines=["profiles"],
+            pipelines=[f"profiles/{self._unit_name}"],
         )
 
     def add_profile_forwarding(self, endpoints: List[ProfilingEndpoint]):
@@ -282,7 +284,7 @@ def add_profile_forwarding(self, endpoints: List[ProfilingEndpoint]):
                         "insecure_skip_verify": self._insecure_skip_verify,
                     },
                 },
-                pipelines=["profiles"],
+                pipelines=[f"profiles/{self._unit_name}"],
             )
 
     def add_self_scrape(self, identifier: str, labels: Dict) -> None:
@@ -367,8 +369,38 @@ def add_remote_write(self, endpoints: List[Dict[str, str]]):
                 pipelines=[f"metrics/{self._unit_name}"],
             )
 
-        # TODO Receive alert rules via remote write
-        # https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/37277
+    def add_otlp_forwarding(self, relation_map: Dict[int, OtlpEndpoint]):
+        """Configure sending OTLP telemetry to an OTLP endpoint.
+
+        There are 2 different OTLP exporters for their respective protocols: gRPC and HTTP. If a
+        gRPC endpoint is provided, it is preferred over the HTTP equivalent.
+
+        Telemetry is sent to all pipelines since OTLP supports all and its computationally
+        inexpensive unless a receiver is connected and receiving telemetry.
+
+        Args:
+            relation_map: a mapping of relation ID to a mapping of unit name to OtlpEndpoint
+        """
+        # https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/otlpexporter
+        # https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/otlphttpexporter
+
+        if not relation_map:
+            return
+
+        # Exporter config
+        for rel_id, otlp_endpoint in relation_map.items():
+            insecure = urlparse(otlp_endpoint.endpoint).scheme == "http"
+            tls_config: Dict[str, Any] = {
+                "insecure": insecure,
+                "insecure_skip_verify": self._insecure_skip_verify,
+            }
+            exporter_type = "otlp" if otlp_endpoint.protocol == "grpc" else "otlphttp"
+            self.config.add_component(
+                Component.exporter,
+                f"{exporter_type}/rel-{rel_id}/{self._unit_name}",
+                {"endpoint": otlp_endpoint.endpoint, "tls": tls_config},
+                pipelines=[f"{_type}/{self._unit_name}" for _type in otlp_endpoint.telemetries],
+            )
 
     def add_traces_ingestion(
         self,

src/integrations.py

@@ -64,6 +64,13 @@
     SERVER_CERT_PATH,
     SERVER_CERT_PRIVATE_KEY_PATH,
 )
+from otlp import (
+    OtlpConsumer,
+    OtlpEndpoint,
+    OtlpProvider,
+    DEFAULT_PROVIDER_RELATION_NAME,
+    DEFAULT_CONSUMER_RELATION_NAME,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -247,8 +254,6 @@ def send_remote_write(charm: CharmBase) -> List[Dict[str, str]]:
         peer_relation_name="peers",
     )
     charm.__setattr__("remote_write", remote_write)
-    # TODO: add alerts from remote write
-    # https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/37277
     # TODO: Luca: probably don't need this anymore
     remote_write.reload_alerts()
     return remote_write.endpoints
@@ -460,6 +465,51 @@ def forward_dashboards(charm: CharmBase):
     # grafana_dashboards_provider._reinitialize_dashboard_data(inject_dropdowns=False)
 
 
+def cyclic_otlp_relations_exist(charm: CharmBase) -> bool:
+    """Check if any application is related on both send-otlp and receive-otlp.
+
+    This function only checks relations for the current charm, i.e. one level deep. If there is
+    another charm in between these applications, but is still cyclic, then it will not be caught.
+    """
+    receive_relations = charm.model.relations.get(DEFAULT_PROVIDER_RELATION_NAME, [])
+    send_relations = charm.model.relations.get(DEFAULT_CONSUMER_RELATION_NAME, [])
+
+    if not receive_relations or not send_relations:
+        return False
+
+    receive_apps = {rel.app.name for rel in receive_relations if rel.app}
+    send_apps = {rel.app.name for rel in send_relations if rel.app}
+
+    return not receive_apps.isdisjoint(send_apps)
+
+
+def receive_otlp(charm: CharmBase, resolved_url: str) -> None:
+    """Instantiate the OtlpProvider.
+
+    The gRPC protocol is not supported because Traefik (ingress) does not support it.
+    """
+    otlp_provider = OtlpProvider(charm)
+    # TODO: We can remove this since the lib doesn't observe events
+    charm.__setattr__("otlp_provider", otlp_provider)
+    otlp_provider.add_endpoint("http", f"{resolved_url}:4318", ["metrics"])
+    otlp_provider.publish()
+
+
+def send_otlp(charm: CharmBase) -> Dict[int, OtlpEndpoint]:
+    """Instantiate the OtlpConsumer.
+
+    This provides otelcol with the remote's OTLP endpoint for each relation.
+    """
+    otlp_consumer = OtlpConsumer(
+        charm,
+        protocols=["grpc", "http"],
+        telemetries=["logs", "metrics"],
+    )
+    # TODO: We can remove this since the lib doesn't observe events
+    charm.__setattr__("otlp_consumer", otlp_consumer)
+    return otlp_consumer.get_remote_otlp_endpoints()
+
+
 # TODO: Luca: move this into the GrafanCloudIntegrator library
 @dataclass
 class CloudIntegratorData:
@@ -630,7 +680,10 @@ def _static_ingress_config() -> dict:
 
 
 def _build_lb_server_config(scheme: str, port: int) -> List[Dict[str, str]]:
-    """Build the server portion of the loadbalancer config of Traefik ingress."""
+    """Build the server portion of the loadbalancer config of Traefik ingress.
+
+    The leader provides the kubernetes service address to Traefik to serve as ingress.
+    """
     return [{"url": f"{scheme}://{socket.getfqdn()}:{port}"}]
 
 
@@ -641,7 +694,7 @@ def is_tls_ready(container: Container) -> bool:
     )
 
 
-@dataclass
+@dataclass(kw_only=True)
 class Address:
     """Provide address information for the charm.
 
@@ -650,6 +703,7 @@ class Address:
         - tls events
     """
 
+    ingress: bool
     internal_scheme: str  # Only TLS context
     internal_url: str  # Only TLS context
     resolved_scheme: str  # TLS & ingress context