Merge feat/core-logic into dev

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]

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: <metric>_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

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)

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()

tests/test_converter.py

@@ -0,0 +1,68 @@
+"""Tests for the unit conversion module."""
+
+import numpy as np
+import pytest
+
+from disk_checker.core.converter import to_unit, UNIT_FACTORS, UNIT_LABELS
+
+
+class TestToUnit:
+    """Tests for to_unit function."""
+
+    def test_nanoseconds_to_microseconds(self):
+        """Test conversion from ns to us."""
+        values = np.array([1000, 2000, 3000])
+        result = to_unit(values, "us")
+        expected = np.array([1.0, 2.0, 3.0])
+        np.testing.assert_array_equal(result, expected)
+
+    def test_nanoseconds_to_milliseconds(self):
+        """Test conversion from ns to ms."""
+        values = np.array([1000000, 2000000])
+        result = to_unit(values, "ms")
+        expected = np.array([1.0, 2.0])
+        np.testing.assert_array_equal(result, expected)
+
+    def test_nanoseconds_to_seconds(self):
+        """Test conversion from ns to s."""
+        values = np.array([1000000000, 2000000000])
+        result = to_unit(values, "s")
+        expected = np.array([1.0, 2.0])
+        np.testing.assert_array_equal(result, expected)
+
+    def test_same_unit_returns_same(self):
+        """Test conversion to same unit."""
+        values = np.array([100, 200, 300])
+        result = to_unit(values, "ns")
+        np.testing.assert_array_equal(result, values)
+
+    def test_empty_array(self):
+        """Test conversion with empty array."""
+        values = np.array([])
+        result = to_unit(values, "us")
+        assert len(result) == 0
+
+    def test_single_value(self):
+        """Test conversion with single value."""
+        result = to_unit(1000, "us")
+        assert result == 1.0
+
+
+class TestConstants:
+    """Tests for module constants."""
+
+    def test_unit_factors(self):
+        """Test UNIT_FACTORS contains all expected units."""
+        assert set(UNIT_FACTORS.keys()) == {"ns", "us", "ms", "s"}
+        assert UNIT_FACTORS["ns"] == 1.0
+        assert UNIT_FACTORS["us"] == 1e3
+        assert UNIT_FACTORS["ms"] == 1e6
+        assert UNIT_FACTORS["s"] == 1e9
+
+    def test_unit_labels(self):
+        """Test UNIT_LABELS contains all expected units."""
+        assert set(UNIT_LABELS.keys()) == {"ns", "us", "ms", "s"}
+        assert UNIT_LABELS["ns"] == "ns"
+        assert UNIT_LABELS["us"] == "µs"
+        assert UNIT_LABELS["ms"] == "ms"
+        assert UNIT_LABELS["s"] == "s"

tests/test_extractor.py

@@ -0,0 +1,72 @@
+"""Tests for the data extraction module."""
+
+import pytest
+import pandas as pd
+from pathlib import Path
+
+from disk_checker.core.extractor import load_latency_csv
+
+
+class TestLoadLatencyCsv:
+    """Tests for load_latency_csv function."""
+
+    def test_load_valid_csv(self, sample_csv_path):
+        """Test loading a valid fio CSV file."""
+        df = load_latency_csv(str(sample_csv_path))
+        assert not df.empty
+        assert "metric" in df.columns
+        assert "latency_ns" in df.columns
+        assert "count" in df.columns
+
+    def test_returns_long_format(self, sample_csv_path):
+        """Test that output is in long format with metric column."""
+        df = load_latency_csv(str(sample_csv_path))
+        metrics = df["metric"].unique()
+        assert "read_clat" in metrics
+
+    def test_empty_csv_raises(self, empty_csv_path):
+        """Test that empty CSV raises ValueError."""
+        with pytest.raises(ValueError, match="no metric had any numeric count data"):
+            load_latency_csv(str(empty_csv_path))
+
+    def test_invalid_format_raises(self, invalid_csv_path):
+        """Test that invalid CSV format raises ValueError."""
+        with pytest.raises(ValueError):
+            load_latency_csv(str(invalid_csv_path))
+
+    def test_missing_nsec_column_raises(self, tmp_path):
+        """Test that missing nsec column raises ValueError."""
+        csv_content = "col1,col2\n1,2\n"
+        path = tmp_path / "invalid.csv"
+        path.write_text(csv_content)
+        with pytest.raises(ValueError, match="expected first column to be 'nsec'"):
+            load_latency_csv(str(path))
+
+
+@pytest.fixture
+def sample_csv_path(tmp_path):
+    """Create a sample CSV file for testing."""
+    csv_content = """nsec,read_clat_ns_count,read_clat_ns_cumulative,read_clat_ns_percentile
+19791872,1,1,0.00016775708773695687
+22413312,4,5,0.0008387854386847844
+22675456,9,14,0.0023485992283173965
+"""
+    path = tmp_path / "test.csv"
+    path.write_text(csv_content)
+    return path
+
+
+@pytest.fixture
+def empty_csv_path(tmp_path):
+    """Create an empty CSV file for testing."""
+    path = tmp_path / "empty.csv"
+    path.write_text("nsec\n")
+    return path
+
+
+@pytest.fixture
+def invalid_csv_path(tmp_path):
+    """Create an invalid CSV file for testing."""
+    path = tmp_path / "invalid.csv"
+    path.write_text("col1,col2\n1,2\n")
+    return path

