feat: add monitoring trends
This commit is contained in:
lussierj committed
1 parent
19fc29c72c
commit
4d360fac42
18 files changed
+1683
No files matched your search
@@ -0,0 +1,237 @@
|
||||
"""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()]
|
||||
Reference in new issue
Block a user