238 lines
8.4 KiB
Python
238 lines
8.4 KiB
Python
"""PostgreSQL persistence for bounded monitoring trends."""
|
|
|
|
from collections.abc import Callable, Iterator
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
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=5,
|
|
)
|
|
|
|
|
|
TREND_SQL = """
|
|
WITH requested AS (
|
|
SELECT %s::text AS scope_type, %s::uuid AS scope_id,
|
|
%s::timestamptz AS window_start, %s::timestamptz AS window_end,
|
|
%s::interval AS bucket_width
|
|
),
|
|
current_locations AS (
|
|
SELECT miners.id AS miner_id,
|
|
COALESCE(position.site_id, miners.site_id) AS site_id,
|
|
COALESCE(position.room_id, miners.room_id) AS room_id
|
|
FROM miners
|
|
LEFT JOIN LATERAL (
|
|
SELECT rooms.site_id, location_positions.room_id
|
|
FROM location_positions
|
|
LEFT JOIN rooms ON rooms.id = location_positions.room_id
|
|
WHERE location_positions.miner_id = miners.id
|
|
ORDER BY location_positions.id
|
|
LIMIT 1
|
|
) AS position ON true
|
|
),
|
|
scope_meta AS (
|
|
SELECT requested.scope_type, requested.scope_id,
|
|
CASE requested.scope_type
|
|
WHEN 'fleet' THEN true
|
|
WHEN 'miner' THEN EXISTS (
|
|
SELECT 1 FROM miners WHERE miners.id = requested.scope_id
|
|
)
|
|
WHEN 'site' THEN EXISTS (
|
|
SELECT 1 FROM sites WHERE sites.id = requested.scope_id
|
|
)
|
|
WHEN 'room' THEN EXISTS (
|
|
SELECT 1 FROM rooms WHERE rooms.id = requested.scope_id
|
|
)
|
|
ELSE false
|
|
END AS scope_exists,
|
|
CASE requested.scope_type
|
|
WHEN 'fleet' THEN 'Fleet'
|
|
WHEN 'miner' THEN (
|
|
SELECT COALESCE(NULLIF(trim(miners.asset_tag), ''), miners.id::text)
|
|
FROM miners WHERE miners.id = requested.scope_id
|
|
)
|
|
WHEN 'site' THEN (
|
|
SELECT sites.name FROM sites WHERE sites.id = requested.scope_id
|
|
)
|
|
WHEN 'room' THEN (
|
|
SELECT rooms.name FROM rooms WHERE rooms.id = requested.scope_id
|
|
)
|
|
END AS scope_label
|
|
FROM requested
|
|
),
|
|
scoped_miners AS (
|
|
SELECT miners.id
|
|
FROM miners
|
|
JOIN current_locations ON current_locations.miner_id = miners.id
|
|
CROSS JOIN requested
|
|
WHERE requested.scope_type = 'fleet'
|
|
OR (requested.scope_type = 'miner' AND miners.id = requested.scope_id)
|
|
OR (requested.scope_type = 'site'
|
|
AND current_locations.site_id = requested.scope_id)
|
|
OR (requested.scope_type = 'room'
|
|
AND current_locations.room_id = requested.scope_id)
|
|
),
|
|
scope_count AS (
|
|
SELECT count(*)::integer AS scoped_miner_count FROM scoped_miners
|
|
),
|
|
buckets AS (
|
|
SELECT bucket_start,
|
|
bucket_start + requested.bucket_width AS bucket_end
|
|
FROM requested
|
|
CROSS JOIN LATERAL generate_series(
|
|
requested.window_start,
|
|
requested.window_end - requested.bucket_width,
|
|
requested.bucket_width
|
|
) AS bucket_start
|
|
),
|
|
per_miner_bucket AS (
|
|
SELECT date_bin(
|
|
requested.bucket_width,
|
|
snapshots.observed_at,
|
|
'1970-01-01 00:00:00+00'::timestamptz
|
|
) AS bucket_start,
|
|
snapshots.miner_id,
|
|
count(*)::integer AS observation_count,
|
|
count(snapshots.mhs_av)::integer AS hashrate_sample_count,
|
|
count(snapshots.power_w)::integer AS power_sample_count,
|
|
avg(snapshots.mhs_av) AS average_mhs_av,
|
|
avg(snapshots.power_w) AS average_power_w
|
|
FROM miner_status_snapshots AS snapshots
|
|
JOIN scoped_miners ON scoped_miners.id = snapshots.miner_id
|
|
CROSS JOIN requested
|
|
WHERE snapshots.observed_at >= requested.window_start
|
|
AND snapshots.observed_at < requested.window_end
|
|
GROUP BY bucket_start, snapshots.miner_id
|
|
),
|
|
bucket_totals AS (
|
|
SELECT per_miner_bucket.bucket_start,
|
|
sum(per_miner_bucket.observation_count)::integer AS observation_count,
|
|
sum(per_miner_bucket.hashrate_sample_count)::integer
|
|
AS hashrate_sample_count,
|
|
sum(per_miner_bucket.power_sample_count)::integer AS power_sample_count,
|
|
count(*)::integer AS observed_miner_count,
|
|
count(*) FILTER (
|
|
WHERE per_miner_bucket.hashrate_sample_count > 0
|
|
)::integer AS hashrate_contributing_miner_count,
|
|
count(*) FILTER (
|
|
WHERE per_miner_bucket.power_sample_count > 0
|
|
)::integer AS power_contributing_miner_count,
|
|
CASE WHEN sum(per_miner_bucket.hashrate_sample_count) > 0
|
|
THEN sum(per_miner_bucket.average_mhs_av) FILTER (
|
|
WHERE per_miner_bucket.hashrate_sample_count > 0
|
|
) / 1000000.0
|
|
END AS hashrate_ths,
|
|
CASE WHEN sum(per_miner_bucket.power_sample_count) > 0
|
|
THEN sum(per_miner_bucket.average_power_w) FILTER (
|
|
WHERE per_miner_bucket.power_sample_count > 0
|
|
)
|
|
END AS power_w
|
|
FROM per_miner_bucket
|
|
GROUP BY per_miner_bucket.bucket_start
|
|
),
|
|
latest AS (
|
|
SELECT max(snapshots.observed_at) AS latest_observed_at
|
|
FROM miner_status_snapshots AS snapshots
|
|
JOIN scoped_miners ON scoped_miners.id = snapshots.miner_id
|
|
CROSS JOIN requested
|
|
WHERE snapshots.observed_at >= requested.window_start
|
|
AND snapshots.observed_at < requested.window_end
|
|
)
|
|
SELECT scope_meta.scope_exists, scope_meta.scope_label,
|
|
buckets.bucket_start, buckets.bucket_end,
|
|
bucket_totals.hashrate_ths, bucket_totals.power_w,
|
|
COALESCE(bucket_totals.observation_count, 0) AS observation_count,
|
|
COALESCE(bucket_totals.hashrate_sample_count, 0) AS hashrate_sample_count,
|
|
COALESCE(bucket_totals.power_sample_count, 0) AS power_sample_count,
|
|
COALESCE(bucket_totals.observed_miner_count, 0) AS observed_miner_count,
|
|
COALESCE(bucket_totals.hashrate_contributing_miner_count, 0)
|
|
AS hashrate_contributing_miner_count,
|
|
COALESCE(bucket_totals.power_contributing_miner_count, 0)
|
|
AS power_contributing_miner_count,
|
|
scope_count.scoped_miner_count, latest.latest_observed_at
|
|
FROM buckets
|
|
CROSS JOIN scope_meta
|
|
CROSS JOIN scope_count
|
|
CROSS JOIN latest
|
|
LEFT JOIN bucket_totals ON bucket_totals.bucket_start = buckets.bucket_start
|
|
ORDER BY buckets.bucket_start
|
|
"""
|
|
|
|
|
|
SCOPE_OPTIONS_SQL = """
|
|
SELECT option_type, id, site_id, name
|
|
FROM (
|
|
SELECT 'site'::text AS option_type, sites.id,
|
|
NULL::uuid AS site_id, sites.name
|
|
FROM sites
|
|
UNION ALL
|
|
SELECT 'room'::text AS option_type, rooms.id,
|
|
rooms.site_id, rooms.name
|
|
FROM rooms
|
|
) AS options
|
|
ORDER BY option_type DESC, lower(name), id
|
|
"""
|
|
|
|
|
|
class MonitoringRepository:
|
|
"""Own the set-based raw-snapshot monitoring read path."""
|
|
|
|
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_trends(
|
|
self,
|
|
*,
|
|
scope: str,
|
|
scope_id: UUID | None,
|
|
window_start: datetime,
|
|
window_end: datetime,
|
|
bucket_width: timedelta,
|
|
) -> list[dict[str, Any]]:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
TREND_SQL,
|
|
(scope, scope_id, window_start, window_end, bucket_width),
|
|
)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
|
|
def list_location_scope_options(self) -> list[dict[str, Any]]:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(SCOPE_OPTIONS_SQL)
|
|
return [dict(row) for row in cur.fetchall()]
|