feat(admin): add stats dashboard observability

Author: masterkain

lib/codex_pooler/admin/activity_read_model.ex

@@ -8,6 +8,34 @@ defmodule CodexPooler.Admin.ActivityReadModel do
   alias CodexPooler.Audit.AuditEvent
   alias CodexPooler.Repo
 
+  @type activity_source_counts :: %{
+          required(:audit_events) => non_neg_integer(),
+          required(:jobs) => non_neg_integer()
+        }
+  @type activity_summary :: %{
+          required(:recent_activity) => [map()],
+          required(:source_counts) => activity_source_counts()
+        }
+
+  @spec activity_summary_for_pool_ids([Ecto.UUID.t()], DateTime.t(), DateTime.t()) ::
+          activity_summary()
+  def activity_summary_for_pool_ids([], _started_at, _ended_at) do
+    %{recent_activity: [], source_counts: %{audit_events: 0, jobs: 0}}
+  end
+
+  def activity_summary_for_pool_ids(pool_ids, started_at, ended_at) do
+    audit_activity = audit_events_for(pool_ids, started_at, ended_at)
+    job_activity = jobs_for(pool_ids, started_at, ended_at)
+
+    %{
+      recent_activity: recent_activity(audit_activity, job_activity),
+      source_counts: %{
+        audit_events: source_total(audit_activity),
+        jobs: source_total(job_activity)
+      }
+    }
+  end
+
   @spec recent_activity_for_pool_ids([Ecto.UUID.t()], DateTime.t(), DateTime.t()) :: [map()]
   def recent_activity_for_pool_ids([], _started_at, _ended_at), do: []
 
@@ -17,13 +45,14 @@ defmodule CodexPooler.Admin.ActivityReadModel do
     |> recent_activity(jobs_for(pool_ids, started_at, ended_at))
   end
 
-  @spec activity_source_counts([Ecto.UUID.t()], DateTime.t(), DateTime.t()) :: map()
+  @spec activity_source_counts([Ecto.UUID.t()], DateTime.t(), DateTime.t()) ::
+          activity_source_counts()
   def activity_source_counts([], _started_at, _ended_at), do: %{audit_events: 0, jobs: 0}
 
   def activity_source_counts(pool_ids, started_at, ended_at) do
     %{
-      audit_events: length(audit_events_for(pool_ids, started_at, ended_at)),
-      jobs: length(jobs_for(pool_ids, started_at, ended_at))
+      audit_events: source_total(audit_events_for(pool_ids, started_at, ended_at)),
+      jobs: source_total(jobs_for(pool_ids, started_at, ended_at))
     }
   end
 
@@ -33,16 +62,20 @@ defmodule CodexPooler.Admin.ActivityReadModel do
         where:
           event.pool_id in ^pool_ids and event.occurred_at >= ^started_at and
             event.occurred_at <= ^ended_at,
+        windows: [all: []],
         order_by: [desc: event.occurred_at, desc: event.id],
-        limit: 5,
+        limit: 10,
         select: %{
           type: :audit_event,
           id: event.id,
           occurred_at: event.occurred_at,
           pool_id: event.pool_id,
           action: event.action,
           target_type: event.target_type,
-          outcome: event.outcome
+          outcome: event.outcome,
+          source_rank: 0,
+          source_order_id: event.id,
+          source_total: over(count(event.id), :all)
         }
     )
   end
@@ -53,22 +86,50 @@ defmodule CodexPooler.Admin.ActivityReadModel do
         where:
           fragment("?->>?", job.args, "pool_id") in ^pool_ids and
             job.inserted_at >= ^started_at and job.inserted_at <= ^ended_at,
+        windows: [all: []],
         order_by: [desc: job.inserted_at, desc: job.id],
-        limit: 5,
+        limit: 10,
         select: %{
           type: :job,
           id: job.id,
           occurred_at: job.inserted_at,
           state: job.state,
           worker: job.worker,
-          queue: job.queue
+          queue: job.queue,
+          source_rank: 1,
+          source_order_id: fragment("lpad(?::text, 20, '0')", job.id),
+          source_total: over(count(job.id), :all)
         }
     )
   end
 
   defp recent_activity(audit_activity, job_activity) do
     (audit_activity ++ job_activity)
