Merge pull request #463 from newrelic/vince/oban-instrumentation

Oban Instrumentation

examples/apps/oban_example/.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

examples/apps/oban_example/.gitignore

@@ -0,0 +1,26 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+oban_example-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/

examples/apps/oban_example/README.md

@@ -0,0 +1,3 @@
+# ObanExample
+
+An example app demonstrating auto-instrumentation of Oban

examples/apps/oban_example/lib/oban_example/application.ex

@@ -0,0 +1,15 @@
+defmodule ObanExample.Application do
+  use Application
+
+  def start(_type, _args) do
+    config = [
+      notifier: Oban.Notifiers.PG,
+      testing: :inline
+    ]
+
+    children = [{Oban, config}]
+
+    opts = [strategy: :one_for_one, name: ObanExample.Supervisor]
+    Supervisor.start_link(children, opts)
+  end
+end

examples/apps/oban_example/lib/oban_example/worker.ex

@@ -0,0 +1,13 @@
+defmodule ObanExample.Worker do
+  use Oban.Worker
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"error" => message}}) do
+    {:error, message}
+  end
+
+  def perform(%Oban.Job{args: _args}) do
+    Process.sleep(:rand.uniform(50))
+    :ok
+  end
+end

examples/apps/oban_example/mix.exs

@@ -0,0 +1,34 @@
+defmodule ObanExample.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :oban_example,
+      version: "0.1.0",
+      build_path: "../../_build",
+      config_path: "../../config/config.exs",
+      deps_path: "../../deps",
+      lockfile: "../../mix.lock",
+      elixir: "~> 1.16",
+      start_permanent: Mix.env() == :prod,
+      deps: deps()
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger],
+      mod: {ObanExample.Application, []}
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      {:new_relic_agent, path: "../../../"},
+      {:test_support, in_umbrella: true},
+      {:oban, "~> 2.0"}
+    ]
+  end
+end

examples/apps/oban_example/test/oban_example_test.exs

@@ -0,0 +1,64 @@
+defmodule ObanExampleTest do
+  use ExUnit.Case
+
+  alias NewRelic.Harvest.Collector
+
+  setup_all context, do: TestSupport.simulate_agent_enabled(context)
+  setup_all context, do: TestSupport.simulate_agent_run(context)
+
+  test "instruments a job" do
+    TestSupport.restart_harvest_cycle(Collector.Metric.HarvestCycle)
+    TestSupport.restart_harvest_cycle(Collector.TransactionEvent.HarvestCycle)
+
+    ObanExample.Worker.new(%{some: "args"}, tags: ["foo", "bar"])
+    |> Oban.insert()
+
+    metrics = TestSupport.gather_harvest(Collector.Metric.Harvester)
+    [event | _] = TestSupport.gather_harvest(Collector.TransactionEvent.Harvester)
+
+    assert TestSupport.find_metric(
+             metrics,
+             "OtherTransaction/Oban/default/ObanExample.Worker/perform",
+             1
+           )
+
+    assert [
+             %{:name => "OtherTransaction/Oban/default/ObanExample.Worker/perform"},
+             %{
+               :"oban.worker" => "ObanExample.Worker",
+               :"oban.queue" => "default",
+               :"oban.job.result" => "success",
+               :"oban.job.tags" => "foo,bar"
+             }
+           ] = event
+  end
+
+  test "instruments a failed job" do
+    TestSupport.restart_harvest_cycle(Collector.Metric.HarvestCycle)
+    TestSupport.restart_harvest_cycle(Collector.TransactionEvent.HarvestCycle)
+
+    ObanExample.Worker.new(%{error: "error!"}, tags: ["foo", "bar"])
+    |> Oban.insert()
+
+    metrics = TestSupport.gather_harvest(Collector.Metric.Harvester)
+    [event | _] = TestSupport.gather_harvest(Collector.TransactionEvent.Harvester)
+
+    assert TestSupport.find_metric(
+             metrics,
+             "OtherTransaction/Oban/default/ObanExample.Worker/perform",
+             1
+           )
+
+    assert [
+             %{:name => "OtherTransaction/Oban/default/ObanExample.Worker/perform"},
+             %{
+               :error => true,
+               :error_kind => :error,
+               :"oban.worker" => "ObanExample.Worker",
+               :"oban.queue" => "default",
+               :"oban.job.result" => "failure",
+               :"oban.job.tags" => "foo,bar"
+             }
+           ] = event
+  end
+end

examples/apps/oban_example/test/test_helper.exs

@@ -0,0 +1 @@
+ExUnit.start()

examples/config/config.exs

@@ -2,14 +2,12 @@ import Config
 
 config :logger, level: :debug
 
