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 pod_spec import make_pod_spec
-LOGGER = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
NBI_PORT = 9999
Args:
event (EventBase): Kafka relation event.
"""
- message_host = event.relation.data[event.unit].get("host")
- message_port = event.relation.data[event.unit].get("port")
+ data_loc = event.unit if event.unit else event.app
+
+ message_host = event.relation.data[data_loc].get("host")
+ message_port = event.relation.data[data_loc].get("port")
if (
message_host
)
):
self.state.message_host = message_host
- self.state.message_port = message_port
+ self.state.message_port = int(message_port)
self.on.configure_pod.emit()
def _on_kafka_relation_departed(self, event: EventBase) -> NoReturn:
Args:
event (EventBase): DB relation event.
"""
- database_uri = event.relation.data[event.unit].get("connection_string")
+ data_loc = event.unit if event.unit else event.app
+
+ database_uri = event.relation.data[data_loc].get("connection_string")
if database_uri and self.state.database_uri != database_uri:
self.state.database_uri = database_uri
Args:
event (EventBase): Keystone relation event.
"""
- keystone_host = event.relation.data[event.unit].get("host")
- keystone_port = event.relation.data[event.unit].get("port")
- keystone_user_domain_name = event.relation.data[event.unit].get(
+ data_loc = event.unit if event.unit else event.app
+
+ keystone_host = event.relation.data[data_loc].get("host")
+ keystone_port = event.relation.data[data_loc].get("port")
+ keystone_user_domain_name = event.relation.data[data_loc].get(
"user_domain_name"
)
- keystone_project_domain_name = event.relation.data[event.unit].get(
+ keystone_project_domain_name = event.relation.data[data_loc].get(
"project_domain_name"
)
- keystone_username = event.relation.data[event.unit].get("username")
- keystone_password = event.relation.data[event.unit].get("password")
- keystone_service = event.relation.data[event.unit].get("service")
+ keystone_username = event.relation.data[data_loc].get("username")
+ keystone_password = event.relation.data[data_loc].get("password")
+ keystone_service = event.relation.data[data_loc].get("service")
if (
keystone_host
)
):
self.state.keystone_host = keystone_host
- self.state.keystone_port = keystone_port
+ self.state.keystone_port = int(keystone_port)
self.state.keystone_user_domain_name = keystone_user_domain_name
self.state.keystone_project_domain_name = keystone_project_domain_name
self.state.keystone_username = keystone_username
Args:
event (EventBase): Prometheus relation event.
"""
- prometheus_host = event.relation.data[event.unit].get("hostname")
- prometheus_port = event.relation.data[event.unit].get("port")
+ data_loc = event.unit if event.unit else event.app
+
+ prometheus_host = event.relation.data[data_loc].get("hostname")
+ prometheus_port = event.relation.data[data_loc].get("port")
if (
prometheus_host
)
):
self.state.prometheus_host = prometheus_host
- self.state.prometheus_port = prometheus_port
+ self.state.prometheus_port = int(prometheus_port)
self.on.configure_pod.emit()
def _on_prometheus_relation_departed(self, event: EventBase) -> NoReturn:
Args:
event (EventBase): NBI relation event.
"""
- if self.unit.is_leader():
- rel_data = {
- "host": self.model.app.name,
- "port": str(NBI_PORT),
- }
- for k, v in rel_data.items():
- event.relation.data[self.model.app][k] = v
+ rel_data = {
+ "host": self.model.app.name,
+ "port": str(NBI_PORT),
+ }
+ for k, v in rel_data.items():
+ event.relation.data[self.unit][k] = v
def _missing_relations(self) -> str:
"""Checks if there missing relations.
self.model.app.name,
self.port,
)
- except ValidationError as exc:
+ except ValueError as exc:
+ logger.exception("Config/Relation data validation error")
self.unit.status = BlockedStatus(str(exc))
return