feat(opentelemetry_oban): add missing OTEL messaging semconv attributes (#598)

Author: yordis

instrumentation/opentelemetry_oban/lib/opentelemetry_oban.ex

@@ -18,6 +18,7 @@ defmodule OpentelemetryOban do
 
   alias Ecto.Changeset
   alias OpenTelemetry.Span
+  alias OpenTelemetry.SemConv.ErrorAttributes
   alias OpenTelemetry.SemConv.Incubating.MessagingAttributes
 
   require Logger
@@ -122,6 +123,7 @@ defmodule OpentelemetryOban do
           ctx = OpenTelemetry.Tracer.current_span_ctx()
           Span.record_exception(ctx, exception, __STACKTRACE__)
           Span.set_status(ctx, OpenTelemetry.status(:error, ""))
+          Span.set_attribute(ctx, ErrorAttributes.error_type(), inspect(exception.__struct__))
           reraise exception, __STACKTRACE__
       end
     end
@@ -134,10 +136,15 @@ defmodule OpentelemetryOban do
   end
 
   def insert_all(name, changesets) when is_list(changesets) do
-    # changesets in insert_all can include different workers and different
-    # queues. This means we cannot provide much information here, but we can
-    # still record the insert and propagate the context information.
-    OpenTelemetry.Tracer.with_span "send", kind: :producer do
+    attrs = %{
+      MessagingAttributes.messaging_system() => :oban,
+      MessagingAttributes.messaging_operation_name() => "send",
+      MessagingAttributes.messaging_operation_type() =>
+        MessagingAttributes.messaging_operation_type_values().create,
+      MessagingAttributes.messaging_batch_message_count() => length(changesets)
+    }
+
+    OpenTelemetry.Tracer.with_span "send", attributes: attrs, kind: :producer do
       changesets = Enum.map(changesets, &add_tracing_information_to_meta/1)
       Oban.insert_all(name, changesets)
     end

instrumentation/opentelemetry_oban/lib/opentelemetry_oban/job_handler.ex

@@ -4,6 +4,7 @@ defmodule OpentelemetryOban.JobHandler do
   alias OpenTelemetry.Ctx
   alias OpenTelemetry.Span
   alias OpenTelemetry.Tracer
+  alias OpenTelemetry.SemConv.ErrorAttributes
   alias OpenTelemetry.SemConv.Incubating.MessagingAttributes
 
   @tracer_id __MODULE__
@@ -123,10 +124,16 @@ defmodule OpentelemetryOban.JobHandler do
 
     Span.record_exception(ctx, error, stacktrace)
     Span.set_status(ctx, OpenTelemetry.status(:error, ""))
+    set_error_type(error)
 
     OpentelemetryTelemetry.end_telemetry_span(@tracer_id, metadata)
   end
 
+  defp set_error_type(%struct_name{} = error) when is_exception(error),
+    do: Tracer.set_attribute(ErrorAttributes.error_type(), inspect(struct_name))
+
+  defp set_error_type(_error), do: :ok
+
   defp put_links(span_opts, []), do: span_opts
   defp put_links(span_opts, links), do: Map.put(span_opts, :links, links)
 

instrumentation/opentelemetry_oban/lib/opentelemetry_oban/plugin_handler.ex

@@ -3,6 +3,7 @@ defmodule OpentelemetryOban.PluginHandler do
 
   alias OpenTelemetry.Tracer
   alias OpenTelemetry.Span
+  alias OpenTelemetry.SemConv.ErrorAttributes
 
   @tracer_id __MODULE__
 
@@ -72,9 +73,16 @@ defmodule OpentelemetryOban.PluginHandler do
       OpenTelemetry.status(:error, Exception.format_banner(kind, reason, stacktrace))
     )
 
+    set_error_type(reason)
+
     OpentelemetryTelemetry.end_telemetry_span(@tracer_id, metadata)
   end
 
+  defp set_error_type(%struct_name{} = error) when is_exception(error),
+    do: Tracer.set_attribute(ErrorAttributes.error_type(), inspect(struct_name))
+
+  defp set_error_type(_error), do: :ok
+
   defp end_span_plugin_attrs(%{plugin: Oban.Plugins.Cron} = metadata) do
     %{"oban.plugins.cron.jobs_count": length(metadata[:jobs])}
   end