-# Pretend the agent is configured
 config :new_relic_agent,
   app_name: "ExampleApps",
-  license_key: "license_key",
   trusted_account_key: "trusted_account_key"
 
 for config <- "../apps/*/config/config.exs" |> Path.expand(__DIR__) |> Path.wildcard() do
   import_config config
 end
 
-if File.exists?("../../config/secret.exs"), do: import_config("secret.exs")
+if File.exists?("config/secret.exs"), do: import_config("secret.exs")

examples/mix.lock

@@ -22,6 +22,7 @@
   "mime": {:hex, :mime, "2.0.5", "dc34c8efd439abe6ae0343edbb8556f4d63f178594894720607772a041b04b02", [:mix], [], "hexpm", "da0d64a365c45bc9935cc5c8a7fc5e49a0e0f9932a761c55d6c52b142780a05c"},
   "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"},
   "myxql": {:hex, :myxql, "0.6.4", "1502ea37ee23c31b79725b95d4cc3553693c2bda7421b1febc50722fd988c918", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:geo, "~> 3.4", [hex: :geo, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a3307f4671f3009d3708283649adf205bfe280f7e036fc8ef7f16dbf821ab8e9"},
+  "oban": {:hex, :oban, "2.18.3", "1608c04f8856c108555c379f2f56bc0759149d35fa9d3b825cb8a6769f8ae926", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "36ca6ca84ef6518f9c2c759ea88efd438a3c81d667ba23b02b062a0aa785475e"},
   "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"},
   "phoenix": {:hex, :phoenix, "1.7.10", "02189140a61b2ce85bb633a9b6fd02dff705a5f1596869547aeb2b2b95edd729", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.6", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5.3", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "cf784932e010fd736d656d7fead6a584a4498efefe5b8227e9f383bf15bb79d0"},
   "phoenix_html": {:hex, :phoenix_html, "3.3.3", "380b8fb45912b5638d2f1d925a3771b4516b9a78587249cabe394e0a5d579dc9", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "923ebe6fec6e2e3b3e569dfbdc6560de932cd54b000ada0208b5f45024bdd76c"},

lib/new_relic/config.ex

@@ -151,6 +151,8 @@ defmodule NewRelic.Config do
     * Controls all Ecto instrumentation
   * `:redix_instrumentation_enabled` (default `true`)
     * Controls all Redix instrumentation
+  * `:oban_instrumentation_enabled` (default `true`)
+    * Controls all Oban instrumentation
   * `:request_queuing_metrics_enabled`
     * Controls collection of request queuing metrics
 
@@ -191,6 +193,10 @@ defmodule NewRelic.Config do
     get(:features, :redix_instrumentation)
   end
 
+  def feature?(:oban_instrumentation) do
+    get(:features, :oban_instrumentation)
+  end
+
   def feature?(:function_argument_collection) do
     get(:features, :function_argument_collection)
   end

lib/new_relic/init.ex

@@ -98,6 +98,11 @@ defmodule NewRelic.Init do
           "NEW_RELIC_PHOENIX_INSTRUMENTATION_ENABLED",
           :phoenix_instrumentation_enabled
         ),
