OSM Libs
- By OSM Charmers
Channel | Revision | Published | Runs on |
---|---|---|---|
latest/edge | 1 | 12 May 2022 |
juju deploy osm-libs --channel edge
Deploy universal operators easily with Juju, the Universal Operator Lifecycle Manager.
Platform:
20.04
charms.osm_libs.v0.osm_config
-
- Last updated 13 May 2022
- Revision Library version 0.1
#!/usr/bin/env python3
# Copyright 2022 Canonical Ltd.
# See LICENSE file for licensing details.
"""OSM Config Library.
This library provides a simple way of getting information injected
by OSM in the osm-config configuration option.
# Schema
This is the JSON schema that is supported by the library
```json
{
"v0": {
"k8s" : {
"services": {
<service_name>: {
"ip": [<ip>],
"ports": {
<port_name>: {
"port": <port_number>,
"protocol": "TCP",
}
}
}
}
}
}
}
```
# Getting started
Execute the following command inside your Charmed Operator folder to fetch the library.
```shell
charmcraft fetch-lib charms.osm_libs.v0.osm_config
```
Import the library, and initialize it you charm constructor.
```python
# ...
from charms.osm_libs.v0.osm_config import OsmConfig
class SomeCharm(CharmBase):
def __init__(self, *args):
# ...
self.osm_config = OsmConfig(self)
# ...
```
# K8s services
The `OsmConfig` class allows you to easily get information about the K8s services
injected by OSM.
This is an example on how to get information the IP and port of a particular K8s service.
```python
class SomeCharm(CharmBase):
def _on_some_action(self, event):
openldap_service = self.osm_config.k8s.get_service("openldap")
openldap_ip = openldap_service.ip
openldap_port = openldap_service.get_port("openldap")
```
"""
import json
from typing import Any, Dict
from ops.charm import CharmBase
# The unique Charmhub library identifier, never change it/search
LIBID = "f5969a6bed59469dbe45f8016133c685"
# 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 = 1
class K8sService:
"""Represents a K8s Service."""
def __init__(self, info):
self.info = info
@property
def ip(self) -> str:
"""Get the first IP of the service."""
return self.info["ip"][0]
def get_port(self, port_name: str):
"""Get port by name.
Args:
port_name: Name of the port.
"""
return self.info["ports"][port_name]["port"]
class K8sConfig:
"""Represents the K8s configuration in OsmConfig."""
def __init__(self, k8s_config: Dict[str, Any]) -> None:
self.config = k8s_config
def get_service(self, service_name: str) -> K8sService:
"""Get K8s service by name.
This function will only return one K8s service, the first that contains
the `service_name` provided as argument.
Args:
service_name: Name of the K8s service.
Returns:
K8sService: the Kubernetes Service.
"""
services = [
service for s_name, service in self.get_services().items() if service_name in s_name
]
return services[0] if len(services) > 0 else None
def get_services(self) -> Dict[str, K8sService]:
"""Get all K8s services.
Returns:
Dict[str, K8sService]: Dictionary with service names as keys,
and K8sService objects as values.
"""
return {
s_name: K8sService(s_values)
for s_name, s_values in self.config.get("services", {}).items()
}
class OsmConfig:
"""Represents the configuration injected by OSM."""
option_name = "osm-config"
version = f"v{LIBAPI}"
def __init__(self, charm: CharmBase) -> None:
# Check if option_name exists in the config options.
if self.option_name not in charm.config:
raise Exception(f"config option {OsmConfig.option_name} not provided.")
self.config = json.loads(charm.config[OsmConfig.option_name])
if OsmConfig.version not in self.config:
raise Exception(f"version {OsmConfig.version} not present in osm-config.")
self.k8s = K8sConfig(self.config[OsmConfig.version].get("k8s", {}))