feat: Use :telemetry.span/3 (#480)

* feat: Uses `:telemetry.span/3`

* fix: Preserves more of the "stop" metadata

* fix: Add more of the original context

Co-authored-by: Isaac Sanders <[email protected]>
Co-authored-by: Jonatan Männchen <[email protected]>

Author: isaacsanders

lib/quantum/executor.ex

@@ -93,17 +93,11 @@ defmodule Quantum.Executor do
           {"Execute started for job", node: Node.self(), name: job_name}
         end)
 
-      # Note: we are intentionally mimicking the ":telemetry.span" here to keep current functionality
-      start_monotonic_time = :erlang.monotonic_time()
-
-      :telemetry.execute([:quantum, :job, :start], %{system_time: start_monotonic_time}, %{
-        job: job,
-        node: node,
-        scheduler: scheduler
-      })
-
       try do
-        execute_task(task)
+        :telemetry.span([:quantum, :job], %{job: job, node: node, scheduler: scheduler}, fn ->
+          result = execute_task(task)
+          {result, %{job: job, node: node, scheduler: scheduler, result: result}}
+        end)
       catch
         type, value ->
           debug_logging &&
@@ -115,31 +109,12 @@ defmodule Quantum.Executor do
             end)
 
           log_exception(type, value, __STACKTRACE__)
-
-          duration = :erlang.monotonic_time() - start_monotonic_time
-
-          :telemetry.execute([:quantum, :job, :exception], %{duration: duration}, %{
-            job: job,
-            node: node,
-            reason: value,
-            stacktrace: __STACKTRACE__,
-            scheduler: scheduler
-          })
       else
         result ->
           debug_logging &&
             Logger.debug(fn ->
               {"Execution ended for job", node: Node.self(), name: job_name, result: result}
             end)
-
-          duration = :erlang.monotonic_time() - start_monotonic_time
-
-          :telemetry.execute([:quantum, :job, :stop], %{duration: duration}, %{
-            job: job,
-            node: node,
-            scheduler: scheduler,
-            result: result
-          })
       end
 
       :ok

mix.exs

@@ -99,7 +99,7 @@ defmodule Quantum.Mixfile do
     [
       {:crontab, "~> 1.1"},
       {:gen_stage, "~> 0.14 or ~> 1.0"},
-      {:telemetry, "~> 0.4"},
+      {:telemetry, ">= 0.4.3 and < 1.0.0"},
       {:tzdata, "~> 1.0", only: [:dev, :test]},
       {:ex_doc, ">= 0.0.0", only: [:dev, :docs], runtime: false},
       {:excoveralls, "~> 0.5", only: [:test], runtime: false},

test/quantum/executor_test.exs

@@ -23,20 +23,40 @@ defmodule Quantum.ExecutorTest do
     def handle_event(
           [:quantum, :job, :start],
           %{system_time: _system_time} = _measurements,
-          %{job: %Job{name: job_name}, node: _node, scheduler: _scheduler} = _metadata,
+          %{
+            job: %Job{name: job_name},
+            node: _node,
+            scheduler: _scheduler,
+            telemetry_span_context: telemetry_span_context
+          } = _metadata,
           %{parent_thread: parent_thread, test_id: test_id}
         ) do
-      send(parent_thread, %{test_id: test_id, job_name: job_name, type: :start})
+      send(parent_thread, %{
+        test_id: test_id,
+        job_name: job_name,
+        type: :start,
+        telemetry_span_context: telemetry_span_context
+      })
     end
 
     def handle_event(
           [:quantum, :job, :stop],
           %{duration: _duration} = _measurements,
-          %{job: %Job{name: job_name}, node: _node, scheduler: _scheduler, result: _result} =
-            _metadata,
+          %{
+            job: %Job{name: job_name},
+            node: _node,
+            scheduler: _scheduler,
+            result: _result,
+            telemetry_span_context: telemetry_span_context
+          } = _metadata,
           %{parent_thread: parent_thread, test_id: test_id}
         ) do
-      send(parent_thread, %{test_id: test_id, job_name: job_name, type: :stop})
+      send(parent_thread, %{
+        test_id: test_id,
+        job_name: job_name,
+        telemetry_span_context: telemetry_span_context,
+        type: :stop
+      })
     end
 
     def handle_event(
@@ -46,15 +66,19 @@ defmodule Quantum.ExecutorTest do
             job: %Job{name: job_name},
             node: _node,
             scheduler: _scheduler,
+            kind: kind,
             reason: reason,
-            stacktrace: stacktrace
+            stacktrace: stacktrace,
+            telemetry_span_context: telemetry_span_context
           } = _metadata,
           %{parent_thread: parent_thread, test_id: test_id}
         ) do
       send(parent_thread, %{
         test_id: test_id,
         job_name: job_name,
+        telemetry_span_context: telemetry_span_context,
         type: :exception,
+        kind: kind,
         reason: reason,
         stacktrace: stacktrace
       })
@@ -316,10 +340,13 @@ defmodule Quantum.ExecutorTest do
       assert_receive %{
                        test_id: ^test_id,
                        type: :exception,
+                       kind: :error,
                        reason: %RuntimeError{message: "failed"},
                        stacktrace: [
                          {Quantum.ExecutorTest, _, _, _},
                          {Quantum.Executor, _, _, _},
+                         {:telemetry, _, _, _},
+                         {Quantum.Executor, _, _, _},
                          {Task.Supervised, _, _, _},
                          {Task.Supervised, _, _, _},
                          {:proc_lib, _, _, _}
@@ -366,10 +393,13 @@ defmodule Quantum.ExecutorTest do
       assert_receive %{
                        test_id: ^test_id,
                        type: :exception,
+                       kind: :error,
                        reason: %RuntimeError{message: "failed"},
                        stacktrace: [
                          {Quantum.ExecutorTest, _, _, _},
                          {Quantum.Executor, _, _, _},
+                         {:telemetry, _, _, _},
+                         {Quantum.Executor, _, _, _},
                          {Task.Supervised, _, _, _},
                          {Task.Supervised, _, _, _},
                          {:proc_lib, _, _, _}
@@ -415,10 +445,13 @@ defmodule Quantum.ExecutorTest do
       assert_receive %{
                        test_id: ^test_id,
                        type: :exception,
+                       kind: :exit,
                        reason: :failure,
                        stacktrace: [
                          {Quantum.ExecutorTest, _, _, _},
                          {Quantum.Executor, _, _, _},
+                         {:telemetry, _, _, _},
+                         {Quantum.Executor, _, _, _},
                          {Task.Supervised, _, _, _},
                          {Task.Supervised, _, _, _},
                          {:proc_lib, _, _, _}
@@ -468,10 +501,13 @@ defmodule Quantum.ExecutorTest do
       assert_receive %{
                        test_id: ^test_id,
                        type: :exception,
+                       kind: :throw,
                        reason: ^ref,
                        stacktrace: [
                          {Quantum.ExecutorTest, _, _, _},
                          {Quantum.Executor, _, _, _},
+                         {:telemetry, _, _, _},
+                         {Quantum.Executor, _, _, _},
                          {Task.Supervised, _, _, _},
                          {Task.Supervised, _, _, _},
                          {:proc_lib, _, _, _}