feat(sync-service): tail-drop empty shape-GET response spans

.changeset/drop-empty-response-spans.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Tail-drop the OpenTelemetry spans of empty/up-to-date shape-GET responses at export time to cut trace volume. Disabled by default; set `ELECTRIC_DROP_EMPTY_RESPONSE_SPANS=true` to enable the drop. Error (5xx) and SSE responses are never dropped.

integration-tests/tests/otel-export.lux

@@ -10,7 +10,7 @@
 
 [invoke setup_pg]
 
-[invoke setup_electric_with_env "ELECTRIC_OTLP_ENDPOINT=http://localhost:4318 ELECTRIC_SYSTEM_METRICS_POLL_INTERVAL=1s ELECTRIC_STACK_TELEMETRY_INIT_DELAY=1s ELECTRIC_OTEL_EXPORT_PERIOD=2s ELECTRIC_OTEL_SAMPLING_RATIO=1.0 DO_NOT_START_CONN_MAN_PING=1 ELECTRIC_LOG_LEVEL=info OTEL_RESOURCE_ATTRIBUTES=custom.attr=electric.val"]
+[invoke setup_electric_with_env "ELECTRIC_OTLP_ENDPOINT=http://localhost:4318 ELECTRIC_SYSTEM_METRICS_POLL_INTERVAL=1s ELECTRIC_STACK_TELEMETRY_INIT_DELAY=1s ELECTRIC_OTEL_EXPORT_PERIOD=2s ELECTRIC_OTEL_SAMPLING_RATIO=1.0 ELECTRIC_DROP_EMPTY_RESPONSE_SPANS=true ELECTRIC_LONG_POLL_TIMEOUT=1000 DO_NOT_START_CONN_MAN_PING=1 ELECTRIC_LOG_LEVEL=info OTEL_RESOURCE_ATTRIBUTES=custom.attr=electric.val"]
 
 # Spawn a process containing off-heap binary references and ensure it's in the top 5 by memory footprint.
 [shell electric]
@@ -58,8 +58,16 @@
 
   [invoke curl_shape "http://localhost:3000/v1/shape?table=items&handle=$handle&offset=$offset"]
   ??HTTP/1.1 200 OK
+  ?electric-offset: ([\w\d_]+)
+  [global latest_offset=$1]
   ??"value":{"val":"3"}
 
