implement timescaledb improvements
This commit is contained in:
parent
d401e0b036
commit
17f4558702
3 changed files with 511 additions and 5 deletions
|
|
@ -41,9 +41,44 @@ defmodule Towerops.Monitoring do
|
|||
|
||||
@doc """
|
||||
Gets hourly statistics for device over a time range.
|
||||
Calculates aggregates on-demand from monitoring_checks.
|
||||
Uses TimescaleDB continuous aggregates for performance when available,
|
||||
falls back to raw query calculation otherwise.
|
||||
"""
|
||||
def get_hourly_stats(device_id, start_time, end_time) do
|
||||
# Try continuous aggregate first (much faster for large datasets)
|
||||
query = """
|
||||
SELECT
|
||||
bucket,
|
||||
total_checks,
|
||||
successful_checks,
|
||||
failed_checks,
|
||||
ROUND(avg_response_time_ms::numeric, 2) as avg_response_time_ms,
|
||||
min_response_time_ms,
|
||||
max_response_time_ms,
|
||||
ROUND(100.0 * successful_checks / NULLIF(total_checks, 0), 2) as uptime_percentage
|
||||
FROM monitoring_checks_hourly
|
||||
WHERE device_id = $1::uuid
|
||||
AND bucket >= $2
|
||||
AND bucket <= $3
|
||||
ORDER BY bucket ASC
|
||||
"""
|
||||
|
||||
case SQL.query(Repo, query, [Ecto.UUID.dump!(device_id), start_time, end_time]) do
|
||||
{:ok, %{rows: rows} = result} when rows != [] ->
|
||||
result
|
||||
|
||||
{:ok, _empty_result} ->
|
||||
# Continuous aggregate may not have recent data yet, fall back to raw query
|
||||
get_hourly_stats_raw(device_id, start_time, end_time)
|
||||
|
||||
{:error, _} ->
|
||||
# Fallback to raw query if continuous aggregate doesn't exist
|
||||
get_hourly_stats_raw(device_id, start_time, end_time)
|
||||
end
|
||||
end
|
||||
|
||||
# Fallback for environments without TimescaleDB continuous aggregates
|
||||
defp get_hourly_stats_raw(device_id, start_time, end_time) do
|
||||
query = """
|
||||
SELECT
|
||||
date_trunc('hour', checked_at) as bucket,
|
||||
|
|
@ -71,9 +106,44 @@ defmodule Towerops.Monitoring do
|
|||
|
||||
@doc """
|
||||
Gets daily statistics for device over a time range.
|
||||
Calculates aggregates on-demand from monitoring_checks.
|
||||
Uses TimescaleDB continuous aggregates for performance when available,
|
||||
falls back to raw query calculation otherwise.
|
||||
"""
|
||||
def get_daily_stats(device_id, start_time, end_time) do
|
||||
# Try continuous aggregate first (much faster for large datasets)
|
||||
query = """
|
||||
SELECT
|
||||
bucket,
|
||||
total_checks,
|
||||
successful_checks,
|
||||
failed_checks,
|
||||
ROUND(avg_response_time_ms::numeric, 2) as avg_response_time_ms,
|
||||
min_response_time_ms,
|
||||
max_response_time_ms,
|
||||
ROUND(100.0 * successful_checks / NULLIF(total_checks, 0), 2) as uptime_percentage
|
||||
FROM monitoring_checks_daily
|
||||
WHERE device_id = $1::uuid
|
||||
AND bucket >= $2
|
||||
AND bucket <= $3
|
||||
ORDER BY bucket ASC
|
||||
"""
|
||||
|
||||
case SQL.query(Repo, query, [Ecto.UUID.dump!(device_id), start_time, end_time]) do
|
||||
{:ok, %{rows: rows} = result} when rows != [] ->
|
||||
result
|
||||
|
||||
{:ok, _empty_result} ->
|
||||
# Continuous aggregate may not have recent data yet, fall back to raw query
|
||||
get_daily_stats_raw(device_id, start_time, end_time)
|
||||
|
||||
{:error, _} ->
|
||||
# Fallback to raw query if continuous aggregate doesn't exist
|
||||
get_daily_stats_raw(device_id, start_time, end_time)
|
||||
end
|
||||
end
|
||||
|
||||
# Fallback for environments without TimescaleDB continuous aggregates
|
||||
defp get_daily_stats_raw(device_id, start_time, end_time) do
|
||||
query = """
|
||||
SELECT
|
||||
date_trunc('day', checked_at) as bucket,
|
||||
|
|
@ -101,11 +171,41 @@ defmodule Towerops.Monitoring do
|
|||
|
||||
@doc """
|
||||
Gets uptime percentage for device over the last N days.
|
||||
Calculates on-demand from monitoring_checks.
|
||||
Uses TimescaleDB continuous aggregates for performance when available,
|
||||
falls back to raw query calculation otherwise.
|
||||
"""
|
||||
def get_uptime_percentage(device_id, days_ago \\ 30) do
|
||||
start_time = DateTime.add(DateTime.utc_now(), -days_ago * 24 * 60 * 60, :second)
|
||||
|
||||
# Try continuous aggregate first (aggregates daily data for fast calculation)
|
||||
query = """
|
||||
SELECT
|
||||
ROUND(100.0 * SUM(successful_checks) / NULLIF(SUM(total_checks), 0), 2) as uptime_percentage
|
||||
FROM monitoring_checks_daily
|
||||
WHERE device_id = $1::uuid
|
||||
AND bucket >= $2
|
||||
"""
|
||||
|
||||
case SQL.query(Repo, query, [Ecto.UUID.dump!(device_id), start_time]) do
|
||||
{:ok, %{rows: [[%Decimal{} = uptime]]}} ->
|
||||
uptime |> Decimal.to_float() |> Float.round(2)
|
||||
|
||||
{:ok, %{rows: [[nil]]}} ->
|
||||
# Continuous aggregate may not have recent data yet, fall back to raw query
|
||||
get_uptime_percentage_raw(device_id, start_time)
|
||||
|
||||
{:ok, %{rows: []}} ->
|
||||
# Continuous aggregate may not have recent data yet, fall back to raw query
|
||||
get_uptime_percentage_raw(device_id, start_time)
|
||||
|
||||
{:error, _} ->
|
||||
# Fallback to raw query if continuous aggregate doesn't exist
|
||||
get_uptime_percentage_raw(device_id, start_time)
|
||||
end
|
||||
end
|
||||
|
||||
# Fallback for environments without TimescaleDB continuous aggregates
|
||||
defp get_uptime_percentage_raw(device_id, start_time) do
|
||||
query = """
|
||||
SELECT
|
||||
ROUND(100.0 * COUNT(*) FILTER (WHERE status = 'success') / NULLIF(COUNT(*), 0), 2) as uptime_percentage
|
||||
|
|
|
|||
|
|
@ -32,8 +32,10 @@ defmodule Towerops.Snmp.MibTranslator do
|
|||
"MIKROTIK-MIB::mtxrBoardName.0" => "1.3.6.1.4.1.14988.1.1.7.9.0",
|
||||
"MIKROTIK-MIB::mtxrLicenseVersion.0" => "1.3.6.1.4.1.14988.1.1.7.5.0",
|
||||
# MikroTik License info (short names used in YAML profiles)
|
||||
"MIKROTIK-MIB::mtxrLicVersion.0" => "1.3.6.1.4.1.14988.1.1.4.3.0",
|
||||
"MIKROTIK-MIB::mtxrLicLevel.0" => "1.3.6.1.4.1.14988.1.1.4.4.0",
|
||||
# mtxrLicLevel (.3) = license level integer (0-6)
|
||||
# mtxrLicVersion (.4) = software version string (e.g. "7.20.6")
|
||||
"MIKROTIK-MIB::mtxrLicVersion.0" => "1.3.6.1.4.1.14988.1.1.4.4.0",
|
||||
"MIKROTIK-MIB::mtxrLicLevel.0" => "1.3.6.1.4.1.14988.1.1.4.3.0",
|
||||
# MikroTik Health sensors
|
||||
"MIKROTIK-MIB::mtxrHlCoreVoltage.0" => "1.3.6.1.4.1.14988.1.1.3.1.0",
|
||||
"MIKROTIK-MIB::mtxrHlThreeDotThreeVoltage.0" => "1.3.6.1.4.1.14988.1.1.3.2.0",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,404 @@
|
|||
defmodule Towerops.Repo.Migrations.EnableTimescaledbOptimizations do
|
||||
@moduledoc """
|
||||
Enables TimescaleDB optimizations for time-series data tables.
|
||||
|
||||
TimescaleDB Community Edition is free for self-hosted use under the Timescale License.
|
||||
All features used here are available in the free Community Edition:
|
||||
- Hypertables (automatic time-based partitioning)
|
||||
- Compression (reduces storage by 90-95%)
|
||||
- Retention policies (automatic data deletion)
|
||||
- Continuous aggregates (pre-computed rollups)
|
||||
|
||||
Chunk intervals are set based on query patterns:
|
||||
- monitoring_checks: 1 day chunks (frequent queries for recent data)
|
||||
- snmp_sensor_readings: 1 day chunks (frequent queries for recent data)
|
||||
- snmp_interface_stats: 1 day chunks (frequent queries for recent data)
|
||||
|
||||
Compression is enabled for data older than 7 days.
|
||||
Retention is set to 90 days for raw data (aggregates kept longer).
|
||||
"""
|
||||
use Ecto.Migration
|
||||
|
||||
@disable_ddl_transaction true
|
||||
@disable_migration_lock true
|
||||
|
||||
def up do
|
||||
# Step 1: Enable TimescaleDB extension
|
||||
execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE")
|
||||
|
||||
# Step 2: Convert monitoring_checks to hypertable
|
||||
# TimescaleDB requires unique constraints to include the partitioning column.
|
||||
# For time-series data, we drop the UUID primary key and rely on indexes instead.
|
||||
# The data is queried by (device_id, time) not by individual row ID.
|
||||
execute("DROP INDEX IF EXISTS monitoring_checks_equipment_id_index")
|
||||
execute("DROP INDEX IF EXISTS monitoring_checks_equipment_id_checked_at_index")
|
||||
execute("DROP INDEX IF EXISTS monitoring_checks_checked_at_index")
|
||||
execute("DROP INDEX IF EXISTS monitoring_checks_device_id_index")
|
||||
execute("DROP INDEX IF EXISTS monitoring_checks_device_id_checked_at_index")
|
||||
execute("ALTER TABLE monitoring_checks DROP CONSTRAINT IF EXISTS monitoring_checks_pkey")
|
||||
|
||||
execute("""
|
||||
SELECT create_hypertable(
|
||||
'monitoring_checks',
|
||||
'checked_at',
|
||||
chunk_time_interval => INTERVAL '1 day',
|
||||
migrate_data => true,
|
||||
if_not_exists => true
|
||||
)
|
||||
""")
|
||||
|
||||
# Recreate indexes (TimescaleDB optimized)
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS monitoring_checks_device_id_checked_at_idx ON monitoring_checks (device_id, checked_at DESC)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS monitoring_checks_checked_at_idx ON monitoring_checks (checked_at DESC)"
|
||||
)
|
||||
|
||||
# Step 3: Convert snmp_sensor_readings to hypertable
|
||||
execute("DROP INDEX IF EXISTS snmp_sensor_readings_sensor_id_index")
|
||||
execute("DROP INDEX IF EXISTS snmp_sensor_readings_sensor_id_checked_at_index")
|
||||
execute("DROP INDEX IF EXISTS snmp_sensor_readings_status_index")
|
||||
execute("DROP INDEX IF EXISTS snmp_sensor_readings_checked_at_index")
|
||||
|
||||
execute(
|
||||
"ALTER TABLE snmp_sensor_readings DROP CONSTRAINT IF EXISTS snmp_sensor_readings_pkey"
|
||||
)
|
||||
|
||||
execute("""
|
||||
SELECT create_hypertable(
|
||||
'snmp_sensor_readings',
|
||||
'checked_at',
|
||||
chunk_time_interval => INTERVAL '1 day',
|
||||
migrate_data => true,
|
||||
if_not_exists => true
|
||||
)
|
||||
""")
|
||||
|
||||
# Recreate indexes
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS snmp_sensor_readings_sensor_id_checked_at_idx ON snmp_sensor_readings (sensor_id, checked_at DESC)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS snmp_sensor_readings_checked_at_idx ON snmp_sensor_readings (checked_at DESC)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS snmp_sensor_readings_status_idx ON snmp_sensor_readings (status)"
|
||||
)
|
||||
|
||||
# Step 4: Convert snmp_interface_stats to hypertable
|
||||
execute("DROP INDEX IF EXISTS snmp_interface_stats_interface_id_index")
|
||||
execute("DROP INDEX IF EXISTS snmp_interface_stats_interface_id_checked_at_index")
|
||||
execute("DROP INDEX IF EXISTS snmp_interface_stats_checked_at_index")
|
||||
|
||||
execute(
|
||||
"ALTER TABLE snmp_interface_stats DROP CONSTRAINT IF EXISTS snmp_interface_stats_pkey"
|
||||
)
|
||||
|
||||
execute("""
|
||||
SELECT create_hypertable(
|
||||
'snmp_interface_stats',
|
||||
'checked_at',
|
||||
chunk_time_interval => INTERVAL '1 day',
|
||||
migrate_data => true,
|
||||
if_not_exists => true
|
||||
)
|
||||
""")
|
||||
|
||||
# Recreate indexes
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS snmp_interface_stats_interface_id_checked_at_idx ON snmp_interface_stats (interface_id, checked_at DESC)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS snmp_interface_stats_checked_at_idx ON snmp_interface_stats (checked_at DESC)"
|
||||
)
|
||||
|
||||
# Step 5: Enable compression on all hypertables
|
||||
# Compress data older than 7 days to save storage (90-95% reduction)
|
||||
|
||||
execute(
|
||||
"ALTER TABLE monitoring_checks SET (timescaledb.compress, timescaledb.compress_segmentby = 'device_id')"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_compression_policy('monitoring_checks', INTERVAL '7 days', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"ALTER TABLE snmp_sensor_readings SET (timescaledb.compress, timescaledb.compress_segmentby = 'sensor_id')"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_compression_policy('snmp_sensor_readings', INTERVAL '7 days', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"ALTER TABLE snmp_interface_stats SET (timescaledb.compress, timescaledb.compress_segmentby = 'interface_id')"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_compression_policy('snmp_interface_stats', INTERVAL '7 days', if_not_exists => true)"
|
||||
)
|
||||
|
||||
# Step 6: Add retention policies
|
||||
# Keep 90 days of raw data, aggregates will have longer retention
|
||||
execute(
|
||||
"SELECT add_retention_policy('monitoring_checks', INTERVAL '90 days', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_retention_policy('snmp_sensor_readings', INTERVAL '90 days', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_retention_policy('snmp_interface_stats', INTERVAL '90 days', if_not_exists => true)"
|
||||
)
|
||||
|
||||
# Step 7: Create continuous aggregates for monitoring_checks
|
||||
# These pre-compute hourly and daily statistics for fast dashboard queries
|
||||
|
||||
# Hourly monitoring aggregate
|
||||
execute("""
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS monitoring_checks_hourly
|
||||
WITH (timescaledb.continuous) AS
|
||||
SELECT
|
||||
device_id,
|
||||
time_bucket('1 hour', checked_at) AS bucket,
|
||||
COUNT(*) AS total_checks,
|
||||
COUNT(*) FILTER (WHERE status = 'success') AS successful_checks,
|
||||
COUNT(*) FILTER (WHERE status IN ('failure', 'timeout')) AS failed_checks,
|
||||
AVG(response_time_ms) AS avg_response_time_ms,
|
||||
MIN(response_time_ms) AS min_response_time_ms,
|
||||
MAX(response_time_ms) AS max_response_time_ms
|
||||
FROM monitoring_checks
|
||||
GROUP BY device_id, bucket
|
||||
WITH NO DATA
|
||||
""")
|
||||
|
||||
# Daily monitoring aggregate
|
||||
execute("""
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS monitoring_checks_daily
|
||||
WITH (timescaledb.continuous) AS
|
||||
SELECT
|
||||
device_id,
|
||||
time_bucket('1 day', checked_at) AS bucket,
|
||||
COUNT(*) AS total_checks,
|
||||
COUNT(*) FILTER (WHERE status = 'success') AS successful_checks,
|
||||
COUNT(*) FILTER (WHERE status IN ('failure', 'timeout')) AS failed_checks,
|
||||
AVG(response_time_ms) AS avg_response_time_ms,
|
||||
MIN(response_time_ms) AS min_response_time_ms,
|
||||
MAX(response_time_ms) AS max_response_time_ms
|
||||
FROM monitoring_checks
|
||||
GROUP BY device_id, bucket
|
||||
WITH NO DATA
|
||||
""")
|
||||
|
||||
# Step 8: Create continuous aggregates for sensor readings
|
||||
# Hourly sensor readings aggregate
|
||||
execute("""
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS snmp_sensor_readings_hourly
|
||||
WITH (timescaledb.continuous) AS
|
||||
SELECT
|
||||
sensor_id,
|
||||
time_bucket('1 hour', checked_at) AS bucket,
|
||||
AVG(value) AS avg_value,
|
||||
MIN(value) AS min_value,
|
||||
MAX(value) AS max_value,
|
||||
COUNT(*) AS reading_count
|
||||
FROM snmp_sensor_readings
|
||||
GROUP BY sensor_id, bucket
|
||||
WITH NO DATA
|
||||
""")
|
||||
|
||||
# Daily sensor readings aggregate
|
||||
execute("""
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS snmp_sensor_readings_daily
|
||||
WITH (timescaledb.continuous) AS
|
||||
SELECT
|
||||
sensor_id,
|
||||
time_bucket('1 day', checked_at) AS bucket,
|
||||
AVG(value) AS avg_value,
|
||||
MIN(value) AS min_value,
|
||||
MAX(value) AS max_value,
|
||||
COUNT(*) AS reading_count
|
||||
FROM snmp_sensor_readings
|
||||
GROUP BY sensor_id, bucket
|
||||
WITH NO DATA
|
||||
""")
|
||||
|
||||
# Step 9: Create continuous aggregates for interface stats
|
||||
# Hourly interface stats aggregate (traffic rates calculated from deltas)
|
||||
execute("""
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS snmp_interface_stats_hourly
|
||||
WITH (timescaledb.continuous) AS
|
||||
SELECT
|
||||
interface_id,
|
||||
time_bucket('1 hour', checked_at) AS bucket,
|
||||
MAX(if_in_octets) AS max_in_octets,
|
||||
MAX(if_out_octets) AS max_out_octets,
|
||||
SUM(if_in_errors) AS total_in_errors,
|
||||
SUM(if_out_errors) AS total_out_errors,
|
||||
SUM(if_in_discards) AS total_in_discards,
|
||||
SUM(if_out_discards) AS total_out_discards,
|
||||
COUNT(*) AS sample_count
|
||||
FROM snmp_interface_stats
|
||||
GROUP BY interface_id, bucket
|
||||
WITH NO DATA
|
||||
""")
|
||||
|
||||
# Daily interface stats aggregate
|
||||
execute("""
|
||||
CREATE MATERIALIZED VIEW IF NOT EXISTS snmp_interface_stats_daily
|
||||
WITH (timescaledb.continuous) AS
|
||||
SELECT
|
||||
interface_id,
|
||||
time_bucket('1 day', checked_at) AS bucket,
|
||||
MAX(if_in_octets) AS max_in_octets,
|
||||
MAX(if_out_octets) AS max_out_octets,
|
||||
SUM(if_in_errors) AS total_in_errors,
|
||||
SUM(if_out_errors) AS total_out_errors,
|
||||
SUM(if_in_discards) AS total_in_discards,
|
||||
SUM(if_out_discards) AS total_out_discards,
|
||||
COUNT(*) AS sample_count
|
||||
FROM snmp_interface_stats
|
||||
GROUP BY interface_id, bucket
|
||||
WITH NO DATA
|
||||
""")
|
||||
|
||||
# Step 10: Add refresh policies for continuous aggregates
|
||||
# Refresh hourly aggregates every hour, looking back 3 hours to catch late data
|
||||
execute(
|
||||
"SELECT add_continuous_aggregate_policy('monitoring_checks_hourly', start_offset => INTERVAL '3 hours', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '1 hour', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_continuous_aggregate_policy('snmp_sensor_readings_hourly', start_offset => INTERVAL '3 hours', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '1 hour', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_continuous_aggregate_policy('snmp_interface_stats_hourly', start_offset => INTERVAL '3 hours', end_offset => INTERVAL '1 hour', schedule_interval => INTERVAL '1 hour', if_not_exists => true)"
|
||||
)
|
||||
|
||||
# Refresh daily aggregates every day at midnight, looking back 3 days
|
||||
execute(
|
||||
"SELECT add_continuous_aggregate_policy('monitoring_checks_daily', start_offset => INTERVAL '3 days', end_offset => INTERVAL '1 day', schedule_interval => INTERVAL '1 day', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_continuous_aggregate_policy('snmp_sensor_readings_daily', start_offset => INTERVAL '3 days', end_offset => INTERVAL '1 day', schedule_interval => INTERVAL '1 day', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_continuous_aggregate_policy('snmp_interface_stats_daily', start_offset => INTERVAL '3 days', end_offset => INTERVAL '1 day', schedule_interval => INTERVAL '1 day', if_not_exists => true)"
|
||||
)
|
||||
|
||||
# Step 11: Keep continuous aggregates longer than raw data
|
||||
# Keep hourly aggregates for 1 year, daily for 5 years
|
||||
execute(
|
||||
"SELECT add_retention_policy('monitoring_checks_hourly', INTERVAL '1 year', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_retention_policy('monitoring_checks_daily', INTERVAL '5 years', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_retention_policy('snmp_sensor_readings_hourly', INTERVAL '1 year', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_retention_policy('snmp_sensor_readings_daily', INTERVAL '5 years', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_retention_policy('snmp_interface_stats_hourly', INTERVAL '1 year', if_not_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT add_retention_policy('snmp_interface_stats_daily', INTERVAL '5 years', if_not_exists => true)"
|
||||
)
|
||||
|
||||
# Step 12: Do an initial refresh of the continuous aggregates with existing data
|
||||
# This populates them from any existing data in the tables
|
||||
execute("CALL refresh_continuous_aggregate('monitoring_checks_hourly', NULL, NULL)")
|
||||
execute("CALL refresh_continuous_aggregate('monitoring_checks_daily', NULL, NULL)")
|
||||
execute("CALL refresh_continuous_aggregate('snmp_sensor_readings_hourly', NULL, NULL)")
|
||||
execute("CALL refresh_continuous_aggregate('snmp_sensor_readings_daily', NULL, NULL)")
|
||||
execute("CALL refresh_continuous_aggregate('snmp_interface_stats_hourly', NULL, NULL)")
|
||||
execute("CALL refresh_continuous_aggregate('snmp_interface_stats_daily', NULL, NULL)")
|
||||
end
|
||||
|
||||
def down do
|
||||
# Remove refresh policies
|
||||
execute(
|
||||
"SELECT remove_continuous_aggregate_policy('monitoring_checks_hourly', if_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT remove_continuous_aggregate_policy('monitoring_checks_daily', if_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT remove_continuous_aggregate_policy('snmp_sensor_readings_hourly', if_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT remove_continuous_aggregate_policy('snmp_sensor_readings_daily', if_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT remove_continuous_aggregate_policy('snmp_interface_stats_hourly', if_exists => true)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"SELECT remove_continuous_aggregate_policy('snmp_interface_stats_daily', if_exists => true)"
|
||||
)
|
||||
|
||||
# Remove retention policies from continuous aggregates
|
||||
execute("SELECT remove_retention_policy('monitoring_checks_hourly', if_exists => true)")
|
||||
execute("SELECT remove_retention_policy('monitoring_checks_daily', if_exists => true)")
|
||||
execute("SELECT remove_retention_policy('snmp_sensor_readings_hourly', if_exists => true)")
|
||||
execute("SELECT remove_retention_policy('snmp_sensor_readings_daily', if_exists => true)")
|
||||
execute("SELECT remove_retention_policy('snmp_interface_stats_hourly', if_exists => true)")
|
||||
execute("SELECT remove_retention_policy('snmp_interface_stats_daily', if_exists => true)")
|
||||
|
||||
# Drop continuous aggregates
|
||||
execute("DROP MATERIALIZED VIEW IF EXISTS monitoring_checks_hourly CASCADE")
|
||||
execute("DROP MATERIALIZED VIEW IF EXISTS monitoring_checks_daily CASCADE")
|
||||
execute("DROP MATERIALIZED VIEW IF EXISTS snmp_sensor_readings_hourly CASCADE")
|
||||
execute("DROP MATERIALIZED VIEW IF EXISTS snmp_sensor_readings_daily CASCADE")
|
||||
execute("DROP MATERIALIZED VIEW IF EXISTS snmp_interface_stats_hourly CASCADE")
|
||||
execute("DROP MATERIALIZED VIEW IF EXISTS snmp_interface_stats_daily CASCADE")
|
||||
|
||||
# Remove retention policies from hypertables
|
||||
execute("SELECT remove_retention_policy('monitoring_checks', if_exists => true)")
|
||||
execute("SELECT remove_retention_policy('snmp_sensor_readings', if_exists => true)")
|
||||
execute("SELECT remove_retention_policy('snmp_interface_stats', if_exists => true)")
|
||||
|
||||
# Remove compression policies
|
||||
execute("SELECT remove_compression_policy('monitoring_checks', if_exists => true)")
|
||||
execute("SELECT remove_compression_policy('snmp_sensor_readings', if_exists => true)")
|
||||
execute("SELECT remove_compression_policy('snmp_interface_stats', if_exists => true)")
|
||||
|
||||
# Note: We don't convert hypertables back to regular tables as that would lose data
|
||||
# The tables will continue to function as hypertables but without the policies
|
||||
# To fully revert, you would need to recreate the tables from scratch
|
||||
|
||||
# Restore original indexes
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS monitoring_checks_equipment_id_index ON monitoring_checks (device_id)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS snmp_sensor_readings_sensor_id_index ON snmp_sensor_readings (sensor_id)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX IF NOT EXISTS snmp_interface_stats_interface_id_index ON snmp_interface_stats (interface_id)"
|
||||
)
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue