427 lines
16 KiB
Elixir
427 lines
16 KiB
Elixir
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
|
|
# Skip TimescaleDB in test environment where it's not available
|
|
if timescaledb_available?() do
|
|
do_up()
|
|
else
|
|
# Log and skip - tables already exist from earlier migrations, just without hypertable optimization
|
|
repo().query!("SELECT 1")
|
|
end
|
|
end
|
|
|
|
defp timescaledb_available? do
|
|
case repo().query("SELECT 1 FROM pg_available_extensions WHERE name = 'timescaledb'") do
|
|
{:ok, %{num_rows: n}} when n > 0 -> true
|
|
_ -> false
|
|
end
|
|
end
|
|
|
|
defp do_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
|
|
if timescaledb_available?() do
|
|
do_down()
|
|
end
|
|
end
|
|
|
|
defp do_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
|