feat: collector tracer bullet (#21)\n\nSmartctl acquisition with validation\nSysfs controller identity acquisition\nIdentity normalization (strip, no case fold, blank handling)\nSQLite store in WAL mode with six entities\nSchema versioning via PRAGMA user_version\nInvariant validation (negative bytes, etc.)\n17 passing tests across two test files\n\nCloses #21
This commit is contained in:
xavierk
@@ -0,0 +1,187 @@
|
||||
"""Observation store: SQLite database for persisting observation history.
|
||||
|
||||
This module handles:
|
||||
- Store initialization with WAL mode
|
||||
- Schema versioning with PRAGMA user_version
|
||||
- The six entities: samples, hour_observations, day_aggregates,
|
||||
monitoring_periods, controller_segments, endurance_baseline
|
||||
"""
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
# Schema version - increment on each migration
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
|
||||
def get_store_path(config: dict) -> Path:
|
||||
"""Get the store path from config."""
|
||||
return Path(config["store_path"])
|
||||
|
||||
|
||||
def init_store(store_path: Path) -> sqlite3.Connection:
|
||||
"""Initialize the observation store if not present.
|
||||
|
||||
Creates the database with WAL mode and all six entities.
|
||||
Returns a connection to the store.
|
||||
"""
|
||||
conn = sqlite3.connect(str(store_path))
|
||||
|
||||
# Enable WAL mode for concurrent reads during writes
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
|
||||
# Check if this is a new database
|
||||
cursor = conn.execute("PRAGMA user_version")
|
||||
current_version = cursor.fetchone()[0]
|
||||
|
||||
if current_version == 0:
|
||||
# New database - create schema
|
||||
_create_schema(conn)
|
||||
conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}")
|
||||
conn.commit()
|
||||
elif current_version > SCHEMA_VERSION:
|
||||
# Unknown newer version - refuse
|
||||
conn.close()
|
||||
raise ValueError(
|
||||
f"Observation store written by a newer Fenris (version {current_version}) "
|
||||
f"— upgrade Fenris"
|
||||
)
|
||||
elif current_version < SCHEMA_VERSION:
|
||||
# Older version - apply migrations
|
||||
_apply_migrations(conn, current_version)
|
||||
conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}")
|
||||
conn.commit()
|
||||
|
||||
return conn
|
||||
|
||||
|
||||
def _create_schema(conn: sqlite3.Connection):
|
||||
"""Create the initial schema with all six entities."""
|
||||
|
||||
# Samples: raw collection runs (14-day retention)
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS samples (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
ts TEXT NOT NULL, -- ISO 8601 UTC timestamp
|
||||
device TEXT NOT NULL,
|
||||
-- Normalized controller-identity fields captured at acquisition
|
||||
subnqn TEXT,
|
||||
sn TEXT,
|
||||
mn TEXT,
|
||||
fr TEXT,
|
||||
capacity_bytes INTEGER,
|
||||
percentage_used INTEGER,
|
||||
available_spare INTEGER,
|
||||
media_errors INTEGER,
|
||||
power_on_hours INTEGER,
|
||||
power_cycles INTEGER,
|
||||
unsafe_shutdowns INTEGER,
|
||||
temperature_c INTEGER,
|
||||
data_units_written INTEGER,
|
||||
data_units_read INTEGER,
|
||||
bytes_written INTEGER,
|
||||
bytes_read INTEGER,
|
||||
critical_warning INTEGER
|
||||
)
|
||||
""")
|
||||
|
||||
# Hour observations: UTC-hour usage-habit split
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS hour_observations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
hour TEXT NOT NULL UNIQUE, -- ISO 8601 UTC hour (e.g., "2026-09-01T12:00:00Z")
|
||||
active_seconds INTEGER DEFAULT 0,
|
||||
idle_seconds INTEGER DEFAULT 0,
|
||||
powered_off_seconds INTEGER DEFAULT 0,
|
||||
unknown_seconds INTEGER DEFAULT 0,
|
||||
bytes_written_delta INTEGER DEFAULT 0,
|
||||
bytes_read_delta INTEGER DEFAULT 0,
|
||||
temperature_min INTEGER,
|
||||
temperature_avg REAL,
|
||||
temperature_max INTEGER,
|
||||
sample_count INTEGER DEFAULT 0,
|
||||
coverage REAL DEFAULT 0.0
|
||||
)
|
||||
""")
|
||||
|
||||
# Day aggregates: derived from hour observations
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS day_aggregates (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
day TEXT NOT NULL UNIQUE, -- ISO 8601 UTC day (e.g., "2026-09-01")
|
||||
active_seconds INTEGER DEFAULT 0,
|
||||
idle_seconds INTEGER DEFAULT 0,
|
||||
powered_off_seconds INTEGER DEFAULT 0,
|
||||
unknown_seconds INTEGER DEFAULT 0,
|
||||
bytes_written_delta INTEGER DEFAULT 0,
|
||||
bytes_read_delta INTEGER DEFAULT 0,
|
||||
sample_count INTEGER DEFAULT 0,
|
||||
coverage REAL DEFAULT 0.0
|
||||
)
|
||||
""")
|
||||
|
||||
# Monitoring periods: tracking when monitoring was enabled/disabled
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS monitoring_periods (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
started_at TEXT NOT NULL, -- ISO 8601 UTC timestamp
|
||||
ended_at TEXT, -- NULL if currently active
|
||||
end_cause TEXT CHECK(end_cause IN ('user_disabled', 'migrated', 'unknown_gap'))
|
||||
)
|
||||
""")
|
||||
|
||||
# Controller segments: identity key plus metadata snapshot
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS controller_segments (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
opened_at TEXT NOT NULL, -- ISO 8601 UTC timestamp
|
||||
identity_key TEXT, -- Normalized identity key (NULL if degraded)
|
||||
identity_degraded BOOLEAN DEFAULT 0,
|
||||
subnqn TEXT,
|
||||
sn TEXT,
|
||||
mn TEXT,
|
||||
fr TEXT,
|
||||
vid TEXT,
|
||||
ssvid TEXT,
|
||||
transport TEXT
|
||||
)
|
||||
""")
|
||||
|
||||
# Endurance baseline: one active row, replaced on edit
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS endurance_baseline (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
tbw_terabytes REAL NOT NULL,
|
||||
source_url TEXT,
|
||||
document_revision TEXT,
|
||||
entry_date TEXT,
|
||||
model_string TEXT,
|
||||
nominal_capacity_bytes INTEGER,
|
||||
validated_by TEXT, -- 'user' or 'machine_match'
|
||||
verified BOOLEAN DEFAULT 0,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
)
|
||||
""")
|
||||
|
||||
|
||||
def _apply_migrations(conn: sqlite3.Connection, current_version: int):
|
||||
"""Apply forward-only migrations from current_version to SCHEMA_VERSION."""
|
||||
# Future migrations will go here
|
||||
# For now, just upgrade to current version
|
||||
pass
|
||||
|
||||
|
||||
def is_store_faulty(store_path: Path) -> bool:
|
||||
"""Check if the store is present but cannot be read or trusted."""
|
||||
if not store_path.exists():
|
||||
return False
|
||||
|
||||
try:
|
||||
conn = sqlite3.connect(f"file:{store_path}?mode=ro", uri=True)
|
||||
conn.execute("PRAGMA user_version")
|
||||
conn.close()
|
||||
return False
|
||||
except sqlite3.Error:
|
||||
return True
|
||||
Reference in New Issue
Block a user