180 lines
7.0 KiB
Python
180 lines
7.0 KiB
Python
"""PostgreSQL diagnostics for System Health."""
|
|
|
|
from collections.abc import Callable, Iterator
|
|
from contextlib import contextmanager
|
|
from typing import Any
|
|
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
|
|
from app.config import settings
|
|
|
|
|
|
ConnectionFactory = Callable[[], psycopg.Connection]
|
|
|
|
|
|
def default_connection_factory() -> psycopg.Connection:
|
|
return psycopg.connect(
|
|
host=settings.postgres_host,
|
|
port=settings.postgres_port,
|
|
dbname=settings.postgres_db,
|
|
user=settings.postgres_user,
|
|
password=settings.postgres_password,
|
|
connect_timeout=3,
|
|
)
|
|
|
|
|
|
MIGRATION_SQL = """
|
|
SELECT clock_timestamp() AS database_time,
|
|
(SELECT version_num FROM alembic_version LIMIT 1) AS current_revision
|
|
"""
|
|
|
|
OPERATIONS_SQL = """
|
|
WITH telemetry AS (
|
|
SELECT count(*)::bigint AS snapshot_count,
|
|
min(observed_at) AS oldest_observed_at,
|
|
max(observed_at) AS latest_observed_at
|
|
FROM miner_status_snapshots
|
|
),
|
|
hourly AS (
|
|
SELECT count(*)::bigint AS row_count, max(bucket_at) AS latest_bucket
|
|
FROM miner_status_hourly_rollups
|
|
),
|
|
daily AS (
|
|
SELECT count(*)::bigint AS row_count, max(bucket_date) AS latest_bucket
|
|
FROM miner_status_daily_rollups
|
|
),
|
|
latest_retention AS (
|
|
SELECT mode, started_at FROM minerops_retention_runs
|
|
ORDER BY started_at DESC LIMIT 1
|
|
),
|
|
largest AS (
|
|
SELECT coalesce(jsonb_agg(to_jsonb(sized) ORDER BY sized.total_bytes DESC), '[]'::jsonb) AS rows
|
|
FROM (
|
|
SELECT relname AS name, pg_total_relation_size(relid)::bigint AS total_bytes
|
|
FROM pg_catalog.pg_statio_user_tables
|
|
WHERE relname = ANY(%s)
|
|
ORDER BY pg_total_relation_size(relid) DESC, relname
|
|
LIMIT 8
|
|
) AS sized
|
|
)
|
|
SELECT
|
|
(SELECT count(*) FROM miners)::bigint AS miners,
|
|
(SELECT count(*) FROM sites)::bigint AS sites,
|
|
(SELECT count(*) FROM rooms)::bigint AS rooms,
|
|
(SELECT count(*) FROM location_positions)::bigint AS location_positions,
|
|
(SELECT count(*) FROM repairs)::bigint AS repairs,
|
|
(SELECT count(*) FROM repair_events)::bigint AS repair_events,
|
|
(SELECT count(*) FROM repair_parts)::bigint AS repair_parts,
|
|
telemetry.snapshot_count, telemetry.oldest_observed_at,
|
|
telemetry.latest_observed_at,
|
|
to_regclass('public.miner_status_hourly_rollups') IS NOT NULL AS hourly_present,
|
|
hourly.row_count AS hourly_rows, hourly.latest_bucket AS hourly_latest,
|
|
to_regclass('public.miner_status_daily_rollups') IS NOT NULL AS daily_present,
|
|
daily.row_count AS daily_rows, daily.latest_bucket AS daily_latest,
|
|
latest_retention.mode AS retention_mode,
|
|
latest_retention.started_at AS retention_started_at,
|
|
largest.rows AS largest_tables
|
|
FROM telemetry CROSS JOIN hourly CROSS JOIN daily CROSS JOIN largest
|
|
LEFT JOIN latest_retention ON true
|
|
"""
|
|
|
|
INTEGRITY_SQL = """
|
|
WITH position_counts AS (
|
|
SELECT miner_id, count(*) AS position_count
|
|
FROM location_positions WHERE miner_id IS NOT NULL GROUP BY miner_id
|
|
),
|
|
active_repairs AS (
|
|
SELECT id, miner_id FROM repairs
|
|
WHERE status IN ('OPEN', 'IN_PROGRESS', 'WAITING_FOR_PARTS', 'READY_FOR_TEST')
|
|
),
|
|
active_repair_groups AS (
|
|
SELECT miner_id FROM active_repairs GROUP BY miner_id HAVING count(*) > 1
|
|
),
|
|
normalized_macs AS (
|
|
SELECT lower(regexp_replace(trim(current_mac), '[^0-9A-Fa-f]', '', 'g')) AS mac
|
|
FROM miners WHERE current_mac IS NOT NULL AND trim(current_mac) <> ''
|
|
GROUP BY lower(regexp_replace(trim(current_mac), '[^0-9A-Fa-f]', '', 'g'))
|
|
HAVING count(*) > 1
|
|
)
|
|
SELECT
|
|
(SELECT count(*) FROM miners WHERE asset_tag IS NULL OR trim(asset_tag) = '')::bigint
|
|
AS miners_without_asset_id,
|
|
(SELECT count(*) FROM miners m WHERE NOT EXISTS (
|
|
SELECT 1 FROM location_positions p WHERE p.miner_id = m.id
|
|
))::bigint AS unmapped_miners,
|
|
(SELECT count(*) FROM position_counts WHERE position_count > 1)::bigint
|
|
AS miners_in_multiple_positions,
|
|
(SELECT count(*) FROM normalized_macs)::bigint AS duplicate_current_mac_groups,
|
|
(SELECT count(*) FROM location_positions WHERE room_id IS NULL)::bigint
|
|
AS unscoped_legacy_positions,
|
|
(SELECT count(*) FROM location_positions p JOIN miners m ON m.id = p.miner_id
|
|
WHERE p.expected_mac IS NOT NULL AND trim(p.expected_mac) <> ''
|
|
AND m.current_mac IS NOT NULL AND trim(m.current_mac) <> ''
|
|
AND lower(regexp_replace(trim(p.expected_mac), '[^0-9A-Fa-f]', '', 'g'))
|
|
<> lower(regexp_replace(trim(m.current_mac), '[^0-9A-Fa-f]', '', 'g'))
|
|
)::bigint AS expected_mac_mismatches,
|
|
(SELECT count(*) FROM location_positions WHERE miner_id IS NOT NULL)::bigint
|
|
AS occupied_positions,
|
|
(SELECT count(*) FROM location_positions WHERE miner_id IS NULL)::bigint
|
|
AS empty_positions,
|
|
(SELECT count(*) FROM active_repairs)::bigint AS active_modular_repairs,
|
|
(SELECT count(*) FROM active_repair_groups)::bigint AS duplicate_active_repair_groups,
|
|
(SELECT count(*) FROM miner_location_states state
|
|
WHERE state.asset_state = 'repair' AND NOT EXISTS (
|
|
SELECT 1 FROM active_repairs repair WHERE repair.miner_id = state.miner_id
|
|
))::bigint AS legacy_repair_only_miners,
|
|
(SELECT count(*) FROM miners m WHERE m.lifecycle_status = 'in_repair'
|
|
AND NOT EXISTS (SELECT 1 FROM active_repairs r WHERE r.miner_id = m.id))::bigint
|
|
AS lifecycle_repair_without_active_repair,
|
|
(SELECT count(*) FROM active_repairs r JOIN miners m ON m.id = r.miner_id
|
|
WHERE m.lifecycle_status <> 'in_repair')::bigint
|
|
AS active_repair_lifecycle_mismatches,
|
|
(SELECT count(*) FROM miner_location_states WHERE asset_state = 'waiting_placement')::bigint
|
|
AS waiting_placement_records
|
|
"""
|
|
|
|
|
|
RELEVANT_TABLES = [
|
|
"miner_status_snapshots", "repair_events", "repair_parts", "repairs",
|
|
"location_positions", "miners", "miner_status_hourly_rollups",
|
|
"miner_status_daily_rollups",
|
|
]
|
|
|
|
|
|
class SystemHealthRepository:
|
|
def __init__(
|
|
self,
|
|
connection_factory: ConnectionFactory | None = None,
|
|
*,
|
|
connection: psycopg.Connection | None = None,
|
|
):
|
|
self._connection_factory = connection_factory or default_connection_factory
|
|
self._connection = connection
|
|
|
|
@contextmanager
|
|
def _connection_scope(self) -> Iterator[psycopg.Connection]:
|
|
if self._connection is not None:
|
|
yield self._connection
|
|
return
|
|
connection = self._connection_factory()
|
|
try:
|
|
yield connection
|
|
finally:
|
|
connection.close()
|
|
|
|
def read_overview(self) -> dict[str, Any]:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(MIGRATION_SQL)
|
|
migration = dict(cur.fetchone())
|
|
cur.execute(OPERATIONS_SQL, (RELEVANT_TABLES,))
|
|
operations = dict(cur.fetchone())
|
|
cur.execute(INTEGRITY_SQL)
|
|
integrity = dict(cur.fetchone())
|
|
return {
|
|
"migration": migration,
|
|
"operations": operations,
|
|
"integrity": integrity,
|
|
}
|