Add minimum tox + unit tests for wrapper

Author: phvalguima

poetry.lock

@@ -0,0 +1,129 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+[tool.poetry]
+package-mode = false
+requires-poetry = ">=2.0.0"
+
+[tool.poetry.dependencies]
+python = "^3.10"
+ops = "^2.17.0"
+tenacity = "^9.0.0"
+jinja2 = "^3.1.4"
+overrides = "7.7.0"
+requests = "2.32.3"
+shortuuid = "1.0.13"
+cryptography = "^43.0.1"
+jsonschema = "^4.23.0"
+prometheus-client = ">=0.19.0"
+pydantic = "^1.10.18, <2"
+# pydantic = ">=2.0,<3.0"
+fastapi = ">=0.115.0"
+uvicorn = ">0.11.5"
+kafka-python = ">=2.0"
+
+[tool.poetry.group.charm-libs.dependencies]
+# data_platform_libs/v0/data_interfaces.py
+ops = "^2.17"
+# data_platform_libs/v0/upgrade.py
+# grafana_agent/v0/cos_agent.py requires pydantic <2
+pydantic = "^1.10, <2"
+# tls_certificates_interface/v1/tls_certificates.py
+cryptography = "^43.0.0"
+jsonschema = "^4.23.0"
+# grafana_agent/v0/cos_agent.py
+cosl = "^0.0.41"
+bcrypt = "^4.1.3"
+
+[tool.poetry.group.format]
+optional = true
+
+[tool.poetry.group.format.dependencies]
+ruff = "^0.6.8"
+
+[tool.poetry.group.lint]
+optional = true
+
+[tool.poetry.group.lint.dependencies]
+codespell = "^2.3.0"
+shellcheck-py = "^0.10.0.1"
+
+[tool.poetry.group.unit.dependencies]
+pytest = "^8.3.3"
+pytest-asyncio = "^0.21.2"
+coverage = {extras = ["toml"], version = "^7.6.1"}
+parameterized = "^0.9.0"
+
+[tool.poetry.group.integration.dependencies]
+pytest = "^8.3.3"
+pytest-github-secrets = {git = "https://github.com/canonical/data-platform-workflows", tag = "v24.0.6", subdirectory = "python/pytest_plugins/github_secrets"}
+pytest-operator = "^0.37.0"
+pytest-operator-cache = {git = "https://github.com/canonical/data-platform-workflows", tag = "v24.0.6", subdirectory = "python/pytest_plugins/pytest_operator_cache"}
+pytest-operator-groups = {git = "https://github.com/canonical/data-platform-workflows", tag = "v24.0.6", subdirectory = "python/pytest_plugins/pytest_operator_groups"}
+pytest-microceph = {git = "https://github.com/canonical/data-platform-workflows", tag = "v24.0.6", subdirectory = "python/pytest_plugins/microceph"}
+juju = "^3.5.2"
+ops = "^2.17.0"
+tenacity = "^9.0.0"
+pyyaml = "^6.0.2"
+urllib3 = "^2.2.3"
+protobuf = "5.28.2"
+
+[tool.coverage.run]
+branch = true
+
+[tool.coverage.report]
+show_missing = true
+
+[tool.pytest.ini_options]
+minversion = "6.0"
+log_cli_level = "INFO"
+markers = ["unstable"]
+asyncio_mode = "auto"
+
+# Formatting tools configuration
+[tool.black]
+line-length = 99
+target-version = ["py310"]
+
+# Linting tools configuration
+[tool.ruff]
+# preview and explicit preview are enabled for CPY001
+preview = true
+target-version = "py310"
+src = ["src", "."]
+line-length = 99
+
+[tool.ruff.lint]
+explicit-preview-rules = true
+select = ["A", "E", "W", "F", "C", "N", "D", "I001", "CPY001"]
+extend-ignore = [
+    "D203",
+    "D204",
+    "D213",
+    "D215",
+    "D400",
+    "D404",
+    "D406",
+    "D407",
+    "D408",
+    "D409",
+    "D413",
+]
+# Ignore E501 because using black creates errors with this
+# Ignore D107 Missing docstring in __init__
+ignore = ["E501", "D107"]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/*" = ["D100", "D101", "D102", "D103", "D104"]
+
+[tool.ruff.lint.flake8-copyright]
+# Check for properly formatted copyright header in each file
+author = "Canonical Ltd."
+notice-rgx = "Copyright\\s\\d{4}([-,]\\d{4})*\\s+"
+min-file-size = 1
+
+[tool.ruff.lint.mccabe]
+max-complexity = 10
+
+[tool.ruff.lint.pydocstyle]
+convention = "google"

pyproject.toml

@@ -0,0 +1,6 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""This module contains the constants and models used by the sysbench charm."""
+
+BENCHMARK_WORKLOAD_PATH = "/root/.benchmark/charmed_parameters"

src/benchmark/literals.py

@@ -60,8 +60,9 @@ class WorkloadCLIArgsModel(BaseModel):
     run_count: int
     report_interval: int
     extra_labels: str
-    log_file: str = "/var/log/dpe_benchmark_workload.log"
     peers: str
+    log_file: str = "/var/log/dpe_benchmark_workload.log"
+    is_coordinator: bool
 
 
 class BenchmarkMetrics:
@@ -88,7 +89,6 @@ def add(self, sample: BaseModel):
             )
 
 
-
 class KafkaBenchmarkSample(BaseModel):
     """Sample from the benchmark tool."""
 
@@ -130,4 +130,4 @@ class KafkaBenchmarkSampleMatcher(Enum):
 
     consume_rate: str = r"Cons rate\s+(.*?)\s+msg/s"
     consume_throughput: str = r"Cons rate\s+\d+.\d+\s+msg/s\s+/\s+(.*?)\s+MB/s"
-    consume_backlog: str = r"Backlog:\s+(.*?)\s+K"
+    consume_backlog: str = r"Backlog:\s+(.*?)\s+K"

src/benchmark/wrapper/core.py

@@ -24,16 +24,14 @@ def run(self):
         """Prepares the workload and runs the benchmark."""
         manager, _ = self.mapping.map(self.args.command)
 
-        logging.basicConfig(
-            filename=self.args.log_file, encoding="utf-8", level=logging.INFO
-        )
+        logging.basicConfig(filename=self.args.log_file, encoding="utf-8", level=logging.INFO)
 
         def _exit(*args, **kwargs):
             manager.stop()
 
         signal.signal(signal.SIGINT, _exit)
         signal.signal(signal.SIGTERM, _exit)
-        start_http_server(8088)
+        start_http_server(8008)
 
         # Start the manager and process the output
         manager.start()
@@ -62,14 +60,14 @@ def _exit(*args, **kwargs):
 #         "--target_hosts", type=str, default="", help="comma-separated list of target hosts"
 #     )
 #     parser.add_argument(
+#         "--log_file", type=str, default="/var/log/dpe_benchmark_workload.log", help="Log file for all threads"
+#     )
+#     parser.add_argument(
 #         "--extra_labels",
 #         type=str,
 #         help="comma-separated list of extra labels to be used.",
 #         default="",
 #     )
-#     parser.add_argument(
-#         "--log_file", type=str, default="/var/log/dpe_benchmark_workload.log", help="Log file for all threads"
-#     )
 #     # Parse the arguments as dictionary, using the same logic as:
 #     # https://github.com/python/cpython/blob/ \
 #     #     47c5a0f307cff3ed477528536e8de095c0752efa/Lib/argparse.py#L134

src/benchmark/wrapper/main.py

@@ -27,6 +27,9 @@
 
 
 logger = logging.getLogger(__name__)
+logging.basicConfig(
+    filename="/var/log/dpe_benchmark_workload.log", encoding="utf-8", level=logging.INFO
+)
 
 
 class BenchmarkProcess:
@@ -42,7 +45,7 @@ class BenchmarkProcess:
 
     def __init__(
         self,
-        model: ProcessModel,
+        model: ProcessModel | None,
         args: WorkloadCLIArgsModel,
         metrics: BenchmarkMetrics,
     ):
@@ -53,6 +56,8 @@ def __init__(
 
     def start(self):
         """Start the process."""
+        if not self.model:
+            return
         self._proc = subprocess.Popen(
             self.model.cmd,
             user=self.model.user,
@@ -73,6 +78,10 @@ def start(self):
     def status(self) -> ProcessStatus:
         """Return the status of the process."""
         stat = ProcessStatus.STOPPED
+        if not self._proc:
+            # We are managing only, we do not run a process
+            return ProcessStatus.RUNNING
+
         if self._proc.poll() is None:
             stat = ProcessStatus.RUNNING
         elif self._proc.returncode != 0:
@@ -95,20 +104,21 @@ async def process(
             or (self.status() == ProcessStatus.RUNNING and self.args.duration == 0)
         ):
             to_wait = True
-            for line in iter(self._proc.stdout.readline, ""):
-                if output := self.process_line(line):
-                    self.metrics.add(output)
+            if self._proc:
+                for line in iter(self._proc.stdout.readline, ""):
+                    if output := self.process_line(line):
+                        self.metrics.add(output)
 
-                    if self.status() != ProcessStatus.RUNNING:
-                        # Process has finished
-                        break
+                        if self.status() != ProcessStatus.RUNNING:
+                            # Process has finished
+                            break
 
-                    to_wait = False
+                        to_wait = False
 
-                # Log the output.
-                # This way, an user can see what the process is doing and
-                # some of the metrics will be readily available without COS.
-                logger.info(f"[workload pid {self._proc.pid}] " + line.rstrip())
+                    # Log the output.
+                    # This way, an user can see what the process is doing and
+                    # some of the metrics will be readily available without COS.
+                    logger.info(f"[workload pid {self._proc.pid}] " + line.rstrip())
 
             if to_wait:
                 # In case the stdout is empty, we ensure we sleep anyways
@@ -126,7 +136,8 @@ async def process(
     def stop(self):
         """Stop the process."""
         try:
-            self._proc.kill()
+            if self._proc:
+                self._proc.kill()
         except Exception as e:
             logger.warning(f"Error stopping worker: {e}")
         self.model.status = ProcessStatus.STOPPED
@@ -214,16 +225,16 @@ def map(self, cmd: BenchmarkCommand) -> tuple[BenchmarkManager, list[BenchmarkPr
         raise ValueError(f"Invalid command: {cmd}")
 
     @abstractmethod
-    def _map_prepare(self) -> tuple[BenchmarkManager, list[BenchmarkProcess] | None]:
+    def _map_prepare(self) -> tuple[BenchmarkManager | None, list[BenchmarkProcess] | None]:
         """Returns the mapping for the prepare phase."""
         ...
 
     @abstractmethod
-    def _map_run(self) -> tuple[BenchmarkManager, list[BenchmarkProcess] | None]:
+    def _map_run(self) -> tuple[BenchmarkManager | None, list[BenchmarkProcess] | None]:
         """Returns the mapping for the run phase."""
         ...
 
     @abstractmethod
-    def _map_clean(self) -> tuple[BenchmarkManager, list[BenchmarkProcess] | None]:
+    def _map_clean(self) -> tuple[BenchmarkManager | None, list[BenchmarkProcess] | None]:
         """Returns the mapping for the clean phase."""
         ...