+      oban_instrumentation:
+        determine_feature(
+          "NEW_RELIC_OBAN_INSTRUMENTATION_ENABLED",
+          :oban_instrumentation_enabled
+        ),
       function_argument_collection:
         determine_feature(
           "NEW_RELIC_FUNCTION_ARGUMENT_COLLECTION_ENABLED",

lib/new_relic/telemetry/oban.ex

@@ -0,0 +1,131 @@
+defmodule NewRelic.Telemetry.Oban do
+  use GenServer
+
+  @moduledoc """
+  Provides `Oban` instrumentation via `telemetry`.
+
+  Oban jobs are auto-discovered and instrumented.
+
+  We automatically gather:
+
+  * Transaction metrics and events
+  * Transaction Traces
+  * Distributed Traces
+
+  You can opt-out of this instrumentation via configuration. See `NewRelic.Config` for details.
+  """
+
+  alias NewRelic.Transaction
+
+  @doc false
+  def start_link(_) do
+    config = %{
+      enabled?: NewRelic.Config.feature?(:oban_instrumentation),
+      handler_id: {:new_relic, :oban}
+    }
+
+    GenServer.start_link(__MODULE__, config, name: __MODULE__)
+  end
+
+  @oban_start [:oban, :job, :start]
+  @oban_stop [:oban, :job, :stop]
+  @oban_exception [:oban, :job, :exception]
+
+  @oban_events [
+    @oban_start,
+    @oban_stop,
+    @oban_exception
+  ]
+
+  @doc false
+  def init(%{enabled?: false}), do: :ignore
+
+  def init(%{enabled?: true} = config) do
+    :telemetry.attach_many(
+      config.handler_id,
+      @oban_events,
+      &__MODULE__.handle_event/4,
+      config
+    )
+
+    Process.flag(:trap_exit, true)
+    {:ok, config}
+  end
+
+  @doc false
+  def terminate(_reason, %{handler_id: handler_id}) do
+    :telemetry.detach(handler_id)
+  end
+
+  @doc false
+  def handle_event(
+        @oban_start,
+        %{system_time: system_time},
+        meta,
+        _config
+      ) do
+    Transaction.Reporter.start_transaction(:other)
+    NewRelic.DistributedTrace.start(:other)
+
+    add_start_attrs(meta, system_time)
+  end
+
+  def handle_event(
+        @oban_stop,
+        %{duration: duration} = meas,
+        meta,
+        _config
+      ) do
+    add_stop_attrs(meas, meta, duration)
+
+    Transaction.Reporter.stop_transaction(:other)
+  end
+
+  def handle_event(
+        @oban_exception,
+        %{duration: duration} = meas,
+        %{kind: kind} = meta,
+        _config
+      ) do
+    add_stop_attrs(meas, meta, duration)
+    {reason, stack} = NewRelic.Util.Telemetry.reason_and_stack(meta)
+
+    Transaction.Reporter.fail(%{kind: kind, reason: reason, stack: stack})
+    Transaction.Reporter.stop_transaction(:other)
+  end
+
+  def handle_event(_event, _measurements, _meta, _config) do
+    :ignore
+  end
+
+  defp add_start_attrs(meta, system_time) do
+    [
+      pid: inspect(self()),
+      system_time: system_time,
+      other_transaction_name: "Oban/#{meta.queue}/#{meta.worker}/perform",
+      "oban.worker": meta.worker,
+      "oban.queue": meta.queue,
+      "oban.job.args": meta.job.args,
+      "oban.job.tags": meta.job.tags |> Enum.join(","),
+      "oban.job.attempt": meta.job.attempt,
+      "oban.job.attempted_by": meta.job.attempted_by |> Enum.join("."),
+      "oban.job.max_attempts": meta.job.max_attempts,
+      "oban.job.priority": meta.job.priority
+    ]
+    |> NewRelic.add_attributes()
+  end
+
+  @kb 1024
+  defp add_stop_attrs(meas, meta, duration) do
+    info = Process.info(self(), [:memory, :reductions])
+
+    [
+      duration: duration,
+      memory_kb: info[:memory] / @kb,
+      reductions: info[:reductions],
+      "oban.job.result": meta.state,
+      "oban.job.queue_time": meas.queue_time
+    ]
+    |> NewRelic.add_attributes()
+  end
+end

lib/new_relic/telemetry/plug.ex

@@ -149,7 +149,7 @@ defmodule NewRelic.Telemetry.Plug do
         _config
       ) do
     add_stop_attrs(meas, meta, server)
-    {reason, stack} = reason_and_stack(meta)
+    {reason, stack} = NewRelic.Util.Telemetry.reason_and_stack(meta)
 
     Transaction.Reporter.fail(%{kind: kind, reason: reason, stack: stack})
     Transaction.Reporter.stop_transaction(:web)
@@ -303,23 +303,6 @@ defmodule NewRelic.Telemetry.Plug do
     status
   end
 
-  defp reason_and_stack(%{reason: %{__exception__: true} = reason, stacktrace: stack}) do
-    {reason, stack}
-  end
-
-  defp reason_and_stack(%{reason: {{reason, stack}, _init_call}}) do
-    {reason, stack}
-  end
-
-  defp reason_and_stack(%{reason: {reason, _init_call}}) do
-    {reason, []}
-  end
-
-  defp reason_and_stack(unexpected_cowboy_exception) do
-    NewRelic.log(:debug, "unexpected_cowboy_exception: #{inspect(unexpected_cowboy_exception)}")
-    {:unexpected_cowboy_exception, []}
-  end
-
   defp plug_name(conn, match_path) do
     "/Plug/#{conn.method}/#{match_path}"
     |> String.replace("/*glob", "")

lib/new_relic/telemetry/supervisor.ex

@@ -12,7 +12,8 @@ defmodule NewRelic.Telemetry.Supervisor do
       NewRelic.Telemetry.Ecto.Supervisor,
       NewRelic.Telemetry.Redix,
       NewRelic.Telemetry.Plug,
-      NewRelic.Telemetry.Phoenix
+      NewRelic.Telemetry.Phoenix,
+      NewRelic.Telemetry.Oban
     ]
 
     Supervisor.init(children, strategy: :one_for_one)