feat(core): extract data extraction, conversion, helpers, and orchestrator
This commit is contained in:
@@ -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]
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user