-    |> Enum.sort_by(& &1.occurred_at, {:desc, DateTime})
+    |> Enum.sort(&activity_before?/2)
     |> Enum.take(10)
+    |> Enum.map(&strip_internal_fields/1)
+  end
+
+  defp activity_before?(left, right) do
+    case DateTime.compare(left.occurred_at, right.occurred_at) do
+      :gt -> true
+      :lt -> false
+      :eq -> activity_tie_before?(left, right)
+    end
+  end
+
+  defp activity_tie_before?(left, right) do
+    cond do
+      left.source_rank < right.source_rank -> true
+      left.source_rank > right.source_rank -> false
+      true -> left.source_order_id > right.source_order_id
+    end
+  end
+
+  defp source_total([]), do: 0
+  defp source_total([%{source_total: source_total} | _rows]), do: source_total
+
+  defp strip_internal_fields(row) do
+    Map.drop(row, [:source_rank, :source_order_id, :source_total])
   end
 end

lib/codex_pooler/admin/stats.ex

@@ -157,19 +157,15 @@ defmodule CodexPooler.Admin.Stats do
     turns =
       GatewayReadModel.turns_for_pool_ids(pool_ids, normalized.started_at, normalized.ended_at)
 
-    recent_activity =
-      ActivityReadModel.recent_activity_for_pool_ids(
+    activity_summary =
+      ActivityReadModel.activity_summary_for_pool_ids(
         pool_ids,
         normalized.started_at,
         normalized.ended_at
       )
 
-    activity_counts =
-      ActivityReadModel.activity_source_counts(
-        pool_ids,
-        normalized.started_at,
-        normalized.ended_at
-      )
+    recent_activity = activity_summary.recent_activity
+    activity_counts = activity_summary.source_counts
 
     quota_accounts =
       Quota.ReadModel.account_summaries_for_pool_ids(pool_ids, normalized.ended_at)

lib/codex_pooler_web/live/admin/pages/stats_live.ex

@@ -11,6 +11,9 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
 
   @stats_reload_debounce_ms 1_000
   @stats_event_topics ~w(request_logs usage upstreams job_status model_sync)
+  @reload_telemetry_event [:codex_pooler, :admin, :stats_live, :reload]
+  @dashboard_build_telemetry_event [:codex_pooler, :admin, :stats, :dashboard, :build]
+  @telemetry_windows ~w(1h 5h 24h 7d)
 
   @window_options [
     {"Last 1 hour", "1h"},
@@ -64,7 +67,7 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
   end
 
   def handle_info(:reload_stats_dashboard, socket) do
-    notify_stats_reload(:reloaded)
+    notify_stats_reload(:executed, socket.assigns.current_params)
 
     {:noreply,
      socket
@@ -179,9 +182,10 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
   defp load_dashboard(socket, params) do
     filters = stats_filters(params)
 
-    case Stats.build_dashboard(socket.assigns.current_scope, filters) do
+    case build_dashboard_with_telemetry(socket.assigns.current_scope, filters) do
       {:ok, dashboard} ->
         socket
+        |> assign(:current_params, params)
         |> reconcile_pool_subscriptions(dashboard)
         |> assign(
           dashboard: dashboard,
@@ -193,6 +197,7 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
 
       {:error, error} ->
         socket
+        |> assign(:current_params, params)
         |> reconcile_pool_subscriptions(nil)
         |> assign(
           dashboard: nil,
@@ -204,6 +209,16 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
     end
   end
 
+  defp build_dashboard_with_telemetry(scope, filters) do
+    started_at = System.monotonic_time()
+    result = Stats.build_dashboard(scope, filters)
+    duration = System.monotonic_time() - started_at
+
+    emit_dashboard_build_telemetry(result, filters, duration)
+
+    result
+  end
+
   defp reconcile_pool_subscriptions(socket, dashboard) do
     if connected?(socket) do
       {socket, stale_pool_ids} =
@@ -232,6 +247,7 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
 
   defp schedule_stats_reload(socket) do
     if is_reference(socket.assigns[:stats_reload_timer]) do
+      notify_stats_reload(:coalesced, socket.assigns.current_params)
       socket
     else
       schedule_new_stats_reload(socket)
@@ -240,13 +256,14 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
 
   defp schedule_new_stats_reload(socket) do
     timer = Process.send_after(self(), :reload_stats_dashboard, @stats_reload_debounce_ms)
-    notify_stats_reload(:scheduled)
+    notify_stats_reload(:scheduled, socket.assigns.current_params)
     assign(socket, :stats_reload_timer, timer)
   end
 
   defp cancel_stats_reload_timer(socket) do
     if is_reference(socket.assigns[:stats_reload_timer]) do
       Process.cancel_timer(socket.assigns.stats_reload_timer, async: false, info: false)
+      notify_stats_reload(:cancelled, socket.assigns.current_params)
     end
 
     assign(socket, :stats_reload_timer, nil)
@@ -257,14 +274,50 @@ defmodule CodexPoolerWeb.Admin.StatsLive do
 
   defp stats_event?(_topics), do: false
 
-  defp notify_stats_reload(stage) do
+  defp notify_stats_reload(stage, params) do
+    filters = stats_filters(params || %{})
+
+    :telemetry.execute(
+      @reload_telemetry_event,
+      %{count: 1},
+      %{stage: stage, window: telemetry_window(filters), scope: telemetry_scope(filters)}
+    )
+  end
+
+  defp emit_dashboard_build_telemetry({:ok, _dashboard}, filters, duration) do
+    :telemetry.execute(
+      @dashboard_build_telemetry_event,
+      %{count: 1, duration: duration},
+      %{outcome: :ok, window: telemetry_window(filters), scope: telemetry_scope(filters)}
+    )
+  end
+
+  defp emit_dashboard_build_telemetry({:error, error}, filters, duration) do
     :telemetry.execute(
-      [:codex_pooler, :admin, :stats_live, :reload],
-      %{},
-      %{stage: stage, pid: self()}
+      @dashboard_build_telemetry_event,
+      %{count: 1, duration: duration},
+      %{
+        outcome: :error,
+        window: telemetry_window(filters),
+        scope: telemetry_scope(filters),
+        error_code: telemetry_error_code(error)
+      }
     )
   end
 
+  defp telemetry_window(%{"window" => window}), do: telemetry_window(window)
+  defp telemetry_window(%{window: window}), do: telemetry_window(window)
+  defp telemetry_window(window) when window in @telemetry_windows, do: window
+  defp telemetry_window(_window), do: "unknown"
+
+  defp telemetry_scope(%{"pool_id" => pool_id}), do: telemetry_scope(pool_id)
+  defp telemetry_scope(%{pool_id: pool_id}), do: telemetry_scope(pool_id)
+  defp telemetry_scope(nil), do: "all_visible_pools"
+  defp telemetry_scope(pool_id) when is_binary(pool_id) and pool_id != "", do: "selected_pool"
+  defp telemetry_scope(_pool_id), do: "unknown"
+
+  defp telemetry_error_code(%{code: code}), do: code
+
   defp stats_filters(params) do
     params = Map.new(params)
 

lib/codex_pooler_web/telemetry.ex

@@ -9,9 +9,16 @@ defmodule CodexPoolerWeb.Telemetry do
   @type http_tags :: %{method: String.t(), status_class: String.t()}
   @type http_route_tags :: %{method: String.t(), route: String.t(), status_class: String.t()}
   @type admission_tags :: %{route_class: String.t(), transport: String.t()}
+  @type admin_stats_reload_tags :: %{stage: String.t(), window: String.t(), scope: String.t()}
+  @type admin_stats_build_tags :: %{outcome: String.t(), window: String.t(), scope: String.t()}
 
   @repo_query_buckets [0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5]
   @admission_queue_buckets [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5]
+  @admin_stats_duration_buckets [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5]
+  @admin_stats_windows ~w(1h 5h 24h 7d)
+  @admin_stats_scopes ~w(selected_pool all_visible_pools)
+  @admin_stats_reload_stages ~w(scheduled coalesced cancelled executed)
+  @admin_stats_build_outcomes ~w(ok error)
   @repo_source_pattern ~r/\A[a-zA-Z0-9_.-]+\z/
   @safe_route_pattern ~r/\A[a-zA-Z0-9_.*:\/{}-]+\z/
   @safe_tag_pattern ~r/\A[a-zA-Z0-9_.:-]+\z/
@@ -149,6 +156,29 @@ defmodule CodexPoolerWeb.Telemetry do
         tag_values: &repo_query_tag_values/1,
         description: "Total Ecto repository queries by source and SQL command."
       ),
+      counter("codex_pooler.admin.stats.reload.count",
+        event_name: [:codex_pooler, :admin, :stats_live, :reload],
+        measurement: :count,
+        tags: [:stage, :window, :scope],
+        tag_values: &admin_stats_reload_tag_values/1,
+        description: "Total admin stats dashboard reload events by stage, window, and scope."
+      ),
+      counter("codex_pooler.admin.stats.dashboard.build.count",
+        event_name: [:codex_pooler, :admin, :stats, :dashboard, :build],
+        measurement: :count,
+        tags: [:outcome, :window, :scope],
+        tag_values: &admin_stats_build_tag_values/1,
+        description: "Total admin stats dashboard builds by outcome, window, and scope."
+      ),
+      distribution("codex_pooler.admin.stats.dashboard.build.duration.seconds",
+        event_name: [:codex_pooler, :admin, :stats, :dashboard, :build],
+        measurement: :duration,
+        unit: {:native, :second},
+        tags: [:outcome, :window, :scope],
+        tag_values: &admin_stats_build_tag_values/1,
+        description: "Admin stats dashboard build duration by outcome, window, and scope.",
+        reporter_options: [buckets: @admin_stats_duration_buckets]
+      ),
       distribution("codex_pooler.repo.query.total_time.seconds",
         event_name: [:codex_pooler, :repo, :query],
         measurement: :total_time,
@@ -446,6 +476,57 @@ defmodule CodexPoolerWeb.Telemetry do
     }
   end
 
+  @spec admin_stats_reload_tag_values(map()) :: admin_stats_reload_tags()
+  defp admin_stats_reload_tag_values(metadata) do
+    %{
+      stage: admin_stats_reload_stage(metadata[:stage]),
+      window: admin_stats_window(metadata[:window]),
+      scope: admin_stats_scope(metadata[:scope])
+    }
+  end
+
+  @spec admin_stats_build_tag_values(map()) :: admin_stats_build_tags()
+  defp admin_stats_build_tag_values(metadata) do
+    %{
+      outcome: admin_stats_build_outcome(metadata[:outcome]),
+      window: admin_stats_window(metadata[:window]),
+      scope: admin_stats_scope(metadata[:scope])
+    }
+  end
+
+  @spec admin_stats_window(term()) :: String.t()
+  defp admin_stats_window(value), do: admin_stats_enum_value(value, @admin_stats_windows)
+
+  @spec admin_stats_scope(term()) :: String.t()
+  defp admin_stats_scope(value), do: admin_stats_enum_value(value, @admin_stats_scopes)
+
+  @spec admin_stats_reload_stage(term()) :: String.t()
+  defp admin_stats_reload_stage(value),
+    do: admin_stats_enum_value(value, @admin_stats_reload_stages)
+
+  @spec admin_stats_build_outcome(term()) :: String.t()
+  defp admin_stats_build_outcome(value),
+    do: admin_stats_enum_value(value, @admin_stats_build_outcomes)
+
+  @spec admin_stats_enum_value(term(), [String.t()]) :: String.t()
+  defp admin_stats_enum_value(value, allowed_values) when is_atom(value) do
+    value
+    |> Atom.to_string()
+    |> admin_stats_enum_value(allowed_values)
+  end
+
+  defp admin_stats_enum_value(value, allowed_values) when is_binary(value) do
+    value = String.trim(value)
+
+    if value in allowed_values do
+      value
+    else
+      "unknown"
+    end
+  end
+
+  defp admin_stats_enum_value(_value, _allowed_values), do: "unknown"
+
   @spec conn_value(term(), atom()) :: term()
   defp conn_value(%{method: method}, :method), do: method
   defp conn_value(%{status: status}, :status), do: status