From 3f0f6e7d5f4e09a36a10cdf2569c7d47f927f303 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Tue, 24 Mar 2026 15:30:43 -0500 Subject: [PATCH] feat: add per-organization data retention with 1-year default (#144) Add data_retention_days field to organizations (default 365, range 30-730). DataRetentionWorker runs nightly at 1 AM to purge expired time-series data across 8 tables, respecting each org's retention setting. TimescaleDB global retention bumped from 90 to 730 days as safety net. Resolved alerts are cleaned up; unresolved alerts kept regardless of age. Reviewed-on: https://git.mcintire.me/graham/towerops-web/pulls/144 --- config/runtime.exs | 4 +- lib/towerops/organizations/organization.ex | 12 +- lib/towerops/workers/data_retention_worker.ex | 174 ++++++++++ ...d_data_retention_days_to_organizations.exs | 9 + ...date_timescaledb_retention_to_730_days.exs | 59 ++++ .../workers/data_retention_worker_test.exs | 303 ++++++++++++++++++ 6 files changed, 559 insertions(+), 2 deletions(-) create mode 100644 lib/towerops/workers/data_retention_worker.ex create mode 100644 priv/repo/migrations/20260324193531_add_data_retention_days_to_organizations.exs create mode 100644 priv/repo/migrations/20260324193820_update_timescaledb_retention_to_730_days.exs create mode 100644 test/towerops/workers/data_retention_worker_test.exs diff --git a/config/runtime.exs b/config/runtime.exs index ce0f240d..9358b978 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -252,7 +252,9 @@ if config_env() == :prod do # Backhaul capacity utilization insights every 15 minutes {"*/15 * * * *", Towerops.Workers.CapacityInsightWorker}, # Sync device usage to Stripe daily at 3 AM UTC - {"0 3 * * *", Towerops.Workers.BillingSyncWorker} + {"0 3 * * *", Towerops.Workers.BillingSyncWorker}, + # Data retention cleanup nightly at 1 AM + {"0 1 * * *", Towerops.Workers.DataRetentionWorker} ]}, # Automatically delete completed jobs after 60 seconds. # Default limit of 10k/30s (333/s) can't keep up with job throughput (~600+/s), diff --git a/lib/towerops/organizations/organization.ex b/lib/towerops/organizations/organization.ex index 1fd8f3c7..15de4b9a 100644 --- a/lib/towerops/organizations/organization.ex +++ b/lib/towerops/organizations/organization.ex @@ -66,6 +66,9 @@ defmodule Towerops.Organizations.Organization do field :last_billing_sync_at, :utc_datetime field :last_synced_device_count, :integer + # Data retention + field :data_retention_days, :integer, default: 365 + # Virtual fields (populated by queries) field :device_count, :integer, virtual: true @@ -112,6 +115,7 @@ defmodule Towerops.Organizations.Organization do payment_method_status: String.t() | nil, last_billing_sync_at: DateTime.t() | nil, last_synced_device_count: integer() | nil, + data_retention_days: integer(), default_agent_token_id: Ecto.UUID.t() | nil, default_agent_token: NotLoaded.t() | AgentToken.t() | nil, default_escalation_policy_id: Ecto.UUID.t() | nil, @@ -155,7 +159,8 @@ defmodule Towerops.Organizations.Organization do :last_billing_sync_at, :last_synced_device_count, :default_escalation_policy_id, - :alert_routing + :alert_routing, + :data_retention_days ]) |> validate_required([:name]) |> validate_length(:name, min: 2, max: 100) @@ -165,6 +170,11 @@ defmodule Towerops.Organizations.Organization do ) |> validate_inclusion(:snmp_version, ["1", "2c", "3"], message: "must be 1, 2c, or 3") |> validate_number(:snmp_port, greater_than: 0, less_than: 65_536) + |> validate_number(:data_retention_days, + greater_than_or_equal_to: 30, + less_than_or_equal_to: 730, + message: "must be between 30 and 730 days" + ) |> validate_snmpv3_fields() |> validate_mikrotik_fields() |> generate_slug() diff --git a/lib/towerops/workers/data_retention_worker.ex b/lib/towerops/workers/data_retention_worker.ex new file mode 100644 index 00000000..9e4181be --- /dev/null +++ b/lib/towerops/workers/data_retention_worker.ex @@ -0,0 +1,174 @@ +defmodule Towerops.Workers.DataRetentionWorker do + @moduledoc """ + Oban worker for per-organization data retention. + + Runs nightly to delete time-series data older than each organization's + `data_retention_days` setting (default: 365 days). + + Tables cleaned: + - monitoring_checks (via device_id -> devices.organization_id) + - snmp_sensor_readings (via sensor -> snmp_device -> device -> org) + - snmp_interface_stats (via interface -> snmp_device -> device -> org) + - snmp_processor_readings (via processor -> snmp_device -> device -> org) + - snmp_storage_readings (via storage -> snmp_device -> device -> org) + - check_results (direct organization_id) + - wireless_client_readings (direct organization_id) + - alerts (direct organization_id, only resolved alerts) + + TimescaleDB global retention (730 days) acts as a safety net. + """ + use Oban.Worker, queue: :maintenance + + alias Towerops.Repo + + require Logger + + @impl Oban.Worker + def perform(_job) do + retention_groups = fetch_retention_groups() + + results = + Enum.flat_map(retention_groups, fn {retention_days, org_ids} -> + cutoff = DateTime.add(DateTime.utc_now(), -retention_days, :day) + purge_for_orgs(org_ids, cutoff) + end) + + total = Enum.reduce(results, 0, fn {_table, count}, acc -> acc + count end) + + if total > 0 do + summary = Enum.reject(results, fn {_table, count} -> count == 0 end) + Logger.info("Data retention: deleted #{total} total rows — #{inspect(summary)}") + end + + {:ok, %{total_deleted: total, details: Map.new(results)}} + end + + defp fetch_retention_groups do + result = + Repo.query!(""" + SELECT data_retention_days, array_agg(id) AS org_ids + FROM organizations + GROUP BY data_retention_days + """) + + Enum.map(result.rows, fn [days, ids] -> + # Keep raw binary UUIDs from PostgreSQL — Postgrex expects binary format + # for uuid[] parameters, not string-formatted UUIDs + {days, ids} + end) + end + + defp purge_for_orgs(org_ids, cutoff) do + # Tables with direct organization_id + direct_results = + Enum.map( + [{"check_results", "checked_at"}, {"wireless_client_readings", "checked_at"}], + fn {table, time_col} -> + result = + Repo.query!( + "DELETE FROM #{table} WHERE organization_id = ANY($1::uuid[]) AND #{time_col} < $2", + [org_ids, cutoff] + ) + + {table, result.num_rows} + end + ) + + # Resolved alerts only (keep unresolved regardless of age) + alert_result = + Repo.query!( + """ + DELETE FROM alerts + WHERE organization_id = ANY($1::uuid[]) + AND inserted_at < $2 + AND resolved_at IS NOT NULL + """, + [org_ids, cutoff] + ) + + # monitoring_checks: device_id -> devices.organization_id + mc_result = + Repo.query!( + """ + DELETE FROM monitoring_checks + WHERE checked_at < $2 + AND device_id IN (SELECT id FROM devices WHERE organization_id = ANY($1::uuid[])) + """, + [org_ids, cutoff] + ) + + # snmp_sensor_readings: sensor_id -> snmp_sensors -> snmp_devices -> devices -> org + sr_result = + Repo.query!( + """ + DELETE FROM snmp_sensor_readings + WHERE checked_at < $2 + AND sensor_id IN ( + SELECT s.id FROM snmp_sensors s + JOIN snmp_devices sd ON sd.id = s.snmp_device_id + JOIN devices d ON d.id = sd.device_id + WHERE d.organization_id = ANY($1::uuid[]) + ) + """, + [org_ids, cutoff] + ) + + # snmp_interface_stats: interface_id -> snmp_interfaces -> snmp_devices -> devices -> org + is_result = + Repo.query!( + """ + DELETE FROM snmp_interface_stats + WHERE checked_at < $2 + AND interface_id IN ( + SELECT i.id FROM snmp_interfaces i + JOIN snmp_devices sd ON sd.id = i.snmp_device_id + JOIN devices d ON d.id = sd.device_id + WHERE d.organization_id = ANY($1::uuid[]) + ) + """, + [org_ids, cutoff] + ) + + # snmp_processor_readings: processor_id -> snmp_processors -> snmp_devices -> devices -> org + pr_result = + Repo.query!( + """ + DELETE FROM snmp_processor_readings + WHERE checked_at < $2 + AND processor_id IN ( + SELECT p.id FROM snmp_processors p + JOIN snmp_devices sd ON sd.id = p.snmp_device_id + JOIN devices d ON d.id = sd.device_id + WHERE d.organization_id = ANY($1::uuid[]) + ) + """, + [org_ids, cutoff] + ) + + # snmp_storage_readings: storage_id -> snmp_storage -> snmp_devices -> devices -> org + str_result = + Repo.query!( + """ + DELETE FROM snmp_storage_readings + WHERE checked_at < $2 + AND storage_id IN ( + SELECT st.id FROM snmp_storage st + JOIN snmp_devices sd ON sd.id = st.snmp_device_id + JOIN devices d ON d.id = sd.device_id + WHERE d.organization_id = ANY($1::uuid[]) + ) + """, + [org_ids, cutoff] + ) + + direct_results ++ + [ + {"alerts_resolved", alert_result.num_rows}, + {"monitoring_checks", mc_result.num_rows}, + {"snmp_sensor_readings", sr_result.num_rows}, + {"snmp_interface_stats", is_result.num_rows}, + {"snmp_processor_readings", pr_result.num_rows}, + {"snmp_storage_readings", str_result.num_rows} + ] + end +end diff --git a/priv/repo/migrations/20260324193531_add_data_retention_days_to_organizations.exs b/priv/repo/migrations/20260324193531_add_data_retention_days_to_organizations.exs new file mode 100644 index 00000000..6d996c0d --- /dev/null +++ b/priv/repo/migrations/20260324193531_add_data_retention_days_to_organizations.exs @@ -0,0 +1,9 @@ +defmodule Towerops.Repo.Migrations.AddDataRetentionDaysToOrganizations do + use Ecto.Migration + + def change do + alter table(:organizations) do + add :data_retention_days, :integer, default: 365, null: false + end + end +end diff --git a/priv/repo/migrations/20260324193820_update_timescaledb_retention_to_730_days.exs b/priv/repo/migrations/20260324193820_update_timescaledb_retention_to_730_days.exs new file mode 100644 index 00000000..ca470757 --- /dev/null +++ b/priv/repo/migrations/20260324193820_update_timescaledb_retention_to_730_days.exs @@ -0,0 +1,59 @@ +defmodule Towerops.Repo.Migrations.UpdateTimescaledbRetentionTo730Days do + @moduledoc """ + Updates TimescaleDB retention policies from 90 days to 730 days. + + Per-organization retention is now handled by DataRetentionWorker (default 365 days). + The 730-day TimescaleDB policy acts as a safety net for orphaned data from + deleted organizations or data that the worker misses. + """ + use Ecto.Migration + + @disable_ddl_transaction true + @disable_migration_lock true + + @hypertables ~w(monitoring_checks snmp_sensor_readings snmp_interface_stats wireless_client_readings) + + def up do + if timescaledb_available?() do + # Update existing retention policies from 90 days to 730 days + for table <- @hypertables do + execute("SELECT remove_retention_policy('#{table}', if_exists => true)") + + execute( + "SELECT add_retention_policy('#{table}', INTERVAL '730 days', if_not_exists => true)" + ) + end + + # Add missing retention policy for check_results + execute( + "SELECT add_retention_policy('check_results', INTERVAL '730 days', if_not_exists => true)" + ) + end + end + + def down do + if timescaledb_available?() do + # Revert to original 90-day retention + for table <- @hypertables do + execute("SELECT remove_retention_policy('#{table}', if_exists => true)") + + execute( + "SELECT add_retention_policy('#{table}', INTERVAL '90 days', if_not_exists => true)" + ) + end + + # Remove check_results retention (it had none before) + execute("SELECT remove_retention_policy('check_results', if_exists => true)") + end + end + + defp timescaledb_available? do + # Check for actual hypertables, not just the extension being installed. + # The extension may be available but tables might not be hypertables + # (e.g., CI/staging environments that skip hypertable creation). + case repo().query("SELECT 1 FROM timescaledb_information.hypertables LIMIT 1") do + {:ok, %{num_rows: n}} when n > 0 -> true + _ -> false + end + end +end diff --git a/test/towerops/workers/data_retention_worker_test.exs b/test/towerops/workers/data_retention_worker_test.exs new file mode 100644 index 00000000..3f41c081 --- /dev/null +++ b/test/towerops/workers/data_retention_worker_test.exs @@ -0,0 +1,303 @@ +defmodule Towerops.Workers.DataRetentionWorkerTest do + use Towerops.DataCase, async: false + + import Towerops.AccountsFixtures + import Towerops.DevicesFixtures + import Towerops.OrganizationsFixtures + + alias Towerops.Organizations + alias Towerops.Repo + alias Towerops.Workers.DataRetentionWorker + + describe "perform/1" do + test "deletes monitoring_checks older than org retention period" do + user = user_fixture() + org = organization_fixture(user.id) + {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30}) + device = device_fixture(%{organization_id: org.id}) + + now = DateTime.truncate(DateTime.utc_now(), :second) + old_time = DateTime.add(now, -31, :day) + recent_time = DateTime.add(now, -10, :day) + + old_id = Ecto.UUID.generate() + recent_id = Ecto.UUID.generate() + + Repo.insert_all("monitoring_checks", [ + %{ + id: Ecto.UUID.dump!(old_id), + device_id: Ecto.UUID.dump!(device.id), + status: "success", + checked_at: old_time, + inserted_at: old_time + }, + %{ + id: Ecto.UUID.dump!(recent_id), + device_id: Ecto.UUID.dump!(device.id), + status: "success", + checked_at: recent_time, + inserted_at: recent_time + } + ]) + + assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}}) + assert result.total_deleted >= 1 + + # Old record should be deleted + remaining = + Repo.all( + from(m in "monitoring_checks", + where: m.device_id == type(^device.id, :binary_id), + select: m.id + ) + ) + + remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1) + refute old_id in remaining_uuids + assert recent_id in remaining_uuids + end + + test "deletes check_results older than org retention (direct org_id)" do + user = user_fixture() + org = organization_fixture(user.id) + {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30}) + + now = DateTime.truncate(DateTime.utc_now(), :second) + old_time = DateTime.add(now, -31, :day) + recent_time = DateTime.add(now, -10, :day) + + # Create a check for the org + {:ok, check} = + Towerops.Monitoring.create_check(%{ + organization_id: org.id, + check_type: "http", + name: "Test Check", + interval_seconds: 60, + config: %{"url" => "http://example.com"} + }) + + old_id = Ecto.UUID.generate() + recent_id = Ecto.UUID.generate() + + Repo.insert_all("check_results", [ + %{ + id: Ecto.UUID.dump!(old_id), + organization_id: Ecto.UUID.dump!(org.id), + check_id: Ecto.UUID.dump!(check.id), + status: 0, + checked_at: old_time + }, + %{ + id: Ecto.UUID.dump!(recent_id), + organization_id: Ecto.UUID.dump!(org.id), + check_id: Ecto.UUID.dump!(check.id), + status: 0, + checked_at: recent_time + } + ]) + + assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}}) + assert result.total_deleted >= 1 + + remaining = + Repo.all( + from(cr in "check_results", + where: cr.organization_id == type(^org.id, :binary_id), + select: cr.id + ) + ) + + remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1) + refute old_id in remaining_uuids + assert recent_id in remaining_uuids + end + + test "respects different retention periods per org" do + user1 = user_fixture() + org1 = organization_fixture(user1.id) + {:ok, org1} = Organizations.update_organization(org1, %{data_retention_days: 60}) + device1 = device_fixture(%{organization_id: org1.id}) + + user2 = user_fixture() + org2 = organization_fixture(user2.id) + {:ok, org2} = Organizations.update_organization(org2, %{data_retention_days: 90}) + device2 = device_fixture(%{organization_id: org2.id}) + + now = DateTime.truncate(DateTime.utc_now(), :second) + # 75 days old: past org1's 60-day retention, within org2's 90-day retention + mid_time = DateTime.add(now, -75, :day) + + org1_check_id = Ecto.UUID.generate() + org2_check_id = Ecto.UUID.generate() + + Repo.insert_all("monitoring_checks", [ + %{ + id: Ecto.UUID.dump!(org1_check_id), + device_id: Ecto.UUID.dump!(device1.id), + status: "success", + checked_at: mid_time, + inserted_at: mid_time + }, + %{ + id: Ecto.UUID.dump!(org2_check_id), + device_id: Ecto.UUID.dump!(device2.id), + status: "success", + checked_at: mid_time, + inserted_at: mid_time + } + ]) + + assert {:ok, _result} = DataRetentionWorker.perform(%Oban.Job{args: %{}}) + + # org1 record should be deleted (75 days > 60 day retention) + org1_remaining = + Repo.all( + from(m in "monitoring_checks", + where: m.device_id == type(^device1.id, :binary_id), + select: m.id + ) + ) + + assert org1_remaining == [] + + # org2 record should remain (75 days < 90 day retention) + org2_remaining = + Repo.all( + from(m in "monitoring_checks", + where: m.device_id == type(^device2.id, :binary_id), + select: m.id + ) + ) + + org2_remaining_uuids = Enum.map(org2_remaining, &Ecto.UUID.cast!/1) + assert org2_check_id in org2_remaining_uuids + end + + test "does NOT delete data within retention window" do + user = user_fixture() + org = organization_fixture(user.id) + # Default 365-day retention + device = device_fixture(%{organization_id: org.id}) + + now = DateTime.truncate(DateTime.utc_now(), :second) + recent_time = DateTime.add(now, -100, :day) + + recent_id = Ecto.UUID.generate() + + Repo.insert_all("monitoring_checks", [ + %{ + id: Ecto.UUID.dump!(recent_id), + device_id: Ecto.UUID.dump!(device.id), + status: "success", + checked_at: recent_time, + inserted_at: recent_time + } + ]) + + assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}}) + + remaining = + Repo.all( + from(m in "monitoring_checks", + where: m.device_id == type(^device.id, :binary_id), + select: m.id + ) + ) + + remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1) + assert recent_id in remaining_uuids + assert result.total_deleted == 0 + end + + test "returns summary with deletion counts" do + user = user_fixture() + org = organization_fixture(user.id) + {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30}) + device = device_fixture(%{organization_id: org.id}) + + now = DateTime.truncate(DateTime.utc_now(), :second) + old_time = DateTime.add(now, -31, :day) + + Repo.insert_all("monitoring_checks", [ + %{ + id: Ecto.UUID.dump!(Ecto.UUID.generate()), + device_id: Ecto.UUID.dump!(device.id), + status: "success", + checked_at: old_time, + inserted_at: old_time + }, + %{ + id: Ecto.UUID.dump!(Ecto.UUID.generate()), + device_id: Ecto.UUID.dump!(device.id), + status: "success", + checked_at: old_time, + inserted_at: old_time + } + ]) + + assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}}) + assert result.total_deleted >= 2 + assert is_map(result.details) + assert Map.get(result.details, "monitoring_checks") >= 2 + end + + test "only deletes resolved alerts, keeps unresolved regardless of age" do + user = user_fixture() + org = organization_fixture(user.id) + {:ok, org} = Organizations.update_organization(org, %{data_retention_days: 30}) + device = device_fixture(%{organization_id: org.id}) + + now = DateTime.truncate(DateTime.utc_now(), :second) + old_time = DateTime.add(now, -31, :day) + + resolved_id = Ecto.UUID.generate() + unresolved_id = Ecto.UUID.generate() + + Repo.insert_all("alerts", [ + %{ + id: Ecto.UUID.dump!(resolved_id), + organization_id: Ecto.UUID.dump!(org.id), + device_id: Ecto.UUID.dump!(device.id), + alert_type: "device_down", + triggered_at: old_time, + resolved_at: old_time, + storm_suppressed: false, + notification_sent: false, + inserted_at: old_time, + updated_at: old_time + }, + %{ + id: Ecto.UUID.dump!(unresolved_id), + organization_id: Ecto.UUID.dump!(org.id), + device_id: Ecto.UUID.dump!(device.id), + alert_type: "device_down", + triggered_at: old_time, + resolved_at: nil, + storm_suppressed: false, + notification_sent: false, + inserted_at: old_time, + updated_at: old_time + } + ]) + + assert {:ok, _result} = DataRetentionWorker.perform(%Oban.Job{args: %{}}) + + remaining = + Repo.all( + from(a in "alerts", + where: a.organization_id == type(^org.id, :binary_id), + select: a.id + ) + ) + + remaining_uuids = Enum.map(remaining, &Ecto.UUID.cast!/1) + refute resolved_id in remaining_uuids + assert unresolved_id in remaining_uuids + end + + test "works when no organizations exist" do + assert {:ok, result} = DataRetentionWorker.perform(%Oban.Job{args: %{}}) + assert result.total_deleted == 0 + end + end +end