diff --git a/config/dev.exs b/config/dev.exs index 6dcea6b0..a6f07b85 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -81,7 +81,9 @@ config :towerops, Oban, # Delete stale violation records daily at 2 AM {"0 2 * * *", Towerops.Workers.StaleViolationCleanupWorker}, # Sync Preseem data every 10 minutes - {"*/10 * * * *", Towerops.Workers.PreseemSyncWorker} + {"*/10 * * * *", Towerops.Workers.PreseemSyncWorker}, + # Compute Preseem baselines and fleet profiles nightly at 2:30 AM + {"30 2 * * *", Towerops.Workers.PreseemBaselineWorker} ]}, # Automatically delete completed jobs after 60 seconds {Oban.Plugins.Pruner, max_age: 60}, diff --git a/config/runtime.exs b/config/runtime.exs index 520fabf2..1c76fdfb 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -194,7 +194,9 @@ if config_env() == :prod do # Delete stale violation records daily at 2 AM {"0 2 * * *", Towerops.Workers.StaleViolationCleanupWorker}, # Sync Preseem data every 10 minutes - {"*/10 * * * *", Towerops.Workers.PreseemSyncWorker} + {"*/10 * * * *", Towerops.Workers.PreseemSyncWorker}, + # Compute Preseem baselines and fleet profiles nightly at 2:30 AM + {"30 2 * * *", Towerops.Workers.PreseemBaselineWorker} ]}, # Automatically delete completed jobs after 60 seconds {Oban.Plugins.Pruner, max_age: 60}, diff --git a/lib/towerops/preseem/baseline.ex b/lib/towerops/preseem/baseline.ex new file mode 100644 index 00000000..e9147da9 --- /dev/null +++ b/lib/towerops/preseem/baseline.ex @@ -0,0 +1,107 @@ +defmodule Towerops.Preseem.Baseline do + @moduledoc """ + Computes rolling 14-day baselines for individual Preseem access points. + """ + import Ecto.Query + + alias Towerops.Preseem.AccessPoint + alias Towerops.Preseem.DeviceBaseline + alias Towerops.Preseem.SubscriberMetric + alias Towerops.Repo + + @baseline_days 14 + @metrics ~w(avg_latency avg_jitter avg_loss avg_throughput p95_latency subscriber_count) + + @doc "Compute baselines for all matched APs in an organization." + def compute_baselines(organization_id) do + cutoff = DateTime.add(DateTime.utc_now(), -@baseline_days * 86_400, :second) + now = DateTime.truncate(DateTime.utc_now(), :second) + + matched_aps = + AccessPoint + |> where(organization_id: ^organization_id) + |> where([ap], not is_nil(ap.device_id)) + |> Repo.all() + + count = + Enum.reduce(matched_aps, 0, fn ap, acc -> + metrics = get_metrics_in_window(ap.id, cutoff) + + if length(metrics) >= 3 do + compute_and_upsert_baselines(ap.id, metrics, now) + acc + 1 + else + acc + end + end) + + {:ok, count} + end + + defp get_metrics_in_window(ap_id, cutoff) do + SubscriberMetric + |> where(preseem_access_point_id: ^ap_id) + |> where([m], m.recorded_at >= ^cutoff) + |> order_by(:recorded_at) + |> Repo.all() + end + + defp compute_and_upsert_baselines(ap_id, metrics, now) do + for metric_name <- @metrics do + values = + metrics + |> Enum.map(&Map.get(&1, String.to_existing_atom(metric_name))) + |> Enum.reject(&is_nil/1) + + if length(values) >= 3 do + stats = compute_stats(values) + + upsert_baseline(ap_id, %{ + metric_name: metric_name, + period: "all", + mean: stats.mean, + stddev: stats.stddev, + p5: stats.p5, + p95: stats.p95, + sample_count: length(values), + computed_at: now + }) + end + end + end + + defp compute_stats(values) do + sorted = Enum.sort(values) + n = length(sorted) + mean = Enum.sum(sorted) / n + + variance = + sorted + |> Enum.map(fn v -> (v - mean) * (v - mean) end) + |> Enum.sum() + |> Kernel./(max(n - 1, 1)) + + stddev = :math.sqrt(variance) + p5_idx = max(round(n * 0.05) - 1, 0) + p95_idx = min(round(n * 0.95) - 1, n - 1) + + %{ + mean: Float.round(mean / 1, 4), + stddev: Float.round(stddev / 1, 4), + p5: sorted |> Enum.at(p5_idx) |> to_float(), + p95: sorted |> Enum.at(p95_idx) |> to_float() + } + end + + defp to_float(v) when is_float(v), do: Float.round(v, 4) + defp to_float(v) when is_integer(v), do: v * 1.0 + + defp upsert_baseline(ap_id, attrs) do + %DeviceBaseline{} + |> DeviceBaseline.changeset(Map.put(attrs, :preseem_access_point_id, ap_id)) + |> Repo.insert( + on_conflict: {:replace_all_except, [:id, :preseem_access_point_id]}, + conflict_target: [:preseem_access_point_id, :metric_name, :period] + ) + end +end diff --git a/lib/towerops/preseem/fleet_intelligence.ex b/lib/towerops/preseem/fleet_intelligence.ex new file mode 100644 index 00000000..0d22dc7b --- /dev/null +++ b/lib/towerops/preseem/fleet_intelligence.ex @@ -0,0 +1,109 @@ +defmodule Towerops.Preseem.FleetIntelligence do + @moduledoc """ + Computes fleet-wide model performance profiles by grouping matched devices + by manufacturer and model. + """ + import Ecto.Query + + alias Towerops.Preseem.AccessPoint + alias Towerops.Preseem.FleetProfile + alias Towerops.Repo + + @doc "Compute fleet profiles for all matched APs in an organization." + def compute_fleet_profiles(organization_id) do + now = DateTime.truncate(DateTime.utc_now(), :second) + + matched_aps = + AccessPoint + |> where(organization_id: ^organization_id) + |> where([ap], not is_nil(ap.device_id)) + |> where([ap], not is_nil(ap.model)) + |> Repo.all() + + groups = + Enum.group_by(matched_aps, fn ap -> + {extract_manufacturer(ap.model), ap.model} + end) + + count = + Enum.reduce(groups, 0, fn {{manufacturer, model}, aps}, acc -> + profile_attrs = compute_model_profile(manufacturer, model, aps, now) + upsert_fleet_profile(organization_id, profile_attrs) + acc + 1 + end) + + {:ok, count} + end + + defp extract_manufacturer(model) when is_binary(model) do + case String.split(model, " ", parts: 2) do + [manufacturer, _rest] -> manufacturer + [single] -> single + end + end + + defp compute_model_profile(manufacturer, model, aps, now) do + subscriber_counts = aps |> Enum.map(& &1.subscriber_count) |> Enum.reject(&is_nil/1) + qoe_scores = aps |> Enum.map(& &1.qoe_score) |> Enum.reject(&is_nil/1) + capacity_scores = aps |> Enum.map(& &1.capacity_score) |> Enum.reject(&is_nil/1) + busy_hours = aps |> Enum.map(& &1.busy_hours) |> Enum.reject(&is_nil/1) + + %{ + manufacturer: manufacturer, + model: model, + firmware_version: nil, + device_count: length(aps), + avg_subscribers: safe_avg(subscriber_counts), + avg_qoe_score: safe_avg(qoe_scores), + avg_capacity_score: safe_avg(capacity_scores), + capacity_ceiling: compute_capacity_ceiling(aps), + avg_busy_hours: safe_avg(busy_hours), + performance_data: %{ + "subscriber_range" => %{ + "min" => Enum.min(subscriber_counts, fn -> nil end), + "max" => Enum.max(subscriber_counts, fn -> nil end) + }, + "firmware_distribution" => aps |> Enum.map(& &1.firmware) |> Enum.reject(&is_nil/1) |> Enum.frequencies() + }, + computed_at: now + } + end + + defp safe_avg([]), do: nil + defp safe_avg(values), do: Float.round(Enum.sum(values) / length(values), 2) + + defp compute_capacity_ceiling(aps) do + aps + |> Enum.filter(fn ap -> ap.qoe_score != nil and ap.subscriber_count != nil end) + |> Enum.filter(fn ap -> ap.qoe_score >= 70 end) + |> Enum.map(& &1.subscriber_count) + |> Enum.max(fn -> nil end) + end + + # Use a manual query + update/insert since the unique index includes nullable firmware_version. + # PostgreSQL treats NULLs as distinct in unique indexes, so ON CONFLICT won't match + # rows with NULL firmware_version. Ecto's get_by also rejects nil values. + defp upsert_fleet_profile(organization_id, attrs) do + full_attrs = Map.put(attrs, :organization_id, organization_id) + + existing = + FleetProfile + |> where(organization_id: ^organization_id) + |> where(manufacturer: ^attrs.manufacturer) + |> where(model: ^attrs.model) + |> where([fp], is_nil(fp.firmware_version)) + |> Repo.one() + + case existing do + nil -> + %FleetProfile{} + |> FleetProfile.changeset(full_attrs) + |> Repo.insert!() + + profile -> + profile + |> FleetProfile.changeset(full_attrs) + |> Repo.update!() + end + end +end diff --git a/lib/towerops/workers/preseem_baseline_worker.ex b/lib/towerops/workers/preseem_baseline_worker.ex new file mode 100644 index 00000000..fe48b2bd --- /dev/null +++ b/lib/towerops/workers/preseem_baseline_worker.ex @@ -0,0 +1,36 @@ +defmodule Towerops.Workers.PreseemBaselineWorker do + @moduledoc """ + Nightly Oban cron worker that computes device baselines and fleet profiles + for all organizations with enabled Preseem integrations. + """ + use Oban.Worker, queue: :maintenance + + alias Towerops.Integrations + alias Towerops.Preseem.Baseline + alias Towerops.Preseem.FleetIntelligence + + require Logger + + @impl Oban.Worker + def perform(%Oban.Job{}) do + integrations = Integrations.list_enabled_integrations("preseem") + + Enum.each(integrations, fn integration -> + org_id = integration.organization_id + compute_for_org(org_id) + end) + + :ok + end + + defp compute_for_org(org_id) do + {:ok, baseline_count} = Baseline.compute_baselines(org_id) + Logger.info("Preseem baselines computed for org #{org_id}: #{baseline_count} devices") + + {:ok, profile_count} = FleetIntelligence.compute_fleet_profiles(org_id) + Logger.info("Preseem fleet profiles computed for org #{org_id}: #{profile_count} models") + rescue + error -> + Logger.error("Preseem baseline/fleet computation failed for org #{org_id}: #{inspect(error)}") + end +end diff --git a/test/towerops/preseem/baseline_test.exs b/test/towerops/preseem/baseline_test.exs new file mode 100644 index 00000000..512f9497 --- /dev/null +++ b/test/towerops/preseem/baseline_test.exs @@ -0,0 +1,204 @@ +defmodule Towerops.Preseem.BaselineTest do + use Towerops.DataCase, async: true + + import Towerops.AccountsFixtures + import Towerops.DevicesFixtures + import Towerops.OrganizationsFixtures + + alias Towerops.Preseem.AccessPoint + alias Towerops.Preseem.Baseline + alias Towerops.Preseem.DeviceBaseline + alias Towerops.Preseem.SubscriberMetric + + setup do + user = user_fixture() + org = organization_fixture(user.id) + %{org: org} + end + + describe "compute_baselines/1" do + test "computes baselines for matched APs with sufficient data", %{org: org} do + ap = insert_matched_ap(org) + + for i <- 1..5 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 10.0 + i, + p95_latency: 20.0 + i, + subscriber_count: 40 + i, + recorded_at: hours_ago(i) + }) + end + + assert {:ok, 1} = Baseline.compute_baselines(org.id) + + baselines = Repo.all(DeviceBaseline) + assert length(baselines) > 0 + + latency_baseline = Enum.find(baselines, &(&1.metric_name == "avg_latency")) + assert latency_baseline + assert latency_baseline.sample_count == 5 + assert latency_baseline.period == "all" + assert latency_baseline.mean + assert latency_baseline.stddev + assert latency_baseline.p5 + assert latency_baseline.p95 + end + + test "skips APs with fewer than 3 data points", %{org: org} do + ap = insert_matched_ap(org) + + for i <- 1..2 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 10.0 + i, + recorded_at: hours_ago(i) + }) + end + + assert {:ok, 0} = Baseline.compute_baselines(org.id) + assert Repo.all(DeviceBaseline) == [] + end + + test "skips unmatched APs", %{org: org} do + ap = insert_unmatched_ap(org.id) + + for i <- 1..5 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 10.0 + i, + recorded_at: hours_ago(i) + }) + end + + assert {:ok, 0} = Baseline.compute_baselines(org.id) + end + + test "handles empty organization", %{org: org} do + assert {:ok, 0} = Baseline.compute_baselines(org.id) + end + + test "only includes metrics within the 14-day window", %{org: org} do + ap = insert_matched_ap(org) + + # 3 recent metrics (within window) + for i <- 1..3 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 10.0 + i, + recorded_at: hours_ago(i) + }) + end + + # 2 old metrics (outside 14-day window) + for i <- 1..2 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 100.0 + i, + recorded_at: days_ago(15 + i) + }) + end + + assert {:ok, 1} = Baseline.compute_baselines(org.id) + + latency_baseline = + DeviceBaseline + |> Repo.all() + |> Enum.find(&(&1.metric_name == "avg_latency")) + + # Should only have 3 samples from the recent window + assert latency_baseline.sample_count == 3 + end + + test "upserts baselines on recomputation", %{org: org} do + ap = insert_matched_ap(org) + + for i <- 1..5 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 10.0 + i, + recorded_at: hours_ago(i) + }) + end + + assert {:ok, 1} = Baseline.compute_baselines(org.id) + initial_count = length(Repo.all(DeviceBaseline)) + + # Recompute - should update existing baselines, not create duplicates + assert {:ok, 1} = Baseline.compute_baselines(org.id) + assert length(Repo.all(DeviceBaseline)) == initial_count + end + + test "computes baselines for multiple matched APs", %{org: org} do + for _idx <- 1..3 do + ap = insert_matched_ap(org) + + for i <- 1..4 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 10.0 + i, + recorded_at: hours_ago(i) + }) + end + end + + assert {:ok, 3} = Baseline.compute_baselines(org.id) + end + + test "skips metrics with nil values for a given metric name", %{org: org} do + ap = insert_matched_ap(org) + + # Insert metrics where avg_jitter is nil but avg_latency has values + for i <- 1..5 do + insert_subscriber_metric(ap.id, %{ + avg_latency: 10.0 + i, + avg_jitter: nil, + recorded_at: hours_ago(i) + }) + end + + assert {:ok, 1} = Baseline.compute_baselines(org.id) + + baselines = Repo.all(DeviceBaseline) + latency_baseline = Enum.find(baselines, &(&1.metric_name == "avg_latency")) + jitter_baseline = Enum.find(baselines, &(&1.metric_name == "avg_jitter")) + + assert latency_baseline + # avg_jitter should not have a baseline since all values are nil (< 3 non-nil) + assert jitter_baseline == nil + end + end + + # Helpers + + defp insert_matched_ap(org) do + device = device_fixture(%{organization_id: org.id}) + + %AccessPoint{} + |> AccessPoint.changeset(%{ + organization_id: org.id, + preseem_id: "ap-#{System.unique_integer([:positive])}", + name: "Test AP", + device_id: device.id, + match_confidence: "auto_ip" + }) + |> Repo.insert!() + end + + defp insert_unmatched_ap(org_id) do + %AccessPoint{} + |> AccessPoint.changeset(%{ + organization_id: org_id, + preseem_id: "ap-unmatched-#{System.unique_integer([:positive])}", + name: "Unmatched AP" + }) + |> Repo.insert!() + end + + defp insert_subscriber_metric(ap_id, attrs) do + %SubscriberMetric{} + |> SubscriberMetric.changeset(Map.put(attrs, :preseem_access_point_id, ap_id)) + |> Repo.insert!() + end + + defp hours_ago(n) do + DateTime.utc_now() |> DateTime.add(-n * 3600, :second) |> DateTime.truncate(:second) + end + + defp days_ago(n) do + DateTime.utc_now() |> DateTime.add(-n * 86_400, :second) |> DateTime.truncate(:second) + end +end diff --git a/test/towerops/preseem/fleet_intelligence_test.exs b/test/towerops/preseem/fleet_intelligence_test.exs new file mode 100644 index 00000000..f4fc88df --- /dev/null +++ b/test/towerops/preseem/fleet_intelligence_test.exs @@ -0,0 +1,187 @@ +defmodule Towerops.Preseem.FleetIntelligenceTest do + use Towerops.DataCase, async: true + + import Towerops.AccountsFixtures + import Towerops.DevicesFixtures + import Towerops.OrganizationsFixtures + + alias Towerops.Preseem.AccessPoint + alias Towerops.Preseem.FleetIntelligence + alias Towerops.Preseem.FleetProfile + + setup do + user = user_fixture() + org = organization_fixture(user.id) + %{org: org} + end + + describe "compute_fleet_profiles/1" do + test "computes profiles grouped by model", %{org: org} do + for i <- 1..3 do + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 30 + i * 5, + qoe_score: 80.0 + i, + capacity_score: 70.0 + i + }) + end + + assert {:ok, 1} = FleetIntelligence.compute_fleet_profiles(org.id) + + profiles = Repo.all(FleetProfile) + assert length(profiles) == 1 + profile = hd(profiles) + assert profile.model == "Ubiquiti AF5XHD" + assert profile.manufacturer == "Ubiquiti" + assert profile.device_count == 3 + assert profile.avg_subscribers + assert profile.avg_qoe_score + assert profile.avg_capacity_score + end + + test "creates separate profiles for different models", %{org: org} do + for _i <- 1..2 do + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 30, + qoe_score: 80.0 + }) + end + + for _i <- 1..2 do + insert_matched_ap_with_model(org, "MikroTik RB4011", %{ + subscriber_count: 20, + qoe_score: 75.0 + }) + end + + assert {:ok, 2} = FleetIntelligence.compute_fleet_profiles(org.id) + + profiles = Repo.all(FleetProfile) + assert length(profiles) == 2 + + manufacturers = profiles |> Enum.map(& &1.manufacturer) |> Enum.sort() + assert manufacturers == ["MikroTik", "Ubiquiti"] + end + + test "handles empty organization", %{org: org} do + assert {:ok, 0} = FleetIntelligence.compute_fleet_profiles(org.id) + end + + test "skips APs without model", %{org: org} do + device = device_fixture(%{organization_id: org.id}) + + %AccessPoint{} + |> AccessPoint.changeset(%{ + organization_id: org.id, + preseem_id: "ap-no-model-#{System.unique_integer([:positive])}", + name: "No Model", + device_id: device.id, + match_confidence: "auto_ip" + }) + |> Repo.insert!() + + assert {:ok, 0} = FleetIntelligence.compute_fleet_profiles(org.id) + end + + test "skips unmatched APs", %{org: org} do + %AccessPoint{} + |> AccessPoint.changeset(%{ + organization_id: org.id, + preseem_id: "ap-unmatched-#{System.unique_integer([:positive])}", + name: "Unmatched", + model: "Ubiquiti AF5XHD" + }) + |> Repo.insert!() + + assert {:ok, 0} = FleetIntelligence.compute_fleet_profiles(org.id) + end + + test "computes capacity ceiling from APs with good QoE", %{org: org} do + # AP with high subscribers and good QoE - should define ceiling + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 50, + qoe_score: 85.0 + }) + + # AP with even higher subscribers but poor QoE - excluded from ceiling + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 80, + qoe_score: 55.0 + }) + + # AP with low subscribers and good QoE + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 20, + qoe_score: 95.0 + }) + + assert {:ok, 1} = FleetIntelligence.compute_fleet_profiles(org.id) + + profile = Repo.one!(FleetProfile) + # Capacity ceiling should be 50 (highest subscriber count with QoE >= 70) + assert profile.capacity_ceiling == 50 + end + + test "upserts profiles on recomputation", %{org: org} do + for _i <- 1..3 do + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 30, + qoe_score: 80.0 + }) + end + + assert {:ok, 1} = FleetIntelligence.compute_fleet_profiles(org.id) + assert length(Repo.all(FleetProfile)) == 1 + + # Recompute - should update, not duplicate + assert {:ok, 1} = FleetIntelligence.compute_fleet_profiles(org.id) + assert length(Repo.all(FleetProfile)) == 1 + end + + test "includes performance_data with subscriber range and firmware distribution", %{org: org} do + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 20, + firmware: "v4.3.1" + }) + + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 50, + firmware: "v4.3.1" + }) + + insert_matched_ap_with_model(org, "Ubiquiti AF5XHD", %{ + subscriber_count: 35, + firmware: "v4.3.2" + }) + + assert {:ok, 1} = FleetIntelligence.compute_fleet_profiles(org.id) + + profile = Repo.one!(FleetProfile) + assert profile.performance_data["subscriber_range"]["min"] == 20 + assert profile.performance_data["subscriber_range"]["max"] == 50 + assert profile.performance_data["firmware_distribution"] == %{"v4.3.1" => 2, "v4.3.2" => 1} + end + end + + # Helpers + + defp insert_matched_ap_with_model(org, model, extra_attrs) do + device = device_fixture(%{organization_id: org.id}) + + attrs = + Map.merge( + %{ + organization_id: org.id, + preseem_id: "ap-#{System.unique_integer([:positive])}", + name: "AP", + model: model, + device_id: device.id, + match_confidence: "auto_ip" + }, + extra_attrs + ) + + %AccessPoint{} + |> AccessPoint.changeset(attrs) + |> Repo.insert!() + end +end diff --git a/test/towerops/workers/preseem_baseline_worker_test.exs b/test/towerops/workers/preseem_baseline_worker_test.exs new file mode 100644 index 00000000..f2829f2b --- /dev/null +++ b/test/towerops/workers/preseem_baseline_worker_test.exs @@ -0,0 +1,31 @@ +defmodule Towerops.Workers.PreseemBaselineWorkerTest do + use Towerops.DataCase, async: true + + import Towerops.AccountsFixtures + import Towerops.IntegrationsFixtures + import Towerops.OrganizationsFixtures + + alias Towerops.Workers.PreseemBaselineWorker + + describe "perform/1" do + test "runs without errors for org with enabled integration" do + user = user_fixture() + org = organization_fixture(user.id) + _integration = integration_fixture(org.id) + + assert :ok = PreseemBaselineWorker.perform(%Oban.Job{}) + end + + test "handles no integrations" do + assert :ok = PreseemBaselineWorker.perform(%Oban.Job{}) + end + + test "skips disabled integrations" do + user = user_fixture() + org = organization_fixture(user.id) + _integration = integration_fixture(org.id, %{enabled: false}) + + assert :ok = PreseemBaselineWorker.perform(%Oban.Job{}) + end + end +end