Repository: plataformatec/broadway
Pull request #157 (2020-01-16 04:42:55)

Added individual processor message telemetry events
Creating a new PR to implement individual processor events per message (as was discussed in #152).

akoutmos: I think my most recent commit addresses the comments that you had. If there is anything else you would like me to tweak feel free to let me know. Thanks!
josevalim: Thanks @akoutmos! I have added two last comments. Also, please don't forget to update the docs with the new event. :)
akoutmos: I went ahead and refactored the metadata maps as you suggested and also updated the docs to include the new events. I also noticed that in 081c1bb `batch_info` was added to the consumer start/stop events. I went ahead and added those entries to the consumer docs and also refactored the existing processor docs to make it clear how these per-message events differ.

Let me know if you want me to tweak/fix anything. Thanks!

Patch:
diff --git a/lib/broadway.ex b/lib/broadway.ex
index e54a830..4beb01b 100644
--- a/lib/broadway.ex
+++ b/lib/broadway.ex
@@ -395,14 +395,15 @@ defmodule Broadway do
 
   Broadway currently exposes following Telemetry events:
 
-    * `[:broadway, :processor, :start]` - Dispatched by a Broadway processor before your
-      `c:handle_message/3` callback is invoked
+    * `[:broadway, :processor, :start]` - Dispatched by a Broadway processor when
+       the internal GenStage `handle_events/3` callback is invoked
 
       * Measurement: `%{time: System.monotonic_time}`
       * Metadata: `%{name: atom, messages: [Broadway.Message.t]}`
 
-    * `[:broadway, :processor, :stop]` -  Dispatched by a Broadway processor after
-      your `c:handle_message/3` callback has returned
+    * `[:broadway, :processor, :stop]` -  Dispatched by a Broadway processor when
+      the internal GenStage `handle_events/3` callback has completed processing
+      all the individual messages
 
       * Measurement: `%{time: System.monotonic_time, duration: native_time}`
 
@@ -417,11 +418,68 @@ defmodule Broadway do
         }
         ```
 
+    * `[:broadway, :processor, :message, :start]` - Dispatched by a Broadway processor
+      before your `c:handle_message/3` callback is invoked
+
+      * Measurement: `%{time: System.monotonic_time}`
+
+      * Metadata:
+
+        ```
+        %{
+          processor_key: atom,
+          name: atom,
+          message: Broadway.Message.t
+        }
+        ```
+
+    * `[:broadway, :processor, :message, :stop]` - Dispatched by a Broadway processor
+      after your `c:handle_message/3` callback has returned
+
+      * Measurement: `%{time: System.monotonic_time, duration: native_time}`
+
+      * Metadata:
+
+        ```
+        %{
+          processor_key: atom,
+          name: atom,
+          message: Broadway.Message.t,
+          updated_message: Broadway.Message.t
+        }
+        ```
+
+    * `[:broadway, :processor, :message, :error]` - Dispatched by a Broadway processor
+      if your `c:handle_message/3` callback encounters an error
+
+      * Measurement: `%{time: System.monotonic_time, duration: native_time}`
+
+      * Metadata:
+
+        ```
+        %{
+          processor_key: atom,
+          name: atom,
+          message: Broadway.Message.t,
+          kind: kind,
+          reason: reason,
+          stacktrace: stacktrace
+        }
+        ```
+
     * `[:broadway, :consumer, :start]` - Dispatched by a Broadway consumer before your
       `c:handle_batch/4` callback is invoked
 
       * Measurement: `%{time: System.monotonic_time}`
-      * Metadata: `%{name: atom, messages: [Broadway.Message.t]}`
+      * Metadata:
+
+        ```
+        %{
+          name: atom,
+          messages: [Broadway.Message.t],
+          batch_info: Broadway.BatchInfo.t
+        }
+        ```
 
     * `[:broadway, :consumer, :stop]` - Dispatched by a Broadway consumer after your
       `c:handle_batch/4` callback has returned
@@ -434,7 +492,8 @@ defmodule Broadway do
         %{
           name: atom,
           successful_messages: [Broadway.Message.t],
-          failed_messages: [Broadway.Message.t]
+          failed_messages: [Broadway.Message.t],
+          batch_info: Broadway.BatchInfo.t
         }
         ```
 
diff --git a/lib/broadway/processor.ex b/lib/broadway/processor.ex
index 50010c4..a6356e1 100644
--- a/lib/broadway/processor.ex
+++ b/lib/broadway/processor.ex
@@ -117,16 +117,34 @@ defmodule Broadway.Processor do
       module: module,
       context: context,
       processor_key: processor_key,
-      batchers: batchers
+      batchers: batchers,
+      name: name
     } = state
 
+    start_time = System.monotonic_time()
+
+    telemetry_metadata = %{
+      processor_key: processor_key,
+      name: name,
+      message: message
+    }
+
+    emit_message_start_event(start_time, telemetry_metadata)
+
     try do