+  # A live request with no new data long-polls until it times out and returns an
+  # empty up-to-date response — the only path that sets is_empty_response.
+  [invoke curl_shape "http://localhost:3000/v1/shape?table=items&handle=$handle&offset=$latest_offset&live"]
+  ??HTTP/1.1 200 OK
+  ??[{"headers":{"control":"up-to-date"
+
 [sleep 10]
 
 [shell otel_collector_startup]
@@ -124,6 +132,17 @@
   # (ELECTRIC_OTEL_SAMPLING_RATIO=1.0); it is sampled at 1% by default.
   [invoke grep_otel_collector_output "Span pg_txn\.replication_client\.transaction_received .*total_processing_time=[0-9]+"]
 
+  # Verify that a span is exported for a non-empty shape response.
+  [invoke grep_otel_collector_output "Span Plug_shape_get .*shape_req\.is_empty_response=false"]
+
+  # Verify that no span is exported for an empty shape response.
+  !test $(docker logs $otel_collector_container_name 2>&1 | python3 $(realpath ../support_files/normalize-otel-output.py) | grep -c "Span Plug_shape_get .*shape_req\.is_empty_response=true") -eq 0 && echo EMPTY_RESPONSE_SPAN_DROPPED
+  ??EMPTY_RESPONSE_SPAN_DROPPED
+
+  # Even though its span was dropped, the empty response is still counted by the
+  # serve_shape request metric, tagged empty=true.
+  [invoke grep_otel_collector_output "Metric electric\.plug\.serve_shape\.requests\.count .*empty=true"]
+
 ###
 
 [cleanup]

packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex

@@ -94,7 +94,7 @@ defmodule ElectricTelemetry.StackTelemetry do
       counter("electric.plug.serve_shape.requests.count",
         event_name: [:electric, :plug, :serve_shape],
         measurement: :count,
-        tags: [:status, :known_error, :live]
+        tags: [:status, :known_error, :live, :empty]
       ),
       distribution("electric.shape.response_size.bytes",
         unit: :byte,

packages/sync-service/config/runtime.exs

@@ -226,6 +226,7 @@ config :electric,
       env!("ELECTRIC_EXPERIMENTAL_MAX_SHAPES", :integer, nil),
   consumer_partitions: env!("ELECTRIC_CONSUMER_PARTITIONS", :integer, nil),
   max_concurrent_requests: max_concurrent_requests,
+  long_poll_timeout: env!("ELECTRIC_LONG_POLL_TIMEOUT", :integer, nil),
   # Used in telemetry
   instance_id: instance_id,
   call_home_telemetry?: env!("ELECTRIC_USAGE_REPORTING", :boolean, config_env() == :prod),
@@ -235,6 +236,7 @@ config :electric,
   otel_export_period: otel_export_period,
   otel_sampling_ratio: env!("ELECTRIC_OTEL_SAMPLING_RATIO", :float, nil),
   metrics_sampling_ratio: env!("ELECTRIC_METRICS_SAMPLING_RATIO", :float, nil),
+  drop_empty_response_spans?: env!("ELECTRIC_DROP_EMPTY_RESPONSE_SPANS", :boolean, nil),
   telemetry_top_process_limit:
     env!("ELECTRIC_TELEMETRY_TOP_PROCESS_LIMIT", &Electric.Config.parse_top_process_limit!/1, nil) ||
       env!(

packages/sync-service/lib/electric/config.ex

@@ -84,6 +84,9 @@ defmodule Electric.Config do
     telemetry_url: URI.new!("https://checkpoint.electric-sql.com"),
     otel_sampling_ratio: 0.01,
     metrics_sampling_ratio: 1,
+    # When true, the OTel spans of empty/up-to-date shape-GET responses are tail-dropped
+    # at export time to cut trace volume. See Electric.Telemetry.EmptyResponseSampler.
+    drop_empty_response_spans?: false,
     ## Memory
     # After this duration of inactivity, consumer processes will hibernate
     # to allow garbage collection

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

@@ -36,6 +36,7 @@ defmodule Electric.Plug.ServeShapePlug do
   alias Electric.ShapeCache
   alias Electric.ShapeCache.ShapeStatus
   alias Electric.Shapes.Api
+  alias Electric.Telemetry.EmptyResponseSampler
   alias Electric.Telemetry.OpenTelemetry
   alias Electric.Utils
   alias Plug.Conn
@@ -443,6 +444,7 @@ defmodule Electric.Plug.ServeShapePlug do
     bytes_sent = assigns[:streaming_bytes_sent] || 0
     is_live = get_live_mode(assigns)
     stack_id = get_in(conn.assigns, [:config, :stack_id])
+    is_empty = response_trace_attrs(conn)[:ot_is_empty_response] || false
 
     OpenTelemetry.execute(
       [:electric, :plug, :serve_shape],
@@ -458,7 +460,8 @@ defmodule Electric.Plug.ServeShapePlug do
         client_ip: conn.remote_ip,
         status: conn.status,
         stack_id: stack_id,
-        known_error: Api.Response.conn_has_known_error?(conn)
+        known_error: Api.Response.conn_has_known_error?(conn),
+        empty: is_empty
       }
     )
 
@@ -504,21 +507,56 @@ defmodule Electric.Plug.ServeShapePlug do
   # the parent-based sampler left no recording span, so `add_span_attributes` is a no-op
   # and nothing is stamped or exported.
   #
+  # Empty/up-to-date responses are additionally tail-dropped: when the drop is enabled
+  # they are stamped with `SampleRate = 0`, which the drop processor recognises as a
+  # sentinel to skip export (see Electric.Telemetry.EmptyResponseSampler).
+  #
   # Called both at span start (status not yet known: the rate hint is stamped as-is) and
   # at emit time, when the final attribute values overwrite the initial ones.
   defp add_span_attrs_from_conn(conn) do
     conn
     |> open_telemetry_attrs()
-    |> Map.merge(TraceContextPlug.sample_rate_attrs(conn, conn.status))
+    |> Map.merge(sample_rate_attrs(conn))
     |> OpenTelemetry.add_span_attributes()
 
     conn
   end
 
-  defp open_telemetry_attrs(%Conn{assigns: assigns} = conn) do
+  defp sample_rate_attrs(conn) do
+    base_attrs = TraceContextPlug.sample_rate_attrs(conn, conn.status)
+    trace_attrs = response_trace_attrs(conn)
+
+    decision =
+      EmptyResponseSampler.sample_rate(
+        conn.status,
+        trace_attrs[:ot_is_empty_response] || false,
+        trace_attrs[:ot_is_sse_response] || false,
+        Electric.Config.get_env(:drop_empty_response_spans?)
+      )
+
+    case decision do
+      :unchanged -> base_attrs
+      sample_rate -> Map.put(base_attrs, TraceContextPlug.sample_rate_attr(), sample_rate)
+    end
+  end
+
+  # Resolves the request and response structs from the conn assigns as bare maps.
+  # The response may live directly on the conn or nested under the request
+  # (its emit-time home), depending on where in the pipeline this runs.
+  defp request_and_response(assigns) do
     request = Map.get(assigns, :request, %{}) |> bare_map()
-    params = Map.get(request, :params, %{}) |> bare_map()
     response = (Map.get(assigns, :response) || Map.get(request, :response) || %{}) |> bare_map()
+    {request, response}
+  end
+
+  defp response_trace_attrs(%Conn{assigns: assigns}) do
+    {_request, response} = request_and_response(assigns)
+    Map.get(response, :trace_attrs, %{})
+  end
+
+  defp open_telemetry_attrs(%Conn{assigns: assigns} = conn) do
+    {request, response} = request_and_response(assigns)
+    params = Map.get(request, :params, %{}) |> bare_map()
     attrs = Map.get(response, :trace_attrs, %{})
     maybe_up_to_date = Map.get(response, :up_to_date, false)
 

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

@@ -0,0 +1,45 @@
+defmodule Electric.Telemetry.EmptyResponseSampler do
+  @moduledoc """
+  Decides the `SampleRate` weight for a shape-GET root span, tail-dropping the spans of
+  empty/up-to-date responses to cut trace volume.
+
+  When the drop is enabled, such spans are stamped with `SampleRate = 0`, which
+  `Electric.Telemetry.OpenTelemetry.EmptyResponseDropProcessor` recognises as a sentinel
+  to drop the span before it is queued for export.
+
+  The decision, in order:
+
+    1. Error responses (`status >= 500`) are kept and stamped with `SampleRate = 1` —
+       keep-on-error wins and is checked first.
+    2. Empty, non-SSE 2xx responses are dropped (`SampleRate = 0`) when the drop is
+       enabled.
+    3. Everything else is left unchanged (`:unchanged`), preserving whatever base
+       `SampleRate` the upstream rate hint produced.
+  """
+
+  @drop_sample_rate 0
+  @error_sample_rate 1
+
+  @doc """
+  The effective `SampleRate` for a shape-GET root span, or `:unchanged` to leave the base
+  rate untouched.
+
+    * `status` - the final HTTP status of the response.
+    * `is_empty_response?` - whether the response was empty/up-to-date.
+    * `is_sse_response?` - whether the response used server-sent events (excluded from the
+      drop).
+    * `drop_enabled?` - whether the empty-response drop is enabled.
+
+  Returns `0` (the drop sentinel) for a dropped empty response, `1` for a kept error
+  response, or `:unchanged` otherwise.
+  """
+  @spec sample_rate(integer() | nil, boolean(), boolean(), boolean()) ::
+          non_neg_integer() | :unchanged
+  def sample_rate(status, is_empty_response?, is_sse_response?, drop_enabled?) do
+    cond do
+      is_integer(status) and status >= 500 -> @error_sample_rate
+      drop_enabled? and is_empty_response? and not is_sse_response? -> @drop_sample_rate
+      true -> :unchanged
+    end
+  end
+end

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

@@ -36,6 +36,14 @@ defmodule Electric.Telemetry.OpenTelemetry.Config do
         Electric.Telemetry.OpenTelemetry.ResourceDetector
       ],
       resource: otel_resource,
-      processors: [otel_batch_processor, otel_simple_processor] |> Enum.reject(&is_nil/1)
+      processors:
+        [
+          # Must run before the exporting processors so that a span it drops is never
+          # queued for export.
+          {Electric.Telemetry.OpenTelemetry.EmptyResponseDropProcessor, %{}},
+          otel_batch_processor,
+          otel_simple_processor
+        ]
+        |> Enum.reject(&is_nil/1)
   end
 end

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

@@ -0,0 +1,52 @@
+# This processor reads the SDK-internal `#span{}` record, whose definition lives in the
+# `opentelemetry` (SDK) application. That app is only a dependency when building for the
+# telemetry target, so the module is only compiled there.
+if Electric.telemetry_enabled?() do
+  defmodule Electric.Telemetry.OpenTelemetry.EmptyResponseDropProcessor do
+    @moduledoc """
+    An OTel span processor that tail-drops spans stamped with the `SampleRate = 0`
+    sentinel by `Electric.Telemetry.EmptyResponseSampler`.
+
+    `on_end/2` returns `:dropped` for such spans. The SDK stops running the remaining
+    processors for a span as soon as one declines it, so registering this processor
+    ahead of the exporting processors (see `Electric.Telemetry.OpenTelemetry.Config`)
+    keeps a dropped span from ever being queued for export.
+
+    Only empty/up-to-date shape-GET response spans carry `SampleRate = 0`; every other span
+    passes through unchanged.
+    """
+
+    @behaviour :otel_span_processor
+
+    require Record
+
+    Record.defrecordp(
+      :span,
+      Record.extract(:span, from_lib: "opentelemetry/include/otel_span.hrl")
+    )
+
+    @sample_rate_attr "SampleRate"
+    @drop_sample_rate 0
+
+    @impl true
+    def on_start(_ctx, span, _config), do: span
+
+    @impl true
+    def on_end(span, _config) do
+      case sample_rate(span) do
+        @drop_sample_rate -> :dropped
+        _ -> true
+      end
+    end
+
+    @impl true
+    def force_flush(_config), do: :ok
+
+    defp sample_rate(span) do
+      case span(span, :attributes) do
+        :undefined -> nil
+        attributes -> attributes |> :otel_attributes.map() |> Map.get(@sample_rate_attr)
+      end
+    end
+  end
+end

packages/sync-service/test/electric/telemetry/empty_response_sampler_test.exs

@@ -0,0 +1,34 @@
+defmodule Electric.Telemetry.EmptyResponseSamplerTest do
+  use ExUnit.Case, async: true
+
+  alias Electric.Telemetry.EmptyResponseSampler
+
+  describe "sample_rate/4 with the drop enabled" do
+    test "empty 2xx responses are dropped (SampleRate = 0)" do
+      assert 0 = EmptyResponseSampler.sample_rate(200, true, false, true)
+    end
+
+    test "error responses (>= 500) are kept and stamped with SampleRate = 1, even when empty" do
+      assert 1 = EmptyResponseSampler.sample_rate(500, true, false, true)
+      assert 1 = EmptyResponseSampler.sample_rate(503, false, false, true)
+    end
+
+    test "SSE empty responses are left unchanged (never dropped)" do
+      assert :unchanged = EmptyResponseSampler.sample_rate(200, true, true, true)
+    end
+
+    test "non-empty 2xx responses are left unchanged" do
+      assert :unchanged = EmptyResponseSampler.sample_rate(200, false, false, true)
+    end
+  end
+
+  describe "sample_rate/4 with the drop disabled" do
+    test "empty 2xx responses are left unchanged (not dropped)" do
+      assert :unchanged = EmptyResponseSampler.sample_rate(200, true, false, false)
+    end
+
+    test "error responses are still kept with SampleRate = 1 (checked before the toggle)" do
+      assert 1 = EmptyResponseSampler.sample_rate(500, true, false, false)
+    end
+  end
+end

packages/sync-service/test/electric/telemetry/open_telemetry/empty_response_drop_processor_test.exs

@@ -0,0 +1,50 @@
+# The processor reads the SDK-internal `#span{}` record, only available on the telemetry
+# target (see the module for details).
+if Electric.telemetry_enabled?() do
+  defmodule Electric.Telemetry.OpenTelemetry.EmptyResponseDropProcessorTest do
+    use ExUnit.Case, async: true
+
+    alias Electric.Telemetry.OpenTelemetry.EmptyResponseDropProcessor, as: Processor
+
+    require Record
+
+    Record.defrecordp(
+      :span,
+      Record.extract(:span, from_lib: "opentelemetry/include/otel_span.hrl")
+    )
+
+    defp span_with_attributes(attrs) do
+      span(attributes: :otel_attributes.new(attrs, 128, :infinity))
+    end
+
+    describe "on_end/2" do
+      test "drops a span stamped with SampleRate = 0" do
+        span = span_with_attributes(%{"SampleRate" => 0})
+        assert Processor.on_end(span, %{}) == :dropped
+      end
+
+      test "keeps a span with a non-zero SampleRate" do
+        span = span_with_attributes(%{"SampleRate" => 20})
+        assert Processor.on_end(span, %{}) == true
+      end
+
+      test "keeps a span with no SampleRate attribute" do
+        span = span_with_attributes(%{"other" => "attr"})
+        assert Processor.on_end(span, %{}) == true
+      end
+
+      test "keeps a span with no attributes at all" do
+        assert Processor.on_end(span(), %{}) == true
+      end
+    end
+
+    test "on_start/3 returns the span unchanged" do
+      span = span_with_attributes(%{"SampleRate" => 0})
+      assert Processor.on_start(:otel_ctx.new(), span, %{}) == span
+    end
+
+    test "force_flush/1 returns :ok" do
+      assert Processor.force_flush(%{}) == :ok
+    end
+  end
+end