openstack/sunbeam-charms
Code Issues Proposed changes

Update traefik ingress library to v1

Browse Source 
Use traefik ingress v1 library for sunbeam
charms.

Change-Id: I74cdf90b736dccae198f680aaca74bd3fe3f4503
This commit is contained in:
Hemanth Nakkina 2022-09-12 09:24:56 +05:30
parent 7684a0db73
commit cd4621887a
6 changed files with 571 additions and 395 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ops-sunbeam/fetch-libs.sh
View File

@ -11,6 +11,6 @@ charmcraft fetch-lib charms.sunbeam_keystone_operator.v0.cloud_credentials
charmcraft fetch-lib charms.sunbeam_rabbitmq_operator.v0.amqp charmcraft fetch-lib charms.sunbeam_rabbitmq_operator.v0.amqp
charmcraft fetch-lib charms.sunbeam_ovn_central_operator.v0.ovsdb charmcraft fetch-lib charms.sunbeam_ovn_central_operator.v0.ovsdb
charmcraft fetch-lib charms.observability_libs.v0.kubernetes_service_patch charmcraft fetch-lib charms.observability_libs.v0.kubernetes_service_patch
charmcraft fetch-lib charms.traefik_k8s.v0.ingress charmcraft fetch-lib charms.traefik_k8s.v1.ingress
echo "Copying libs to to unit_test dir" echo "Copying libs to to unit_test dir"
rsync --recursive --delete lib/ unit_tests/lib/ rsync --recursive --delete lib/ unit_tests/lib/

29
ops-sunbeam/ops_sunbeam/relation_handlers.py
View File

@ -22,6 +22,13 @@ from urllib.parse import urlparse
import ops.charm import ops.charm
import ops.framework import ops.framework
try:
from charms.traefik_k8s.v1.ingress import (
IngressPerAppRequirer,
IngressPerAppReadyEvent,
IngressPerAppRevokedEvent)
except ModuleNotFoundError:
pass
import ops_sunbeam.interfaces as sunbeam_interfaces import ops_sunbeam.interfaces as sunbeam_interfaces
@ -122,23 +129,20 @@ class IngressHandler(RelationHandler):
def setup_event_handler(self) -> ops.charm.Object: def setup_event_handler(self) -> ops.charm.Object:
"""Configure event handlers for an Ingress relation.""" """Configure event handlers for an Ingress relation."""
logger.debug("Setting up ingress event handler") logger.debug("Setting up ingress event handler")
# Lazy import to ensure this lib is only required if the charm interface = IngressPerAppRequirer(
# has this relation.
import charms.traefik_k8s.v0.ingress as ingress
interface = ingress.IngressPerAppRequirer(
self.charm, self.charm,
self.relation_name, self.relation_name,
port=self.default_ingress_port, port=self.default_ingress_port,
) )
_rname = self.relation_name.replace("-", "_") self.framework.observe(
ingress_relation_event = getattr( interface.on.ready, self._on_ingress_ready
self.charm.on, f"{_rname}_relation_changed" )
self.framework.observe(
interface.on.revoked, self._on_ingress_revoked
) )
self.framework.observe(ingress_relation_event,
self._on_ingress_changed)
return interface return interface
def _on_ingress_changed(self, event: ops.framework.EventBase) -> None: def _on_ingress_ready(self, event: IngressPerAppReadyEvent) -> None:
"""Handle ingress relation changed events.""" """Handle ingress relation changed events."""
url = self.url url = self.url
logger.debug(f'Received url: {url}') logger.debug(f'Received url: {url}')
@ -147,6 +151,11 @@ class IngressHandler(RelationHandler):
self.callback_f(event) self.callback_f(event)
def _on_ingress_revoked(self, event: IngressPerAppRevokedEvent) -> None:
"""Handle ingress relation revoked event."""
# Callback call to update keystone endpoints
self.callback_f(event)
@property @property
def ready(self) -> bool: def ready(self) -> bool:
"""Whether the handler is ready for use.""" """Whether the handler is ready for use."""

