diff --git a/lib/prom_ex/plugins/oban.ex b/lib/prom_ex/plugins/oban.ex index 03218b4f..8907c586 100644 --- a/lib/prom_ex/plugins/oban.ex +++ b/lib/prom_ex/plugins/oban.ex @@ -438,7 +438,7 @@ if Code.ensure_loaded?(Oban) do config |> Oban.Repo.all(query) - |> include_zeros_for_missing_queue_states() + |> include_zeros_for_missing_queue_states(config) |> Enum.each(fn {{queue, state}, count} -> measurements = %{count: count} metadata = %{name: normalize_module_name(oban_supervisor), queue: queue, state: state} @@ -447,10 +447,10 @@ if Code.ensure_loaded?(Oban) do end) end - defp include_zeros_for_missing_queue_states(query_result) do + defp include_zeros_for_missing_queue_states(query_result, config) do {_, opts} = - Oban.config().plugins - |> Enum.find({nil, [queues: Oban.config().queues]}, fn {plugin, _} -> + config.plugins + |> Enum.find({nil, [queues: config.queues]}, fn {plugin, _} -> plugin == Oban.Pro.Plugins.DynamicQueues end) diff --git a/test/prom_ex/plugins/oban_test.exs b/test/prom_ex/plugins/oban_test.exs index 267f3dd3..494e5f0e 100644 --- a/test/prom_ex/plugins/oban_test.exs +++ b/test/prom_ex/plugins/oban_test.exs @@ -14,6 +14,34 @@ defmodule PromEx.Plugins.ObanTest do end end + defmodule TestRepo do + def child_spec(_opts) do + %{ + id: __MODULE__, + start: {__MODULE__, :start_link, []}, + type: :supervisor + } + end + + def start_link do + Agent.start_link(fn -> :ok end, name: __MODULE__) + end + + def all(_query), do: [] + def all(_query, _opts), do: [] + def get_dynamic_repo, do: nil + + # Required by Oban - provides repository configuration + def config do + [ + otp_app: :prom_ex, + adapter: Ecto.Adapters.Postgres, + database: "test_db", + hostname: "localhost" + ] + end + end + test "telemetry events are accumulated" do start_supervised!(WebApp.PromEx) @@ -39,4 +67,54 @@ defmodule PromEx.Plugins.ObanTest do assert [] == ObanPlugin.manual_metrics([]) end end + + describe "Integration test - named supervisor" do + test "execute_queue_metrics uses config of named supervisor if Oban instance is not running" do + start_supervised!(TestRepo) + + oban_config = [ + # Named supervisor - this is key! + name: CustomApp.Jobs.Oban, + repo: TestRepo, + queues: [default: 10, high: 5], + plugins: [], + # Prevent automatic job processing + testing: :manual, + # Use a minimal notifier for testing + notifier: {Oban.Notifiers.PG, []}, + # Use isolated peer for testing + peer: {Oban.Peers.Isolated, [leader?: false]} + ] + + {:ok, _oban_pid} = start_supervised({Oban, oban_config}) + :timer.sleep(200) + + # Verify our named supervisor is running + named_supervisor_pid = Oban.Registry.whereis(CustomApp.Jobs.Oban) + assert is_pid(named_supervisor_pid), "CustomApp.Jobs.Oban supervisor should be running" + + # Verify we can get its config + config = Oban.Registry.config(CustomApp.Jobs.Oban) + assert config.name == CustomApp.Jobs.Oban + + # Verify there's no default "Oban" instance + default_supervisor_pid = Oban.Registry.whereis(Oban) + assert is_nil(default_supervisor_pid), "Default Oban instance should NOT exist" + + named_supervisors = MapSet.new([CustomApp.Jobs.Oban]) + + refute_error(fn -> + ObanPlugin.execute_queue_metrics(named_supervisors) + end) + end + end + + defp refute_error(fun) do + try do + fun.() + :ok + rescue + error -> flunk("Expected no error, but got: #{inspect(error)}") + end + end end