From 29c02c27fbe2c3b81c6fcf3b6936dea809bf578e Mon Sep 17 00:00:00 2001 From: Maxim Vershinin Date: Mon, 28 Sep 2026 19:04:05 +0400 Subject: [PATCH] feat(core): extract data extraction, conversion, helpers, and orchestrator --- src/disk_checker/core/converter.py | 32 +++++ src/disk_checker/core/extractor.py | 88 ++++++++++++++ src/disk_checker/core/helpers.py | 62 ++++++++++ src/disk_checker/core/orchestrator.py | 164 ++++++++++++++++++++++++++ 4 files changed, 346 insertions(+) create mode 100644 src/disk_checker/core/converter.py create mode 100644 src/disk_checker/core/extractor.py create mode 100644 src/disk_checker/core/helpers.py create mode 100644 src/disk_checker/core/orchestrator.py diff --git a/src/disk_checker/core/converter.py b/src/disk_checker/core/converter.py new file mode 100644 index 0000000..215e925 --- /dev/null +++ b/src/disk_checker/core/converter.py @@ -0,0 +1,32 @@ +"""Unit conversion utilities for latency data.""" + +from typing import Dict +import numpy as np + +UNIT_FACTORS: Dict[str, float] = { + "ns": 1.0, + "us": 1e3, + "ms": 1e6, + "s": 1e9, +} + +UNIT_LABELS: Dict[str, str] = { + "ns": "ns", + "us": "µs", + "ms": "ms", + "s": "s", +} + + +def to_unit(values_ns, unit: str) -> np.ndarray: + """ + Convert nanosecond values to the specified unit. + + Args: + values_ns: Values in nanoseconds + unit: Target unit (ns, us, ms, s) + + Returns: + Converted values as numpy array + """ + return np.asarray(values_ns, dtype=float) / UNIT_FACTORS[unit] \ No newline at end of file diff --git a/src/disk_checker/core/extractor.py b/src/disk_checker/core/extractor.py new file mode 100644 index 0000000..4b80e50 --- /dev/null +++ b/src/disk_checker/core/extractor.py @@ -0,0 +1,88 @@ +"""CSV data extraction module for fio latency data.""" + +from pathlib import Path +import pandas as pd +from typing import Optional + + +def load_latency_csv(path: str) -> pd.DataFrame: + """ + Parse wide-format fio JSON+ CSV into long-format DataFrame. + + Transforms wide-format columns (e.g., read_clat_ns_count) into rows with + a metric identifier, making it easier to group and analyze by operation type. + + Args: + path: Path to CSV file generated by fio_jsonplus_clat2csv + + Returns: + DataFrame with columns: metric, latency_ns, count, cumulative, percentile + + Raises: + ValueError: If CSV format is invalid or contains no valid data + """ + df = pd.read_csv(path, skipinitialspace=True) + df.columns = [str(c).strip() for c in df.columns] + + # Drop trailing unnamed / empty columns + df = df.loc[:, ~df.columns.str.match(r"^Unnamed")] + df = df.loc[:, df.columns != ""] + + nsec_col = df.columns[0] + if "nsec" not in nsec_col.lower(): + raise ValueError( + f"expected first column to be 'nsec', got {nsec_col!r}" + ) + + # Coerce the nsec column to numeric + df[nsec_col] = pd.to_numeric(df[nsec_col], errors="coerce") + df = df.dropna(subset=[nsec_col]) + + # Discover metric triples: _count / _cumulative / _percentile + metrics = {} + for col in df.columns: + if col == nsec_col: + continue + for suffix in ("_count", "_cumulative", "_percentile"): + if col.endswith(suffix): + base = col[: -len(suffix)] + metrics.setdefault(base, {})[suffix[1:]] = col + + if not metrics: + raise ValueError("no *_count columns found in CSV") + + frames = [] + for base, cols in metrics.items(): + count_col = cols["count"] + + # Coerce count to numeric + counts = pd.to_numeric(df[count_col], errors="coerce") + + # Keep only rows where the count column actually has a number + mask = counts.notna() + if not mask.any(): + continue + + sub = pd.DataFrame({ + "latency_ns": df.loc[mask, nsec_col].values, + "count": counts[mask].astype(int).values, + }) + if "cumulative" in cols: + sub["cumulative"] = pd.to_numeric( + df.loc[mask, cols["cumulative"]], errors="coerce" + ).values + if "percentile" in cols: + sub["percentile"] = pd.to_numeric( + df.loc[mask, cols["percentile"]], errors="coerce" + ).values + + # read_clat_ns -> read_clat + sub["metric"] = base.replace("_ns", "") + frames.append(sub) + + if not frames: + raise ValueError("no metric had any numeric count data") + + out = pd.concat(frames, ignore_index=True) + out = out[out["count"] > 0].reset_index(drop=True) + return out \ No newline at end of file diff --git a/src/disk_checker/core/helpers.py b/src/disk_checker/core/helpers.py new file mode 100644 index 0000000..20529ae --- /dev/null +++ b/src/disk_checker/core/helpers.py @@ -0,0 +1,62 @@ +"""Helper functions for formatting and display.""" + +from typing import List +import numpy as np +import pandas as pd +from matplotlib.ticker import FuncFormatter + +from .converter import UNIT_LABELS + + +def human_count(x, _pos=None) -> str: + """ + Format large numbers with K/M/G suffixes for axis labels. + + Args: + x: Numeric value to format + _pos: Matplotlib tick position (unused) + + Returns: + Formatted string (e.g., "1.5M", "3.2G") + """ + if x >= 1e9: + return f"{x/1e9:.1f}G" + if x >= 1e6: + return f"{x/1e6:.1f}M" + if x >= 1e3: + return f"{x/1e3:.0f}k" + return f"{x:.0f}" + + +def percentile_table(df: pd.DataFrame, metric: str, unit: str, pcts: List[float]) -> str: + """ + Generate a formatted table showing latency values at specified percentiles. + + Args: + df: Long-format DataFrame from load_latency_csv + metric: Metric name (e.g., 'read_clat') + unit: Display unit (ns, us, ms, s) + pcts: List of percentile values (e.g., [50, 90, 99]) + + Returns: + Formatted string table + """ + g = df[df["metric"] == metric].sort_values("latency_ns") + if g.empty: + return f"(no data for metric {metric})" + + counts = g["count"].values.astype(float) + total = int(counts.sum()) + cdf = np.cumsum(counts) / total * 100.0 + lat = to_unit(g["latency_ns"].values, unit) + + lines = [ + f"\nMetric: {metric} (total I/Os: {total:,})", + f" {'percentile':>12} {'latency':>14}", + ] + + for p in pcts: + idx = min(np.searchsorted(cdf, p), len(lat) - 1) + lines.append(f" {p:>11.2f}% {lat[idx]:>12.3f} {UNIT_LABELS[unit]}") + + return "\n".join(lines) \ No newline at end of file diff --git a/src/disk_checker/core/orchestrator.py b/src/disk_checker/core/orchestrator.py new file mode 100644 index 0000000..656385b --- /dev/null +++ b/src/disk_checker/core/orchestrator.py @@ -0,0 +1,164 @@ +"""Tool orchestration module for running disk benchmarks.""" + +import asyncio +import subprocess +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Dict, Optional + + +class ToolStatus(Enum): + """Status of a benchmark job.""" + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + + +@dataclass +class JobResult: + """Result of a benchmark job.""" + job_id: str + status: ToolStatus + output_path: Optional[Path] = None + error: Optional[str] = None + metadata: Dict = field(default_factory=dict) + + +class Orchestrator: + """ + Orchestrates disk benchmarking tools (fio, badblocks, etc.) + + Provides async interfaces for running benchmarks and collecting results. + """ + + def __init__(self, work_dir: Path = Path(".")): + self.work_dir = work_dir + self._jobs: Dict[str, JobResult] = {} + + async def run_fio( + self, + device: str, + job_file: Optional[Path] = None, + output_format: str = "jsonplus", + **kwargs, + ) -> Path: + """ + Run fio benchmark and return path to output file. + + Args: + device: Block device path (e.g., '/dev/sda') + job_file: Optional fio job file + output_format: Output format (json, jsonplus, csv) + **kwargs: Additional fio options + + Returns: + Path to output file + + Raises: + subprocess.CalledProcessError: If fio command fails + """ + job_id = f"fio_{asyncio.get_event_loop().time()}" + output_path = self.work_dir / f"{job_id}.{output_format}" + + cmd = ["fio", "--output-format=jsonplus"] + if job_file: + cmd.extend(["--filename", str(device)]) + else: + cmd.extend(["--name=test", "--rw=randread", "--bs=4k"]) + + cmd.extend(["--output", str(output_path)]) + cmd.extend([f"--output-format={output_format}"]) + + for k, v in kwargs.items(): + cmd.extend([f"--{k}={v}"]) + + self._jobs[job_id] = JobResult( + job_id=job_id, + status=ToolStatus.RUNNING, + output_path=output_path, + ) + + try: + await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + self._jobs[job_id] = JobResult( + job_id=job_id, + status=ToolStatus.COMPLETED, + output_path=output_path, + ) + return output_path + except subprocess.CalledProcessError as e: + self._jobs[job_id] = JobResult( + job_id=job_id, + status=ToolStatus.FAILED, + error=str(e), + ) + raise + + async def run_badblocks( + self, + device: str, + blocks_count: Optional[int] = None, + **kwargs, + ) -> Path: + """ + Run badblocks scan on device. + + Args: + device: Block device path + blocks_count: Number of blocks to scan + **kwargs: Additional badblocks options + + Returns: + Path to badblocks output file + """ + job_id = f"badblocks_{asyncio.get_event_loop().time()}" + output_path = self.work_dir / f"{job_id}.txt" + + cmd = ["badblocks", "-w", "-s", "-p", "0", "-t", "RANDOM", "-o"] + cmd.extend([str(output_path), str(device)]) + + if blocks_count: + cmd.extend(["-b", str(blocks_count)]) + + for k, v in kwargs.items(): + cmd.extend([f"-{k}", str(v)]) + + self._jobs[job_id] = JobResult( + job_id=job_id, + status=ToolStatus.RUNNING, + output_path=output_path, + ) + + try: + await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + self._jobs[job_id] = JobResult( + job_id=job_id, + status=ToolStatus.COMPLETED, + output_path=output_path, + ) + return output_path + except subprocess.CalledProcessError as e: + self._jobs[job_id] = JobResult( + job_id=job_id, + status=ToolStatus.FAILED, + error=str(e), + ) + raise + + def get_job_status(self, job_id: str) -> Optional[JobResult]: + """Get status of a submitted job.""" + return self._jobs.get(job_id) + + def list_jobs(self) -> Dict[str, JobResult]: + """List all jobs.""" + return self._jobs.copy() \ No newline at end of file