fix: add undo disk selection capability #2

Closed
bot wants to merge 26 commits from fio_workflow into main
4 changed files with 90 additions and 210 deletions
Showing only changes of commit e0c0da6def - Show all commits
+57
View File
@@ -0,0 +1,57 @@
"""Tool detection module for disk-checker."""
import shutil
from typing import Dict, List
from dataclasses import dataclass
@dataclass
class ToolInfo:
"""Information about a benchmarking tool."""
name: str
available: bool
version: str = ""
# Tools to scan for
TOOLS_TO_SCAN = [
"fio",
"badblocks",
"smartctl",
"iozone",
"hdparm",
]
def scan_tools() -> Dict[str, ToolInfo]:
"""
Scan host system for available benchmarking tools.
Returns:
Dictionary mapping tool names to ToolInfo objects
"""
results = {}
for tool_name in TOOLS_TO_SCAN:
available = shutil.which(tool_name) is not None
version = ""
if available:
# Try to get version (optional enhancement)
try:
import subprocess
result = subprocess.run(
[tool_name, "--version"],
capture_output=True,
text=True,
timeout=2
)
version = result.stdout.split("\n")[0][:50]
except Exception:
# If we can't even run '--version', then
# a tool is not considered as available
available = False
results[tool_name] = ToolInfo(
name=tool_name,
available=available,
version=version
)
return results
-43
View File
@@ -20,38 +20,6 @@ class JobStatus(str, Enum):
FAILED = "failed"
class MetricInfo(BaseModel):
"""Information about a metric."""
name: str
total_ios: int
min_latency_ns: Optional[float] = None
max_latency_ns: Optional[float] = None
avg_latency_ns: Optional[float] = None
class PercentileInfo(BaseModel):
"""Latency value at a specific percentile."""
percentile: float
latency_ns: float
latency_display: str
class MetricResult(BaseModel):
"""Complete result for a metric."""
metric: str
total_ios: int
percentiles: List[PercentileInfo]
min_latency_ns: float
max_latency_ns: float
avg_latency_ns: float
class ScanResponse(BaseModel):
"""Response from CSV scan endpoint."""
metrics: List[MetricResult]
summary: Dict[str, Any]
class BenchmarkRequest(BaseModel):
"""Request to start a benchmark job."""
device: str = Field(..., description="Block device path (e.g., /dev/sda)")
@@ -82,14 +50,3 @@ class JobStatusResponse(BaseModel):
error: Optional[str] = None
created_at: Optional[str] = None
completed_at: Optional[str] = None
class PlotRequest(BaseModel):
"""Request to generate a plot."""
metric: Optional[str] = None
unit: str = Field(default="us", pattern="^(ns|us|ms|s)$")
log_x: bool = False
log_y: bool = False
cdf: bool = False
normalize: bool = False
title: Optional[str] = None
+21 -164
View File
@@ -3,7 +3,7 @@
import asyncio
import tempfile
from pathlib import Path
from typing import List, Optional
from typing import List, Optional, Dict
from fastapi import APIRouter, UploadFile, File, HTTPException, Depends, Request
from fastapi.responses import JSONResponse, FileResponse
@@ -11,177 +11,18 @@ import pandas as pd
from disk_checker.core.extractor import load_latency_csv
from disk_checker.core.helpers import percentile_table
from disk_checker.core.tool_scanner import ToolInfo
from disk_checker.plotting.plots import generate_plot
from disk_checker.web.models import (
ScanResponse,
BenchmarkRequest,
JobStatusResponse,
PlotRequest,
MetricInfo,
PercentileInfo,
MetricResult,
)
router = APIRouter()
def _calculate_metric_results(df: pd.DataFrame, metric: str, pcts: List[float]) -> MetricResult:
"""Calculate detailed results for a metric."""
metric_df = df[df["metric"] == metric]
counts = metric_df["count"].values.astype(float)
total_ios = int(counts.sum())
latencies_ns = metric_df["latency_ns"].values
# Calculate percentiles
percentiles = []
cdf = __import__('numpy').cumsum(counts) / total_ios * 100.0
for p in pcts:
idx = min(__import__('numpy').searchsorted(cdf, p), len(latencies_ns) - 1)
percentiles.append(PercentileInfo(
percentile=p,
latency_ns=float(latencies_ns[idx]),
latency_display=f"{latencies_ns[idx]:.3f} ns"
))
return MetricResult(
metric=metric,
total_ios=total_ios,
percentiles=percentiles,
min_latency_ns=float(latencies_ns.min()),
max_latency_ns=float(latencies_ns.max()),
avg_latency_ns=float(latencies_ns.mean()),
)
@router.post("/scan", response_model=ScanResponse)
async def scan_csv(file: UploadFile = File(...)):
"""
Upload a fio JSON+ CSV and return analysis.
"""
with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as tmp:
content = await file.read()
tmp.write(content)
tmp_path = tmp.name
try:
df = load_latency_csv(tmp_path)
if df.empty:
raise HTTPException(status_code=400, detail="No valid data in CSV")
# Calculate metrics
metrics = []
pcts = [50, 90, 99, 99.9, 99.99]
for metric in sorted(df["metric"].unique()):
metrics.append(_calculate_metric_results(df, metric, pcts))
summary = {
"total_metrics": len(metrics),
"total_ios": sum(m.total_ios for m in metrics),
}
return ScanResponse(metrics=metrics, summary=summary)
finally:
Path(tmp_path).unlink(missing_ok=True)
@router.post("/plot")
async def generate_plot_endpoint(
file: UploadFile = File(...),
metric: Optional[str] = None,
unit: str = "us",
log_x: bool = False,
log_y: bool = False,
cdf: bool = False,
normalize: bool = False,
title: Optional[str] = None,
):
"""
Generate and return a latency plot.
"""
with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as tmp:
content = await file.read()
tmp.write(content)
tmp_path = tmp.name
try:
df = load_latency_csv(tmp_path)
if df.empty:
raise HTTPException(status_code=400, detail="No valid data in CSV")
# Determine metrics
available = sorted(df["metric"].unique())
if metric:
if metric not in available:
raise HTTPException(
status_code=400,
detail=f"Metric {metric} not found"
)
metrics = [metric]
else:
totals = df.groupby("metric")["count"].sum().sort_values(ascending=False)
metrics = [totals.index[0]]
# Generate plot
with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as img:
output_path = Path(img.name)
generate_plot(
df=df,
metrics=metrics,
output_path=output_path,
unit=unit,
log_x=log_x,
log_y=log_y,
normalize=normalize,
show_cdf=cdf,
title=title or "",
)
return FileResponse(output_path, media_type="image/png")
finally:
Path(tmp_path).unlink(missing_ok=True)
@router.post("/benchmark", response_model=JobStatusResponse)
async def start_benchmark(request: BenchmarkRequest):
"""
Start a disk benchmark job.
"""
orchestrator = request.app.state.orchestrator
job_id = f"bench_{asyncio.get_event_loop().time()}"
async def run_job():
try:
if request.tool == "fio":
output_path = await orchestrator.run_fio(
device=request.device,
**request.options,
)
elif request.tool == "badblocks":
output_path = await orchestrator.run_badblocks(
device=request.device,
**request.options,
)
else:
raise ValueError(f"Unknown tool: {request.tool}")
return {
"job_id": job_id,
"status": "completed",
"output_path": str(output_path),
}
except Exception as e:
return {"job_id": job_id, "status": "failed", "error": str(e)}
request.app.state.jobs[job_id] = {
"status": "running",
"task": asyncio.create_task(run_job()),
}
return JobStatusResponse(job_id=job_id, status="running")
@router.get("/jobs/{job_id}", response_model=JobStatusResponse)
async def get_job_status(job_id: str, request: Request):
"""
@@ -207,3 +48,19 @@ async def list_jobs(request: Request):
List all benchmark jobs.
"""
return list(request.app.state.jobs.keys())
@router.get("/tools")
async def get_available_tools(request: Request):
"""
Get list of available benchmarking tools.
"""
tools = request.app.state.available_tools
return {
tool_name: {
"name" : tool.name,
"available": tool.available,
"version" : tool.version
}
for tool_name, tool in tools.items()
}
+10 -1
View File
@@ -1,11 +1,14 @@
"""FastAPI application for disk-checker web API."""
import asyncio
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import logging
from disk_checker.core.orchestrator import Orchestrator
from disk_checker.core.tool_scanner import scan_tools
logger = logging.getLogger(__name__)
@@ -14,9 +17,15 @@ logger = logging.getLogger(__name__)
async def lifespan(app: FastAPI):
"""Application lifespan handler."""
# Startup: initialize orchestrator
app.state.orchestrator = Orchestrator(work_dir=app.state.work_dir)
#app.state.orchestrator = Orchestrator(work_dir=app.state.work_dir)
app.state.jobs = {}
# Scan for available benchmarking tools
app.state.available_tools = await asyncio.to_thread(scan_tools)
logger.info(f"Available tools: {list(app.state.available_tools.keys())}")
logger.info("Disk Checker API started")
yield
# Shutdown: cleanup
logger.info("Disk Checker API stopped")