8
ops-sunbeam/ops_sunbeam/test_utils.py
View File

@ -250,13 +250,12 @@ def add_ingress_relation_data(
"""Add ingress data to ingress relation.""" """Add ingress data to ingress relation."""
app_name = 'traefik-' + endpoint_type app_name = 'traefik-' + endpoint_type
url = 'http://' + endpoint_type + "-url" url = 'http://' + endpoint_type + "-url"
ingress_data = {"ingress": {"url": url}}
ingress_data = {"url": url}
harness.update_relation_data( harness.update_relation_data(
rel_id, rel_id,
app_name, app_name,
{ {"ingress": json.dumps(ingress_data)})
"data": json.dumps(ingress_data),
"_supported_versions": yaml.dump(["v1"])})
def add_complete_ingress_relation(harness: Harness) -> None: def add_complete_ingress_relation(harness: Harness) -> None:
@ -670,6 +669,7 @@ def get_harness(
harness._framework = framework.Framework( harness._framework = framework.Framework(
":memory:", harness._charm_dir, harness._meta, harness._model ":memory:", harness._charm_dir, harness._meta, harness._model
) )
harness.set_model_name("test-model")
if initial_charm_config: if initial_charm_config:
harness.update_config(initial_charm_config) harness.update_config(initial_charm_config)
else: else:

4
ops-sunbeam/requirements.txt
View File

@ -1,11 +1,9 @@
charmhelpers charmhelpers
jinja2 jinja2
jsonschema
kubernetes kubernetes
ops ops
python-keystoneclient python-keystoneclient
git+https://github.com/openstack/charm-ops-interface-ceph-client#egg=interface_ceph_client git+https://github.com/openstack/charm-ops-interface-ceph-client#egg=interface_ceph_client
lightkube lightkube
lightkube-models lightkube-models
# included for interfacing with traefik
serialized-data-interface>=0.4.0

377
ops-sunbeam/unit_tests/lib/charms/traefik_k8s/v0/ingress.py
View File

@ -1,377 +0,0 @@
# Copyright 2022 Canonical Ltd.
# See LICENSE file for licensing details.
r"""# Interface Library for ingress.
This library wraps relation endpoints using the `ingress` interface
and provides a Python API for both requesting and providing per-application
ingress, with load-balancing occurring across all units.
## Getting Started
To get started using the library, you just need to fetch the library using `charmcraft`.
**Note that you also need to add the `serialized_data_interface` dependency to your
charm's `requirements.txt`.**
```shell
cd some-charm
charmcraft fetch-lib charms.traefik_k8s.v0.ingress
echo -e "serialized_data_interface\n" >> requirements.txt
```
In the `metadata.yaml` of the charm, add the following:
```yaml
requires:
ingress:
interface: ingress
limit: 1
```
Then, to initialise the library:
```python
# ...
from charms.traefik_k8s.v0.ingress import IngressPerAppRequirer
class SomeCharm(CharmBase):
def __init__(self, *args):
# ...
self.ingress = IngressPerAppRequirer(self, port=80)
# The following event is triggered when the ingress URL to be used
# by this deployment of the `SomeCharm` changes or there is no longer
# an ingress URL available, that is, `self.ingress_per_unit` would
# return `None`.
self.framework.observe(
self.ingress.on.ingress_changed, self._handle_ingress
)
# ...
def _handle_ingress(self, event):
logger.info("This app's ingress URL: %s", self.ingress.url)
```
"""
import logging
from typing import Optional
from ops.charm import CharmBase, RelationBrokenEvent, RelationEvent, RelationRole
from ops.framework import EventSource, StoredState
from ops.model import Relation
try:
from serialized_data_interface import EndpointWrapper
from serialized_data_interface.errors import RelationDataError, UnversionedRelation
from serialized_data_interface.events import EndpointWrapperEvents
except ImportError:
import os
library_name = os.path.basename(__file__)
raise ModuleNotFoundError(
"To use the '{}' library, you must include "
"the '{}' package in your dependencies".format(library_name, "serialized_data_interface")
) from None # Suppress original ImportError
# The unique Charmhub library identifier, never change it
LIBID = "e6de2a5cd5b34422a204668f3b8f90d2"
# Increment this major API version when introducing breaking changes
LIBAPI = 0
# Increment this PATCH version before using `charmcraft publish-lib` or reset
# to 0 if you are raising the major API version
LIBPATCH = 5
log = logging.getLogger(__name__)
INGRESS_SCHEMA = {
"v1": {
"requires": {
"app": {
"type": "object",
"properties": {
"model": {"type": "string"},
"name": {"type": "string"},
"host": {"type": "string"},
"port": {"type": "integer"},
},
"required": ["model", "name", "host", "port"],
},
},
"provides": {
"app": {
"type": "object",
"properties": {
"ingress": {
"type": "object",
"properties": {
"url": {"type": "string"},
},
}
},
"required": ["ingress"],
},
},
}
}
class IngressPerAppRequestEvent(RelationEvent):
"""Event representing an incoming request.
This is equivalent to the "ready" event, but is more semantically meaningful.
"""
class IngressPerAppProviderEvents(EndpointWrapperEvents):
"""Container for IUP events."""
request = EventSource(IngressPerAppRequestEvent)
class IngressPerAppProvider(EndpointWrapper):
"""Implementation of the provider of ingress."""
ROLE = RelationRole.provides.name
INTERFACE = "ingress"
SCHEMA = INGRESS_SCHEMA
on = IngressPerAppProviderEvents()
def __init__(self, charm: CharmBase, endpoint: str = None):
"""Constructor for IngressPerAppProvider.
Args:
charm: The charm that is instantiating the instance.
endpoint: The name of the relation endpoint to bind to
(defaults to "ingress").
"""
super().__init__(charm, endpoint)
self.framework.observe(self.on.ready, self._emit_request_event)
def _emit_request_event(self, event):
self.on.request.emit(event.relation)
def get_request(self, relation: Relation):
"""Get the IngressPerAppRequest for the given Relation."""
return IngressPerAppRequest(self, relation)
def is_failed(self, relation: Relation = None):
"""Checks whether the given relation, or any relation if not specified, has an error."""
if relation is None:
return any(self.is_failed(relation) for relation in self.relations)
if super().is_failed(relation):
return True
try:
data = self.unwrap(relation)
except UnversionedRelation:
return False
prev_fields = None
other_app = relation.app
new_fields = {
field: data[other_app][field]
for field in ("model", "port")
if field in data[other_app]
}
if prev_fields is None:
prev_fields = new_fields
if new_fields != prev_fields:
raise RelationDataMismatchError(relation, other_app)
return False
@property
def proxied_endpoints(self):
"""Returns the ingress settings provided to applications by this IngressPerAppProvider.
For example, when this IngressPerAppProvider has provided the
`http://foo.bar/my-model.my-app` URL to the my-app application, the returned dictionary
will be:
```
{
"my-app": {
"url": "http://foo.bar/my-model.my-app"
}
}
```
"""
return {
ingress_relation.app.name: self.unwrap(ingress_relation)[self.charm.app].get(
"ingress", {}
)
for ingress_relation in self.charm.model.relations[self.endpoint]
}
class IngressPerAppRequest:
"""A request for per-application ingress."""
def __init__(self, provider: IngressPerAppProvider, relation: Relation):
"""Construct an IngressRequest."""
self._provider = provider
self._relation = relation
self._data = provider.unwrap(relation)
@property
def model(self):
"""The name of the model the request was made from."""
return self._data[self.app].get("model")
@property
def app(self):
"""The remote application."""
return self._relation.app
@property
def app_name(self):
"""The name of the remote app.
Note: This is not the same as `self.app.name` when using CMR relations,
since `self.app.name` is replaced by a `remote-{UUID}` pattern.
"""
return self._relation.app.name
@property
def host(self):
"""The hostname to be used to route to the application."""
return self._data[self.app].get("host")
@property
def port(self):
"""The port to be used to route to the application."""
return self._data[self.app].get("port")
def respond(self, url: str):
"""Send URL back for the application.
Note: only the leader can send URLs.
"""
ingress = self._data[self._provider.charm.app].setdefault("ingress", {})
ingress["url"] = url
self._provider.wrap(self._relation, self._data)
class RelationDataMismatchError(RelationDataError):
"""Data from different units do not match where they should."""
class IngressPerAppConfigurationChangeEvent(RelationEvent):
"""Event representing a change in the data sent by the ingress."""
class IngressPerAppRequirerEvents(EndpointWrapperEvents):
"""Container for IUP events."""
ingress_changed = EventSource(IngressPerAppConfigurationChangeEvent)
class IngressPerAppRequirer(EndpointWrapper):
"""Implementation of the requirer of the ingress relation."""
on = IngressPerAppRequirerEvents()
_stored = StoredState()
ROLE = RelationRole.requires.name
INTERFACE = "ingress"
SCHEMA = INGRESS_SCHEMA
LIMIT = 1
def __init__(
self,
charm: CharmBase,
endpoint: str = None,
*,
host: str = None,
port: int = None,
):
"""Constructor for IngressRequirer.
The request args can be used to specify the ingress properties when the
instance is created. If any are set, at least `port` is required, and
they will be sent to the ingress provider as soon as it is available.
All request args must be given as keyword args.
Args:
charm: the charm that is instantiating the library.
endpoint: the name of the relation endpoint to bind to (defaults to `ingress`);
relation must be of interface type `ingress` and have "limit: 1")
host: Hostname to be used by the ingress provider to address the requiring
application; if unspecified, the default Kubernetes service name will be used.
Request Args:
port: the port of the service
"""
super().__init__(charm, endpoint)
# Workaround for SDI not marking the EndpointWrapper as not
# ready upon a relation broken event
self.is_relation_broken = False
self._stored.set_default(current_url=None)
if port and charm.unit.is_leader():
self.auto_data = self._complete_request(host or "", port)
self.framework.observe(
self.charm.on[self.endpoint].relation_changed, self._emit_ingress_change_event
)
self.framework.observe(
self.charm.on[self.endpoint].relation_broken, self._emit_ingress_change_event
)
def _emit_ingress_change_event(self, event):
if isinstance(event, RelationBrokenEvent):
self.is_relation_broken = True
# Avoid spurious events, emit only when URL changes
new_url = self.url
if self._stored.current_url != new_url:
self._stored.current_url = new_url
self.on.ingress_changed.emit(self.relation)
def _complete_request(self, host: Optional[str], port: int):
if not host:
# TODO Make host mandatory?
host = "{app_name}.{model_name}.svc.cluster.local".format(
app_name=self.app.name,
model_name=self.model.name,
)
return {
self.app: {
"model": self.model.name,
"name": self.charm.unit.name,
"host": host,
"port": port,
}
}
def request(self, *, host: str = None, port: int):
"""Request ingress to this application.
Args:
host: Hostname to be used by the ingress provider to address the requirer; if
unspecified, the Kubernetes service address is used.
port: the port of the service (required)
"""
self.wrap(self.relation, self._complete_request(host, port))
@property
def relation(self):
"""The established Relation instance, or None."""
return self.relations[0] if self.relations else None
@property
def url(self):
"""The full ingress URL to reach the current unit.
May return None if the URL isn't available yet.
"""
if self.is_relation_broken or not self.is_ready():
return {}
data = self.unwrap(self.relation)
ingress = data[self.relation.app].get("ingress", {})
return ingress.get("url")

546
ops-sunbeam/unit_tests/lib/charms/traefik_k8s/v1/ingress.py Normal file
View File

@ -0,0 +1,546 @@
# Copyright 2022 Canonical Ltd.
# See LICENSE file for licensing details.
r"""# Interface Library for ingress.
This library wraps relation endpoints using the `ingress` interface
and provides a Python API for both requesting and providing per-application
ingress, with load-balancing occurring across all units.
## Getting Started
To get started using the library, you just need to fetch the library using `charmcraft`.
```shell
cd some-charm
charmcraft fetch-lib charms.traefik_k8s.v1.ingress
```
In the `metadata.yaml` of the charm, add the following:
```yaml
requires:
ingress:
interface: ingress
limit: 1
```
Then, to initialise the library:
```python
from charms.traefik_k8s.v1.ingress import (IngressPerAppRequirer,
IngressPerAppReadyEvent, IngressPerAppRevokedEvent)
class SomeCharm(CharmBase):
def __init__(self, *args):
# ...
self.ingress = IngressPerAppRequirer(self, port=80)
# The following event is triggered when the ingress URL to be used
# by this deployment of the `SomeCharm` is ready (or changes).
self.framework.observe(
self.ingress.on.ready, self._on_ingress_ready
)
self.framework.observe(
self.ingress.on.revoked, self._on_ingress_revoked
)
def _on_ingress_ready(self, event: IngressPerAppReadyEvent):
logger.info("This app's ingress URL: %s", event.url)
def _on_ingress_revoked(self, event: IngressPerAppRevokedEvent):
logger.info("This app no longer has ingress")
"""
import logging
import socket
import typing
from typing import Any, Dict, Optional, Tuple
import yaml
from ops.charm import CharmBase, RelationBrokenEvent, RelationEvent
from ops.framework import EventSource, Object, ObjectEvents, StoredState
from ops.model import ModelError, Relation
# The unique Charmhub library identifier, never change it
LIBID = "e6de2a5cd5b34422a204668f3b8f90d2"
# Increment this major API version when introducing breaking changes
LIBAPI = 1
# Increment this PATCH version before using `charmcraft publish-lib` or reset
# to 0 if you are raising the major API version
LIBPATCH = 3
DEFAULT_RELATION_NAME = "ingress"
RELATION_INTERFACE = "ingress"
log = logging.getLogger(__name__)
try:
import jsonschema
DO_VALIDATION = True
except ModuleNotFoundError:
log.warning(
"The `ingress` library needs the `jsonschema` package to be able "
"to do runtime data validation; without it, it will still work but validation "
"will be disabled. \n"
"It is recommended to add `jsonschema` to the 'requirements.txt' of your charm, "
"which will enable this feature."
)
DO_VALIDATION = False
INGRESS_REQUIRES_APP_SCHEMA = {
"type": "object",
"properties": {
"model": {"type": "string"},
"name": {"type": "string"},
"host": {"type": "string"},
"port": {"type": "string"},
},
"required": ["model", "name", "host", "port"],
}
INGRESS_PROVIDES_APP_SCHEMA = {
"type": "object",
"properties": {
"ingress": {"type": "object", "properties": {"url": {"type": "string"}}},
},
"required": ["ingress"],
}
try:
from typing import TypedDict
except ImportError:
from typing_extensions import TypedDict # py35 compat
# Model of the data a unit implementing the requirer will need to provide.
RequirerData = TypedDict("RequirerData", {"model": str, "name": str, "host": str, "port": int})
# Provider ingress data model.
ProviderIngressData = TypedDict("ProviderIngressData", {"url": str})
# Provider application databag model.
ProviderApplicationData = TypedDict("ProviderApplicationData", {"ingress": ProviderIngressData})
def _validate_data(data, schema):
"""Checks whether `data` matches `schema`.
Will raise DataValidationError if the data is not valid, else return None.
"""
if not DO_VALIDATION:
return
try:
jsonschema.validate(instance=data, schema=schema)
except jsonschema.ValidationError as e:
raise DataValidationError(data, schema) from e
class DataValidationError(RuntimeError):
"""Raised when data validation fails on IPU relation data."""
class _IngressPerAppBase(Object):
"""Base class for IngressPerUnit interface classes."""
def __init__(self, charm: CharmBase, relation_name: str = DEFAULT_RELATION_NAME):
super().__init__(charm, relation_name)
self.charm: CharmBase = charm
self.relation_name = relation_name
self.app = self.charm.app
self.unit = self.charm.unit
observe = self.framework.observe
rel_events = charm.on[relation_name]
observe(rel_events.relation_created, self._handle_relation)
observe(rel_events.relation_joined, self._handle_relation)
observe(rel_events.relation_changed, self._handle_relation)
observe(rel_events.relation_broken, self._handle_relation_broken)
observe(charm.on.leader_elected, self._handle_upgrade_or_leader)
observe(charm.on.upgrade_charm, self._handle_upgrade_or_leader)
@property
def relations(self):
"""The list of Relation instances associated with this endpoint."""
return list(self.charm.model.relations[self.relation_name])
def _handle_relation(self, event):
"""Subclasses should implement this method to handle a relation update."""
pass
def _handle_relation_broken(self, event):
"""Subclasses should implement this method to handle a relation breaking."""
pass
def _handle_upgrade_or_leader(self, event):
"""Subclasses should implement this method to handle upgrades or leadership change."""
pass
class _IPAEvent(RelationEvent):
__args__ = () # type: Tuple[str, ...]
__optional_kwargs__ = {} # type: Dict[str, Any]
@classmethod
def __attrs__(cls):
return cls.__args__ + tuple(cls.__optional_kwargs__.keys())
def __init__(self, handle, relation, *args, **kwargs):
super().__init__(handle, relation)
if not len(self.__args__) == len(args):
raise TypeError("expected {} args, got {}".format(len(self.__args__), len(args)))
for attr, obj in zip(self.__args__, args):
setattr(self, attr, obj)
for attr, default in self.__optional_kwargs__.items():
obj = kwargs.get(attr, default)
setattr(self, attr, obj)
def snapshot(self) -> dict:
dct = super().snapshot()
for attr in self.__attrs__():
obj = getattr(self, attr)
try:
dct[attr] = obj
except ValueError as e:
raise ValueError(
"cannot automagically serialize {}: "
"override this method and do it "
"manually.".format(obj)
) from e
return dct
def restore(self, snapshot: dict) -> None:
super().restore(snapshot)
for attr, obj in snapshot.items():
setattr(self, attr, obj)
class IngressPerAppDataProvidedEvent(_IPAEvent):
"""Event representing that ingress data has been provided for an app."""
__args__ = ("name", "model", "port", "host")
if typing.TYPE_CHECKING:
name = None # type: str
model = None # type: str
port = None # type: int
host = None # type: str
class IngressPerAppDataRemovedEvent(RelationEvent):
"""Event representing that ingress data has been removed for an app."""
class IngressPerAppProviderEvents(ObjectEvents):
"""Container for IPA Provider events."""
data_provided = EventSource(IngressPerAppDataProvidedEvent)
data_removed = EventSource(IngressPerAppDataRemovedEvent)
class IngressPerAppProvider(_IngressPerAppBase):
"""Implementation of the provider of ingress."""
on = IngressPerAppProviderEvents()
def __init__(self, charm: CharmBase, relation_name: str = DEFAULT_RELATION_NAME):
"""Constructor for IngressPerAppProvider.
Args:
charm: The charm that is instantiating the instance.
relation_name: The name of the relation endpoint to bind to
(defaults to "ingress").
"""
super().__init__(charm, relation_name)
def _handle_relation(self, event):
# created, joined or changed: if remote side has sent the required data:
# notify listeners.
if self.is_ready(event.relation):
data = self._get_requirer_data(event.relation)
self.on.data_provided.emit(
event.relation,
data["name"],
data["model"],
data["port"],
data["host"],
)
def _handle_relation_broken(self, event):
self.on.data_removed.emit(event.relation)
def wipe_ingress_data(self, relation: Relation):
"""Clear ingress data from relation."""
assert self.unit.is_leader(), "only leaders can do this"
try:
relation.data
except ModelError as e:
log.warning(
"error {} accessing relation data for {!r}. "
"Probably a ghost of a dead relation is still "
"lingering around.".format(e, relation.name)
)
return
del relation.data[self.app]["ingress"]
def _get_requirer_data(self, relation: Relation) -> RequirerData:
"""Fetch and validate the requirer's app databag.
For convenience, we convert 'port' to integer.
"""
if not all((relation.app, relation.app.name)):
# Handle edge case where remote app name can be missing, e.g.,
# relation_broken events.
# FIXME https://github.com/canonical/traefik-k8s-operator/issues/34
return {}
databag = relation.data[relation.app]
try:
remote_data = {k: databag[k] for k in ("model", "name", "host", "port")}
except KeyError as e:
# incomplete data / invalid data
log.debug("error {}; ignoring...".format(e))
return {}
except TypeError as e:
raise DataValidationError("Error casting remote data: {}".format(e))
_validate_data(remote_data, INGRESS_REQUIRES_APP_SCHEMA)
remote_data["port"] = int(remote_data["port"])
return remote_data
def get_data(self, relation: Relation) -> RequirerData:
"""Fetch the remote app's databag, i.e. the requirer data."""
return self._get_requirer_data(relation)
def is_ready(self, relation: Relation = None):
"""The Provider is ready if the requirer has sent valid data."""
if not relation:
return any(map(self.is_ready, self.relations))
try:
return bool(self._get_requirer_data(relation))
except DataValidationError as e:
log.warning("Requirer not ready; validation error encountered: %s" % str(e))
return False
def _provided_url(self, relation: Relation) -> ProviderIngressData:
"""Fetch and validate this app databag; return the ingress url."""
if not all((relation.app, relation.app.name, self.unit.is_leader())):
# Handle edge case where remote app name can be missing, e.g.,
# relation_broken events.
# Also, only leader units can read own app databags.
# FIXME https://github.com/canonical/traefik-k8s-operator/issues/34
return {} # noqa
# fetch the provider's app databag
raw_data = relation.data[self.app].get("ingress")
if not raw_data:
raise RuntimeError("This application did not `publish_url` yet.")
ingress: ProviderIngressData = yaml.safe_load(raw_data)
_validate_data({"ingress": ingress}, INGRESS_PROVIDES_APP_SCHEMA)
return ingress
def publish_url(self, relation: Relation, url: str):
"""Publish to the app databag the ingress url."""
ingress = {"url": url}
ingress_data = {"ingress": ingress}
_validate_data(ingress_data, INGRESS_PROVIDES_APP_SCHEMA)
relation.data[self.app]["ingress"] = yaml.safe_dump(ingress)
@property
def proxied_endpoints(self):
"""Returns the ingress settings provided to applications by this IngressPerAppProvider.
For example, when this IngressPerAppProvider has provided the
`http://foo.bar/my-model.my-app` URL to the my-app application, the returned dictionary
will be:
```
{
"my-app": {
"url": "http://foo.bar/my-model.my-app"
}
}
```
"""
results = {}
for ingress_relation in self.relations:
results[ingress_relation.app.name] = self._provided_url(ingress_relation)
return results
class IngressPerAppReadyEvent(_IPAEvent):
"""Event representing that ingress for an app is ready."""
__args__ = ("url",)
if typing.TYPE_CHECKING:
url = None # type: str
class IngressPerAppRevokedEvent(RelationEvent):
"""Event representing that ingress for an app has been revoked."""
class IngressPerAppRequirerEvents(ObjectEvents):
"""Container for IPA Requirer events."""
ready = EventSource(IngressPerAppReadyEvent)
revoked = EventSource(IngressPerAppRevokedEvent)
class IngressPerAppRequirer(_IngressPerAppBase):
"""Implementation of the requirer of the ingress relation."""
on = IngressPerAppRequirerEvents()
# used to prevent spur1ious urls to be sent out if the event we're currently
# handling is a relation-broken one.
_stored = StoredState()
def __init__(
self,
charm: CharmBase,
relation_name: str = DEFAULT_RELATION_NAME,
*,
host: str = None,
port: int = None,
):
"""Constructor for IngressRequirer.
The request args can be used to specify the ingress properties when the
instance is created. If any are set, at least `port` is required, and
they will be sent to the ingress provider as soon as it is available.
All request args must be given as keyword args.
Args:
charm: the charm that is instantiating the library.
relation_name: the name of the relation endpoint to bind to (defaults to `ingress`);
relation must be of interface type `ingress` and have "limit: 1")
host: Hostname to be used by the ingress provider to address the requiring
application; if unspecified, the default Kubernetes service name will be used.
Request Args:
port: the port of the service
"""
super().__init__(charm, relation_name)
self.charm: CharmBase = charm
self.relation_name = relation_name
self._stored.set_default(current_url=None)
# if instantiated with a port, and we are related, then
# we immediately publish our ingress data to speed up the process.
if port:
self._auto_data = host, port
else:
self._auto_data = None
def _handle_relation(self, event):
# created, joined or changed: if we have auto data: publish it
self._publish_auto_data(event.relation)
if self.is_ready():
# Avoid spurious events, emit only when there is a NEW URL available
new_url = (
None
if isinstance(event, RelationBrokenEvent)
else self._get_url_from_relation_data()
)
if self._stored.current_url != new_url:
self._stored.current_url = new_url
self.on.ready.emit(event.relation, new_url)
def _handle_relation_broken(self, event):
self._stored.current_url = None
self.on.revoked.emit(event.relation)
def _handle_upgrade_or_leader(self, event):
"""On upgrade/leadership change: ensure we publish the data we have."""
for relation in self.relations:
self._publish_auto_data(relation)
def is_ready(self):
"""The Requirer is ready if the Provider has sent valid data."""
try:
return bool(self._get_url_from_relation_data())
except DataValidationError as e:
log.warning("Requirer not ready; validation error encountered: %s" % str(e))
return False
def _publish_auto_data(self, relation: Relation):
if self._auto_data and self.unit.is_leader():
host, port = self._auto_data
self.provide_ingress_requirements(host=host, port=port)
def provide_ingress_requirements(self, *, host: str = None, port: int):
"""Publishes the data that Traefik needs to provide ingress.
NB only the leader unit is supposed to do this.
Args:
host: Hostname to be used by the ingress provider to address the
requirer unit; if unspecified, FQDN will be used instead
port: the port of the service (required)
"""
# get only the leader to publish the data since we only
# require one unit to publish it -- it will not differ between units,
# unlike in ingress-per-unit.
assert self.unit.is_leader(), "only leaders should do this."
assert self.relation, "no relation"
if not host:
host = socket.getfqdn()
data = {
"model": self.model.name,
"name": self.app.name,
"host": host,
"port": str(port),
}
_validate_data(data, INGRESS_REQUIRES_APP_SCHEMA)
self.relation.data[self.app].update(data)
@property
def relation(self):
"""The established Relation instance, or None."""
return self.relations[0] if self.relations else None
def _get_url_from_relation_data(self) -> Optional[str]:
"""The full ingress URL to reach the current unit.
Returns None if the URL isn't available yet.
"""
relation = self.relation
if not relation:
return None
# fetch the provider's app databag
try:
raw = relation.data.get(relation.app, {}).get("ingress")
except ModelError as e:
log.debug(
f"Error {e} attempting to read remote app data; "
f"probably we are in a relation_departed hook"
)
return None
if not raw:
return None
ingress: ProviderIngressData = yaml.safe_load(raw)
_validate_data({"ingress": ingress}, INGRESS_PROVIDES_APP_SCHEMA)
return ingress["url"]
@property
def url(self) -> Optional[str]:
"""The full ingress URL to reach the current unit.
Returns None if the URL isn't available yet.
"""
data = self._stored.current_url or None # type: ignore
assert isinstance(data, (str, type(None))) # for static checker
return data