feat(telemetry): add :telemetry.span for handle_failed/2 callback

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

lib/broadway.ex

@@ -673,8 +673,7 @@ defmodule Broadway do
           processor_key: atom,
           index: non_neg_integer,
           messages: [Broadway.Message.t],
-          telemetry_span_context: reference,
-          producer: {atom, list}
+          telemetry_span_context: reference
         }
         ```
 
@@ -695,8 +694,7 @@ defmodule Broadway do
           successful_messages_to_ack: [Broadway.Message.t],
           successful_messages_to_forward: [Broadway.Message.t],
           failed_messages: [Broadway.Message.t],
-          telemetry_span_context: reference,
-          producer: {atom, list}
+          telemetry_span_context: reference
         }
         ```
 
@@ -795,6 +793,57 @@ defmodule Broadway do
         }
         ```
 
+    * `[:broadway, :handle_failed, :start]` - Dispatched before the `c:handle_failed/2`
+      callback is invoked
+
+      * Measurement: `%{system_time: integer}`
+
+      * Metadata:
+
+        ```
+        %{
+          module: atom,
+          messages: [Broadway.Message.t],
+          context: term,
+          telemetry_span_context: reference
+        }
+        ```
+
+    * `[:broadway, :handle_failed, :stop]` - Dispatched after the `c:handle_failed/2`
+      callback has returned
+
+      * Measurement: `%{duration: native_time}`
+
+      * Metadata:
+
+        ```
+        %{
+          module: atom,
+          messages: [Broadway.Message.t],
+          context: term,
+          telemetry_span_context: reference
+        }
+        ```
+
+    * `[:broadway, :handle_failed, :exception]` - Dispatched if the `c:handle_failed/2`
+      callback raises an exception
+
+      * Measurement: `%{duration: native_time}`
+
+      * Metadata:
+
+        ```
+        %{
+          module: atom,
+          messages: [Broadway.Message.t],
+          context: term,
+          kind: :error | :exit | :throw,
+          reason: term,
+          stacktrace: list,
+          telemetry_span_context: reference
+        }
+        ```
+
     * `[:broadway, :batcher, :start]` - Dispatched by a Broadway batcher before
       handling events
 

lib/broadway/acknowledger.ex

@@ -102,6 +102,19 @@ defmodule Broadway.Acknowledger do
   end
 
   defp handle_failed_messages(messages, module, context) do
+    metadata = %{module: module, messages: messages, context: context}
+
+    :telemetry.span(
+      [:broadway, :handle_failed],
+      metadata,
+      fn ->
+        result = do_handle_failed_messages(messages, module, context)
+        {result, Map.put(metadata, :messages, result)}
+      end
+    )
+  end
+
+  defp do_handle_failed_messages(messages, module, context) do
     module.handle_failed(messages, context)
   catch
     kind, reason ->

test/broadway_test.exs

@@ -1939,6 +1939,98 @@ defmodule BroadwayTest do
     end
   end
 
+  describe "handle_failed telemetry" do
+    test "emits [:broadway, :handle_failed] span events from a processor" do
+      broadway_name = new_unique_name()
+      test_pid = self()
+
+      :telemetry.attach_many(
+        "#{broadway_name}-handle-failed",
+        [[:broadway, :handle_failed, :start], [:broadway, :handle_failed, :stop]],
+        fn event, _measurements, metadata, _config ->
+          send(test_pid, {:telemetry_event, event, metadata})
+        end,
+        %{}
+      )
+
+      {:ok, _broadway} =
+        Broadway.start_link(CustomHandlersWithHandleFailed,
+          name: broadway_name,
+          context: %{
+            handle_message: fn message, _ -> Message.failed(message, :failed) end,
+            handle_failed: fn messages, _ ->
+              send(test_pid, :handle_failed_called)
+              messages
+            end
+          },
+          producer: [module: {ManualProducer, []}],
+          processors: [default: []]
+        )
+
+      ref = Broadway.test_batch(broadway_name, [:fail], batch_mode: :flush)
+
+      assert_receive {:telemetry_event, [:broadway, :handle_failed, :start], start_metadata}
+      assert start_metadata.module == CustomHandlersWithHandleFailed
+      assert [%Message{}] = start_metadata.messages
+
+      assert_receive {:telemetry_event, [:broadway, :handle_failed, :stop], stop_metadata}
+      assert stop_metadata.module == CustomHandlersWithHandleFailed
+      assert [%Message{}] = stop_metadata.messages
+
+      assert_receive :handle_failed_called
+      assert_receive {:ack, ^ref, _successful, _failed}
+
+      :telemetry.detach("#{broadway_name}-handle-failed")
+    end
+
+    test "emits [:broadway, :handle_failed] span events from a batch processor" do
+      broadway_name = new_unique_name()
+      test_pid = self()
+
+      :telemetry.attach_many(
+        "#{broadway_name}-handle-failed",
+        [[:broadway, :handle_failed, :start], [:broadway, :handle_failed, :stop]],
+        fn event, _measurements, metadata, _config ->
+          send(test_pid, {:telemetry_event, event, metadata})
+        end,
+        %{}
+      )
+
+      {:ok, _broadway} =
+        Broadway.start_link(CustomHandlersWithHandleFailed,
+          name: broadway_name,
+          context: %{
+            handle_message: fn message, _ -> message end,
+            handle_batch: fn _, messages, _, _ ->
+              Enum.map(messages, &Message.failed(&1, :failed))
+            end,
+            handle_failed: fn messages, _ ->
+              send(test_pid, :handle_failed_called)
+              messages
+            end
+          },
+          producer: [module: {ManualProducer, []}],
+          processors: [default: []],
+          batchers: [default: []]
+        )
+
+      ref = Broadway.test_batch(broadway_name, [:fail], batch_mode: :flush)
+
+      assert_receive {:telemetry_event, [:broadway, :handle_failed, :start], start_metadata}
+      assert start_metadata.module == CustomHandlersWithHandleFailed
+      assert [%Message{}] = start_metadata.messages
+
+      assert_receive {:telemetry_event, [:broadway, :handle_failed, :stop], stop_metadata}
+      assert stop_metadata.module == CustomHandlersWithHandleFailed
+      assert [%Message{}] = stop_metadata.messages
+
+      assert_receive :handle_failed_called
+      assert_receive {:ack, ^ref, _successful, _failed}
+
+      :telemetry.detach("#{broadway_name}-handle-failed")
+    end
+  end
+
   describe "handle producer crash" do
     setup do
       test_pid = self()