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 diff --git a/tests/test_converter.py b/tests/test_converter.py new file mode 100644 index 0000000..f9ac54f --- /dev/null +++ b/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" \ No newline at end of file diff --git a/tests/test_extractor.py b/tests/test_extractor.py new file mode 100644 index 0000000..3a6a351 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..69fe405 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py new file mode 100644 index 0000000..7f45682 --- /dev/null +++ b/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 \ No newline at end of file