diff --git a/lib/charms/observability_libs/v1/kubernetes_service_patch.py b/lib/charms/observability_libs/v1/kubernetes_service_patch.py index 061ca18e..64dd13ce 100644 --- a/lib/charms/observability_libs/v1/kubernetes_service_patch.py +++ b/lib/charms/observability_libs/v1/kubernetes_service_patch.py @@ -89,6 +89,26 @@ def __init__(self, *args): # ... ``` +Bound with custom events by providing `refresh_event` argument: +For example, you would like to have a configurable port in your charm and want to apply +service patch every time charm config is changed. + +```python +from charms.observability_libs.v1.kubernetes_service_patch import KubernetesServicePatch +from lightkube.models.core_v1 import ServicePort + +class SomeCharm(CharmBase): + def __init__(self, *args): + # ... + port = ServicePort(int(self.config["charm-config-port"]), name=f"{self.app.name}") + self.service_patcher = KubernetesServicePatch( + self, + [port], + refresh_event=self.on.config_changed + ) + # ... +``` + Additionally, you may wish to use mocks in your charm's unit testing to ensure that the library does not try to make any API calls, or open any files during testing that are unlikely to be present, and could break your tests. The easiest way to do this is during your test `setUp`: @@ -105,7 +125,7 @@ def setUp(self, *unused): import logging from types import MethodType -from typing import List, Literal +from typing import List, Literal, Optional, Union from lightkube import ApiError, Client from lightkube.core import exceptions @@ -114,7 +134,7 @@ def setUp(self, *unused): from lightkube.resources.core_v1 import Service from lightkube.types import PatchType from ops.charm import CharmBase -from ops.framework import Object +from ops.framework import BoundEvent, Object logger = logging.getLogger(__name__) @@ -126,7 +146,7 @@ def setUp(self, *unused): # Increment this PATCH version before using `charmcraft publish-lib` or reset # to 0 if you are raising the major API version -LIBPATCH = 3 +LIBPATCH = 7 ServiceType = Literal["ClusterIP", "LoadBalancer"] @@ -138,11 +158,13 @@ def __init__( self, charm: CharmBase, ports: List[ServicePort], - service_name: str = None, + service_name: Optional[str] = None, service_type: ServiceType = "ClusterIP", - additional_labels: dict = None, - additional_selectors: dict = None, - additional_annotations: dict = None, + additional_labels: Optional[dict] = None, + additional_selectors: Optional[dict] = None, + additional_annotations: Optional[dict] = None, + *, + refresh_event: Optional[Union[BoundEvent, List[BoundEvent]]] = None, ): """Constructor for KubernetesServicePatch. @@ -158,6 +180,9 @@ def __init__( additional_selectors: Selectors to be added to the kubernetes service (by default only "app.kubernetes.io/name" is set to the service name) additional_annotations: Annotations to be added to the kubernetes service. + refresh_event: an optional bound event or list of bound events which + will be observed to re-apply the patch (e.g. on port change). + The `install` and `upgrade-charm` events would be observed regardless. """ super().__init__(charm, "kubernetes-service-patch") self.charm = charm @@ -176,15 +201,24 @@ def __init__( # Ensure this patch is applied during the 'install' and 'upgrade-charm' events self.framework.observe(charm.on.install, self._patch) self.framework.observe(charm.on.upgrade_charm, self._patch) + self.framework.observe(charm.on.update_status, self._patch) + + # apply user defined events + if refresh_event: + if not isinstance(refresh_event, list): + refresh_event = [refresh_event] + + for evt in refresh_event: + self.framework.observe(evt, self._patch) def _service_object( self, ports: List[ServicePort], - service_name: str = None, + service_name: Optional[str] = None, service_type: ServiceType = "ClusterIP", - additional_labels: dict = None, - additional_selectors: dict = None, - additional_annotations: dict = None, + additional_labels: Optional[dict] = None, + additional_selectors: Optional[dict] = None, + additional_annotations: Optional[dict] = None, ) -> Service: """Creates a valid Service representation. @@ -276,9 +310,8 @@ def _is_patched(self, client: Client) -> bool: except ApiError as e: if e.status.code == 404 and self.service_name != self._app: return False - else: - logger.error("Kubernetes service get failed: %s", str(e)) - raise + logger.error("Kubernetes service get failed: %s", str(e)) + raise # Construct a list of expected ports, should the patch be applied expected_ports = [(p.port, p.targetPort) for p in self.service.spec.ports]