fix: Ensure shape info is added to OTEL spans in shape requests (#2358)

Fixes electric-sql/stratovolt#299

It's a bit confusing because we use the same attirbute names elsewhere
and use different values (e.g. `"shape.root_table"` is also added as a
span in `Electric.Shapes.Consumer`, but there it uses the namespaced
tuple instead of the request provided table).

Right now they are coming in as `nil` because there is no such thing as
a `:shape` key in the parameters, so I'm changing this to at least
always try and include something.

This populating of attributes happens both at the start of the request
and at the end - I always try to use the raw query parameter value,
because it should always be available and it is what we will try to
correlate to request spans from proxies etc. I have the parsed parameter
fallback but I'm not sure if that is even necessary or if it makes sense
to be there.

For example, for the `replica` parameter, if it is not present in the
query string, it will still be populated with the parsed value which
defaults to `"default"` - but for the purpose of this span, should we
not leave it as `nil` if the requester didn't include it? Or should we
include what we parsed the request as? And should priority go to the
parsed version or the raw version?

I've also removed the `shape.definition` attribute as it feels a bit too
verbose - we can technically add it with a json safe version of the
shape definition, but it feels like overkill.

Unsure about these changes but figured it'd be good to at least get the
basics working again - this would benefit from tests but I think we need
an OTEL dependency injection to get it working, if anyone has a better
idea I'd appreciate that.

.changeset/wet-zebras-laugh.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Ensure shape properties are added to OTEL spans in shape requests.

packages/sync-service/lib/electric/plug/serve_shape_plug.ex

@@ -153,24 +153,40 @@ defmodule Electric.Plug.ServeShapePlug do
     params = Map.get(request, :params, %{}) |> bare_map()
     response = (Map.get(assigns, :response) || Map.get(request, :response) || %{}) |> bare_map()
     attrs = Map.get(response, :trace_attrs, %{})
+    maybe_up_to_date = Map.get(response, :up_to_date, false)
 
-    shape_handle =
-      conn.query_params["handle"] || params[:handle] || request[:handle]
+    is_live_req =
+      if is_nil(params[:live]),
+        do: !is_nil(conn.query_params["live"]) && conn.query_params["live"] != "false",
+        else: params[:live]
 
-    maybe_up_to_date = Map.get(response, :up_to_date, false)
+    replica =
+      if is_nil(params[:replica]),
+        do: conn.query_params["replica"],
+        else: to_string(params[:replica])
+
+    columns =
+      if is_nil(params[:columns]),
+        do: conn.query_params["columns"],
+        else: Enum.join(params[:columns], ",")
 
     Electric.Telemetry.OpenTelemetry.get_stack_span_attrs(
       get_in(conn.assigns, [:config, :stack_id])
     )
     |> Map.merge(Electric.Plug.Utils.common_open_telemetry_attrs(conn))
     |> Map.merge(%{
-      "shape.handle" => shape_handle,
-      "shape.where" => get_in(params, [:shape, :where]),
-      "shape.root_table" => get_in(params, [:shape, :root_table]),
-      "shape.definition" => get_in(params, [:shape]),
-      "shape.replica" => get_in(params, [:shape, :replica]),
-      "shape_req.is_live" => params[:live],
-      "shape_req.offset" => params[:offset],
+      "shape.handle" => conn.query_params["handle"] || params[:handle] || request[:handle],
+      "shape.where" => conn.query_params["where"] || params[:where],
+      "shape.root_table" => conn.query_params["table"] || params[:table],
+      "shape.columns" => columns,
+      # # Very verbose info to add to spans - keep out unless we explicitly need it
+      # "shape.definition" =>
+      #   if(not is_nil(params[:shape_definition]),
+      #     do: Electric.Shapes.Shape.to_json_safe(params[:shape_definition])
+      #   ),
+      "shape.replica" => replica,
+      "shape_req.is_live" => is_live_req,
+      "shape_req.offset" => conn.query_params["offset"],
       "shape_req.is_shape_rotated" => attrs[:ot_is_shape_rotated] || false,
       "shape_req.is_long_poll_timeout" => attrs[:ot_is_long_poll_timeout] || false,
       "shape_req.is_empty_response" => attrs[:ot_is_empty_response] || false,

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

@@ -283,7 +283,11 @@ defmodule Electric.Postgres.ReplicationClient do
         # individual storage write can timeout the entire batch.
         OpenTelemetry.with_span(
           "pg_txn.replication_client.transaction_received",
-          [num_changes: txn.num_changes, num_relations: MapSet.size(txn.affected_relations)],
+          [
+            num_changes: txn.num_changes,
+            num_relations: MapSet.size(txn.affected_relations),
+            xid: txn.xid
+          ],
           stack_id,
           fn -> apply(m, f, [txn | args]) end
         )