improve oban logging

lib/cannery/application.ex

@@ -4,6 +4,7 @@ defmodule Cannery.Application do
   @moduledoc false
 
   use Application
+  alias Cannery.ErrorReporter
 
   @impl true
   def start(_type, _args) do
@@ -17,16 +18,24 @@ defmodule Cannery.Application do
       # Start the Endpoint (http/https)
       CanneryWeb.Endpoint,
       # Add Oban
-      {Oban, oban_config()}
+      {Oban, oban_config()},
+      Cannery.Repo.Migrator
       # Start a worker by calling: Cannery.Worker.start_link(arg)
       # {Cannery.Worker, arg}
     ]
 
-    # Automatically migrate on start in prod
-    children =
-      if Application.get_env(:cannery, Cannery.Application, automigrate: false)[:automigrate],
-        do: children ++ [Cannery.Repo.Migrator],
-        else: children
+    # Oban events logging https://hexdocs.pm/oban/Oban.html#module-reporting-errors
+    :ok =
+      :telemetry.attach_many(
+        "oban-logger",
+        [
+          [:oban, :job, :exception],
+          [:oban, :job, :start],
+          [:oban, :job, :stop]
+        ],
+        &ErrorReporter.handle_event/4,
+        []
+      )
 
     # See https://hexdocs.pm/elixir/Supervisor.html
     # for other strategies and supported options

lib/cannery/error_reporter.ex

@@ -0,0 +1,57 @@
+defmodule Cannery.ErrorReporter do
+  @moduledoc """
+  Custom logger for telemetry events
+
+  Oban implementation taken from
+  https://hexdocs.pm/oban/Oban.html#module-reporting-errors
+  """
+
+  require Logger
+
+  def handle_event([:oban, :job, :exception], measure, %{stacktrace: stacktrace} = meta, _config) do
+    data =
+      get_oban_job_data(meta, measure)
+      |> Map.put(:stacktrace, Exception.format_stacktrace(stacktrace))
+
+    Logger.error(meta.reason, data: pretty_encode(data))
+  end
+
+  def handle_event([:oban, :job, :start], measure, meta, _config) do
+    Logger.info("Started oban job", data: get_oban_job_data(meta, measure) |> pretty_encode())
+  end
+
+  def handle_event([:oban, :job, :stop], measure, meta, _config) do
+    Logger.info("Finished oban job", data: get_oban_job_data(meta, measure) |> pretty_encode())
+  end
+
+  def handle_event([:oban, :job, unhandled_event], measure, meta, _config) do
+    data =
+      get_oban_job_data(meta, measure)
+      |> Map.put(:event, unhandled_event)
+
+    Logger.warning("Unhandled oban job event", data: pretty_encode(data))
+  end
+
+  def handle_event(unhandled_event, measure, meta, config) do
+    data = %{
+      event: unhandled_event,
+      meta: meta,
+      measurements: measure,
+      config: config
+    }
+
+    Logger.warning("Unhandled telemetry event", data: pretty_encode(data))
+  end
+
+  defp get_oban_job_data(%{job: job}, measure) do
+    job
+    |> Map.take([:id, :args, :meta, :queue, :worker])
+    |> Map.merge(measure)
+  end
+
+  defp pretty_encode(data) do
+    data
+    |> Jason.encode!()
+    |> Jason.Formatter.pretty_print()
+  end
+end

lib/cannery/repo/migrator.ex

@@ -11,12 +11,15 @@ defmodule Cannery.Repo.Migrator do
   end
 
   def init(_opts) do
-    migrate!()
-    {:ok, nil}
+    {:ok, if(automigrate_enabled?(), do: migrate!())}
   end
 
   def migrate! do
     path = Application.app_dir(:cannery, "priv/repo/migrations")
     Ecto.Migrator.run(Cannery.Repo, path, :up, all: true)
   end
+
+  defp automigrate_enabled? do
+    Application.get_env(:cannery, Cannery.Application, automigrate: false)[:automigrate]
+  end
 end

lib/cannery_web/telemetry.ex

@@ -57,6 +57,30 @@ defmodule CanneryWeb.Telemetry do
           "The time the connection spent waiting before being checked out for the query"
       ),
 
+      # Oban Metrics
+      counter("oban.job.exception",
+        tags: [:queue, :worker],
+        event_name: [:oban, :job, :exception],
+        measurement: :duration,
+        description: "Number of oban jobs that raised an exception"
+      ),
+      counter("oban.job.start",
+        tags: [:queue, :worker],
+        event_name: [:oban, :job, :start],
+        measurement: :system_time,
+        description: "Number of oban jobs started"
+      ),
+      summary("oban.job.stop.duration",
+        tags: [:queue, :worker],
+        unit: {:native, :millisecond},
+        description: "Length of time spent processing the oban job"
+      ),
+      summary("oban.job.stop.queue_time",
+        tags: [:queue, :worker],
+        unit: {:native, :millisecond},
+        description: "Time the oban job spent waiting in milliseconds"
+      ),
+
       # VM Metrics
       summary("vm.memory.total", unit: {:byte, :kilobyte}),
       summary("vm.total_run_queue_lengths.total"),