instrumentation/opentelemetry_oban/test/opentelemetry_oban/plugin_handler_test.exs

@@ -98,19 +98,21 @@ defmodule OpentelemetryOban.PluginHandlerTest do
     assert_receive {:span,
                     span(
                       name: "Elixir.Oban.Plugins.Stager process",
+                      attributes: span_attributes,
                       events: events,
                       status: ^expected_status
                     )}
 
+    assert %{"error.type": "UndefinedFunctionError"} =
+             :otel_attributes.map(span_attributes)
+
     [
       event(
         name: :exception,
         attributes: event_attributes
       )
     ] = :otel_events.list(events)
 
-    # When using :otel_span.record_exception/5 with kind/reason/stacktrace,
-    # it doesn't extract the message field even from exception structs
     assert [:"exception.stacktrace", :"exception.type"] ==
              Enum.sort(Map.keys(:otel_attributes.map(event_attributes)))
   end
@@ -142,18 +144,20 @@ defmodule OpentelemetryOban.PluginHandlerTest do
     assert_receive {:span,
                     span(
                       name: "Elixir.Oban.Plugins.Stager process",
+                      attributes: span_attributes,
                       events: events,
                       status: ^expected_status
                     )}
 
+    refute Map.has_key?(:otel_attributes.map(span_attributes), :"error.type")
+
     [
       event(
         name: :exception,
         attributes: event_attributes
       )
     ] = :otel_events.list(events)
 
-    # Non-exception error reasons don't have exception.message
     assert [:"exception.stacktrace", :"exception.type"] ==
              Enum.sort(Map.keys(:otel_attributes.map(event_attributes)))
   end

instrumentation/opentelemetry_oban/test/opentelemetry_oban_test.exs

@@ -179,6 +179,8 @@ defmodule OpentelemetryObanTest do
                       status: ^expected_status
                     )}
 
+    attrs = :otel_attributes.map(attributes)
+
     assert %{
              "oban.job.attempt": 1,
              "oban.job.inserted_at": _inserted_at,
@@ -192,8 +194,9 @@ defmodule OpentelemetryObanTest do
              "messaging.operation.type": :process,
              "messaging.system": :oban,
              "messaging.message.id": ^id,
-             "messaging.client.id": "TestJobThatReturnsError"
-           } = :otel_attributes.map(attributes)
+             "messaging.client.id": "TestJobThatReturnsError",
+             "error.type": "Oban.PerformError"
+           } = attrs
 
     [
       event(
@@ -259,6 +262,8 @@ defmodule OpentelemetryObanTest do
                       status: ^expected_status
                     )}
 
+    attrs = :otel_attributes.map(attributes)
+
     assert %{
              "messaging.destination.name": "events",
              "messaging.operation.name": "process",
@@ -272,8 +277,9 @@ defmodule OpentelemetryObanTest do
              "messaging.operation.type": :process,
              "messaging.message.id": ^id,
              "messaging.client.id": "TestJobThatThrowsException",
-             "messaging.system": :oban
-           } = :otel_attributes.map(attributes)
+             "messaging.system": :oban,
+             "error.type": "UndefinedFunctionError"
+           } = attrs
 
     [
       event(
@@ -327,10 +333,14 @@ defmodule OpentelemetryObanTest do
     assert_receive {:span,
                     span(
                       name: "send events",
+                      attributes: send_attributes,
                       events: events,
                       status: ^expected_status
                     )}
 
+    assert %{"error.type": "Ecto.InvalidChangesetError"} =
+             :otel_attributes.map(send_attributes)
+
     [
       event(
         name: :exception,
@@ -355,13 +365,22 @@ defmodule OpentelemetryObanTest do
     assert_receive {:span,
                     span(
                       name: "send",
-                      attributes: _attributes,
+                      attributes: send_attributes,
                       trace_id: send_trace_id,
                       span_id: send_span_id,
                       kind: :producer,
                       status: :undefined
                     )}
 
+    send_attrs = :otel_attributes.map(send_attributes)
+
+    assert %{
+             "messaging.system": :oban,
+             "messaging.operation.name": "send",
+             "messaging.operation.type": :create,
+             "messaging.batch.message_count": 2
+           } = send_attrs
+
     assert_receive {:span,
                     span(
                       name: "process events",