chore (sync service): replace span for decode_message by attribute …

…on parent span (#2082)

This PR fixes #2078.

packages/sync-service/lib/electric/postgres/replication_client.ex

@@ -213,8 +213,7 @@ defmodule Electric.Postgres.ReplicationClient do
   end
 
   defp process_x_log_data(data, wal_end, %State{} = state) do
-    data
-    |> decode_message()
+    OpenTelemetry.timed_fun("decode_message_duration", fn -> decode_message(data) end)
     # # Useful for debugging:
     # |> tap(fn %struct{} = msg ->
     #   message_type = struct |> to_string() |> String.split(".") |> List.last()
@@ -284,11 +283,7 @@ defmodule Electric.Postgres.ReplicationClient do
   end
 
   defp decode_message(data) do
-    OpenTelemetry.with_span(
-      "pg_txn.replication_client.decode_message",
-      [msg_size: byte_size(data)],
-      fn -> Decoder.decode(data) end
-    )
+    Decoder.decode(data)
   end
 
   defp encode_standby_status_update(state) do

packages/sync-service/lib/electric/telemetry/open_telemetry.ex

@@ -33,6 +33,7 @@ defmodule Electric.Telemetry.OpenTelemetry do
   require OpenTelemetry.SemanticConventions.Trace
 
   @typep span_name :: String.t()
+  @typep attr_name :: String.t()
   @typep span_attrs :: :opentelemetry.attributes_map()
   @typep span_ctx :: :opentelemetry.span_ctx()
 
@@ -69,6 +70,17 @@ defmodule Electric.Telemetry.OpenTelemetry do
     end)
   end
 
+  @doc """
+  Executes the provided function and records its duration in microseconds.
+  The duration is added to the current span as a span attribute named with the given `name`.
+  """
+  @spec timed_fun(span_ctx() | nil, attr_name(), (-> term)) :: term
+  def timed_fun(span_ctx \\ nil, name, fun) when is_binary(name) do
+    {duration, result} = :timer.tc(fun)
+    add_span_attributes(span_ctx, %{name => duration})
+    result
+  end
+
   @doc """
   Wrap the given `fun` in an anonymous function, attaching the current process' span context to it.