installers/charm/kafka-exporter/src/charm.py

#!/usr/bin/env python3
# Copyright 2021 Canonical Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
# For those usages not covered by the Apache License, Version 2.0 please
# contact: legal@canonical.com
#
# To get in touch with the maintainers, please contact:
# osm-charmers@lists.launchpad.net
##

import logging
from pathlib import Path
from typing import Dict, List, NoReturn
from urllib.parse import urlparse

from ops.charm import CharmBase
from ops.framework import EventBase, StoredState
from ops.main import main
from ops.model import ActiveStatus, Application, BlockedStatus, MaintenanceStatus, Unit
from oci_image import OCIImageResource, OCIImageResourceError

from pod_spec import make_pod_spec

logger = logging.getLogger(__name__)

KAFKA_EXPORTER_PORT = 9308


class RelationsMissing(Exception):
    def __init__(self, missing_relations: List):
        self.message = ""
        if missing_relations and isinstance(missing_relations, list):
            self.message += f'Waiting for {", ".join(missing_relations)} relation'
            if "," in self.message:
                self.message += "s"


class RelationDefinition:
    def __init__(self, relation_name: str, keys: List, source_type):
        if source_type != Application and source_type != Unit:
            raise TypeError(
                "source_type should be ops.model.Application or ops.model.Unit"
            )
        self.relation_name = relation_name
        self.keys = keys
        self.source_type = source_type


def check_missing_relation_data(
    data: Dict,
    expected_relations_data: List[RelationDefinition],
):
    missing_relations = []
    for relation_data in expected_relations_data:
        if not all(
            f"{relation_data.relation_name}_{k}" in data for k in relation_data.keys
        ):
            missing_relations.append(relation_data.relation_name)
    if missing_relations:
        raise RelationsMissing(missing_relations)


def get_relation_data(
    charm: CharmBase,
    relation_data: RelationDefinition,
) -> Dict:
    data = {}
    relation = charm.model.get_relation(relation_data.relation_name)
    if relation:
        self_app_unit = (
            charm.app if relation_data.source_type == Application else charm.unit
        )
        expected_type = relation_data.source_type
        for app_unit in relation.data:
            if app_unit != self_app_unit and isinstance(app_unit, expected_type):
                if all(k in relation.data[app_unit] for k in relation_data.keys):
                    for k in relation_data.keys:
                        data[f"{relation_data.relation_name}_{k}"] = relation.data[
                            app_unit
                        ].get(k)
                    break
    return data


class KafkaExporterCharm(CharmBase):
    """Kafka Exporter Charm."""

    state = StoredState()

    def __init__(self, *args) -> NoReturn:
        """Kafka Exporter Charm constructor."""
        super().__init__(*args)

        # Internal state initialization
        self.state.set_default(pod_spec=None)

        self.port = KAFKA_EXPORTER_PORT
        self.image = OCIImageResource(self, "image")

        # Registering regular events
        self.framework.observe(self.on.start, self.configure_pod)
        self.framework.observe(self.on.config_changed, self.configure_pod)

        # Registering required relation events
        self.framework.observe(self.on.kafka_relation_changed, self.configure_pod)

        # Registering required relation departed events
        self.framework.observe(self.on.kafka_relation_departed, self.configure_pod)

        # Registering provided relation events
        self.framework.observe(
            self.on.prometheus_scrape_relation_joined, self._publish_scrape_info
        )
        self.framework.observe(
            self.on.grafana_dashboard_relation_joined, self._publish_dashboard_info
        )

    def _publish_scrape_info(self, event: EventBase) -> NoReturn:
        """Publishes scrape information.

        Args:
            event (EventBase): Exporter relation event.
        """
        rel_data = {
            "hostname": urlparse(self.model.config["site_url"]).hostname
            if self.model.config["site_url"]
            else self.model.app.name,
            "port": "80" if self.model.config["site_url"] else str(KAFKA_EXPORTER_PORT),
            "metrics_path": "/metrics",
            "scrape_interval": "30s",
            "scrape_timeout": "15s",
        }
        for k, v in rel_data.items():
            event.relation.data[self.unit][k] = v

    def _publish_dashboard_info(self, event: EventBase) -> NoReturn:
        """Publishes dashboard information.

        Args:
            event (EventBase): Exporter relation event.
        """
        rel_data = {
            "name": "osm-kafka",
            "dashboard": Path("files/kafka_exporter_dashboard.json").read_text(),
        }
        for k, v in rel_data.items():
            event.relation.data[self.unit][k] = v

    @property
    def relations_requirements(self):
        return [RelationDefinition("kafka", ["host", "port"], Unit)]

    def get_relation_state(self):
        relation_state = {}
        for relation_requirements in self.relations_requirements:
            data = get_relation_data(self, relation_requirements)
            relation_state = {**relation_state, **data}
        check_missing_relation_data(relation_state, self.relations_requirements)
        return relation_state

    def configure_pod(self, _=None) -> NoReturn:
        """Assemble the pod spec and apply it, if possible.

        Args:
            event (EventBase): Hook or Relation event that started the
                               function.
        """
        if not self.unit.is_leader():
            self.unit.status = ActiveStatus("ready")
            return

        relation_state = None
        try:
            relation_state = self.get_relation_state()
        except RelationsMissing as exc:
            logger.exception("Relation missing error")
            self.unit.status = BlockedStatus(exc.message)
            return

        self.unit.status = MaintenanceStatus("Assembling pod spec")

        # Fetch image information
        try:
            self.unit.status = MaintenanceStatus("Fetching image information")
            image_info = self.image.fetch()
        except OCIImageResourceError:
            self.unit.status = BlockedStatus("Error fetching image information")
            return

        try:
            pod_spec = make_pod_spec(
                image_info,
                self.model.config,
                relation_state,
                self.model.app.name,
                self.port,
            )
        except ValueError as exc:
            logger.exception("Config/Relation data validation error")
            self.unit.status = BlockedStatus(str(exc))
            return

        if self.state.pod_spec != pod_spec:
            self.model.pod.set_spec(pod_spec)
            self.state.pod_spec = pod_spec

        self.unit.status = ActiveStatus("ready")


if __name__ == "__main__":
    main(KafkaExporterCharm)