1856 lines
52 KiB
Markdown
1856 lines
52 KiB
Markdown
# Job Monitoring Dashboard Implementation Plan
|
|
|
|
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
|
|
|
|
**Goal:** Build a live monitoring dashboard at `/admin/monitoring` showing all polling and discovery job activity across all organizations with real-time updates, health metrics, and diagnostic drill-down.
|
|
|
|
**Architecture:** LiveView dashboard with hybrid update strategy (PubSub for job lifecycle events + periodic polling for metrics). Context layer queries Oban jobs with device/agent associations. Workers broadcast lifecycle events to PubSub topic.
|
|
|
|
**Tech Stack:** Phoenix LiveView, Oban, PubSub, Ecto, TailwindCSS
|
|
|
|
---
|
|
|
|
## Task 1: Create JobMonitoring Context (Foundation)
|
|
|
|
**Files:**
|
|
- Create: `lib/towerops/job_monitoring.ex`
|
|
- Test: `test/towerops/job_monitoring_test.exs`
|
|
|
|
**Step 1: Write failing test for list_active_jobs**
|
|
|
|
```elixir
|
|
# test/towerops/job_monitoring_test.exs
|
|
defmodule Towerops.JobMonitoringTest do
|
|
use Towerops.DataCase, async: true
|
|
|
|
alias Towerops.JobMonitoring
|
|
alias Towerops.Workers.{DevicePollerWorker, DiscoveryWorker}
|
|
|
|
describe "list_active_jobs/0" do
|
|
test "returns jobs in executing state for polling and discovery workers" do
|
|
device = insert(:device)
|
|
|
|
# Create executing polling job
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.utc_now()
|
|
)
|
|
|
|
# Create executing discovery job
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.utc_now()
|
|
)
|
|
|
|
# Create completed job (should not be returned)
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "completed",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
jobs = JobMonitoring.list_active_jobs()
|
|
|
|
assert length(jobs) == 2
|
|
assert Enum.all?(jobs, fn j -> j.state == "executing" end)
|
|
end
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 2: Run test to verify it fails**
|
|
|
|
Run: `mix test test/towerops/job_monitoring_test.exs`
|
|
Expected: FAIL with "module JobMonitoring not defined"
|
|
|
|
**Step 3: Create Oban job factory**
|
|
|
|
```elixir
|
|
# test/support/factory.ex (add to existing factory)
|
|
|
|
def oban_job_factory do
|
|
%Oban.Job{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
queue: "pollers",
|
|
args: %{"device_id" => 1},
|
|
state: "available",
|
|
attempt: 0,
|
|
max_attempts: 3,
|
|
inserted_at: DateTime.utc_now(),
|
|
scheduled_at: DateTime.utc_now()
|
|
}
|
|
end
|
|
```
|
|
|
|
**Step 4: Write minimal implementation**
|
|
|
|
```elixir
|
|
# lib/towerops/job_monitoring.ex
|
|
defmodule Towerops.JobMonitoring do
|
|
@moduledoc """
|
|
Context for monitoring Oban jobs, specifically polling and discovery operations.
|
|
|
|
Provides queries for active, stuck, failed, and completed jobs with device
|
|
and agent context for operational monitoring.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
alias Towerops.Repo
|
|
alias Oban.Job
|
|
|
|
@worker_names [
|
|
"Towerops.Workers.DevicePollerWorker",
|
|
"Towerops.Workers.DiscoveryWorker"
|
|
]
|
|
|
|
@doc """
|
|
Lists all currently executing polling and discovery jobs.
|
|
"""
|
|
def list_active_jobs do
|
|
from(j in Job,
|
|
where: j.state == "executing",
|
|
where: j.worker in ^@worker_names,
|
|
order_by: [asc: j.attempted_at]
|
|
)
|
|
|> Repo.all()
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 5: Run test to verify it passes**
|
|
|
|
Run: `mix test test/towerops/job_monitoring_test.exs`
|
|
Expected: PASS
|
|
|
|
**Step 6: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/job_monitoring.ex test/towerops/job_monitoring_test.exs test/support/factory.ex
|
|
git commit -m "feat: add JobMonitoring context with list_active_jobs query"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 2: Add Stuck Jobs Detection
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops/job_monitoring.ex`
|
|
- Test: `test/towerops/job_monitoring_test.exs`
|
|
|
|
**Step 1: Write failing test for list_stuck_jobs**
|
|
|
|
```elixir
|
|
# test/towerops/job_monitoring_test.exs (add to describe block)
|
|
|
|
describe "list_stuck_jobs/0" do
|
|
test "returns polling jobs executing longer than 2 minutes" do
|
|
device = insert(:device)
|
|
|
|
# Stuck polling job (3 minutes ago)
|
|
stuck_job = insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
|
|
)
|
|
|
|
# Recent polling job (30 seconds ago) - not stuck
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -30, :second)
|
|
)
|
|
|
|
stuck = JobMonitoring.list_stuck_jobs()
|
|
|
|
assert length(stuck) == 1
|
|
assert hd(stuck).id == stuck_job.id
|
|
end
|
|
|
|
test "returns discovery jobs executing longer than 5 minutes" do
|
|
device = insert(:device)
|
|
|
|
# Stuck discovery job (6 minutes ago)
|
|
stuck_job = insert(:oban_job,
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -360, :second)
|
|
)
|
|
|
|
# Recent discovery job (3 minutes ago) - not stuck
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
|
|
)
|
|
|
|
stuck = JobMonitoring.list_stuck_jobs()
|
|
|
|
assert length(stuck) == 1
|
|
assert hd(stuck).id == stuck_job.id
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 2: Run test to verify it fails**
|
|
|
|
Run: `mix test test/towerops/job_monitoring_test.exs::list_stuck_jobs`
|
|
Expected: FAIL with "function list_stuck_jobs/0 undefined"
|
|
|
|
**Step 3: Write minimal implementation**
|
|
|
|
```elixir
|
|
# lib/towerops/job_monitoring.ex (add to module)
|
|
|
|
@polling_threshold_seconds 120 # 2 minutes
|
|
@discovery_threshold_seconds 300 # 5 minutes
|
|
|
|
@doc """
|
|
Lists jobs that are executing longer than expected thresholds.
|
|
|
|
Thresholds:
|
|
- Polling: 2 minutes
|
|
- Discovery: 5 minutes
|
|
"""
|
|
def list_stuck_jobs do
|
|
now = DateTime.utc_now()
|
|
polling_threshold = DateTime.add(now, -@polling_threshold_seconds, :second)
|
|
discovery_threshold = DateTime.add(now, -@discovery_threshold_seconds, :second)
|
|
|
|
from(j in Job,
|
|
where: j.state == "executing",
|
|
where:
|
|
(j.worker == "Towerops.Workers.DevicePollerWorker" and j.attempted_at < ^polling_threshold) or
|
|
(j.worker == "Towerops.Workers.DiscoveryWorker" and j.attempted_at < ^discovery_threshold),
|
|
order_by: [asc: j.attempted_at]
|
|
)
|
|
|> Repo.all()
|
|
end
|
|
```
|
|
|
|
**Step 4: Run test to verify it passes**
|
|
|
|
Run: `mix test test/towerops/job_monitoring_test.exs`
|
|
Expected: PASS (all tests)
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/job_monitoring.ex test/towerops/job_monitoring_test.exs
|
|
git commit -m "feat: add stuck job detection with per-worker thresholds"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 3: Add Failed and Recent Jobs Queries
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops/job_monitoring.ex`
|
|
- Test: `test/towerops/job_monitoring_test.exs`
|
|
|
|
**Step 1: Write failing tests**
|
|
|
|
```elixir
|
|
# test/towerops/job_monitoring_test.exs (add describe blocks)
|
|
|
|
describe "list_failed_jobs/0" do
|
|
test "returns retryable and cancelled jobs" do
|
|
device = insert(:device)
|
|
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "retryable",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "cancelled",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
# Should not include completed
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "completed",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
failed = JobMonitoring.list_failed_jobs()
|
|
|
|
assert length(failed) == 2
|
|
assert Enum.all?(failed, fn j -> j.state in ["retryable", "cancelled"] end)
|
|
end
|
|
end
|
|
|
|
describe "list_recent_completions/1" do
|
|
test "returns completed, cancelled, and discarded jobs up to limit" do
|
|
device = insert(:device)
|
|
|
|
# Create 5 completed jobs
|
|
for _ <- 1..5 do
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "completed",
|
|
args: %{"device_id" => device.id},
|
|
completed_at: DateTime.utc_now()
|
|
)
|
|
end
|
|
|
|
# Should not include executing
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
recent = JobMonitoring.list_recent_completions(3)
|
|
|
|
assert length(recent) == 3
|
|
assert Enum.all?(recent, fn j -> j.state == "completed" end)
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 2: Run test to verify it fails**
|
|
|
|
Run: `mix test test/towerops/job_monitoring_test.exs`
|
|
Expected: FAIL with "function list_failed_jobs/0 undefined"
|
|
|
|
**Step 3: Write minimal implementation**
|
|
|
|
```elixir
|
|
# lib/towerops/job_monitoring.ex (add to module)
|
|
|
|
@doc """
|
|
Lists jobs that have failed and are retrying or have been cancelled.
|
|
|
|
Limited to last 100 jobs.
|
|
"""
|
|
def list_failed_jobs do
|
|
from(j in Job,
|
|
where: j.state in ["retryable", "cancelled"],
|
|
where: j.worker in ^@worker_names,
|
|
order_by: [desc: j.updated_at],
|
|
limit: 100
|
|
)
|
|
|> Repo.all()
|
|
end
|
|
|
|
@doc """
|
|
Lists recently completed jobs (completed, cancelled, or discarded).
|
|
|
|
## Options
|
|
- `:limit` - Maximum number of jobs to return (default: 100)
|
|
"""
|
|
def list_recent_completions(limit \\ 100) do
|
|
from(j in Job,
|
|
where: j.state in ["completed", "cancelled", "discarded"],
|
|
where: j.worker in ^@worker_names,
|
|
order_by: [desc: j.completed_at],
|
|
limit: ^limit
|
|
)
|
|
|> Repo.all()
|
|
end
|
|
```
|
|
|
|
**Step 4: Run test to verify it passes**
|
|
|
|
Run: `mix test test/towerops/job_monitoring_test.exs`
|
|
Expected: PASS (all tests)
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/job_monitoring.ex test/towerops/job_monitoring_test.exs
|
|
git commit -m "feat: add failed jobs and recent completions queries"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 4: Add Health Metrics Module
|
|
|
|
**Files:**
|
|
- Create: `lib/towerops/job_monitoring/metrics.ex`
|
|
- Test: `test/towerops/job_monitoring/metrics_test.exs`
|
|
|
|
**Step 1: Write failing test for calculate_all**
|
|
|
|
```elixir
|
|
# test/towerops/job_monitoring/metrics_test.exs
|
|
defmodule Towerops.JobMonitoring.MetricsTest do
|
|
use Towerops.DataCase, async: true
|
|
|
|
alias Towerops.JobMonitoring.Metrics
|
|
|
|
describe "calculate_all/0" do
|
|
test "returns comprehensive metrics structure" do
|
|
device = insert(:device)
|
|
|
|
# Executing jobs
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
# Queued jobs
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "scheduled",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
# Completed jobs (last hour)
|
|
now = DateTime.utc_now()
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "completed",
|
|
args: %{"device_id" => device.id},
|
|
completed_at: DateTime.add(now, -1800, :second),
|
|
attempted_at: DateTime.add(now, -1810, :second)
|
|
)
|
|
|
|
metrics = Metrics.calculate_all()
|
|
|
|
assert metrics.executing_count == 1
|
|
assert metrics.queued_count == 1
|
|
assert metrics.completed_last_hour == 1
|
|
assert is_float(metrics.polling_success_rate_1h)
|
|
assert is_map(metrics.queue_depths)
|
|
end
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 2: Run test to verify it fails**
|
|
|
|
Run: `mix test test/towerops/job_monitoring/metrics_test.exs`
|
|
Expected: FAIL with "module Metrics not defined"
|
|
|
|
**Step 3: Write minimal implementation**
|
|
|
|
```elixir
|
|
# lib/towerops/job_monitoring/metrics.ex
|
|
defmodule Towerops.JobMonitoring.Metrics do
|
|
@moduledoc """
|
|
Calculates health metrics for job monitoring dashboard.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
alias Towerops.Repo
|
|
alias Oban.Job
|
|
|
|
@worker_names [
|
|
"Towerops.Workers.DevicePollerWorker",
|
|
"Towerops.Workers.DiscoveryWorker"
|
|
]
|
|
|
|
@doc """
|
|
Calculates all metrics for the monitoring dashboard.
|
|
|
|
Returns a map with:
|
|
- executing_count: Number of currently executing jobs
|
|
- queued_count: Number of scheduled/available jobs
|
|
- completed_last_hour: Completed jobs in last hour
|
|
- failed_last_hour: Failed jobs in last hour
|
|
- polling_success_rate_1h: % successful polls (last hour)
|
|
- polling_success_rate_24h: % successful polls (last 24 hours)
|
|
- discovery_success_rate_1h: % successful discoveries (last hour)
|
|
- discovery_success_rate_24h: % successful discoveries (last 24 hours)
|
|
- avg_poll_duration: Average poll duration in seconds
|
|
- avg_discovery_duration: Average discovery duration in seconds
|
|
- jobs_per_minute: Jobs per minute (last 10 minutes)
|
|
- queue_depths: Map of queue name to depth
|
|
"""
|
|
def calculate_all do
|
|
now = DateTime.utc_now()
|
|
one_hour_ago = DateTime.add(now, -3600, :second)
|
|
twenty_four_hours_ago = DateTime.add(now, -86400, :second)
|
|
ten_minutes_ago = DateTime.add(now, -600, :second)
|
|
|
|
%{
|
|
executing_count: count_by_state("executing"),
|
|
queued_count: count_by_state(["scheduled", "available"]),
|
|
completed_last_hour: count_completed_since(one_hour_ago),
|
|
failed_last_hour: count_failed_since(one_hour_ago),
|
|
polling_success_rate_1h: success_rate("Towerops.Workers.DevicePollerWorker", one_hour_ago),
|
|
polling_success_rate_24h: success_rate("Towerops.Workers.DevicePollerWorker", twenty_four_hours_ago),
|
|
discovery_success_rate_1h: success_rate("Towerops.Workers.DiscoveryWorker", one_hour_ago),
|
|
discovery_success_rate_24h: success_rate("Towerops.Workers.DiscoveryWorker", twenty_four_hours_ago),
|
|
avg_poll_duration: avg_duration("Towerops.Workers.DevicePollerWorker", one_hour_ago),
|
|
avg_discovery_duration: avg_duration("Towerops.Workers.DiscoveryWorker", one_hour_ago),
|
|
jobs_per_minute: jobs_per_minute(ten_minutes_ago),
|
|
queue_depths: queue_depths()
|
|
}
|
|
end
|
|
|
|
defp count_by_state(state) when is_binary(state) do
|
|
from(j in Job,
|
|
where: j.state == ^state,
|
|
where: j.worker in ^@worker_names,
|
|
select: count(j.id)
|
|
)
|
|
|> Repo.one() || 0
|
|
end
|
|
|
|
defp count_by_state(states) when is_list(states) do
|
|
from(j in Job,
|
|
where: j.state in ^states,
|
|
where: j.worker in ^@worker_names,
|
|
select: count(j.id)
|
|
)
|
|
|> Repo.one() || 0
|
|
end
|
|
|
|
defp count_completed_since(since) do
|
|
from(j in Job,
|
|
where: j.state == "completed",
|
|
where: j.completed_at > ^since,
|
|
where: j.worker in ^@worker_names,
|
|
select: count(j.id)
|
|
)
|
|
|> Repo.one() || 0
|
|
end
|
|
|
|
defp count_failed_since(since) do
|
|
from(j in Job,
|
|
where: j.state in ["retryable", "cancelled", "discarded"],
|
|
where: j.updated_at > ^since,
|
|
where: j.worker in ^@worker_names,
|
|
select: count(j.id)
|
|
)
|
|
|> Repo.one() || 0
|
|
end
|
|
|
|
defp success_rate(worker, since) do
|
|
total = from(j in Job,
|
|
where: j.worker == ^worker,
|
|
where: j.completed_at > ^since or j.updated_at > ^since,
|
|
select: count(j.id)
|
|
) |> Repo.one() || 0
|
|
|
|
successful = from(j in Job,
|
|
where: j.worker == ^worker,
|
|
where: j.state == "completed",
|
|
where: j.completed_at > ^since,
|
|
select: count(j.id)
|
|
) |> Repo.one() || 0
|
|
|
|
case total do
|
|
0 -> 0.0
|
|
_ -> (successful / total) * 100
|
|
end
|
|
end
|
|
|
|
defp avg_duration(worker, since) do
|
|
result = from(j in Job,
|
|
where: j.worker == ^worker,
|
|
where: j.state == "completed",
|
|
where: j.completed_at > ^since,
|
|
select: avg(fragment("EXTRACT(EPOCH FROM (? - ?))", j.completed_at, j.attempted_at))
|
|
)
|
|
|> Repo.one()
|
|
|
|
case result do
|
|
nil -> 0.0
|
|
duration -> Float.round(duration, 2)
|
|
end
|
|
end
|
|
|
|
defp jobs_per_minute(since) do
|
|
completed = count_completed_since(since)
|
|
minutes_elapsed = 10.0
|
|
Float.round(completed / minutes_elapsed, 2)
|
|
end
|
|
|
|
defp queue_depths do
|
|
queues = ["pollers", "discovery", "maintenance"]
|
|
|
|
Enum.into(queues, %{}, fn queue ->
|
|
depth = from(j in Job,
|
|
where: j.queue == ^queue,
|
|
where: j.state in ["scheduled", "available"],
|
|
select: count(j.id)
|
|
) |> Repo.one() || 0
|
|
|
|
{queue, depth}
|
|
end)
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 4: Run test to verify it passes**
|
|
|
|
Run: `mix test test/towerops/job_monitoring/metrics_test.exs`
|
|
Expected: PASS
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/job_monitoring/metrics.ex test/towerops/job_monitoring/metrics_test.exs
|
|
git commit -m "feat: add health metrics calculation module"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 5: Add PubSub Events Module
|
|
|
|
**Files:**
|
|
- Create: `lib/towerops/job_monitoring/events.ex`
|
|
- Test: `test/towerops/job_monitoring/events_test.exs`
|
|
|
|
**Step 1: Write failing test**
|
|
|
|
```elixir
|
|
# test/towerops/job_monitoring/events_test.exs
|
|
defmodule Towerops.JobMonitoring.EventsTest do
|
|
use Towerops.DataCase, async: true
|
|
|
|
alias Towerops.JobMonitoring.Events
|
|
|
|
describe "broadcast_job_event/3" do
|
|
test "broadcasts job lifecycle event to PubSub" do
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "job:lifecycle")
|
|
|
|
job = %Oban.Job{
|
|
id: 123,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
args: %{"device_id" => 456}
|
|
}
|
|
|
|
Events.broadcast_job_event(job, :started, %{})
|
|
|
|
assert_receive %{
|
|
job_id: 123,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
device_id: 456,
|
|
event: :started,
|
|
metadata: %{},
|
|
timestamp: %DateTime{}
|
|
}
|
|
end
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 2: Run test to verify it fails**
|
|
|
|
Run: `mix test test/towerops/job_monitoring/events_test.exs`
|
|
Expected: FAIL with "module Events not defined"
|
|
|
|
**Step 3: Write minimal implementation**
|
|
|
|
```elixir
|
|
# lib/towerops/job_monitoring/events.ex
|
|
defmodule Towerops.JobMonitoring.Events do
|
|
@moduledoc """
|
|
PubSub event broadcasting for job lifecycle events.
|
|
"""
|
|
|
|
@topic "job:lifecycle"
|
|
|
|
@doc """
|
|
Broadcasts a job lifecycle event to PubSub.
|
|
|
|
## Events
|
|
- `:started` - Job began executing
|
|
- `:completed` - Job finished successfully
|
|
- `:failed` - Job failed with error
|
|
|
|
## Metadata
|
|
Optional map with additional context:
|
|
- `duration`: Execution time in seconds
|
|
- `error`: Error message/reason
|
|
- `items_collected`: Number of items discovered/polled
|
|
"""
|
|
def broadcast_job_event(%Oban.Job{} = job, event, metadata \\ %{}) do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
@topic,
|
|
%{
|
|
job_id: job.id,
|
|
worker: job.worker,
|
|
device_id: get_in(job.args, ["device_id"]),
|
|
event: event,
|
|
metadata: metadata,
|
|
timestamp: DateTime.utc_now()
|
|
}
|
|
)
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 4: Run test to verify it passes**
|
|
|
|
Run: `mix test test/towerops/job_monitoring/events_test.exs`
|
|
Expected: PASS
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/job_monitoring/events.ex test/towerops/job_monitoring/events_test.exs
|
|
git commit -m "feat: add PubSub event broadcasting for job lifecycle"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 6: Integrate Events into DevicePollerWorker
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops/workers/device_poller_worker.ex`
|
|
|
|
**Step 1: Add event broadcasting to perform/1**
|
|
|
|
```elixir
|
|
# lib/towerops/workers/device_poller_worker.ex
|
|
|
|
# Add alias at top of module
|
|
alias Towerops.JobMonitoring.Events
|
|
|
|
# Modify perform/1 function to broadcast events
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do
|
|
Events.broadcast_job_event(job, :started)
|
|
start_time = System.monotonic_time(:second)
|
|
|
|
result = case Devices.get_device(device_id) do
|
|
nil ->
|
|
Logger.debug("Device #{device_id} no longer exists, skipping poll")
|
|
:ok
|
|
|
|
device ->
|
|
maybe_poll_device(device)
|
|
schedule_next_poll_with_error_handling(device_id, device)
|
|
:ok
|
|
end
|
|
|
|
duration = System.monotonic_time(:second) - start_time
|
|
|
|
case result do
|
|
:ok ->
|
|
Events.broadcast_job_event(job, :completed, %{duration: duration})
|
|
{:error, reason} ->
|
|
Events.broadcast_job_event(job, :failed, %{error: inspect(reason), duration: duration})
|
|
end
|
|
|
|
result
|
|
end
|
|
```
|
|
|
|
**Step 2: Manually test (no automated test for worker integration)**
|
|
|
|
Run Phoenix server and verify events are broadcast by subscribing in iex:
|
|
|
|
```elixir
|
|
iex> Phoenix.PubSub.subscribe(Towerops.PubSub, "job:lifecycle")
|
|
iex> # Trigger a poll job
|
|
iex> # Should receive event messages
|
|
```
|
|
|
|
**Step 3: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/workers/device_poller_worker.ex
|
|
git commit -m "feat: integrate lifecycle event broadcasting into DevicePollerWorker"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 7: Integrate Events into DiscoveryWorker
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops/workers/discovery_worker.ex`
|
|
|
|
**Step 1: Add event broadcasting to perform/1**
|
|
|
|
```elixir
|
|
# lib/towerops/workers/discovery_worker.ex
|
|
|
|
# Add alias at top of module
|
|
alias Towerops.JobMonitoring.Events
|
|
|
|
# Modify perform/1 to broadcast events (find existing perform/1 and wrap it)
|
|
@impl Oban.Worker
|
|
def perform(%Oban.Job{args: %{"device_id" => device_id}} = job) do
|
|
Events.broadcast_job_event(job, :started)
|
|
start_time = System.monotonic_time(:second)
|
|
|
|
result = case Devices.get_device_with_details(device_id) do
|
|
nil ->
|
|
Logger.warning("Device #{device_id} not found, discarding discovery job")
|
|
:discard
|
|
|
|
device ->
|
|
# Existing discovery logic wrapped in Task.async/await
|
|
task = Task.async(fn ->
|
|
if device.agent_id && agent_online?(device.agent_id) do
|
|
perform_agent_discovery(device)
|
|
else
|
|
perform_direct_discovery(device)
|
|
end
|
|
end)
|
|
|
|
case Task.await(task, @job_timeout_ms) do
|
|
:ok -> :discard
|
|
{:error, reason} ->
|
|
Logger.error("Discovery failed for device #{device_id}: #{inspect(reason)}")
|
|
:discard
|
|
end
|
|
end
|
|
|
|
duration = System.monotonic_time(:second) - start_time
|
|
|
|
case result do
|
|
:discard ->
|
|
# Could be success or failure - check if it's an error case
|
|
Events.broadcast_job_event(job, :completed, %{duration: duration})
|
|
{:error, reason} ->
|
|
Events.broadcast_job_event(job, :failed, %{error: inspect(reason), duration: duration})
|
|
end
|
|
|
|
result
|
|
end
|
|
```
|
|
|
|
**Step 2: Manually test**
|
|
|
|
Same as previous task - verify in iex that discovery events are broadcast.
|
|
|
|
**Step 3: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/workers/discovery_worker.ex
|
|
git commit -m "feat: integrate lifecycle event broadcasting into DiscoveryWorker"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 8: Create MonitoringLive Base Structure
|
|
|
|
**Files:**
|
|
- Create: `lib/towerops_web/live/admin/monitoring_live.ex`
|
|
- Create: `lib/towerops_web/live/admin/monitoring_live.html.heex`
|
|
- Test: `test/towerops_web/live/admin/monitoring_live_test.exs`
|
|
|
|
**Step 1: Write failing mount test**
|
|
|
|
```elixir
|
|
# test/towerops_web/live/admin/monitoring_live_test.exs
|
|
defmodule ToweropsWeb.Admin.MonitoringLiveTest do
|
|
use ToweropsWeb.ConnCase, async: true
|
|
|
|
import Phoenix.LiveViewTest
|
|
|
|
setup do
|
|
superuser = insert(:user, superuser: true)
|
|
%{superuser: superuser}
|
|
end
|
|
|
|
describe "mount" do
|
|
test "renders monitoring dashboard for superuser", %{conn: conn, superuser: superuser} do
|
|
conn = log_in_user(conn, superuser)
|
|
{:ok, _view, html} = live(conn, ~p"/admin/monitoring")
|
|
|
|
assert html =~ "Job Monitoring"
|
|
assert html =~ "Active Operations"
|
|
assert html =~ "Health Metrics"
|
|
end
|
|
|
|
test "requires superuser authentication", %{conn: conn} do
|
|
regular_user = insert(:user, superuser: false)
|
|
conn = log_in_user(conn, regular_user)
|
|
|
|
assert {:error, {:redirect, %{to: "/"}}} = live(conn, ~p"/admin/monitoring")
|
|
end
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 2: Run test to verify it fails**
|
|
|
|
Run: `mix test test/towerops_web/live/admin/monitoring_live_test.exs`
|
|
Expected: FAIL with "no route found for GET /admin/monitoring"
|
|
|
|
**Step 3: Add route**
|
|
|
|
```elixir
|
|
# lib/towerops_web/router.ex (find admin scope and add)
|
|
|
|
scope "/admin", ToweropsWeb.Admin, as: :admin do
|
|
pipe_through [:browser, :require_authenticated_user, :require_superadmin]
|
|
|
|
live_session :admin,
|
|
on_mount: [{ToweropsWeb.UserAuth, :ensure_authenticated}, {ToweropsWeb.UserAuth, :ensure_superadmin}] do
|
|
live "/", DashboardLive, :index
|
|
live "/monitoring", MonitoringLive, :index # Add this line
|
|
# ... existing routes
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 4: Create minimal LiveView**
|
|
|
|
```elixir
|
|
# lib/towerops_web/live/admin/monitoring_live.ex
|
|
defmodule ToweropsWeb.Admin.MonitoringLive do
|
|
@moduledoc """
|
|
Live monitoring dashboard for all polling and discovery jobs.
|
|
|
|
Shows active operations, health metrics, recent activity, and problem detection
|
|
across all organizations.
|
|
"""
|
|
use ToweropsWeb, :live_view
|
|
|
|
alias Towerops.JobMonitoring
|
|
alias Towerops.JobMonitoring.Metrics
|
|
|
|
@impl true
|
|
def mount(_params, _session, socket) do
|
|
if connected?(socket) do
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "job:lifecycle")
|
|
schedule_metrics_refresh()
|
|
end
|
|
|
|
{:ok,
|
|
socket
|
|
|> assign(:page_title, "Job Monitoring")
|
|
|> load_initial_data()}
|
|
end
|
|
|
|
@impl true
|
|
def handle_params(_params, _url, socket) do
|
|
{:noreply, socket}
|
|
end
|
|
|
|
defp load_initial_data(socket) do
|
|
socket
|
|
|> assign(:executing_jobs, JobMonitoring.list_active_jobs())
|
|
|> assign(:stuck_jobs, JobMonitoring.list_stuck_jobs())
|
|
|> assign(:failed_jobs, JobMonitoring.list_failed_jobs())
|
|
|> assign(:recent_events, JobMonitoring.list_recent_completions(100))
|
|
|> assign(:metrics, Metrics.calculate_all())
|
|
|> assign(:diagnostic_modal_open, false)
|
|
|> assign(:diagnostic_job, nil)
|
|
end
|
|
|
|
defp schedule_metrics_refresh do
|
|
Process.send_after(self(), :refresh_metrics, 10_000)
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(:refresh_metrics, socket) do
|
|
schedule_metrics_refresh()
|
|
{:noreply, assign(socket, :metrics, Metrics.calculate_all())}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(%{event: :started} = event, socket) do
|
|
# Job started - add to executing jobs (reload from DB to get full context)
|
|
job = Towerops.Repo.get(Oban.Job, event.job_id)
|
|
|
|
socket =
|
|
if job do
|
|
update(socket, :executing_jobs, fn jobs -> [job | jobs] end)
|
|
else
|
|
socket
|
|
end
|
|
|
|
{:noreply, socket}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(%{event: :completed} = event, socket) do
|
|
# Job completed - remove from executing, add to recent events
|
|
socket =
|
|
socket
|
|
|> update(:executing_jobs, fn jobs ->
|
|
Enum.reject(jobs, fn j -> j.id == event.job_id end)
|
|
end)
|
|
|> update(:recent_events, fn events ->
|
|
job = Towerops.Repo.get(Oban.Job, event.job_id)
|
|
if job, do: [job | Enum.take(events, 99)], else: events
|
|
end)
|
|
|
|
{:noreply, socket}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(%{event: :failed} = event, socket) do
|
|
# Job failed - remove from executing, add to failed and recent
|
|
job = Towerops.Repo.get(Oban.Job, event.job_id)
|
|
|
|
socket =
|
|
if job do
|
|
socket
|
|
|> update(:executing_jobs, fn jobs ->
|
|
Enum.reject(jobs, fn j -> j.id == event.job_id end)
|
|
end)
|
|
|> update(:failed_jobs, fn jobs -> [job | jobs] end)
|
|
|> update(:recent_events, fn events -> [job | Enum.take(events, 99)] end)
|
|
else
|
|
socket
|
|
end
|
|
|
|
{:noreply, socket}
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 5: Create minimal template**
|
|
|
|
```heex
|
|
<!-- lib/towerops_web/live/admin/monitoring_live.html.heex -->
|
|
<Layouts.admin flash={@flash} timezone={@timezone}>
|
|
<div class="space-y-6">
|
|
<div>
|
|
<h1 class="text-2xl font-bold text-gray-900 dark:text-white">Job Monitoring</h1>
|
|
<p class="text-gray-600 dark:text-gray-400">
|
|
Real-time monitoring of polling and discovery operations
|
|
</p>
|
|
</div>
|
|
|
|
<div class="grid grid-cols-1 lg:grid-cols-5 gap-6">
|
|
<!-- Left column (60%) -->
|
|
<div class="lg:col-span-3 space-y-6">
|
|
<!-- Problems Section -->
|
|
<%= if length(@stuck_jobs) > 0 or length(@failed_jobs) > 0 do %>
|
|
<div class="bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 rounded-lg p-4">
|
|
<h2 class="text-lg font-semibold text-red-900 dark:text-red-100 mb-2">
|
|
⚠️ Problems Detected
|
|
</h2>
|
|
<%= if length(@stuck_jobs) > 0 do %>
|
|
<p class="text-sm text-red-800 dark:text-red-200">
|
|
{length(@stuck_jobs)} stuck jobs
|
|
</p>
|
|
<% end %>
|
|
<%= if length(@failed_jobs) > 0 do %>
|
|
<p class="text-sm text-red-800 dark:text-red-200">
|
|
{length(@failed_jobs)} failed jobs
|
|
</p>
|
|
<% end %>
|
|
</div>
|
|
<% end %>
|
|
|
|
<!-- Active Operations -->
|
|
<div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
|
|
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
|
|
Active Operations
|
|
</h2>
|
|
<p class="text-gray-600 dark:text-gray-400">
|
|
{length(@executing_jobs)} jobs currently executing
|
|
</p>
|
|
</div>
|
|
</div>
|
|
|
|
<!-- Right column (40%) -->
|
|
<div class="lg:col-span-2 space-y-6">
|
|
<!-- Health Metrics -->
|
|
<div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
|
|
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
|
|
Health Metrics
|
|
</h2>
|
|
<dl class="space-y-2">
|
|
<div>
|
|
<dt class="text-sm text-gray-600 dark:text-gray-400">Executing</dt>
|
|
<dd class="text-2xl font-bold text-gray-900 dark:text-white">
|
|
{@metrics.executing_count}
|
|
</dd>
|
|
</div>
|
|
<div>
|
|
<dt class="text-sm text-gray-600 dark:text-gray-400">Queued</dt>
|
|
<dd class="text-2xl font-bold text-gray-900 dark:text-white">
|
|
{@metrics.queued_count}
|
|
</dd>
|
|
</div>
|
|
</dl>
|
|
</div>
|
|
|
|
<!-- Recent Activity -->
|
|
<div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
|
|
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
|
|
Recent Activity
|
|
</h2>
|
|
<p class="text-gray-600 dark:text-gray-400">
|
|
Last {length(@recent_events)} events
|
|
</p>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
</Layouts.admin>
|
|
```
|
|
|
|
**Step 6: Run test to verify it passes**
|
|
|
|
Run: `mix test test/towerops_web/live/admin/monitoring_live_test.exs`
|
|
Expected: PASS
|
|
|
|
**Step 7: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops_web/router.ex lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex test/towerops_web/live/admin/monitoring_live_test.exs
|
|
git commit -m "feat: create MonitoringLive base structure with PubSub subscription"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 9: Add Link to Monitoring Dashboard from Admin Dashboard
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops_web/live/admin/dashboard_live.html.heex`
|
|
|
|
**Step 1: Add monitoring card to dashboard**
|
|
|
|
```heex
|
|
<!-- lib/towerops_web/live/admin/dashboard_live.html.heex -->
|
|
<!-- Add after the LiveDashboard card (around line 59) -->
|
|
|
|
<div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-6">
|
|
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-2">
|
|
<.icon name="hero-cpu-chip" class="w-5 h-5 inline mr-1" /> Job Monitoring
|
|
</h2>
|
|
<p class="text-sm text-gray-600 dark:text-gray-400 mb-4">
|
|
Real-time polling and discovery job monitoring
|
|
</p>
|
|
<.link
|
|
navigate={~p"/admin/monitoring"}
|
|
class="text-blue-600 dark:text-blue-400 hover:underline text-sm"
|
|
>
|
|
Open Monitoring →
|
|
</.link>
|
|
</div>
|
|
```
|
|
|
|
**Step 2: Manual test**
|
|
|
|
Start Phoenix server: `mix phx.server`
|
|
Navigate to: `http://localhost:4000/admin`
|
|
Verify: "Job Monitoring" card appears and link works
|
|
|
|
**Step 3: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops_web/live/admin/dashboard_live.html.heex
|
|
git commit -m "feat: add job monitoring link to admin dashboard"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 10: Enhance Active Operations Display
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex`
|
|
- Modify: `lib/towerops/job_monitoring.ex`
|
|
|
|
**Step 1: Add device preloading to queries**
|
|
|
|
```elixir
|
|
# lib/towerops/job_monitoring.ex
|
|
|
|
# Update list_active_jobs to preload device
|
|
def list_active_jobs do
|
|
from(j in Job,
|
|
where: j.state == "executing",
|
|
where: j.worker in ^@worker_names,
|
|
order_by: [asc: j.attempted_at]
|
|
)
|
|
|> Repo.all()
|
|
|> preload_device_context()
|
|
end
|
|
|
|
# Add helper to fetch device context
|
|
defp preload_device_context(jobs) do
|
|
Enum.map(jobs, fn job ->
|
|
device_id = get_in(job.args, ["device_id"])
|
|
device = device_id && Towerops.Repo.get(Towerops.Devices.Device, device_id)
|
|
Map.put(job, :device, device)
|
|
end)
|
|
end
|
|
```
|
|
|
|
**Step 2: Enhance active operations template**
|
|
|
|
```heex
|
|
<!-- lib/towerops_web/live/admin/monitoring_live.html.heex -->
|
|
<!-- Replace Active Operations section -->
|
|
|
|
<div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
|
|
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
|
|
Active Operations ({length(@executing_jobs)})
|
|
</h2>
|
|
|
|
<%= if length(@executing_jobs) == 0 do %>
|
|
<p class="text-gray-600 dark:text-gray-400 text-sm">No jobs currently executing</p>
|
|
<% else %>
|
|
<div class="space-y-3">
|
|
<%= for job <- @executing_jobs do %>
|
|
<div class="border border-gray-200 dark:border-gray-700 rounded p-3">
|
|
<div class="flex items-start justify-between">
|
|
<div class="flex-1">
|
|
<p class="font-medium text-gray-900 dark:text-white">
|
|
<%= if job.device do %>
|
|
<%= job.device.name %>
|
|
<% else %>
|
|
Device #{get_in(job.args, ["device_id"])}
|
|
<% end %>
|
|
</p>
|
|
<p class="text-sm text-gray-600 dark:text-gray-400">
|
|
<%= worker_name(job.worker) %>
|
|
</p>
|
|
<p class="text-xs text-gray-500 dark:text-gray-500 mt-1">
|
|
Started <%= ToweropsWeb.TimeHelpers.time_ago(job.attempted_at) %>
|
|
</p>
|
|
</div>
|
|
<div class="flex items-center">
|
|
<.icon name="hero-arrow-path" class="w-4 h-4 animate-spin text-blue-600" />
|
|
</div>
|
|
</div>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
```
|
|
|
|
**Step 3: Add helper function to LiveView**
|
|
|
|
```elixir
|
|
# lib/towerops_web/live/admin/monitoring_live.ex (add to module)
|
|
|
|
defp worker_name("Towerops.Workers.DevicePollerWorker"), do: "Device Poll"
|
|
defp worker_name("Towerops.Workers.DiscoveryWorker"), do: "SNMP Discovery"
|
|
defp worker_name(worker), do: worker
|
|
```
|
|
|
|
**Step 4: Manual test**
|
|
|
|
Start Phoenix server and trigger some polling jobs. Verify they appear in Active Operations.
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops/job_monitoring.ex lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex
|
|
git commit -m "feat: enhance active operations display with device context"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 11: Build Problems Section UI
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex`
|
|
|
|
**Step 1: Replace problems section with detailed view**
|
|
|
|
```heex
|
|
<!-- lib/towerops_web/live/admin/monitoring_live.html.heex -->
|
|
<!-- Replace Problems Section -->
|
|
|
|
<%= if length(@stuck_jobs) > 0 or length(@failed_jobs) > 0 do %>
|
|
<div class="bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 rounded-lg p-4">
|
|
<h2 class="text-lg font-semibold text-red-900 dark:text-red-100 mb-4">
|
|
⚠️ Problems Detected
|
|
</h2>
|
|
|
|
<%= if length(@stuck_jobs) > 0 do %>
|
|
<div class="mb-4">
|
|
<h3 class="font-medium text-red-800 dark:text-red-200 mb-2">
|
|
Stuck Jobs ({length(@stuck_jobs)})
|
|
</h3>
|
|
<div class="space-y-2">
|
|
<%= for job <- @stuck_jobs do %>
|
|
<div class="bg-white dark:bg-gray-800 rounded p-3 border border-red-300 dark:border-red-700">
|
|
<div class="flex items-start justify-between">
|
|
<div>
|
|
<p class="font-medium text-gray-900 dark:text-white">
|
|
<%= if job.device do %>
|
|
<%= job.device.name %>
|
|
<% else %>
|
|
Device #{get_in(job.args, ["device_id"])}
|
|
<% end %>
|
|
</p>
|
|
<p class="text-sm text-gray-600 dark:text-gray-400">
|
|
<%= worker_name(job.worker) %>
|
|
</p>
|
|
<p class="text-xs text-red-600 dark:text-red-400 mt-1">
|
|
Running for <%= duration_in_words(job.attempted_at) %>
|
|
</p>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
</div>
|
|
<% end %>
|
|
|
|
<%= if length(@failed_jobs) > 0 do %>
|
|
<div>
|
|
<h3 class="font-medium text-red-800 dark:text-red-200 mb-2">
|
|
Failed Jobs ({length(@failed_jobs)})
|
|
</h3>
|
|
<div class="space-y-2">
|
|
<%= for job <- Enum.take(@failed_jobs, 5) do %>
|
|
<div class="bg-white dark:bg-gray-800 rounded p-3 border border-red-300 dark:border-red-700">
|
|
<div class="flex items-start justify-between">
|
|
<div>
|
|
<p class="font-medium text-gray-900 dark:text-white">
|
|
<%= if job.device do %>
|
|
<%= job.device.name %>
|
|
<% else %>
|
|
Device #{get_in(job.args, ["device_id"])}
|
|
<% end %>
|
|
</p>
|
|
<p class="text-sm text-gray-600 dark:text-gray-400">
|
|
<%= worker_name(job.worker) %> - Attempt <%= job.attempt %>/<%= job.max_attempts %>
|
|
</p>
|
|
<%= if job.errors && length(job.errors) > 0 do %>
|
|
<p class="text-xs text-red-600 dark:text-red-400 mt-1">
|
|
<%= hd(job.errors)["error"] || "Unknown error" %>
|
|
</p>
|
|
<% end %>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
<% end %>
|
|
```
|
|
|
|
**Step 2: Add helper function**
|
|
|
|
```elixir
|
|
# lib/towerops_web/live/admin/monitoring_live.ex (add to module)
|
|
|
|
defp duration_in_words(started_at) do
|
|
seconds = DateTime.diff(DateTime.utc_now(), started_at)
|
|
minutes = div(seconds, 60)
|
|
|
|
cond do
|
|
minutes < 1 -> "#{seconds}s"
|
|
minutes < 60 -> "#{minutes}m"
|
|
true -> "#{div(minutes, 60)}h #{rem(minutes, 60)}m"
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 3: Manual test**
|
|
|
|
Create a long-running job or modify thresholds temporarily to see stuck jobs.
|
|
|
|
**Step 4: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex
|
|
git commit -m "feat: build detailed problems section UI for stuck and failed jobs"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 12: Build Health Metrics Panel
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex`
|
|
|
|
**Step 1: Replace health metrics section with full display**
|
|
|
|
```heex
|
|
<!-- lib/towerops_web/live/admin/monitoring_live.html.heex -->
|
|
<!-- Replace Health Metrics section -->
|
|
|
|
<div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
|
|
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
|
|
Health Metrics
|
|
</h2>
|
|
|
|
<!-- Current Activity -->
|
|
<div class="mb-6">
|
|
<h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
|
|
Current Activity
|
|
</h3>
|
|
<div class="grid grid-cols-2 gap-3">
|
|
<div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
|
|
<p class="text-xs text-gray-600 dark:text-gray-400">Executing</p>
|
|
<p class="text-2xl font-bold text-gray-900 dark:text-white">
|
|
{@metrics.executing_count}
|
|
</p>
|
|
</div>
|
|
<div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
|
|
<p class="text-xs text-gray-600 dark:text-gray-400">Queued</p>
|
|
<p class="text-2xl font-bold text-gray-900 dark:text-white">
|
|
{@metrics.queued_count}
|
|
</p>
|
|
</div>
|
|
<div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
|
|
<p class="text-xs text-gray-600 dark:text-gray-400">Completed (1h)</p>
|
|
<p class="text-2xl font-bold text-green-600 dark:text-green-400">
|
|
{@metrics.completed_last_hour}
|
|
</p>
|
|
</div>
|
|
<div class="bg-gray-50 dark:bg-gray-800 rounded p-3">
|
|
<p class="text-xs text-gray-600 dark:text-gray-400">Failed (1h)</p>
|
|
<p class="text-2xl font-bold text-red-600 dark:text-red-400">
|
|
{@metrics.failed_last_hour}
|
|
</p>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
|
|
<!-- Success Rates -->
|
|
<div class="mb-6">
|
|
<h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
|
|
Success Rates
|
|
</h3>
|
|
<div class="space-y-3">
|
|
<div>
|
|
<div class="flex justify-between text-xs mb-1">
|
|
<span class="text-gray-600 dark:text-gray-400">Polling (1h)</span>
|
|
<span class="font-medium text-gray-900 dark:text-white">
|
|
<%= Float.round(@metrics.polling_success_rate_1h, 1) %>%
|
|
</span>
|
|
</div>
|
|
<div class="w-full bg-gray-200 dark:bg-gray-700 rounded-full h-2">
|
|
<div
|
|
class={"bg-green-600 h-2 rounded-full transition-all"}
|
|
style={"width: #{@metrics.polling_success_rate_1h}%"}
|
|
>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
|
|
<div>
|
|
<div class="flex justify-between text-xs mb-1">
|
|
<span class="text-gray-600 dark:text-gray-400">Discovery (1h)</span>
|
|
<span class="font-medium text-gray-900 dark:text-white">
|
|
<%= Float.round(@metrics.discovery_success_rate_1h, 1) %>%
|
|
</span>
|
|
</div>
|
|
<div class="w-full bg-gray-200 dark:bg-gray-700 rounded-full h-2">
|
|
<div
|
|
class={"bg-blue-600 h-2 rounded-full transition-all"}
|
|
style={"width: #{@metrics.discovery_success_rate_1h}%"}
|
|
>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
|
|
<!-- Performance -->
|
|
<div>
|
|
<h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
|
|
Performance
|
|
</h3>
|
|
<dl class="space-y-2 text-sm">
|
|
<div class="flex justify-between">
|
|
<dt class="text-gray-600 dark:text-gray-400">Avg Poll Duration</dt>
|
|
<dd class="font-medium text-gray-900 dark:text-white">
|
|
{@metrics.avg_poll_duration}s
|
|
</dd>
|
|
</div>
|
|
<div class="flex justify-between">
|
|
<dt class="text-gray-600 dark:text-gray-400">Avg Discovery Duration</dt>
|
|
<dd class="font-medium text-gray-900 dark:text-white">
|
|
{@metrics.avg_discovery_duration}s
|
|
</dd>
|
|
</div>
|
|
<div class="flex justify-between">
|
|
<dt class="text-gray-600 dark:text-gray-400">Jobs/Minute</dt>
|
|
<dd class="font-medium text-gray-900 dark:text-white">
|
|
{@metrics.jobs_per_minute}
|
|
</dd>
|
|
</div>
|
|
</dl>
|
|
</div>
|
|
|
|
<div class="mt-4 pt-4 border-t border-gray-200 dark:border-gray-700">
|
|
<p class="text-xs text-gray-500 dark:text-gray-500">
|
|
Updates every 10 seconds
|
|
</p>
|
|
</div>
|
|
</div>
|
|
```
|
|
|
|
**Step 2: Manual test**
|
|
|
|
View dashboard and verify all metrics display correctly.
|
|
|
|
**Step 3: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops_web/live/admin/monitoring_live.html.heex
|
|
git commit -m "feat: build comprehensive health metrics panel with success rates"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 13: Build Recent Activity Timeline
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex`
|
|
|
|
**Step 1: Replace recent activity section**
|
|
|
|
```heex
|
|
<!-- lib/towerops_web/live/admin/monitoring_live.html.heex -->
|
|
<!-- Replace Recent Activity section -->
|
|
|
|
<div class="bg-white dark:bg-gray-800/50 rounded-lg border border-gray-200 dark:border-white/10 p-4">
|
|
<h2 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">
|
|
Recent Activity
|
|
</h2>
|
|
|
|
<%= if length(@recent_events) == 0 do %>
|
|
<p class="text-gray-600 dark:text-gray-400 text-sm">No recent activity</p>
|
|
<% else %>
|
|
<div class="space-y-2 max-h-96 overflow-y-auto">
|
|
<%= for job <- Enum.take(@recent_events, 50) do %>
|
|
<div class="text-sm border-l-2 pl-3 py-2 <%= event_border_color(job.state) %>">
|
|
<div class="flex items-start justify-between">
|
|
<div class="flex-1">
|
|
<p class="font-medium text-gray-900 dark:text-white">
|
|
<%= if job.device do %>
|
|
<%= job.device.name %>
|
|
<% else %>
|
|
Device #{get_in(job.args, ["device_id"])}
|
|
<% end %>
|
|
</p>
|
|
<p class="text-xs text-gray-600 dark:text-gray-400">
|
|
<%= worker_name(job.worker) %> · <%= event_outcome(job) %>
|
|
</p>
|
|
</div>
|
|
<span class="text-xs text-gray-500 dark:text-gray-500 whitespace-nowrap ml-2">
|
|
<%= ToweropsWeb.TimeHelpers.time_ago(job.completed_at || job.updated_at) %>
|
|
</span>
|
|
</div>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
<% end %>
|
|
</div>
|
|
```
|
|
|
|
**Step 2: Add helper functions**
|
|
|
|
```elixir
|
|
# lib/towerops_web/live/admin/monitoring_live.ex (add to module)
|
|
|
|
defp event_border_color("completed"), do: "border-green-500"
|
|
defp event_border_color("cancelled"), do: "border-red-500"
|
|
defp event_border_color("discarded"), do: "border-gray-400"
|
|
defp event_border_color(_), do: "border-gray-300"
|
|
|
|
defp event_outcome(job) do
|
|
case job.state do
|
|
"completed" -> "Completed"
|
|
"cancelled" -> "Failed"
|
|
"discarded" -> "Discarded"
|
|
_ -> String.capitalize(job.state)
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 3: Manual test**
|
|
|
|
View dashboard and verify recent activity timeline shows recent job completions.
|
|
|
|
**Step 4: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops_web/live/admin/monitoring_live.ex lib/towerops_web/live/admin/monitoring_live.html.heex
|
|
git commit -m "feat: build recent activity timeline with color-coded events"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 14: Add Queue Depth Display
|
|
|
|
**Files:**
|
|
- Modify: `lib/towerops_web/live/admin/monitoring_live.html.heex`
|
|
|
|
**Step 1: Add queue depth section to metrics panel**
|
|
|
|
```heex
|
|
<!-- lib/towerops_web/live/admin/monitoring_live.html.heex -->
|
|
<!-- Add after Performance section in Health Metrics -->
|
|
|
|
<!-- Queue Depths -->
|
|
<div class="mt-6">
|
|
<h3 class="text-sm font-medium text-gray-700 dark:text-gray-300 mb-3">
|
|
Queue Depths
|
|
</h3>
|
|
<dl class="space-y-2 text-sm">
|
|
<div class="flex justify-between">
|
|
<dt class="text-gray-600 dark:text-gray-400">Pollers</dt>
|
|
<dd class="font-medium text-gray-900 dark:text-white">
|
|
{@metrics.queue_depths["pollers"]}
|
|
</dd>
|
|
</div>
|
|
<div class="flex justify-between">
|
|
<dt class="text-gray-600 dark:text-gray-400">Discovery</dt>
|
|
<dd class="font-medium text-gray-900 dark:text-white">
|
|
{@metrics.queue_depths["discovery"]}
|
|
</dd>
|
|
</div>
|
|
<div class="flex justify-between">
|
|
<dt class="text-gray-600 dark:text-gray-400">Maintenance</dt>
|
|
<dd class="font-medium text-gray-900 dark:text-white">
|
|
{@metrics.queue_depths["maintenance"]}
|
|
</dd>
|
|
</div>
|
|
</dl>
|
|
</div>
|
|
```
|
|
|
|
**Step 2: Manual test**
|
|
|
|
View dashboard and verify queue depths display correctly.
|
|
|
|
**Step 3: Commit**
|
|
|
|
```bash
|
|
git add lib/towerops_web/live/admin/monitoring_live.html.heex
|
|
git commit -m "feat: add queue depth display to health metrics"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 15: Final Testing and Polish
|
|
|
|
**Files:**
|
|
- Test: `test/towerops_web/live/admin/monitoring_live_test.exs`
|
|
|
|
**Step 1: Add comprehensive integration tests**
|
|
|
|
```elixir
|
|
# test/towerops_web/live/admin/monitoring_live_test.exs (add tests)
|
|
|
|
describe "real-time updates" do
|
|
test "updates when job starts via PubSub", %{conn: conn, superuser: superuser} do
|
|
conn = log_in_user(conn, superuser)
|
|
{:ok, view, _html} = live(conn, ~p"/admin/monitoring")
|
|
|
|
device = insert(:device)
|
|
job = insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
# Simulate PubSub event
|
|
send(view.pid, %{
|
|
event: :started,
|
|
job_id: job.id,
|
|
worker: job.worker,
|
|
device_id: device.id
|
|
})
|
|
|
|
assert render(view) =~ device.name
|
|
end
|
|
end
|
|
|
|
describe "metrics display" do
|
|
test "shows current activity metrics", %{conn: conn, superuser: superuser} do
|
|
device = insert(:device)
|
|
insert(:oban_job,
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id}
|
|
)
|
|
|
|
conn = log_in_user(conn, superuser)
|
|
{:ok, _view, html} = live(conn, ~p"/admin/monitoring")
|
|
|
|
assert html =~ "Executing"
|
|
assert html =~ "1" # One executing job
|
|
end
|
|
end
|
|
```
|
|
|
|
**Step 2: Run all tests**
|
|
|
|
Run: `mix test test/towerops_web/live/admin/monitoring_live_test.exs`
|
|
Expected: PASS (all tests)
|
|
|
|
**Step 3: Run full test suite to ensure no regressions**
|
|
|
|
Run: `mix test`
|
|
Expected: PASS (all tests)
|
|
|
|
**Step 4: Manual end-to-end test**
|
|
|
|
1. Start Phoenix server: `mix phx.server`
|
|
2. Navigate to `/admin/monitoring`
|
|
3. Verify all sections render
|
|
4. Trigger some polls/discoveries
|
|
5. Verify real-time updates work
|
|
6. Check metrics refresh after 10 seconds
|
|
|
|
**Step 5: Commit**
|
|
|
|
```bash
|
|
git add test/towerops_web/live/admin/monitoring_live_test.exs
|
|
git commit -m "test: add comprehensive integration tests for monitoring dashboard"
|
|
```
|
|
|
|
---
|
|
|
|
## Task 16: Documentation and Cleanup
|
|
|
|
**Files:**
|
|
- Create: `docs/features/job-monitoring-dashboard.md`
|
|
|
|
**Step 1: Write feature documentation**
|
|
|
|
```markdown
|
|
# Job Monitoring Dashboard
|
|
|
|
## Overview
|
|
|
|
The Job Monitoring Dashboard (`/admin/monitoring`) provides real-time visibility into all polling and discovery operations across all organizations.
|
|
|
|
## Features
|
|
|
|
### Real-Time Updates
|
|
- PubSub broadcasts job lifecycle events (start, complete, fail)
|
|
- LiveView updates immediately when jobs change state
|
|
- Metrics refresh every 10 seconds
|
|
|
|
### Problem Detection
|
|
- Stuck jobs (executing longer than threshold)
|
|
- Polling: >2 minutes
|
|
- Discovery: >5 minutes
|
|
- Failed/retrying jobs
|
|
- Discarded jobs
|
|
|
|
### Health Metrics
|
|
- Current activity (executing, queued counts)
|
|
- Success rates (1h and 24h windows)
|
|
- Performance metrics (avg duration, jobs/minute)
|
|
- Queue depths per queue
|
|
|
|
### Recent Activity
|
|
- Last 100 job completions
|
|
- Color-coded by outcome
|
|
- Shows device, operation type, timestamp
|
|
|
|
## Usage
|
|
|
|
### Accessing the Dashboard
|
|
|
|
1. Log in as superuser
|
|
2. Navigate to `/admin` dashboard
|
|
3. Click "Job Monitoring" card
|
|
4. Dashboard loads with current state
|
|
|
|
### Monitoring Jobs
|
|
|
|
**Active Operations** shows currently executing jobs:
|
|
- Device name
|
|
- Operation type (Poll vs Discovery)
|
|
- Duration so far
|
|
- Real-time spinner
|
|
|
|
**Problems** section only appears when issues exist:
|
|
- Stuck jobs with duration
|
|
- Failed jobs with error messages
|
|
- Click any job for diagnostic details (future enhancement)
|
|
|
|
**Health Metrics** panel updates every 10 seconds:
|
|
- Current counts
|
|
- Success rate progress bars
|
|
- Performance statistics
|
|
|
|
**Recent Activity** timeline shows last 100 events:
|
|
- Scrollable list
|
|
- Color-coded borders (green=success, red=failed)
|
|
- Relative timestamps
|
|
|
|
## Technical Details
|
|
|
|
### Architecture
|
|
|
|
```
|
|
MonitoringLive (LiveView)
|
|
├─ Subscribes to "job:lifecycle" PubSub topic
|
|
├─ Loads initial data from JobMonitoring context
|
|
├─ Handles lifecycle events (started, completed, failed)
|
|
└─ Refreshes metrics every 10 seconds
|
|
|
|
JobMonitoring (Context)
|
|
├─ list_active_jobs/0 - executing jobs
|
|
├─ list_stuck_jobs/0 - jobs exceeding threshold
|
|
├─ list_failed_jobs/0 - retrying/cancelled
|
|
└─ list_recent_completions/1 - last N completions
|
|
|
|
Metrics (Module)
|
|
└─ calculate_all/0 - aggregates all metrics
|
|
|
|
Events (Module)
|
|
└─ broadcast_job_event/3 - PubSub broadcasting
|
|
|
|
Workers (DevicePollerWorker, DiscoveryWorker)
|
|
└─ Broadcast lifecycle events on state changes
|
|
```
|
|
|
|
### Performance Considerations
|
|
|
|
- Queries limited to polling/discovery workers only
|
|
- Recent events capped at 100 items
|
|
- Metrics calculation runs every 10s (not per event)
|
|
- Device context preloaded for active jobs only
|
|
|
|
## Future Enhancements
|
|
|
|
- Diagnostic modal with full job details
|
|
- Filtering by organization/device/queue
|
|
- Job cancellation from UI
|
|
- Historical trends and sparklines
|
|
- Alerting when thresholds exceeded
|
|
```
|
|
|
|
**Step 2: Commit**
|
|
|
|
```bash
|
|
git add docs/features/job-monitoring-dashboard.md
|
|
git commit -m "docs: add job monitoring dashboard feature documentation"
|
|
```
|
|
|
|
---
|
|
|
|
## Final Checklist
|
|
|
|
Before marking complete, verify:
|
|
|
|
- [ ] All tests pass: `mix test`
|
|
- [ ] Code is formatted: `mix format`
|
|
- [ ] No compilation warnings: `mix compile --warnings-as-errors`
|
|
- [ ] Credo passes: `mix credo --strict`
|
|
- [ ] Manual testing completed
|
|
- [ ] Dashboard accessible at `/admin/monitoring`
|
|
- [ ] Real-time updates work via PubSub
|
|
- [ ] Metrics refresh every 10 seconds
|
|
- [ ] Problems section appears/disappears correctly
|
|
- [ ] All commits follow conventional commit format
|
|
- [ ] Feature documentation written
|
|
|
|
## Success Criteria
|
|
|
|
✅ Dashboard shows all active polling and discovery jobs
|
|
✅ Real-time updates via PubSub when jobs start/complete/fail
|
|
✅ Health metrics display and refresh automatically
|
|
✅ Problems section detects stuck and failed jobs
|
|
✅ Recent activity timeline shows last 100 events
|
|
✅ Accessible from admin dashboard
|
|
✅ Superadmin-only access enforced
|
|
✅ All tests passing
|