import logging
from typing import NoReturn, Optional
+from charms.kafka_k8s.v0.kafka import KafkaEvents, KafkaRequires
from ops.main import main
from opslib.osm.charm import CharmedOsmBase, RelationsMissing
-from opslib.osm.interfaces.kafka import KafkaClient
from opslib.osm.interfaces.mongo import MongoClient
from opslib.osm.pod import (
ContainerV3Builder,
class PlaCharm(CharmedOsmBase):
+
+ on = KafkaEvents()
+
def __init__(self, *args) -> NoReturn:
super().__init__(*args, oci_image="image")
- self.kafka_client = KafkaClient(self, "kafka")
- self.framework.observe(self.on["kafka"].relation_changed, self.configure_pod)
- self.framework.observe(self.on["kafka"].relation_broken, self.configure_pod)
+ self.kafka = KafkaRequires(self)
+ self.framework.observe(self.on.kafka_available, self.configure_pod)
+ self.framework.observe(self.on.kafka_broken, self.configure_pod)
self.mongodb_client = MongoClient(self, "mongodb")
self.framework.observe(self.on["mongodb"].relation_changed, self.configure_pod)
def _check_missing_dependencies(self, config: ConfigModel):
missing_relations = []
- if self.kafka_client.is_missing_data_in_unit():
+ if not self.kafka.host or not self.kafka.port:
missing_relations.append("kafka")
if not config.mongodb_uri and self.mongodb_client.is_missing_data_in_unit():
missing_relations.append("mongodb")
"OSMPLA_GLOBAL_LOG_LEVEL": config.log_level,
# Kafka configuration
"OSMPLA_MESSAGE_DRIVER": "kafka",
- "OSMPLA_MESSAGE_HOST": self.kafka_client.host,
- "OSMPLA_MESSAGE_PORT": self.kafka_client.port,
+ "OSMPLA_MESSAGE_HOST": self.kafka.host,
+ "OSMPLA_MESSAGE_PORT": self.kafka.port,
# Database configuration
"OSMPLA_DATABASE_DRIVER": "mongo",
}
# Add Pod restart policy
restart_policy = PodRestartPolicy()
- restart_policy.add_secrets(secret_names=(mongodb_secret_name))
+ restart_policy.add_secrets(secret_names=(mongodb_secret_name,))
pod_spec_builder.set_restart_policy(restart_policy)
# Add container to pod spec