-      module.handle_message(processor_key, message, context)
-      |> validate_message(batchers)
+      handle_message_result =
+        module.handle_message(processor_key, message, context)
+        |> validate_message(batchers)
+
+      emit_message_stop_event(start_time, telemetry_metadata, handle_message_result)
+
+      handle_message_result
     catch
       kind, reason ->
         reason = Exception.normalize(kind, reason, __STACKTRACE__)
 
+        emit_message_error_event(start_time, telemetry_metadata, kind, reason, __STACKTRACE__)
+
         Logger.error(Exception.format(kind, reason, __STACKTRACE__),
           crash_reason: Acknowledger.crash_reason(kind, reason, __STACKTRACE__)
         )
@@ -146,6 +164,40 @@ defmodule Broadway.Processor do
     {Enum.reverse(successful), Enum.reverse(failed)}
   end
 
+  defp emit_message_start_event(start_time, metadata) do
+    :telemetry.execute([:broadway, :processor, :message, :start], %{time: start_time}, metadata)
+  end
+
+  defp emit_message_stop_event(start_time, base_metadata, handle_message_result) do
+    stop_time = System.monotonic_time()
+    measurements = %{time: stop_time, duration: stop_time - start_time}
+
+    metadata = %{
+      processor_key: base_metadata.processor_key,
+      name: base_metadata.name,
+      message: base_metadata.message,
+      updated_message: handle_message_result
+    }
+
+    :telemetry.execute([:broadway, :processor, :message, :stop], measurements, metadata)
+  end
+
+  defp emit_message_error_event(start_time, base_metadata, kind, reason, stacktrace) do
+    stop_time = System.monotonic_time()
+    measurements = %{time: stop_time, duration: stop_time - start_time}
+
+    metadata = %{
+      processor_key: base_metadata.processor_key,
+      name: base_metadata.name,
+      message: base_metadata.message,
+      kind: kind,
+      reason: reason,
+      stacktrace: stacktrace
+    }
+
+    :telemetry.execute([:broadway, :processor, :message, :error], measurements, metadata)
+  end
+
   defp validate_message(%Message{batcher: batcher, status: status} = message, batchers) do
     if status == :ok and batchers != :none and batcher not in batchers do
       raise "message was set to unknown batcher #{inspect(batcher)}. " <>


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index 35b39bc..f552725 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -602,7 +602,10 @@ defmodule BroadwayTest do
             [:broadway, :batcher, :start],
             [:broadway, :batcher, :stop],
             [:broadway, :consumer, :start],
-            [:broadway, :consumer, :stop]
+            [:broadway, :consumer, :stop],
+            [:broadway, :processor, :message, :start],
+            [:broadway, :processor, :message, :stop],
+            [:broadway, :processor, :message, :error]
           ],
           fn name, measurements, metadata, _ ->
             send(self, {:telemetry_event, name, measurements, metadata})
@@ -610,7 +613,9 @@ defmodule BroadwayTest do
           nil
         )
 
-      ref = Broadway.test_messages(broadway, [1, 2, :fail, :fail_batcher], batch_mode: :bulk)
+      ref =
+        Broadway.test_messages(broadway, [1, 2, :fail, :fail_batcher, :raise], batch_mode: :bulk)
+
       assert_receive {:batch_handled, [%{data: 1, status: :ok}, %{data: 2, status: :ok}]}
       assert_receive {:batch_handled, [%{data: :fail_batcher, status: :ok}]}
 
@@ -619,10 +624,14 @@ defmodule BroadwayTest do
       assert_receive {:ack, ^ref, [], [%{status: {:failed, "Failed batcher"}}]}
 
       assert_receive {:telemetry_event, [:broadway, :processor, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :stop], %{}, %{}}
       assert_receive {:telemetry_event, [:broadway, :processor, :stop], %{}, metadata}
       assert [] = metadata.failed_messages
 
       assert_receive {:telemetry_event, [:broadway, :processor, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :stop], %{}, %{}}
 
       assert_receive {:telemetry_event, [:broadway, :processor, :stop], %{},
                       %{failed_messages: []}}
@@ -630,10 +639,14 @@ defmodule BroadwayTest do
       assert [] = metadata.failed_messages
 
       assert_receive {:telemetry_event, [:broadway, :processor, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :stop], %{}, %{}}
       assert_receive {:telemetry_event, [:broadway, :processor, :stop], %{}, metadata}
       assert [%{data: :fail}] = metadata.failed_messages
 
       assert_receive {:telemetry_event, [:broadway, :processor, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :stop], %{}, %{}}
       assert_receive {:telemetry_event, [:broadway, :processor, :stop], %{}, metadata}
       assert [] = metadata.failed_messages
 
@@ -655,6 +668,12 @@ defmodule BroadwayTest do
 
       assert_receive {:telemetry_event, [:broadway, :consumer, :stop], %{},
                       %{failed_messages: [%{data: :fail_batcher}], batch_info: %BatchInfo{}}}
+
+      assert_receive {:telemetry_event, [:broadway, :processor, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :start], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :message, :error], %{}, %{}}
+      assert_receive {:telemetry_event, [:broadway, :processor, :stop], %{}, metadata}
+      assert_receive {:ack, ^ref, [], [%{status: {:error, _, _}}]}
     end
   end