Add Cron support for Quantum jobs (#699)

Author: whatyouhide

lib/sentry/application.ex

@@ -51,5 +51,9 @@ defmodule Sentry.Application do
     if config[:oban][:cron][:enabled] do
       Sentry.Cron.Oban.attach_telemetry_handler()
     end
+
+    if config[:quantum][:cron][:enabled] do
+      Sentry.Cron.Quantum.attach_telemetry_handler()
+    end
   end
 end

lib/sentry/config.ex

@@ -27,6 +27,32 @@ defmodule Sentry.Config do
           ]
         ]
       ]
+    ],
+    quantum: [
+      type: :keyword_list,
+      doc: """
+      Configuration for the [Quantum](https://github.com/quantum-elixir/quantum-core) integration.
+      *Available since v10.2.0*.
+      """,
+      keys: [
+        cron: [
+          doc: """
+          Configuration options for configuring [*crons*](https://docs.sentry.io/product/crons/)
+          for Quantum.
+          """,
+          type: :keyword_list,
+          keys: [
+            enabled: [
+              type: :boolean,
+              default: false,
+              doc: """
+              Whether to enable the Quantum integration. When enabled, the Sentry SDK will
+              capture check-ins for Quantum jobs. *Available since v10.2.0*.
+              """
+            ]
+          ]
+        ]
+      ]
     ]
   ]
 

lib/sentry/cron/quantum.ex

@@ -0,0 +1,95 @@
+defmodule Sentry.Cron.Quantum do
+  @moduledoc false
+
+  @events [
+    [:quantum, :job, :start],
+    [:quantum, :job, :stop],
+    [:quantum, :job, :exception]
+  ]
+
+  @spec attach_telemetry_handler() :: :ok
+  def attach_telemetry_handler do
+    _ = :telemetry.attach_many(__MODULE__, @events, &__MODULE__.handle_event/4, :no_config)
+    :ok
+  end
+
+  @spec handle_event([atom()], term(), term(), :no_config) :: :ok
+  def handle_event(event, measurements, metadata, _config)
+
+  def handle_event(
+        [:quantum, :job, event],
+        measurements,
+        %{job: %mod{schedule: schedule}} = metadata,
+        _config
+      )
+      when event in [:start, :stop, :exception] and mod == Quantum.Job and not is_nil(schedule) do
+    _ = handle_event(event, measurements, metadata)
+    :ok
+  end
+
+  def handle_event([:quantum, :job, event], _measurements, _metadata, _config)
+      when event in [:start, :stop, :exception] do
+    :ok
+  end
+
+  ## Helpers
+
+  defp handle_event(:start, _measurements, metadata) do
+    if opts = check_in_opts(metadata) do
+      opts
+      |> Keyword.merge(status: :in_progress)
+      |> Sentry.capture_check_in()
+    end
+  end
+
+  defp handle_event(:stop, measurements, metadata) do
+    if opts = check_in_opts(metadata) do
+      opts
+      |> Keyword.merge(status: :ok, duration: duration_in_seconds(measurements))
+      |> Sentry.capture_check_in()
+    end
+  end
+
+  defp handle_event(:exception, measurements, metadata) do
+    if opts = check_in_opts(metadata) do
+      opts
+      |> Keyword.merge(status: :error, duration: duration_in_seconds(measurements))
+      |> Sentry.capture_check_in()
+    end
+  end
+
+  defp check_in_opts(%{job: job} = metadata) when is_struct(job, Quantum.Job) do
+    if schedule_opts = schedule_opts(job) do
+      id = metadata.telemetry_span_context |> :erlang.phash2() |> Integer.to_string()
+
+      [
+        check_in_id: "quantum-#{id}",
+        # This is already a binary.
+        monitor_slug: "quantum-#{inspect(job.name)}",
+        monitor_config: [schedule: schedule_opts]
+      ]
+    else
+      nil
+    end
+  end
+
+  defp schedule_opts(job) when is_struct(job, Quantum.Job) do
+    case apply(Crontab.CronExpression.Composer, :compose, [job.schedule]) do
+      "@hourly" -> [type: :interval, value: 1, unit: :hour]
+      "@daily" -> [type: :interval, value: 1, unit: :day]
+      "@weekly" -> [type: :interval, value: 1, unit: :week]
+      "@monthly" -> [type: :interval, value: 1, unit: :month]
+      "@yearly" -> [type: :interval, value: 1, unit: :year]
+      "@annually" -> [type: :interval, value: 1, unit: :year]
+      "@reboot" -> nil
+      cron_expr when is_binary(cron_expr) -> [type: :crontab, value: cron_expr]
+      _other -> nil
+    end
+  end
+
+  defp duration_in_seconds(%{duration: duration} = _measurements) do
+    duration
+    |> System.convert_time_unit(:native, :millisecond)
+    |> Kernel./(1000)
+  end
+end

