diff --git a/config/runtime.exs b/config/runtime.exs index 14beafeb..93fe9c16 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -159,7 +159,7 @@ if config_env() == :prod do default: 10, discovery: 10, # SNMP polling jobs - one per device - pollers: 50, + pollers: String.to_integer(System.get_env("POLLER_CONCURRENCY") || "50"), # Device monitoring jobs - health checks monitors: 50, # Service checks - HTTP/TCP/DNS @@ -184,6 +184,8 @@ if config_env() == :prod do {"30 0,8,16 * * *", Towerops.Workers.AgentLatencyEvaluator}, # Health check for missing jobs every 10 minutes {"*/10 * * * *", Towerops.Workers.JobHealthCheckWorker}, + # Send notification digests for rate-limited alerts every 5 minutes + {"*/5 * * * *", Towerops.Workers.AlertDigestWorker, args: %{"user_id" => "__cron__"}}, # Mark stale backup requests as timeout every 10 minutes {"*/10 * * * *", Towerops.Workers.BackupTimeoutWorker}, # Clean up old login history daily at 2 AM @@ -295,7 +297,7 @@ if config_env() == :prod do config :towerops, Towerops.Repo, ssl: ssl_config, url: database_url, - pool_size: String.to_integer(System.get_env("POOL_SIZE") || "10"), + pool_size: String.to_integer(System.get_env("POOL_SIZE") || "20"), # For machines with several cores, consider starting multiple pools of `pool_size` # pool_count: 4, socket_options: maybe_ipv6 diff --git a/lib/towerops/devices/event_logger.ex b/lib/towerops/devices/event_logger.ex index 18441d0a..086562b2 100644 --- a/lib/towerops/devices/event_logger.ex +++ b/lib/towerops/devices/event_logger.ex @@ -2,6 +2,10 @@ defmodule Towerops.Devices.EventLogger do @moduledoc """ GenServer that subscribes to device events via PubSub and logs them to the database. + Subscribes to org-scoped `device_events:org:{org_id}` topics instead of a + single global topic, so PubSub traffic is partitioned per organization and + LiveView processes only receive events for their tenant. + This decouples event detection from event storage, allowing: - Faster polling without database write blocking - Multiple event consumers (logging, alerts, webhooks) @@ -10,6 +14,7 @@ defmodule Towerops.Devices.EventLogger do use GenServer alias Towerops.Devices + alias Towerops.Organizations require Logger @@ -19,14 +24,37 @@ defmodule Towerops.Devices.EventLogger do GenServer.start_link(__MODULE__, %{}, name: __MODULE__) end + @doc """ + Subscribe to a new org's device events topic at runtime (e.g. when an org is created). + """ + def subscribe_org(org_id) when is_binary(org_id) do + GenServer.cast(__MODULE__, {:subscribe_org, org_id}) + end + # Server Callbacks @impl true def init(state) do - # device events topic - _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events") - Logger.info("EventLogger started and subscribed to device events") - {:ok, state} + # Subscribe to all existing org-scoped device event topics + org_ids = Organizations.list_organization_ids() + + for org_id <- org_ids do + _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device_events:org:#{org_id}") + end + + Logger.info("EventLogger started, subscribed to #{length(org_ids)} org device event topics") + {:ok, Map.put(state, :subscribed_orgs, MapSet.new(org_ids))} + end + + @impl true + def handle_cast({:subscribe_org, org_id}, state) do + if MapSet.member?(state.subscribed_orgs, org_id) do + {:noreply, state} + else + _ = Phoenix.PubSub.subscribe(Towerops.PubSub, "device_events:org:#{org_id}") + Logger.info("EventLogger subscribed to device_events:org:#{org_id}") + {:noreply, %{state | subscribed_orgs: MapSet.put(state.subscribed_orgs, org_id)}} + end end @impl true diff --git a/lib/towerops/organizations.ex b/lib/towerops/organizations.ex index 8b20b6f0..cac5dd2d 100644 --- a/lib/towerops/organizations.ex +++ b/lib/towerops/organizations.ex @@ -19,6 +19,14 @@ defmodule Towerops.Organizations do ## Organizations + @doc """ + Returns all organization IDs. Used by EventLogger to subscribe to org-scoped PubSub topics. + """ + @spec list_organization_ids() :: [String.t()] + def list_organization_ids do + Repo.all(from(o in Organization, select: o.id)) + end + @doc """ Returns the list of organizations for a user. """ diff --git a/lib/towerops/snmp/discovery.ex b/lib/towerops/snmp/discovery.ex index 9aca0c72..0275ecf3 100644 --- a/lib/towerops/snmp/discovery.ex +++ b/lib/towerops/snmp/discovery.ex @@ -1429,6 +1429,7 @@ defmodule Towerops.Snmp.Discovery do event_attrs = %{ device_id: device.id, + org_id: device.organization_id, event_type: event_type, severity: "info", message: message, @@ -1442,6 +1443,13 @@ defmodule Towerops.Snmp.Discovery do "device:events", {:device_event, event_attrs} ) + + # Also publish to org-scoped topic for targeted LiveView subscriptions + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "device_events:org:#{device.organization_id}", + {:device_event, event_attrs} + ) end # Sanitizes data structures to ensure they are JSON-encodable diff --git a/lib/towerops/snmp/sensor_change_detector.ex b/lib/towerops/snmp/sensor_change_detector.ex index 1ec08c9b..30d68777 100644 --- a/lib/towerops/snmp/sensor_change_detector.ex +++ b/lib/towerops/snmp/sensor_change_detector.ex @@ -272,10 +272,26 @@ defmodule Towerops.Snmp.SensorChangeDetector do _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{event.device_id}", {:device_event, event}) _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event}) + case get_org_id_from_device(event.device_id) do + nil -> + :ok + + org_id -> + event_with_org = Map.put(event, :org_id, org_id) + _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device_events:org:#{org_id}", {:device_event, event_with_org}) + end + Logger.debug("Sensor event: #{event.message}") end) end + defp get_org_id_from_device(device_id) do + case Towerops.Repo.get(Towerops.Devices.Device, device_id) do + nil -> nil + device -> device.organization_id + end + end + defp get_device_id_from_sensor(sensor) do case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do nil -> nil diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index e66e40d7..a8428fd2 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -296,7 +296,7 @@ defmodule Towerops.Workers.DevicePollerWorker do end defp poll_device_state_sensors(device, snmp_device, client_opts, now) do - poll_state_sensors(snmp_device.state_sensors, client_opts, now, device.id) + poll_state_sensors(snmp_device.state_sensors, client_opts, now, device.id, device.organization_id) Logger.debug("Polled #{length(snmp_device.state_sensors)} state sensors for #{device.name}") _ = @@ -745,12 +745,12 @@ defmodule Towerops.Workers.DevicePollerWorker do end) end - defp poll_state_sensors(state_sensors, client_opts, timestamp, device_id) do + defp poll_state_sensors(state_sensors, client_opts, timestamp, device_id, org_id) do state_sensors |> Task.async_stream( fn state_sensor -> result = poll_state_sensor_value(state_sensor, client_opts) - handle_state_sensor_poll_result(state_sensor, result, timestamp, device_id) + handle_state_sensor_poll_result(state_sensor, result, timestamp, device_id, org_id) state_sensor.sensor_descr end, max_concurrency: 2, @@ -776,7 +776,7 @@ defmodule Towerops.Workers.DevicePollerWorker do end end - defp handle_state_sensor_poll_result(state_sensor, {:ok, new_state_value}, timestamp, device_id) do + defp handle_state_sensor_poll_result(state_sensor, {:ok, new_state_value}, timestamp, device_id, org_id) do old_state_value = state_sensor.state_value old_state_descr = state_sensor.state_descr old_status = state_sensor.status @@ -803,11 +803,11 @@ defmodule Towerops.Workers.DevicePollerWorker do new_status: new_status } - broadcast_state_sensor_change(state_sensor, state_changes, device_id, timestamp) + broadcast_state_sensor_change(state_sensor, state_changes, device_id, org_id, timestamp) end end - defp handle_state_sensor_poll_result(state_sensor, {:error, reason}, timestamp, _device_id) do + defp handle_state_sensor_poll_result(state_sensor, {:error, reason}, timestamp, _device_id, _org_id) do Logger.debug("Failed to poll state sensor #{state_sensor.sensor_descr}: #{inspect(reason)}") Snmp.update_state_sensor(state_sensor, %{ @@ -816,7 +816,7 @@ defmodule Towerops.Workers.DevicePollerWorker do }) end - defp broadcast_state_sensor_change(state_sensor, state_changes, device_id, timestamp) do + defp broadcast_state_sensor_change(state_sensor, state_changes, device_id, org_id, timestamp) do %{ old_state_value: old_state_value, new_state_value: new_state_value, @@ -835,6 +835,7 @@ defmodule Towerops.Workers.DevicePollerWorker do event = %{ device_id: device_id, + org_id: org_id, event_type: event_type, severity: severity, message: message, @@ -855,6 +856,10 @@ defmodule Towerops.Workers.DevicePollerWorker do _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{device_id}", {:device_event, event}) _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event}) + if org_id do + _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device_events:org:#{org_id}", {:device_event, event}) + end + Logger.info("State sensor change: #{event.message}") end @@ -1041,7 +1046,7 @@ defmodule Towerops.Workers.DevicePollerWorker do fn interface -> case get_interface_attributes(client_opts, interface.if_index) do {:ok, current_attrs} -> - detect_and_log_changes(interface, current_attrs, device.id, timestamp) + detect_and_log_changes(interface, current_attrs, device.id, device.organization_id, timestamp) {:error, reason} -> Logger.debug("Failed to get interface attributes for #{interface.if_name}: #{inspect(reason)}") @@ -1078,7 +1083,7 @@ defmodule Towerops.Workers.DevicePollerWorker do end end - defp detect_and_log_changes(interface, current_attrs, device_id, timestamp) do + defp detect_and_log_changes(interface, current_attrs, device_id, org_id, timestamp) do events = [] |> maybe_add_oper_status_event(interface, current_attrs, device_id, timestamp) @@ -1087,7 +1092,7 @@ defmodule Towerops.Workers.DevicePollerWorker do |> maybe_add_mac_change_event(interface, current_attrs, device_id, timestamp) if events != [] do - broadcast_interface_events(events) + broadcast_interface_events(events, org_id) Snmp.update_interface(interface, current_attrs) end end @@ -1233,22 +1238,33 @@ defmodule Towerops.Workers.DevicePollerWorker do "Interface #{if_name} MAC address changed from #{old_mac} to #{new_mac}" end - defp broadcast_interface_events(events) do + defp broadcast_interface_events(events, org_id) do Enum.each(events, fn {:event, event_attrs} -> + event_with_org = Map.put(event_attrs, :org_id, org_id) + _ = Phoenix.PubSub.broadcast( Towerops.PubSub, "device:#{event_attrs.device_id}", - {:device_event, event_attrs} + {:device_event, event_with_org} ) _ = Phoenix.PubSub.broadcast( Towerops.PubSub, "device:events", - {:device_event, event_attrs} + {:device_event, event_with_org} ) + if org_id do + _ = + Phoenix.PubSub.broadcast( + Towerops.PubSub, + "device_events:org:#{org_id}", + {:device_event, event_with_org} + ) + end + Logger.debug("Broadcast event: #{event_attrs.message}") end) end diff --git a/lib/towerops_web/live/dashboard_live.ex b/lib/towerops_web/live/dashboard_live.ex index 329f696f..2a0901bf 100644 --- a/lib/towerops_web/live/dashboard_live.ex +++ b/lib/towerops_web/live/dashboard_live.ex @@ -29,7 +29,8 @@ defmodule ToweropsWeb.DashboardLive do |> assign(:device_count, device_count) |> assign(:insight_source, nil) |> assign(:timezone, socket.assigns.current_scope.timezone) - |> assign(:time_format, socket.assigns.current_scope.time_format)} + |> assign(:time_format, socket.assigns.current_scope.time_format) + |> assign(:pending_dashboard_reload_ref, nil)} end @impl true @@ -81,15 +82,27 @@ defmodule ToweropsWeb.DashboardLive do end @impl true - def handle_info({:new_alert, _device_id, _alert_type}, socket) do - {:noreply, load_dashboard_data(socket, socket.assigns.current_scope.organization.id)} + def handle_info({event, _device_id, _alert_type}, socket) + when event in [:new_alert, :alert_resolved] do + {:noreply, schedule_debounced_dashboard_reload(socket)} end @impl true - def handle_info({:alert_resolved, _device_id, _alert_type}, socket) do + def handle_info(:debounced_dashboard_reload, socket) do {:noreply, load_dashboard_data(socket, socket.assigns.current_scope.organization.id)} end + @dashboard_debounce_ms 1_500 + + defp schedule_debounced_dashboard_reload(socket) do + if ref = socket.assigns.pending_dashboard_reload_ref do + Process.cancel_timer(ref) + end + + ref = Process.send_after(self(), :debounced_dashboard_reload, @dashboard_debounce_ms) + assign(socket, :pending_dashboard_reload_ref, ref) + end + defp load_dashboard_data(socket, organization_id) do organization = socket.assigns.current_scope.organization summary = Dashboard.get_dashboard_summary(organization_id)