installers/charm/ng-ui/src/charm.py

#!/usr/bin/env python3
# Copyright 2020 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 typing import Any, Dict, NoReturn
from pydantic import ValidationError

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

from pod_spec import make_pod_spec

logger = logging.getLogger(__name__)

NGUI_PORT = 80


class ConfigurePodEvent(EventBase):
    """Configure Pod event"""

    pass


class NgUiEvents(CharmEvents):
    """NGUI Events"""

    configure_pod = EventSource(ConfigurePodEvent)


class NgUiCharm(CharmBase):
    """NGUI Charm."""

    state = StoredState()
    on = NgUiEvents()

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

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

        # North bound interface initialization
        self.state.set_default(nbi_host=None)
        self.state.set_default(nbi_port=None)

        self.http_port = NGUI_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)
        # self.framework.observe(self.on.upgrade_charm, self.configure_pod)

        # Registering custom internal events
        self.framework.observe(self.on.configure_pod, self.configure_pod)

        # Registering required relation changed events
        self.framework.observe(
            self.on.nbi_relation_changed, self._on_nbi_relation_changed
        )

        # Registering required relation departed events
        self.framework.observe(
            self.on.nbi_relation_departed, self._on_nbi_relation_departed
        )

    def _on_nbi_relation_changed(self, event: EventBase) -> NoReturn:
        """Reads information about the nbi relation.

        Args:
            event (EventBase): NBI relation event.
        """
        if not event.unit in event.relation.data:
            return
        relation_data = event.relation.data[event.unit]
        nbi_host = relation_data.get("host")
        nbi_port = relation_data.get("port")

        if (
            nbi_host
            and nbi_port
            and (self.state.nbi_host != nbi_host or self.state.nbi_port != nbi_port)
        ):
            self.state.nbi_host = nbi_host
            self.state.nbi_port = nbi_port
            self.on.configure_pod.emit()

    def _on_nbi_relation_departed(self, event: EventBase) -> NoReturn:
        """Clears data from nbi relation.

        Args:
            event (EventBase): NBI relation event.
        """
        self.state.nbi_host = None
        self.state.nbi_port = None
        self.on.configure_pod.emit()

    def _missing_relations(self) -> str:
        """Checks if there missing relations.

        Returns:
            str: string with missing relations
        """
        data_status = {
            "nbi": self.state.nbi_host,
        }

        missing_relations = [k for k, v in data_status.items() if not v]

        return ", ".join(missing_relations)

    @property
    def relation_state(self) -> Dict[str, Any]:
        """Collects relation state configuration for pod spec assembly.

        Returns:
            Dict[str, Any]: relation state information.
        """
        relation_state = {
            "nbi_host": self.state.nbi_host,
            "nbi_port": self.state.nbi_port,
        }
        return relation_state

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

        Args:
            event (EventBase): Hook or Relation event that started the
                               function.
        """
        if missing := self._missing_relations():
            self.unit.status = BlockedStatus(
                f"Waiting for {missing} relation{'s' if ',' in missing else ''}"
            )
            return

        if not self.unit.is_leader():
            self.unit.status = ActiveStatus("ready")
            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.config,
                self.relation_state,
                self.model.app.name,
            )
        except ValidationError 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(NgUiCharm)