mix.exs

@@ -97,10 +97,10 @@ defmodule Sentry.Mixfile do
       {:excoveralls, "~> 0.17.1", only: [:test]},
       {:phoenix, "~> 1.5", only: [:test]},
       {:phoenix_html, "~> 2.0", only: [:test]}
-    ] ++ maybe_oban_optional_dependency()
+    ] ++ maybe_oban_optional_dependency() ++ maybe_quantum_optional_dependency()
   end
 
-  # TODO: Remove this once we drop support for Elixir 1.13.
+  # TODO: Remove this once we drop support for Elixir < 1.13.
   defp maybe_oban_optional_dependency do
     if Version.match?(System.version(), "~> 1.13") do
       [{:oban, "~> 2.17 and >= 2.17.6", only: [:test]}]
@@ -109,6 +109,15 @@ defmodule Sentry.Mixfile do
     end
   end
 
+  # TODO: Remove this once we drop support for Elixir < 1.12.
+  defp maybe_quantum_optional_dependency do
+    if Version.match?(System.version(), "~> 1.12") do
+      [{:quantum, "~> 3.0", only: [:test]}]
+    else
+      []
+    end
+  end
+
   defp package do
     [
       files: ["lib", "LICENSE", "mix.exs", "README.md", "CHANGELOG.md"],

mix.lock

@@ -4,6 +4,7 @@
   "cowboy": {:hex, :cowboy, "2.8.0", "f3dc62e35797ecd9ac1b50db74611193c29815401e53bac9a5c0577bd7bc667d", [:rebar3], [{:cowlib, "~> 2.9.1", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.7.1", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "4643e4fba74ac96d4d152c75803de6fad0b3fa5df354c71afdd6cbeeb15fac8a"},
   "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.3.1", "ebd1a1d7aff97f27c66654e78ece187abdc646992714164380d8a041eda16754", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "3a6efd3366130eab84ca372cbd4a7d3c3a97bdfcfb4911233b035d117063f0af"},
   "cowlib": {:hex, :cowlib, "2.9.1", "61a6c7c50cf07fdd24b2f45b89500bb93b6686579b069a89f88cb211e1125c78", [:rebar3], [], "hexpm", "e4175dc240a70d996156160891e1c62238ede1729e45740bdd38064dad476170"},
+  "crontab": {:hex, :crontab, "1.1.13", "3bad04f050b9f7f1c237809e42223999c150656a6b2afbbfef597d56df2144c5", [:mix], [{:ecto, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :ecto, repo: "hexpm", optional: true]}], "hexpm", "d67441bec989640e3afb94e123f45a2bc42d76e02988c9613885dc3d01cf7085"},
   "db_connection": {:hex, :db_connection, "2.6.0", "77d835c472b5b67fc4f29556dee74bf511bbafecdcaf98c27d27fa5918152086", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c2f992d15725e721ec7fbc1189d4ecdb8afef76648c746a8e1cad35e3b8a35f3"},
   "decimal": {:hex, :decimal, "2.1.1", "5611dca5d4b2c3dd497dec8f68751f1f1a54755e8ed2a966c2633cf885973ad6", [:mix], [], "hexpm", "53cfe5f497ed0e7771ae1a475575603d77425099ba5faef9394932b35020ffcc"},
   "dialyxir": {:hex, :dialyxir, "1.1.0", "c5aab0d6e71e5522e77beff7ba9e08f8e02bad90dfbeffae60eaf0cb47e29488", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "07ea8e49c45f15264ebe6d5b93799d4dd56a44036cf42d0ad9c960bc266c0b9a"},
@@ -13,6 +14,7 @@
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
   "ex_doc": {:hex, :ex_doc, "0.29.4", "6257ecbb20c7396b1fe5accd55b7b0d23f44b6aa18017b415cb4c2b91d997729", [:mix], [{:earmark_parser, "~> 1.4.31", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "2c6699a737ae46cb61e4ed012af931b57b699643b24dabe2400a8168414bc4f5"},
   "excoveralls": {:hex, :excoveralls, "0.17.1", "83fa7906ef23aa7fc8ad7ee469c357a63b1b3d55dd701ff5b9ce1f72442b2874", [:mix], [{:castore, "~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "95bc6fda953e84c60f14da4a198880336205464e75383ec0f570180567985ae0"},
+  "gen_stage": {:hex, :gen_stage, "1.2.1", "19d8b5e9a5996d813b8245338a28246307fd8b9c99d1237de199d21efc4c76a1", [:mix], [], "hexpm", "83e8be657fa05b992ffa6ac1e3af6d57aa50aace8f691fcf696ff02f8335b001"},
   "hackney": {:hex, :hackney, "1.17.4", "99da4674592504d3fb0cfef0db84c3ba02b4508bae2dff8c0108baa0d6e0977c", [:rebar3], [{:certifi, "~>2.6.1", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~>6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~>1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~>1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~>0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "de16ff4996556c8548d512f4dbe22dd58a587bf3332e7fd362430a7ef3986b16"},
   "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~>0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"},
   "jason": {:hex, :jason, "1.4.1", "af1504e35f629ddcdd6addb3513c3853991f694921b1b9368b0bd32beb9f1b63", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "fbb01ecdfd565b56261302f7e1fcc27c4fb8f32d56eab74db621fc154604a7a1"},
@@ -33,8 +35,10 @@
   "plug": {:hex, :plug, "1.11.1", "f2992bac66fdae679453c9e86134a4201f6f43a687d8ff1cd1b2862d53c80259", [:mix], [{:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "23524e4fefbb587c11f0833b3910bfb414bf2e2534d61928e920f54e3a1b881f"},
   "plug_cowboy": {:hex, :plug_cowboy, "2.4.1", "779ba386c0915027f22e14a48919a9545714f849505fa15af2631a0d298abf0f", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "d72113b6dff7b37a7d9b2a5b68892808e3a9a752f2bf7e503240945385b70507"},
   "plug_crypto": {:hex, :plug_crypto, "1.2.2", "05654514ac717ff3a1843204b424477d9e60c143406aa94daf2274fdd280794d", [:mix], [], "hexpm", "87631c7ad914a5a445f0a3809f99b079113ae4ed4b867348dd9eec288cecb6db"},
+  "quantum": {:hex, :quantum, "3.5.3", "ee38838a07761663468145f489ad93e16a79440bebd7c0f90dc1ec9850776d99", [:mix], [{:crontab, "~> 1.1", [hex: :crontab, repo: "hexpm", optional: false]}, {:gen_stage, "~> 0.14 or ~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_registry, "~> 0.2", [hex: :telemetry_registry, repo: "hexpm", optional: false]}], "hexpm", "500fd3fa77dcd723ed9f766d4a175b684919ff7b6b8cfd9d7d0564d58eba8734"},
   "ranch": {:hex, :ranch, "1.7.1", "6b1fab51b49196860b733a49c07604465a47bdb78aa10c1c16a3d199f7f8c881", [:rebar3], [], "hexpm", "451d8527787df716d99dc36162fca05934915db0b6141bbdac2ea8d3c7afc7d7"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
   "telemetry": {:hex, :telemetry, "0.4.3", "a06428a514bdbc63293cd9a6263aad00ddeb66f608163bdec7c8995784080818", [:rebar3], [], "hexpm", "eb72b8365ffda5bed68a620d1da88525e326cb82a75ee61354fc24b844768041"},
+  "telemetry_registry": {:hex, :telemetry_registry, "0.2.1", "fe648a691f2128e4279d993cd010994c67f282354dc061e697bf070d4b87b480", [:mix, :rebar3], [{:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "4221cefbcadd0b3e7076960339223742d973f1371bc20f3826af640257bc3690"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"},
 }

test/sentry/cron/quantum_test.exs

@@ -0,0 +1,174 @@
+# TODO: Quantum requires Elixir 1.12+, remove this once we depend on that too.
+if Version.match?(System.version(), "~> 1.12") do
+  defmodule Sentry.Cron.QuantumTest do
+    use Sentry.Case, async: false
+
+    import Sentry.TestHelpers
+
+    defmodule Scheduler do
+      use Quantum, otp_app: :sentry
+    end
+
+    setup_all do
+      Sentry.Cron.Quantum.attach_telemetry_handler()
+    end
+
+    setup do
+      bypass = Bypass.open()
+
+      put_test_config(
+        dsn: "http://public:secret@localhost:#{bypass.port}/1",
+        dedup_events: false,
+        environment_name: "test"
+      )
+
+      %{bypass: bypass}
+    end
+
+    for event_type <- [:start, :stop, :exception] do
+      test "ignores #{event_type} events without a cron meta", %{bypass: bypass} do
+        Bypass.down(bypass)
+
+        :telemetry.execute([:quantum, :job, unquote(event_type)], %{}, %{
+          job: Scheduler.new_job(name: :test_job)
+        })
+      end
+
+      test "ignores #{event_type} events with a cron expr of @reboot", %{bypass: bypass} do
+        Bypass.down(bypass)
+
+        :telemetry.execute([:quantum, :job, unquote(event_type)], %{}, %{
+          job:
+            Scheduler.new_job(
+              name: :reboot_job,
+              schedule: Crontab.CronExpression.Parser.parse!("@reboot")
+            )
+        })
+      end
+    end
+
+    test "captures start events with monitor config", %{bypass: bypass} do
+      test_pid = self()
+      ref = make_ref()
+
+      Bypass.expect_once(bypass, "POST", "/api/1/envelope/", fn conn ->
+        {:ok, body, conn} = Plug.Conn.read_body(conn)
+        assert [{headers, check_in_body}] = decode_envelope!(body)
+
+        assert headers["type"] == "check_in"
+
+        assert check_in_body["check_in_id"] == "quantum-#{:erlang.phash2(ref)}"
+        assert check_in_body["status"] == "in_progress"
+        assert check_in_body["monitor_slug"] == "quantum-:test_job"
+        assert check_in_body["duration"] == nil
+        assert check_in_body["environment"] == "test"
+
+        assert check_in_body["monitor_config"] == %{
+                 "schedule" => %{
+                   "type" => "crontab",
+                   "value" => "0 0 * * * *"
+                 }
+               }
+
+        send(test_pid, {ref, :done})
+
+        Plug.Conn.send_resp(conn, 200, ~s<{"id": "1923"}>)
+      end)
+
+      :telemetry.execute([:quantum, :job, :start], %{}, %{
+        job:
+          Scheduler.new_job(
+            name: :test_job,
+            schedule: Crontab.CronExpression.Parser.parse!("@daily")
+          ),
+        telemetry_span_context: ref
+      })
+
+      assert_receive {^ref, :done}, 1000
+    end
+
+    test "captures exception events with monitor config", %{bypass: bypass} do
+      test_pid = self()
+      ref = make_ref()
+
+      Bypass.expect_once(bypass, "POST", "/api/1/envelope/", fn conn ->
+        {:ok, body, conn} = Plug.Conn.read_body(conn)
+        assert [{headers, check_in_body}] = decode_envelope!(body)
+
+        assert headers["type"] == "check_in"
+
+        assert check_in_body["check_in_id"] == "quantum-#{:erlang.phash2(ref)}"
+        assert check_in_body["status"] == "error"
+        assert check_in_body["monitor_slug"] == "quantum-:test_job"
+        assert check_in_body["duration"] == 12.099
+        assert check_in_body["environment"] == "test"
+
+        assert check_in_body["monitor_config"] == %{
+                 "schedule" => %{
+                   "type" => "crontab",
+                   "value" => "0 0 * * * *"
+                 }
+               }
+
+        send(test_pid, {ref, :done})
+
+        Plug.Conn.send_resp(conn, 200, ~s<{"id": "1923"}>)
+      end)
+
+      duration = System.convert_time_unit(12_099, :millisecond, :native)
+
+      :telemetry.execute([:quantum, :job, :exception], %{duration: duration}, %{
+        job:
+          Scheduler.new_job(
+            name: :test_job,
+            schedule: Crontab.CronExpression.Parser.parse!("@daily")
+          ),
+        telemetry_span_context: ref
+      })
+
+      assert_receive {^ref, :done}, 1000
+    end
+
+    test "captures stop events with monitor config", %{bypass: bypass} do
+      test_pid = self()
+      ref = make_ref()
+
+      Bypass.expect_once(bypass, "POST", "/api/1/envelope/", fn conn ->
+        {:ok, body, conn} = Plug.Conn.read_body(conn)
+        assert [{headers, check_in_body}] = decode_envelope!(body)
+
+        assert headers["type"] == "check_in"
+
+        assert check_in_body["check_in_id"] == "quantum-#{:erlang.phash2(ref)}"
+        assert check_in_body["status"] == "ok"
+        assert check_in_body["monitor_slug"] == "quantum-:test_job"
+        assert check_in_body["duration"] == 12.099
+        assert check_in_body["environment"] == "test"
+
+        assert check_in_body["monitor_config"] == %{
+                 "schedule" => %{
+                   "type" => "crontab",
+                   "value" => "0 0 * * * *"
+                 }
+               }
+
+        send(test_pid, {ref, :done})
+
+        Plug.Conn.send_resp(conn, 200, ~s<{"id": "1923"}>)
+      end)
+
+      duration = System.convert_time_unit(12_099, :millisecond, :native)
+
+      :telemetry.execute([:quantum, :job, :stop], %{duration: duration}, %{
+        job:
+          Scheduler.new_job(
+            name: :test_job,
+            schedule: Crontab.CronExpression.Parser.parse!("@daily")
+          ),
+        telemetry_span_context: ref
+      })
+
+      assert_receive {^ref, :done}, 1000
+    end
+  end
+end