Simplifies service management (#19)

src/charm.py

@@ -11,7 +11,6 @@
     LTECoreAvailableEvent,
     LTECoreRequires,
 )
-from jinja2 import Template
 from ops.charm import (
     ActionEvent,
     CharmBase,
@@ -22,31 +21,13 @@
 from ops.main import main
 from ops.model import ActiveStatus, BlockedStatus, MaintenanceStatus, WaitingStatus
 
-from utils import (
-    get_iface_ip_address,
-    ip_from_default_iface,
-    service_active,
-    service_enable,
-    service_restart,
-    service_stop,
-    shell,
-    systemctl_daemon_reload,
-    wait_for_condition,
-)
+from linux_service import Service
+from utils import get_iface_ip_address, ip_from_default_iface, shell, wait_for_condition
 
 logger = logging.getLogger(__name__)
 
 CONFIG_PATH = "/snap/srsran/current/config"
-
-SRS_ENB_SERVICE = "srsenb"
-SRS_ENB_SERVICE_TEMPLATE = "./templates/srsenb.service"
-SRS_ENB_SERVICE_PATH = "/etc/systemd/system/srsenb.service"
-
-SRS_UE_SERVICE = "srsue"
-SRS_UE_SERVICE_TEMPLATE = "./templates/srsue.service"
-SRS_UE_SERVICE_PATH = "/etc/systemd/system/srsue.service"
-
-WAIT_FOR_UE_IP_TIMEOUT = 10
+WAIT_FOR_UE_IP_TIMEOUT = 20
 
 
 class SrsRANCharm(CharmBase):
@@ -55,6 +36,8 @@ class SrsRANCharm(CharmBase):
     def __init__(self, *args):
         """Observes various events."""
         super().__init__(*args)
+        self.ue_service = Service("srsue")
+        self.enb_service = Service("srsenb")
 
         # Basic hooks
         self.framework.observe(self.on.install, self._on_install)
@@ -96,9 +79,13 @@ def _on_config_changed(self, _: Union[ConfigChangedEvent, LTECoreAvailableEvent]
             self.unit.status = WaitingStatus("Waiting for MME address to be available")
             return
         self.unit.status = MaintenanceStatus("Configuring srsenb")
-        self._configure_srsenb_service()
-        service_enable(SRS_ENB_SERVICE)
-        service_restart(SRS_ENB_SERVICE)
+        self.enb_service.create(
+            command=self._get_srsenb_command(),
+            user="root",
+            description="SRS eNodeB Emulator Service",
+        )
+        self.enb_service.enable()
+        self.enb_service.restart()
         self.unit.status = ActiveStatus("srsenb started")
 
     def _on_attach_ue_action(self, event: ActionEvent) -> None:
@@ -112,18 +99,23 @@ def _on_attach_ue_action(self, event: ActionEvent) -> None:
         if not self._lte_core_mme_address_is_available:
             event.fail("MME address is not available")
             return
-        if not service_active(SRS_ENB_SERVICE):
+        if not self.enb_service.is_active():
             event.fail("Failed to attach. The EnodeB is not running.")
             return
-        if service_active(SRS_UE_SERVICE):
+        if self.ue_service.is_active():
             event.fail("Failed to attach. UE already running, please detach first.")
             return
-        self._configure_srsue_service(
-            ue_usim_imsi=event.params["usim-imsi"],
-            ue_usim_k=event.params["usim-k"],
-            ue_usim_opc=event.params["usim-opc"],
+        self.ue_service.create(
+            command=self._get_srsue_command(
+                ue_usim_imsi=event.params["usim-imsi"],
+                ue_usim_k=event.params["usim-k"],
+                ue_usim_opc=event.params["usim-opc"],
+            ),
+            user="ubuntu",
+            description="SRS UE Emulator Service",
+            exec_stop_post="service srsenb restart",
         )
-        service_restart(SRS_UE_SERVICE)
+        self.ue_service.restart()
         if not wait_for_condition(
             lambda: get_iface_ip_address("tun_srsue"), timeout=WAIT_FOR_UE_IP_TIMEOUT
         ):
@@ -141,8 +133,8 @@ def _on_attach_ue_action(self, event: ActionEvent) -> None:
 
     def _on_detach_ue_action(self, event: ActionEvent) -> None:
         """Triggered on detach_ue action."""
-        service_stop(SRS_UE_SERVICE)
-        self._configure_srsue_service(None, None, None)  # type: ignore[arg-type]
+        self.ue_service.stop()
+        self.ue_service.delete()
         self.unit.status = ActiveStatus("ue detached")
         event.set_results({"status": "ok", "message": "Detached successfully"})
 
@@ -151,14 +143,6 @@ def _on_remove_default_gw_action(self, event: ActionEvent) -> None:
         shell("route del default")
         event.set_results({"status": "ok", "message": "Default route removed!"})
 
-    def _configure_srsenb_service(self) -> None:
-        """Configures srs enb service."""
-        self._configure_service(
-            command=self._get_srsenb_command(),
-            service_template=SRS_ENB_SERVICE_TEMPLATE,
-            service_path=SRS_ENB_SERVICE_PATH,
-        )
-
     @staticmethod
     def _install_srsran() -> None:
         """Installs srsRAN snap."""
@@ -171,16 +155,6 @@ def _uninstall_srsran() -> None:
         shell("snap remove srsran --purge")
         logger.info("Removed srsRAN snap")
 
-    def _configure_srsue_service(
-        self, ue_usim_imsi: str, ue_usim_k: str, ue_usim_opc: str
-    ) -> None:
-        """Configures srs ue service."""
-        self._configure_service(
-            command=self._get_srsue_command(ue_usim_imsi, ue_usim_k, ue_usim_opc),
-            service_template=SRS_UE_SERVICE_TEMPLATE,
-            service_path=SRS_UE_SERVICE_PATH,
-        )
-
     @property
     def _lte_core_relation_is_created(self) -> bool:
         """Checks if the relation with the LTE core is created."""
@@ -200,19 +174,6 @@ def _lte_core_mme_address_is_available(self) -> bool:
         """Checks if the MME address is available."""
         return self._mme_address is not None
 
-    @staticmethod
-    def _configure_service(
-        command: str,
-        service_template: str,
-        service_path: str,
-    ) -> None:
-        """Renders service template and reload daemon service."""
-        with open(service_template, "r") as template:
-            service_content = Template(template.read()).render(command=command)
-        with open(service_path, "w") as service:
-            service.write(service_content)
-        systemctl_daemon_reload()
-
     def _get_srsenb_command(self) -> str:
         """Returns srs enb command."""
         srsenb_command = ["/snap/bin/srsran.srsenb"]

src/linux_service.py

@@ -0,0 +1,76 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Set of utilities related to managing Linux services."""
+
+import logging
+import os
+from subprocess import CalledProcessError
+from typing import Optional
+
+from jinja2 import Template
+
+from utils import shell
+
+SERVICE_TEMPLATE = "./templates/service.j2"
+
+logger = logging.getLogger(__name__)
+
+
+class Service:
+    """Class for representing a Linux Service."""
+
+    def __init__(self, name: str):
+        """Sets service name."""
+        self.name = name
+
+    def is_active(self) -> bool:
+        """Returns whether service is active."""
+        try:
+            response = self._systemctl("is-active")
+            return response == "active\n"
+        except CalledProcessError:
+            return False
+
+    def create(
+        self, command: str, user: str, description: str, exec_stop_post: Optional[str] = None
+    ) -> None:
+        """Creates a linux service."""
+        with open(SERVICE_TEMPLATE, "r") as template:
+            service_content = Template(template.read()).render(
+                command=command, user=user, description=description, exec_stop_post=exec_stop_post
+            )
+        with open(f"/etc/systemd/system/{self.name}.service", "w") as service:
+            service.write(service_content)
+        self._systemctl_daemon_reload()
+
+    def delete(self) -> None:
+        """Deletes a linux service."""
+        try:
+            os.remove(f"/etc/systemd/system/{self.name}.service")
+        except FileNotFoundError:
+            pass
+
+    def restart(self) -> None:
+        """Restarts a linux service."""
+        self._systemctl("restart")
+        logger.info("Service %s restarted", self.name)
+
+    def stop(self) -> None:
+        """Stops a linux service."""
+        self._systemctl("stop")
+        logger.info("Service %s stopped", self.name)
+
+    def enable(self) -> None:
+        """Enables a linux service."""
+        self._systemctl("enable")
+        logger.info("Service %s enabled", self.name)
+
+    def _systemctl(self, action: str) -> str:
+        return shell(f"systemctl {action} {self.name}")
+
+    @staticmethod
+    def _systemctl_daemon_reload() -> None:
+        """Runs `systemctl daemon-reload`."""
+        shell("systemctl daemon-reload")
+        logger.info("Systemd manager configuration reloaded")

src/utils.py

@@ -6,86 +6,22 @@
 import logging
 import subprocess
 import time
-from subprocess import CalledProcessError
-from typing import Callable, List, Optional
+from typing import Callable, Optional
 
 import netifaces  # type: ignore[import]
-from netaddr import IPAddress, IPNetwork  # type: ignore[import]
-from netaddr.core import AddrFormatError  # type: ignore[import]
 
 logger = logging.getLogger(__name__)
 
 
-def service_active(service_name: str) -> bool:
-    """Returns whether a given service is active."""
-    try:
-        response = shell(f"systemctl is-active {service_name}")
-        return response == "active\n"
-    except CalledProcessError:
-        return False
-
-
 def shell(command: str) -> str:
     """Runs a shell command."""
     response = subprocess.run(command, shell=True, stdout=subprocess.PIPE, encoding="utf-8")
     response.check_returncode()
     return response.stdout
 
 
-def get_local_ipv4_networks() -> List[IPNetwork]:
-    """Returns list of IPv4 networks."""
-    networks = []
-    interfaces = netifaces.interfaces()
-    for interface in interfaces:
-        addresses = netifaces.ifaddresses(interface)
-        if netifaces.AF_INET in addresses:
-            ipv4_addr = addresses[netifaces.AF_INET][0]
-            network = IPNetwork(f'{ipv4_addr["addr"]}/{ipv4_addr["netmask"]}')
-            networks.append(network)
-    return networks
-
-
-def is_ipv4(ip: str) -> bool:
-    """Returns whether an IP address is IPv4."""
-    try:
-        if not isinstance(ip, str) or len(ip.split(".")) != 4:
-            return False
-        IPAddress(ip)
-        return True
-    except AddrFormatError:
-        return False
-
-
-def _systemctl(action: str, service_name: str) -> None:
-    shell(f"systemctl {action} {service_name}")
-
-
-def service_restart(service_name: str) -> None:
-    """Restarts a given service."""
-    _systemctl("restart", service_name)
-    logger.info("Service %s restarted", service_name)
-
-
-def service_stop(service_name: str) -> None:
-    """Stops a given service."""
-    _systemctl("stop", service_name)
-    logger.info("Service %s stopped", service_name)
-
-
-def service_enable(service_name: str) -> None:
-    """Enables a given service."""
-    _systemctl("enable", service_name)
-    logger.info("Service %s enabled", service_name)
-
-
-def systemctl_daemon_reload() -> None:
-    """Runs `systemctl daemon-reload`."""
-    shell("systemctl daemon-reload")
-    logger.info("Systemd manager configuration reloaded")
-
-
 def ip_from_default_iface() -> Optional[str]:
-    """Returns a Ip address from the default interface."""
+    """Returns the default interface's IP address."""
     default_gateway = netifaces.gateways()["default"]
     if netifaces.AF_INET in default_gateway:
         _, iface = netifaces.gateways()["default"][netifaces.AF_INET]
@@ -95,19 +31,6 @@ def ip_from_default_iface() -> Optional[str]:
     return None
 
 
-def ip_from_iface(subnet: str) -> Optional[str]:
-    """Returns Ip address from a given subnet."""
-    try:
-        target_network = IPNetwork(subnet)
-        networks = get_local_ipv4_networks()
-        return next(
-            (network.ip.format() for network in networks if network.ip in target_network), None
-        )
-
-    except AddrFormatError:
-        return None
-
-
 def get_iface_ip_address(iface: str) -> Optional[str]:
     """Get the IP address of the given interface.
 

templates/srsue.service → templates/service.j2

@@ -1,16 +1,18 @@
 [Unit]
-Description=Srs User Emulator Service
+Description={{ description }}
 After=network.target
 StartLimitIntervalSec=0
 [Service]
 Type=simple
 Restart=always
 RestartSec=1
+User={{ user }}
 ExecStart={{ command }}
-User=ubuntu
 KillSignal=SIGINT
 TimeoutStopSec=10
-ExecStopPost=service srsenb restart
+{%- if exec_stop_post %}
+ExecStopPost={{ exec_stop_post }}
+{%- endif %}
 
 [Install]
 WantedBy=multi-user.target