feat(telemetry): include producer metadata in per-message events

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

Author: yordis

lib/broadway.ex

@@ -714,7 +714,8 @@ defmodule Broadway do
           name: atom,
           index: non_neg_integer,
           message: Broadway.Message.t,
-          telemetry_span_context: reference
+          telemetry_span_context: reference,
+          producer: {atom, list}
         }
         ```
 
@@ -732,7 +733,8 @@ defmodule Broadway do
           name: atom,
           index: non_neg_integer,
           message: Broadway.Message.t,
-          telemetry_span_context: reference
+          telemetry_span_context: reference,
+          producer: {atom, list}
         }
         ```
 
@@ -753,7 +755,8 @@ defmodule Broadway do
           kind: kind,
           reason: reason,
           stacktrace: stacktrace,
-          telemetry_span_context: reference
+          telemetry_span_context: reference,
+          producer: {atom, list}
         }
         ```
 

lib/broadway/topology/processor_stage.ex

@@ -164,7 +164,8 @@ defmodule Broadway.Topology.ProcessorStage do
             index: state.partition,
             name: state.name,
             message: message,
-            context: state.context
+            context: state.context,
+            producer: state.producer
           },
           fn ->
             updated_message =
@@ -179,7 +180,8 @@ defmodule Broadway.Topology.ProcessorStage do
                index: state.partition,
                name: state.name,
                message: updated_message,
-               context: state.context
+               context: state.context,
+               producer: state.producer
              }}
           end
         )

test/broadway/telemetry_test.exs

@@ -0,0 +1,95 @@
+defmodule Broadway.TelemetryTest do
+  use ExUnit.Case
+
+  defmodule MessageServer do
+    use Broadway
+
+    def start_link(opts) do
+      Broadway.start_link(__MODULE__,
+        name: opts[:name],
+        producer: [
+          module: {Broadway.DummyProducer, opts[:producer_options]}
+        ],
+        processors: [
+          default: []
+        ],
+        context: opts[:context]
+      )
+    end
+
+    @impl true
+    def handle_message(_, message, %{test_pid: test_pid}) do
+      send(test_pid, {:message_handled, message})
+      message
+    end
+  end
+
+  test "[:broadway, :processor, :message, :start] and :stop events include producer metadata" do
+    test_pid = self()
+    producer_options = [queue_url: "https://sqs.amazonaws.com/123456789/my-queue"]
+
+    :telemetry.attach_many(
+      "test-producer-metadata",
+      [
+        [:broadway, :processor, :message, :start],
+        [:broadway, :processor, :message, :stop]
+      ],
+      fn event_name, _measurements, metadata, _config ->
+        send(test_pid, {:telemetry_event, event_name, metadata})
+      end,
+      %{}
+    )
+
+    {:ok, _pid} =
+      start_supervised(
+        {MessageServer,
+         name: :test_broadway_producer_metadata,
+         producer_options: producer_options,
+         context: %{test_pid: test_pid}}
+      )
+
+    ref = Broadway.test_message(:test_broadway_producer_metadata, "test_data")
+
+    assert_receive {:telemetry_event, [:broadway, :processor, :message, :start], metadata}
+    assert {Broadway.DummyProducer, ^producer_options} = metadata.producer
+
+    assert_receive {:telemetry_event, [:broadway, :processor, :message, :stop], metadata}
+    assert {Broadway.DummyProducer, ^producer_options} = metadata.producer
+
+    assert_receive {:message_handled, _message}
+    assert_receive {:ack, ^ref, _successful, _failed}
+
+    :telemetry.detach("test-producer-metadata")
+  end
+
+  test "producer metadata in telemetry events can be used to extract adapter options" do
+    test_pid = self()
+    queue_url = "https://sqs.amazonaws.com/123456789/my-special-queue"
+    producer_options = [queue_url: queue_url, some_other_option: "value"]
+
+    :telemetry.attach(
+      "test-extract-queue-name",
+      [:broadway, :processor, :message, :start],
+      fn _event_name, _measurements, metadata, _config ->
+        {_module, adapter_options} = metadata.producer
+        queue_name = adapter_options |> Keyword.fetch!(:queue_url) |> String.split("/") |> List.last()
+        send(test_pid, {:queue_name, queue_name})
+      end,
+      %{}
+    )
+
+    {:ok, _pid} =
+      start_supervised(
+        {MessageServer,
+         name: :test_broadway_queue_extraction,
+         producer_options: producer_options,
+         context: %{test_pid: test_pid}}
+      )
+
+    Broadway.test_message(:test_broadway_queue_extraction, "test_data")
+
+    assert_receive {:queue_name, "my-special-queue"}
+
+    :telemetry.detach("test-extract-queue-name")
+  end
+end