584 lines
20 KiB
Python
584 lines
20 KiB
Python
"""Collector: acquires counters and identity, writes to observation store.
|
|
|
|
This module implements the thinnest complete write path:
|
|
- Acquire counters and thermal evidence from smartctl -a -j
|
|
- Acquire controller identity from sysfs
|
|
- Normalize identity exactly once at write time
|
|
- Validate every row against store invariants
|
|
- Stage valid observations privately before derivation
|
|
- Publish the sample and derived evidence in one collection-owned transaction
|
|
|
|
No code path outside the collector interrogates the device.
|
|
"""
|
|
import json
|
|
import sqlite3
|
|
from collections.abc import Callable
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
from .derive import derive_hours_from_interval, find_previous_sample
|
|
from .monitoring_periods import ensure_period_open
|
|
from .store import init_store, get_store_path
|
|
|
|
|
|
class AcquisitionError(Exception):
|
|
"""Raised when acquisition fails - whole run is refused."""
|
|
pass
|
|
|
|
|
|
class InvariantViolationError(Exception):
|
|
"""Raised when a row would violate store invariants - writes nothing."""
|
|
pass
|
|
|
|
|
|
class _PendingDerivationError(Exception):
|
|
"""A valid staged observation failed while deriving dependent evidence."""
|
|
|
|
def __init__(self, cause: Exception):
|
|
super().__init__(str(cause))
|
|
self.cause = cause
|
|
|
|
|
|
class _PendingRecoveryFailure(Exception):
|
|
"""A publication attempt failed after earlier pending rows committed."""
|
|
|
|
def __init__(self, error: Exception, published_count: int):
|
|
derivation_error = isinstance(error, _PendingDerivationError)
|
|
cause = error.cause if derivation_error else error
|
|
super().__init__(str(cause))
|
|
self.cause = cause
|
|
self.published_count = published_count
|
|
self.derivation_error = derivation_error
|
|
|
|
|
|
PENDING_PUBLICATION_LIMIT = 6720
|
|
|
|
|
|
def acquire_from_smartctl(smartctl_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Acquire counters and thermal evidence from smartctl -a -j data.
|
|
|
|
Validates that all required fields are present.
|
|
Raises AcquisitionError on any failure.
|
|
"""
|
|
required_fields = [
|
|
"nvme_smart_health_information_log",
|
|
"user_capacity",
|
|
"model_name",
|
|
"serial_number",
|
|
"firmware_version",
|
|
]
|
|
|
|
for field in required_fields:
|
|
if field not in smartctl_data:
|
|
raise AcquisitionError(f"Missing required field in smartctl data: {field}")
|
|
|
|
log = smartctl_data["nvme_smart_health_information_log"]
|
|
required_log_fields = [
|
|
"data_units_written",
|
|
"data_units_read",
|
|
"percentage_used",
|
|
"power_on_hours",
|
|
"temperature",
|
|
]
|
|
|
|
for field in required_log_fields:
|
|
if field not in log:
|
|
raise AcquisitionError(f"Missing required field in SMART log: {field}")
|
|
|
|
return {
|
|
"model": smartctl_data["model_name"],
|
|
"serial": smartctl_data["serial_number"],
|
|
"firmware_rev": smartctl_data["firmware_version"],
|
|
"capacity_bytes": smartctl_data["user_capacity"]["bytes"],
|
|
"percentage_used": log["percentage_used"],
|
|
"available_spare": log.get("available_spare"),
|
|
"media_errors": log.get("media_errors", 0),
|
|
"power_on_hours": log["power_on_hours"],
|
|
"power_cycles": log.get("power_cycles"),
|
|
"unsafe_shutdowns": log.get("unsafe_shutdowns"),
|
|
"temperature_c": log["temperature"],
|
|
"data_units_written": log["data_units_written"],
|
|
"data_units_read": log["data_units_read"],
|
|
"bytes_written": log["data_units_written"] * 512000,
|
|
"bytes_read": log["data_units_read"] * 512000,
|
|
"critical_warning": log.get("critical_warning", 0),
|
|
}
|
|
|
|
|
|
def acquire_from_sysfs(sysfs_path: Path) -> Dict[str, Any]:
|
|
"""Acquire controller identity from sysfs.
|
|
|
|
Reads identity from:
|
|
- /sys/class/nvme/<ctrl>/subsysnqn (primary)
|
|
- /sys/class/nvme/<ctrl>/model
|
|
- /sys/class/nvme/<ctrl>/serial
|
|
- /sys/class/nvme/<ctrl>/firmware_rev
|
|
- /sys/class/nvme/<ctrl>/transport/ (optional)
|
|
|
|
Raises AcquisitionError on any failure.
|
|
"""
|
|
identity_files = {
|
|
"subnqn": "subsysnqn",
|
|
"mn": "model",
|
|
"sn": "serial",
|
|
"fr": "firmware_rev",
|
|
}
|
|
|
|
identity = {}
|
|
for key, filename in identity_files.items():
|
|
filepath = sysfs_path / filename
|
|
if not filepath.exists():
|
|
raise AcquisitionError(f"Missing sysfs file: {filepath}")
|
|
|
|
try:
|
|
value = filepath.read_text().strip()
|
|
identity[key] = value if value else ""
|
|
except Exception as e:
|
|
raise AcquisitionError(f"Failed to read {filepath}: {e}")
|
|
|
|
# Transport info (optional)
|
|
transport_dir = sysfs_path / "transport"
|
|
if transport_dir.exists():
|
|
try:
|
|
transport_file = transport_dir / "trstring"
|
|
if transport_file.exists():
|
|
identity["transport"] = transport_file.read_text().strip()
|
|
else:
|
|
identity["transport"] = None
|
|
except Exception:
|
|
identity["transport"] = None
|
|
else:
|
|
identity["transport"] = None
|
|
|
|
# vid/ssvid from PCI node (optional, metadata only - never key components)
|
|
# PCI device directory is the sysfs_path itself (the controller dir is a symlink to PCI)
|
|
pci_device = sysfs_path
|
|
for attr, key in [("vendor", "vid"), ("subsystem_vendor", "ssvid")]:
|
|
filepath = pci_device / attr
|
|
if filepath.exists():
|
|
try:
|
|
value = filepath.read_text().strip()
|
|
identity[key] = value if value else None
|
|
except Exception:
|
|
identity[key] = None
|
|
else:
|
|
identity[key] = None
|
|
|
|
return identity
|
|
|
|
|
|
def normalize_identity(identity: Dict[str, Any]) -> str:
|
|
"""Normalize identity exactly once at write time.
|
|
|
|
Rules:
|
|
- Strip trailing spaces and newlines
|
|
- No case folding
|
|
- Empty-after-strip stored blank
|
|
|
|
Returns normalized identity key.
|
|
"""
|
|
# Primary key: normalized kernel-exposed subsystem NQN
|
|
key = identity.get("subnqn", "")
|
|
if key:
|
|
key = key.rstrip()
|
|
return key
|
|
|
|
# Fallback 1: kernel composite (not implemented yet)
|
|
# Fallback 2: model|serial
|
|
mn = identity.get("mn", "").rstrip()
|
|
sn = identity.get("sn", "").rstrip()
|
|
if mn or sn:
|
|
return f"{mn}|{sn}"
|
|
|
|
# All keys blank - degraded identity
|
|
return ""
|
|
|
|
|
|
def compute_identity_degraded(identity: Dict[str, Any]) -> bool:
|
|
"""Check if identity is degraded (all key rungs empty)."""
|
|
key = normalize_identity(identity)
|
|
return key == ""
|
|
|
|
|
|
def validate_sample_invariants(sample: Dict[str, Any], conn: sqlite3.Connection) -> None:
|
|
"""Validate sample against store invariants.
|
|
|
|
Raises InvariantViolationError if any invariant is violated.
|
|
"""
|
|
# TODO: Implement more complex invariants as needed
|
|
# For now, just check basic constraints
|
|
|
|
if sample.get("bytes_written", 0) < 0:
|
|
raise InvariantViolationError("Negative bytes_written")
|
|
|
|
if sample.get("bytes_read", 0) < 0:
|
|
raise InvariantViolationError("Negative bytes_read")
|
|
|
|
|
|
def write_sample(
|
|
sample: Dict[str, Any],
|
|
identity: Dict[str, Any],
|
|
conn: sqlite3.Connection,
|
|
observed_at: datetime,
|
|
) -> Dict[str, Any]:
|
|
"""Write one sample to the observation store.
|
|
|
|
Identity normalization happens exactly once here.
|
|
Returns segment info for the caller. Caller owns the transaction.
|
|
"""
|
|
from .segment import find_current_segment, should_open_new_segment, open_segment
|
|
|
|
# Normalize identity exactly once at write time
|
|
identity_key = normalize_identity(identity)
|
|
identity_degraded = compute_identity_degraded(identity)
|
|
|
|
# Find current segment
|
|
current_segment = find_current_segment(conn)
|
|
|
|
# Determine if we need a new segment
|
|
should_open, reason = should_open_new_segment(
|
|
current_segment, identity_key, sample["bytes_written"], conn
|
|
)
|
|
|
|
# Open new segment if needed
|
|
segment_opened = False
|
|
if should_open:
|
|
open_segment(conn, observed_at, identity, identity_key, identity_degraded)
|
|
segment_opened = True
|
|
|
|
# Get current segment_id for provenance
|
|
current_segment = find_current_segment(conn)
|
|
segment_id = current_segment["id"] if current_segment else None
|
|
|
|
# Insert sample with segment_id
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT INTO samples (
|
|
ts, device, subnqn, sn, mn, fr, capacity_bytes,
|
|
percentage_used, available_spare, media_errors, power_on_hours,
|
|
power_cycles, unsafe_shutdowns, temperature_c,
|
|
data_units_written, data_units_read, bytes_written, bytes_read,
|
|
critical_warning, segment_id, local_tz
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
sample["ts"],
|
|
sample["device"],
|
|
identity.get("subnqn", ""),
|
|
identity.get("sn", ""),
|
|
identity.get("mn", ""),
|
|
identity.get("fr", ""),
|
|
sample["capacity_bytes"],
|
|
sample["percentage_used"],
|
|
sample["available_spare"],
|
|
sample["media_errors"],
|
|
sample["power_on_hours"],
|
|
sample["power_cycles"],
|
|
sample["unsafe_shutdowns"],
|
|
sample["temperature_c"],
|
|
sample["data_units_written"],
|
|
sample["data_units_read"],
|
|
sample["bytes_written"],
|
|
sample["bytes_read"],
|
|
sample["critical_warning"],
|
|
segment_id,
|
|
sample.get("local_tz"),
|
|
),
|
|
)
|
|
|
|
return {
|
|
"segment_opened": segment_opened,
|
|
"segment_reason": reason,
|
|
"identity_key": identity_key,
|
|
"identity_degraded": identity_degraded,
|
|
"segment_id": segment_id,
|
|
"sample_id": cursor.lastrowid,
|
|
}
|
|
|
|
|
|
def _observation_time(sample: Dict[str, Any]) -> datetime:
|
|
"""Read an observation's original timestamp for ordered recovery."""
|
|
observed_at = datetime.fromisoformat(sample["ts"])
|
|
if observed_at.tzinfo is None:
|
|
return observed_at.replace(tzinfo=timezone.utc)
|
|
return observed_at.astimezone(timezone.utc)
|
|
|
|
|
|
def _publish_observation(
|
|
conn: sqlite3.Connection,
|
|
sample: Dict[str, Any],
|
|
identity: Dict[str, Any],
|
|
tz_name: str,
|
|
) -> None:
|
|
"""Publish one staged sample and all dependent evidence in caller transaction."""
|
|
observed_at = _observation_time(sample)
|
|
sample["local_tz"] = tz_name
|
|
validate_sample_invariants(sample, conn)
|
|
ensure_period_open(conn, observed_at)
|
|
seg_info = write_sample(sample, identity, conn, observed_at)
|
|
|
|
current_id = seg_info["sample_id"]
|
|
prev = find_previous_sample(conn, seg_info.get("segment_id"), current_id)
|
|
try:
|
|
if prev is not None:
|
|
current = {
|
|
"id": current_id,
|
|
"ts": sample["ts"],
|
|
"bytes_written": sample["bytes_written"],
|
|
"bytes_read": sample["bytes_read"],
|
|
"power_on_hours": sample["power_on_hours"],
|
|
"temperature_c": sample["temperature_c"],
|
|
"data_units_written": sample["data_units_written"],
|
|
"data_units_read": sample["data_units_read"],
|
|
"local_tz": tz_name,
|
|
}
|
|
derive_hours_from_interval(conn, prev, current)
|
|
from .local_day import record_local_activity_interval
|
|
record_local_activity_interval(
|
|
conn,
|
|
prev,
|
|
current,
|
|
start_sample_id=prev["id"],
|
|
end_sample_id=current_id,
|
|
segment_id=seg_info.get("segment_id"),
|
|
)
|
|
else:
|
|
previous_any_segment = find_previous_sample(conn, None, current_id)
|
|
if previous_any_segment is not None:
|
|
from .local_day import mark_local_activity_gap
|
|
mark_local_activity_gap(
|
|
conn,
|
|
sample["ts"],
|
|
tz_name,
|
|
previous_any_segment,
|
|
)
|
|
|
|
from .day_aggregate import derive_all_days, persist_day_aggregate
|
|
for aggregate in derive_all_days(conn):
|
|
persist_day_aggregate(conn, aggregate)
|
|
|
|
from .local_day import derive_local_day_summary, persist_local_day
|
|
local_summary = derive_local_day_summary(conn, tz_name, observed_at)
|
|
if local_summary is not None:
|
|
persist_local_day(conn, local_summary)
|
|
except Exception as exc: # noqa: BLE001 - preserve valid staged evidence for retry.
|
|
raise _PendingDerivationError(exc) from exc
|
|
|
|
|
|
def _pending_count(conn: sqlite3.Connection) -> int:
|
|
return conn.execute("SELECT COUNT(*) FROM pending_publications").fetchone()[0]
|
|
|
|
|
|
def _stage_observation(
|
|
conn: sqlite3.Connection,
|
|
sample: Dict[str, Any],
|
|
identity: Dict[str, Any],
|
|
tz_name: str,
|
|
) -> int:
|
|
"""Durably stage valid acquired evidence before attempting publication."""
|
|
payload = json.dumps(
|
|
{"sample": sample, "identity": identity, "tz_name": tz_name},
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
)
|
|
cursor = conn.execute(
|
|
"INSERT INTO pending_publications (sample_ts, payload) VALUES (?, ?)",
|
|
(sample["ts"], payload),
|
|
)
|
|
return cursor.lastrowid
|
|
|
|
|
|
def _recover_pending(conn: sqlite3.Connection) -> int:
|
|
"""Publish pending observations oldest first; stop at first failure."""
|
|
recovered = 0
|
|
while True:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT id, payload FROM pending_publications ORDER BY id LIMIT 1"
|
|
).fetchone()
|
|
if row is None:
|
|
conn.rollback()
|
|
return recovered
|
|
|
|
pending_id, payload = row
|
|
observation = json.loads(payload)
|
|
try:
|
|
_publish_observation(
|
|
conn,
|
|
observation["sample"],
|
|
observation["identity"],
|
|
observation["tz_name"],
|
|
)
|
|
except Exception as exc:
|
|
raise _PendingRecoveryFailure(exc, recovered) from exc
|
|
conn.execute("DELETE FROM pending_publications WHERE id = ?", (pending_id,))
|
|
conn.commit()
|
|
recovered += 1
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
|
|
|
|
def _is_store_or_invariant_failure(exc: Exception) -> bool:
|
|
return (
|
|
isinstance(exc, sqlite3.Error)
|
|
and not isinstance(exc, sqlite3.IntegrityError)
|
|
) or isinstance(exc, InvariantViolationError)
|
|
|
|
|
|
def _pending_capacity_error(
|
|
reason: str,
|
|
*,
|
|
acquisition_skipped: bool = False,
|
|
) -> RuntimeError:
|
|
outcome = "; no new observation acquired" if acquisition_skipped else ""
|
|
return RuntimeError(
|
|
"pending publication capacity full "
|
|
f"({PENDING_PUBLICATION_LIMIT} observations){outcome}; "
|
|
f"{reason}"
|
|
)
|
|
|
|
|
|
def _recover_pending_for_collection(conn: sqlite3.Connection) -> int:
|
|
"""Retry queued work and report exhausted capacity without masking store faults."""
|
|
try:
|
|
return _recover_pending(conn)
|
|
except Exception as exc: # noqa: BLE001 - preserve pending evidence on recovery errors.
|
|
cause = exc.cause if isinstance(exc, _PendingRecoveryFailure) else exc
|
|
if _is_store_or_invariant_failure(cause):
|
|
raise cause
|
|
try:
|
|
capacity_full = _pending_count(conn) >= PENDING_PUBLICATION_LIMIT
|
|
except sqlite3.Error:
|
|
raise cause
|
|
if capacity_full:
|
|
raise _pending_capacity_error(f"recovery failed: {cause}") from cause
|
|
raise cause
|
|
|
|
|
|
def _recover_pending_for_admission(conn: sqlite3.Connection) -> int:
|
|
"""Recover in order, then admit one observation if bounded space remains."""
|
|
try:
|
|
recovered = _recover_pending(conn)
|
|
except _PendingRecoveryFailure as failure:
|
|
if (
|
|
not failure.derivation_error
|
|
or _is_store_or_invariant_failure(failure.cause)
|
|
):
|
|
raise failure.cause
|
|
try:
|
|
pending_count = _pending_count(conn)
|
|
except sqlite3.Error:
|
|
raise failure.cause
|
|
if pending_count >= PENDING_PUBLICATION_LIMIT:
|
|
raise _pending_capacity_error(
|
|
f"recovery failed: {failure.cause}",
|
|
acquisition_skipped=True,
|
|
) from failure.cause
|
|
return failure.published_count
|
|
|
|
if _pending_count(conn) >= PENDING_PUBLICATION_LIMIT:
|
|
raise _pending_capacity_error(
|
|
"recovery left the queue full",
|
|
acquisition_skipped=True,
|
|
)
|
|
return recovered
|
|
|
|
|
|
def run_collection(
|
|
smartctl_data: Optional[Dict[str, Any]] = None,
|
|
sysfs_path: Optional[Path] = None,
|
|
config: Optional[Dict[str, Any]] = None,
|
|
clock=None,
|
|
*,
|
|
acquire: Optional[Callable[[], tuple[Dict[str, Any], Path]]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Recover old work, acquire one observation, then publish it atomically.
|
|
|
|
Production callers pass ``acquire`` so recovery and capacity checks run
|
|
before device interrogation. Direct sample arguments remain useful for
|
|
deterministic collector tests.
|
|
"""
|
|
conn = None
|
|
try:
|
|
if config is None or clock is None:
|
|
raise ValueError("config and clock are required")
|
|
|
|
store_path = get_store_path(config)
|
|
conn = init_store(store_path)
|
|
|
|
from .legacy import import_legacy_history
|
|
history_path = Path(config.get("data_dir", ".")) / "history.jsonl"
|
|
if history_path.exists():
|
|
import_legacy_history(conn, history_path, clock=clock)
|
|
|
|
published_count = _recover_pending_for_admission(conn)
|
|
checked_pending_count = _pending_count(conn)
|
|
|
|
# Hold the writer reservation across the capacity check and acquisition.
|
|
# Recheck recovery if another collector appended work during preflight.
|
|
while True:
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
waiting = _pending_count(conn)
|
|
if (
|
|
waiting >= PENDING_PUBLICATION_LIMIT
|
|
or waiting > checked_pending_count
|
|
):
|
|
conn.rollback()
|
|
published_count += _recover_pending_for_admission(conn)
|
|
checked_pending_count = _pending_count(conn)
|
|
continue
|
|
|
|
from .tz_util import detect_system_tz
|
|
tz_name = detect_system_tz()
|
|
if acquire is not None:
|
|
smartctl_data, sysfs_path = acquire()
|
|
if smartctl_data is None or sysfs_path is None:
|
|
raise AcquisitionError("No acquired SMART data or sysfs identity")
|
|
|
|
counters = acquire_from_smartctl(smartctl_data)
|
|
identity = acquire_from_sysfs(sysfs_path)
|
|
sample = {
|
|
"ts": clock.utcnow().isoformat(),
|
|
"device": config["device"],
|
|
**counters,
|
|
**identity,
|
|
}
|
|
validate_sample_invariants(sample, conn)
|
|
_stage_observation(conn, sample, identity, tz_name)
|
|
conn.commit()
|
|
break
|
|
|
|
published_count += _recover_pending_for_collection(conn)
|
|
retention_error = None
|
|
try:
|
|
if _pending_count(conn) == 0:
|
|
from .pruning import prune_old_samples
|
|
|
|
prune_old_samples(conn, clock.utcnow())
|
|
except Exception as exc: # noqa: BLE001 - publication already committed.
|
|
# Publication already committed. Keep collection successful and
|
|
# retry atomic retention on the next normal collection.
|
|
conn.rollback()
|
|
retention_error = str(exc)
|
|
return {
|
|
"ok": True,
|
|
"sample_count": published_count,
|
|
"retention_error": retention_error,
|
|
"store_path": str(store_path),
|
|
}
|
|
|
|
except Exception as exc:
|
|
if conn is not None:
|
|
conn.rollback()
|
|
return {
|
|
"ok": False,
|
|
"error": str(exc),
|
|
"error_type": type(exc).__name__,
|
|
}
|
|
finally:
|
|
if conn is not None:
|
|
conn.close()
|