diff --git a/lib/towerops_web/live/admin/monitoring_live.ex b/lib/towerops_web/live/admin/monitoring_live.ex new file mode 100644 index 00000000..508deb7c --- /dev/null +++ b/lib/towerops_web/live/admin/monitoring_live.ex @@ -0,0 +1,73 @@ +defmodule ToweropsWeb.Admin.MonitoringLive do + @moduledoc """ + Real-time job monitoring dashboard for Oban workers. + + Displays: + - Active operations (currently executing jobs) + - Problems (stuck/failed jobs) + - Health metrics (throughput, errors, latency) + - Recent activity timeline + """ + use ToweropsWeb, :live_view + + alias Towerops.JobMonitoring + alias Towerops.JobMonitoring.Metrics + + @topic "job:lifecycle" + + @impl Phoenix.LiveView + def mount(_params, _session, socket) do + if connected?(socket) do + Phoenix.PubSub.subscribe(Towerops.PubSub, @topic) + end + + socket = + socket + |> assign(:page_title, "Job Monitoring") + |> assign(:active_jobs, []) + |> assign(:stuck_jobs_count, 0) + |> assign(:failed_jobs_count, 0) + |> assign(:recent_jobs, []) + |> assign(:health_metrics, %{}) + |> load_monitoring_data() + + {:ok, socket} + end + + @impl Phoenix.LiveView + def handle_info(%{event: :started} = event, socket) do + # Job started - add to active jobs list + active_jobs = [event | socket.assigns.active_jobs] + {:noreply, assign(socket, :active_jobs, active_jobs)} + end + + def handle_info(%{event: event} = job_event, socket) when event in [:completed, :failed] do + # Job finished - remove from active jobs, reload data + active_jobs = Enum.reject(socket.assigns.active_jobs, fn job -> job.job_id == job_event.job_id end) + + socket = + socket + |> assign(:active_jobs, active_jobs) + |> load_monitoring_data() + + {:noreply, socket} + end + + def handle_info(_event, socket) do + {:noreply, socket} + end + + @spec load_monitoring_data(Phoenix.LiveView.Socket.t()) :: Phoenix.LiveView.Socket.t() + defp load_monitoring_data(socket) do + stuck_jobs = JobMonitoring.list_stuck_jobs() + failed_jobs = JobMonitoring.list_failed_jobs() + recent_jobs = JobMonitoring.list_recent_completions(20) + metrics = Metrics.calculate_all() + + socket + |> assign(:stuck_jobs_count, length(stuck_jobs)) + |> assign(:failed_jobs_count, length(failed_jobs)) + |> assign(:recent_jobs, recent_jobs) + |> assign(:health_metrics, metrics) + end +end diff --git a/lib/towerops_web/live/admin/monitoring_live.html.heex b/lib/towerops_web/live/admin/monitoring_live.html.heex new file mode 100644 index 00000000..c83cb4eb --- /dev/null +++ b/lib/towerops_web/live/admin/monitoring_live.html.heex @@ -0,0 +1,121 @@ + +
+
+

+ Job Monitoring Dashboard +

+

+ Real-time monitoring of polling and discovery jobs +

+
+ + +
+
+

Completed (1h)

+

+ {Map.get(@health_metrics, :completed_last_hour, 0)} +

+

jobs completed

+
+ +
+

Failed (1h)

+

+ {Map.get(@health_metrics, :failed_last_hour, 0)} +

+

failures

+
+ +
+

Avg Duration

+

+ <%= if Map.get(@health_metrics, :avg_execution_time_seconds) do %> + {Map.get(@health_metrics, :avg_execution_time_seconds) |> Float.round(1)}s + <% else %> + N/A + <% end %> +

+

last hour

+
+ +
+

Active Jobs

+

+ {length(@active_jobs)} +

+

executing now

+
+
+ + +
+
+

Active Operations

+
+
+ <%= if @active_jobs == [] do %> +

+ No active jobs +

+ <% else %> +
+ <%= for job <- @active_jobs do %> +
+
+ + {job.worker |> String.split(".") |> List.last()} + + + Device ID: {job.device_id} + +
+ + Started: {Calendar.strftime(job.timestamp, "%H:%M:%S")} + +
+ <% end %> +
+ <% end %> +
+
+ + +
+
+

Problems

+
+ + {if @stuck_jobs_count > 0, do: "#{@stuck_jobs_count} stuck", else: ""} + + + {if @failed_jobs_count > 0, do: "#{@failed_jobs_count} failed", else: ""} + +
+
+
+ <%= if @stuck_jobs_count == 0 and @failed_jobs_count == 0 do %> +

+ No problems detected +

+ <% else %> +

+ Problems section coming soon +

+ <% end %> +
+
+ + +
+
+

Health Metrics

+
+
+

+ Detailed metrics coming soon +

+
+
+
+ diff --git a/lib/towerops_web/router.ex b/lib/towerops_web/router.ex index 893b6c35..47873496 100644 --- a/lib/towerops_web/router.ex +++ b/lib/towerops_web/router.ex @@ -231,6 +231,7 @@ defmodule ToweropsWeb.Router do live "/users", UserLive.Index, :index live "/organizations", OrgLive.Index, :index live "/audit", AuditLive.Index, :index + live "/monitoring", MonitoringLive, :index end end diff --git a/test/towerops_web/live/admin/monitoring_live_test.exs b/test/towerops_web/live/admin/monitoring_live_test.exs new file mode 100644 index 00000000..62388c99 --- /dev/null +++ b/test/towerops_web/live/admin/monitoring_live_test.exs @@ -0,0 +1,38 @@ +defmodule ToweropsWeb.Admin.MonitoringLiveTest do + use ToweropsWeb.ConnCase + + import Phoenix.LiveViewTest + + describe "MonitoringLive" do + setup [:register_and_log_in_superuser] + + test "renders monitoring dashboard", %{conn: conn} do + {:ok, _view, html} = live(conn, ~p"/admin/monitoring") + + assert html =~ "Job Monitoring Dashboard" + assert html =~ "Active Operations" + assert html =~ "Problems" + assert html =~ "Health Metrics" + end + + test "subscribes to job lifecycle events", %{conn: conn} do + {:ok, view, _html} = live(conn, ~p"/admin/monitoring") + + # Verify PubSub subscription by checking it's alive + assert view.pid |> Process.alive?() + end + + test "displays active operations section", %{conn: conn} do + {:ok, _view, html} = live(conn, ~p"/admin/monitoring") + + assert html =~ "Active Operations" + assert html =~ "No active jobs" + end + end + + defp register_and_log_in_superuser(%{conn: conn}) do + user = Towerops.AccountsFixtures.user_fixture() + user = Towerops.Repo.update!(Ecto.Changeset.change(user, is_superuser: true)) + %{conn: log_in_user(conn, user), user: user} + end +end