installers/charm/osm-nbi/lib/charms/osm_nbi/v0/nbi.py

   1 #!/usr/bin/env python3
   2 # Copyright 2022 Canonical Ltd.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: legal@canonical.com
  18 #
  19 # To get in touch with the maintainers, please contact:
  20 # osm-charmers@lists.launchpad.net
  21 #
  22 #
  23 # Learn more at: https://juju.is/docs/sdk
  24
  25 """Nbi library.
  26
  27 This [library](https://juju.is/docs/sdk/libraries) implements both sides of the
  28 `nbi` [interface](https://juju.is/docs/sdk/relations).
  29
  30 The *provider* side of this interface is implemented by the
  31 [osm-nbi Charmed Operator](https://charmhub.io/osm-nbi).
  32
  33 Any Charmed Operator that *requires* NBI for providing its
  34 service should implement the *requirer* side of this interface.
  35
  36 In a nutshell using this library to implement a Charmed Operator *requiring*
  37 NBI would look like
  38
  39 ```
  40 $ charmcraft fetch-lib charms.osm_nbi.v0.nbi
  41 ```
  42
  43 `metadata.yaml`:
  44
  45 ```
  46 requires:
  47   nbi:
  48     interface: nbi
  49     limit: 1
  50 ```
  51
  52 `src/charm.py`:
  53
  54 ```
  55 from charms.osm_nbi.v0.nbi import NbiRequires
  56 from ops.charm import CharmBase
  57
  58
  59 class MyCharm(CharmBase):
  60
  61     def __init__(self, *args):
  62         super().__init__(*args)
  63         self.nbi = NbiRequires(self)
  64         self.framework.observe(
  65             self.on["nbi"].relation_changed,
  66             self._on_nbi_relation_changed,
  67         )
  68         self.framework.observe(
  69             self.on["nbi"].relation_broken,
  70             self._on_nbi_relation_broken,
  71         )
  72         self.framework.observe(
  73             self.on["nbi"].relation_broken,
  74             self._on_nbi_broken,
  75         )
  76
  77     def _on_nbi_relation_broken(self, event):
  78         # Get NBI host and port
  79         host: str = self.nbi.host
  80         port: int = self.nbi.port
  81         # host => "osm-nbi"
  82         # port => 9999
  83
  84     def _on_nbi_broken(self, event):
  85         # Stop service
  86         # ...
  87         self.unit.status = BlockedStatus("need nbi relation")
  88 ```
  89
  90 You can file bugs
  91 [here](https://osm.etsi.org/bugzilla/enter_bug.cgi), selecting the `devops` module!
  92 """
  93 from typing import Optional
  94
  95 from ops.charm import CharmBase, CharmEvents
  96 from ops.framework import EventBase, EventSource, Object
  97 from ops.model import Relation
  98
  99
 100 # The unique Charmhub library identifier, never change it
 101 LIBID = "8c888f7c869949409e12c16d78ec068b"
 102
 103 # Increment this major API version when introducing breaking changes
 104 LIBAPI = 0
 105
 106 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 107 # to 0 if you are raising the major API version
 108 LIBPATCH = 1
 109
 110 NBI_HOST_APP_KEY = "host"
 111 NBI_PORT_APP_KEY = "port"
 112
 113
 114 class NbiRequires(Object):  # pragma: no cover
 115     """Requires-side of the Nbi relation."""
 116
 117     def __init__(self, charm: CharmBase, endpoint_name: str = "nbi") -> None:
 118         super().__init__(charm, endpoint_name)
 119         self.charm = charm
 120         self._endpoint_name = endpoint_name
 121
 122         # Observe relation events
 123         event_observe_mapping = {
 124             charm.on[self._endpoint_name].relation_changed: self._on_relation_changed,
 125         }
 126         for event, observer in event_observe_mapping.items():
 127             self.framework.observe(event, observer)
 128
 129     @property
 130     def host(self) -> str:
 131         """Get nbi hostname."""
 132         relation: Relation = self.model.get_relation(self._endpoint_name)
 133         return (
 134             relation.data[relation.app].get(NBI_HOST_APP_KEY)
 135             if relation and relation.app
 136             else None
 137         )
 138
 139     @property
 140     def port(self) -> int:
 141         """Get nbi port number."""
 142         relation: Relation = self.model.get_relation(self._endpoint_name)
 143         return (
 144             int(relation.data[relation.app].get(NBI_PORT_APP_KEY))
 145             if relation and relation.app
 146             else None
 147         )
 148
 149
 150 class NbiProvides(Object):
 151     """Provides-side of the Nbi relation."""
 152
 153     def __init__(self, charm: CharmBase, endpoint_name: str = "nbi") -> None:
 154         super().__init__(charm, endpoint_name)
 155         self._endpoint_name = endpoint_name
 156
 157     def set_host_info(self, host: str, port: int, relation: Optional[Relation] = None) -> None:
 158         """Set Nbi host and port.
 159
 160         This function writes in the application data of the relation, therefore,
 161         only the unit leader can call it.
 162
 163         Args:
 164             host (str): Nbi hostname or IP address.
 165             port (int): Nbi port.
 166             relation (Optional[Relation]): Relation to update.
 167                                            If not specified, all relations will be updated.
 168
 169         Raises:
 170             Exception: if a non-leader unit calls this function.
 171         """
 172         if not self.model.unit.is_leader():
 173             raise Exception("only the leader set host information.")
 174
 175         if relation:
 176             self._update_relation_data(host, port, relation)
 177             return
 178
 179         for relation in self.model.relations[self._endpoint_name]:
 180             self._update_relation_data(host, port, relation)
 181
 182     def _update_relation_data(self, host: str, port: int, relation: Relation) -> None:
 183         """Update data in relation if needed."""
 184         relation.data[self.model.app][NBI_HOST_APP_KEY] = host
 185         relation.data[self.model.app][NBI_PORT_APP_KEY] = str(port)