chore: added more telemetry for benchmarks

sync_service/config/runtime.exs

@@ -6,6 +6,7 @@ if config_env() in [:dev, :test] do
 end
 
 config :logger, level: :debug
+config :telemetry_poller, :default, period: 500
 
 if Config.config_env() == :test do
   config :electric,

sync_service/lib/electric/in_mem_shape_cache.ex

@@ -83,7 +83,15 @@ defmodule Electric.InMemShapeCache do
               Shapes.query_snapshot(conn, shape_definition)
             end)
 
+          start = System.monotonic_time()
           :ets.insert(@ets_table, {hash, shape_id, snapshot})
+
+          :telemetry.execute(
+            [:electric, :snapshot],
+            %{storage: System.monotonic_time() - start},
+            %{}
+          )
+
           GenServer.cast(parent, {:snapshot_ready, hash})
         rescue
           error -> GenServer.cast(parent, {:snapshot_failed, hash, error})

sync_service/lib/electric/plug/router.ex

@@ -3,6 +3,7 @@ defmodule Electric.Plug.Router do
 
   plug :match
   plug Plug.Logger
+  plug Plug.RequestId
   plug :dispatch
 
   forward "/shape", to: Electric.Plug.Shapes

sync_service/lib/electric/plug/shapes.ex

@@ -100,12 +100,9 @@ defmodule Electric.Plug.Shapes do
                }, offset}
             end)
 
-          conn
-          |> put_resp_header("x-electric-shape-id", shape_id)
-          |> put_resp_content_type("application/json")
-          |> put_resp_header("etag", "#{shape_id}-#{max_offset}")
-          |> send_resp(
-            200,
+          start = System.monotonic_time()
+
+          encoded =
             Jason.encode_to_iodata!(
               initial_rows ++
                 active_log ++
@@ -115,7 +112,18 @@ defmodule Electric.Plug.Shapes do
                   }
                 ]
             )
+
+          :telemetry.execute(
+            [:electric, :snapshot],
+            %{encoding: System.monotonic_time() - start},
+            %{}
           )
+
+          conn
+          |> put_resp_header("x-electric-shape-id", shape_id)
+          |> put_resp_content_type("application/json")
+          |> put_resp_header("etag", "#{shape_id}-#{max_offset}")
+          |> send_resp(200, encoded)
         end
     end
   end

sync_service/lib/electric/shapes.ex

@@ -3,17 +3,31 @@ defmodule Electric.Shapes do
   require Logger
 
   def query_snapshot(conn, table) do
+    start = System.monotonic_time()
     query = Postgrex.query!(conn, "SELECT * FROM #{table}", [])
+    query_stopped = System.monotonic_time()
     Logger.debug("Querying a snapshot for #{inspect(table)}")
 
-    query.rows
-    |> Enum.map(fn row ->
-      Enum.zip_with(query.columns, row, fn
-        "id", val -> {"id", Utils.encode_uuid(val)}
-        col, val -> {col, val}
+    results =
+      query.rows
+      |> Enum.map(fn row ->
+        Enum.zip_with(query.columns, row, fn
+          "id", val -> {"id", Utils.encode_uuid(val)}
+          col, val -> {col, val}
+        end)
+        |> Map.new()
       end)
-      |> Map.new()
-    end)
+
+    :telemetry.execute(
+      [:electric, :query],
+      %{
+        duration: query_stopped - start,
+        serialization_duration: System.monotonic_time() - query_stopped
+      },
+      %{}
+    )
+
+    results
   end
 
   def get_or_create_shape(table, opts \\ []) do

sync_service/lib/electric/telemetry.ex

@@ -45,7 +45,11 @@ defmodule Electric.Telemetry do
       summary("plug.router_dispatch.exception.duration",
         tags: [:route],
         unit: {:native, :millisecond}
-      )
+      ),
+      summary("electric.query.duration", unit: {:native, :millisecond}),
+      summary("electric.query.serialization_duration", unit: {:native, :millisecond}),
+      summary("electric.snapshot.storage", unit: {:native, :millisecond}),
+      summary("electric.snapshot.encoding", unit: {:native, :millisecond})
     ]
     |> Enum.map(&%{&1 | tags: [:instance_id | &1.tags]})
   end