tests/test_helpers.py

@@ -0,0 +1,75 @@
+"""Tests for helper functions."""
+
+import pandas as pd
+import pytest
+
+from disk_checker.core.helpers import human_count, percentile_table
+
+
+class TestHumanCount:
+    """Tests for human_count function."""
+
+    def test_small_number(self):
+        """Test formatting of small numbers."""
+        assert human_count(100) == "100"
+        assert human_count(50) == "50"
+
+    def test_thousands(self):
+        """Test formatting of thousands."""
+        assert human_count(1500) == "1k"
+        assert human_count(2500) == "3k"
+
+    def test_millions(self):
+        """Test formatting of millions."""
+        assert human_count(1500000) == "1.5M"
+        assert human_count(2750000) == "2.8M"
+
+    def test_billions(self):
+        """Test formatting of billions."""
+        assert human_count(1500000000) == "1.5G"
+        assert human_count(2750000000) == "2.8G"
+
+    def test_zero(self):
+        """Test formatting of zero."""
+        assert human_count(0) == "0"
+
+    def test_position_argument_ignored(self):
+        """Test that _pos argument is ignored."""
+        assert human_count(1000, 0) == "1k"
+
+
+class TestPercentileTable:
+    """Tests for percentile_table function."""
+
+    def test_basic_percentiles(self):
+        """Test basic percentile table generation."""
+        df = pd.DataFrame({
+            "metric": ["read_clat"] * 3,
+            "latency_ns": [1000, 2000, 3000],
+            "count": [1, 2, 1],
+        })
+        result = percentile_table(df, "read_clat", "us", [50, 90])
+        assert "Metric: read_clat" in result
+        assert "total I/Os: 4" in result
+        assert "percentile" in result
+        assert "latency" in result
+
+    def test_missing_metric(self):
+        """Test percentile table with missing metric."""
+        df = pd.DataFrame({
+            "metric": ["read_clat"],
+            "latency_ns": [1000],
+            "count": [1],
+        })
+        result = percentile_table(df, "write_clat", "us", [50])
+        assert "(no data for metric write_clat)" in result
+
+    def test_unit_display(self):
+        """Test that correct unit is displayed."""
+        df = pd.DataFrame({
+            "metric": ["read_clat"],
+            "latency_ns": [1000000],  # 1ms in ns
+            "count": [1],
+        })
+        result = percentile_table(df, "read_clat", "ms", [50])
+        assert "ms" in result

tests/test_orchestrator.py

@@ -0,0 +1,51 @@
+"""Tests for the orchestrator module."""
+
+import pytest
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+from disk_checker.core.orchestrator import Orchestrator, ToolStatus, JobResult
+
+
+class TestOrchestrator:
+    """Tests for Orchestrator class."""
+
+    @pytest.fixture
+    def orchestrator(self, tmp_path):
+        """Create an orchestrator instance."""
+        return Orchestrator(work_dir=tmp_path)
+
+    def test_initial_state(self, orchestrator):
+        """Test initial orchestrator state."""
+        assert orchestrator.list_jobs() == {}
+
+    def test_run_fio_creates_job(self, orchestrator, tmp_path):
+        """Test that run_fio creates a job entry."""
+        with patch("asyncio.create_subprocess_exec") as mock_exec:
+            mock_exec.return_value = AsyncMock()
+            asyncio.get_event_loop().run_until_complete(
+                orchestrator.run_fio("/dev/sda")
+            )
+
+        jobs = orchestrator.list_jobs()
+        assert len(jobs) == 1
+        job = list(jobs.values())[0]
+        assert job.status == ToolStatus.COMPLETED
+        assert job.output_path.exists()
+
+    def test_get_job_status(self, orchestrator, tmp_path):
+        """Test retrieving job status."""
+        with patch("asyncio.create_subprocess_exec") as mock_exec:
+            mock_exec.return_value = AsyncMock()
+            job_id = asyncio.get_event_loop().run_until_complete(
+                orchestrator.run_fio("/dev/sda")
+            )
+
+        job = orchestrator.get_job_status(job_id)
+        assert job is not None
+        assert job.status == ToolStatus.COMPLETED
+
+    def test_get_nonexistent_job(self, orchestrator):
+        """Test getting status of non-existent job."""
+        job = orchestrator.get_job_status("nonexistent")
+        assert job is None