Reaggregate metrics between failed exports

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- Reaggregate pending metric data between failed exports so the payload size
+  stays bounded and does not trip the internal 20s send timeout under load
+
 ## [0.3.6] - 2025-04-08
 
 - Fix protobuf encoding of `:logger.report()` events

lib/otel_metric_exporter/metric_store.ex

@@ -232,7 +232,16 @@ defmodule OtelMetricExporter.MetricStore do
         x -> x
       end
 
-    earliest_gen..current_gen//1
+    # If previous exports failed, there may be multiple pending generations
+    # accumulated in the ETS table. Reaggregate them into a single generation
+    # before building the payload so that the request contains at most one data
+    # point per metric/tag combination regardless of how many export attempts
+    # have failed previously. Without this, the payload grows linearly with
+    # every failed export and eventually trips the send timeout
+    # (see electric-sql/stratovolt#1455).
+    reaggregate_failed_generations(state, earliest_gen, current_gen)
+
+    earliest_gen..earliest_gen//1
     |> Enum.reduce(%{}, fn gen, acc ->
       {_, start, finish} = List.first(:ets.lookup(state.generations_table, gen), {nil, nil, nil})
 
@@ -259,20 +268,96 @@ defmodule OtelMetricExporter.MetricStore do
     end)
     |> case do
       :ok ->
-        # Clear exported metrics
-        for x <- earliest_gen..current_gen//1 do
-          :ets.match_delete(state.metrics_table, {{x, :_, :_, :_, :_}, :_, :_})
-          :ets.delete(state.generations_table, x)
-        end
-
+        # Clear the exported generation. After reaggregation all pending data
+        # lives in earliest_gen, so this is the only generation that needs
+        # clearing.
+        :ets.match_delete(state.metrics_table, {{earliest_gen, :_, :_, :_, :_}, :_, :_})
+        :ets.delete(state.generations_table, earliest_gen)
         :ok
 
       {:error, reason} ->
         Logger.error("Failed to export metrics: #{inspect(reason)}")
+        # The data is left in earliest_gen. The next export cycle will
+        # reaggregate it together with any newly-recorded metrics before
+        # attempting another send.
         {:error, reason}
     end
   end
 
+  # Collapses generations in the range earliest_gen..current_gen into earliest_gen,
+  # merging ETS rows by (name, type, tags, bucket) key. Counters/sums/distribution
+  # bucket counts are added together, last_value keeps the newest value, and
+  # distribution min/max are merged with min/max. The generations_table entry for
+  # earliest_gen has its finish timestamp extended to that of current_gen.
+  defp reaggregate_failed_generations(%State{} = _state, earliest_gen, current_gen)
+       when earliest_gen == current_gen,
+       do: :ok
+
+  defp reaggregate_failed_generations(%State{} = state, earliest_gen, current_gen) do
+    for gen <- (earliest_gen + 1)..current_gen//1 do
+      rows = :ets.match_object(state.metrics_table, {{gen, :_, :_, :_, :_}, :_, :_})
+
+      for {{^gen, name, type, tags, bucket}, value, extra} <- rows do
+        target_key = {earliest_gen, name, type, tags, bucket}
+        merge_row(state.metrics_table, type, bucket, target_key, value, extra)
+      end
+
+      :ets.match_delete(state.metrics_table, {{gen, :_, :_, :_, :_}, :_, :_})
+    end
+
+    # Extend earliest generation's finish time to that of the last rotated generation
+    case :ets.lookup(state.generations_table, current_gen) do
+      [{_, _, finish}] when not is_nil(finish) ->
+        :ets.update_element(state.generations_table, earliest_gen, {3, finish})
+
+      _ ->
+        :ok
+    end
+
+    # Drop the now-empty generation entries
+    for gen <- (earliest_gen + 1)..current_gen//1 do
+      :ets.delete(state.generations_table, gen)
+    end
+
+    :ok
+  end
+
+  # Counter/sum: add values.
+  defp merge_row(table, type, nil, target_key, value, _extra) when type in [:counter, :sum] do
+    :ets.update_counter(table, target_key, value, {target_key, 0, nil})
+  end
+
+  # Last value: newest generation wins. Since we iterate from older to newer,
+  # simply overwriting is correct.
+  defp merge_row(table, :last_value, nil, target_key, value, _extra) do
+    :ets.insert(table, {target_key, value, nil})
+  end
+
+  # Distribution min/max sentinel rows: take min or max as appropriate.
+  defp merge_row(table, :distribution, :min, target_key, value, _extra) do
+    case :ets.lookup(table, target_key) do
+      [{_, current, _}] when current <= value -> :ok
+      _ -> :ets.insert(table, {target_key, value, nil})
+    end
+  end
+
+  defp merge_row(table, :distribution, :max, target_key, value, _extra) do
+    case :ets.lookup(table, target_key) do
+      [{_, current, _}] when current >= value -> :ok
+      _ -> :ets.insert(table, {target_key, value, nil})
+    end
+  end
+
+  # Distribution bucket row: add count and sum.
+  defp merge_row(table, :distribution, _bucket_idx, target_key, count, sum) do
+    :ets.update_counter(
+      table,
+      target_key,
+      [{2, count}, {3, sum}],
+      {target_key, 0, 0}
+    )
+  end
+
   defp convert_metric(
          %{name: name, description: description, unit: unit} = metric,
          values

test/otel_metric_exporter/metric_store_test.exs

@@ -191,7 +191,7 @@ defmodule OtelMetricExporter.MetricStoreTest do
       assert MetricStore.get_metrics(@name, 0) == metrics
     end
 
-    test "preserves metrics across generations on failed exports", %{
+    test "reaggregates metrics across generations on failed exports", %{
       bypass: bypass,
       store_config: config
     } do
@@ -212,29 +212,66 @@ defmodule OtelMetricExporter.MetricStoreTest do
       # Second generation
       MetricStore.write_metric(@name, metric, 2, tags)
 
-      # Second export succeeds and should include both generations
+      # Second export succeeds and should contain a single reaggregated data point
+      # spanning both generations (sum = 3) instead of one point per generation.
+      # This prevents the request payload from growing unboundedly across repeated
+      # export failures (see electric-sql/stratovolt#1455).
       Bypass.expect_once(bypass, "POST", "/v1/metrics", fn conn ->
         {:ok, body, conn} = Plug.Conn.read_body(conn)
         metrics = ExportMetricsServiceRequest.decode(body)
 
-        # Verify that we have one metric with sum = 3 (1 from first generation + 2 from second)
         assert [%{scope_metrics: [%{metrics: [metric]}]}] = metrics.resource_metrics
 
-        assert {:sum, %{data_points: [point1, point2]}} = metric.data
-        assert {:as_int, 1} = point1.value
-        assert {:as_int, 2} = point2.value
-
-        assert point1.time_unix_nano < point2.time_unix_nano
-        assert point2.start_time_unix_nano > point1.time_unix_nano
+        assert {:sum, %{data_points: [point]}} = metric.data
+        assert {:as_int, 3} = point.value
+        assert point.start_time_unix_nano <= point.time_unix_nano
 
         Plug.Conn.resp(conn, 200, "")
       end)
 
       assert :ok = MetricStore.export_sync(@name)
 
-      # Both generations should be cleared after successful export
+      # All generations should be cleared after successful export
       assert MetricStore.get_metrics(@name, 0) == %{}
       assert MetricStore.get_metrics(@name, 1) == %{}
+      assert MetricStore.get_metrics(@name, 2) == %{}
+    end
+
+    test "reaggregation across failures keeps payload size bounded", %{
+      bypass: bypass,
+      store_config: config
+    } do
+      metric = Metrics.sum("test.sum")
+      tags = %{test: "value"}
+      start_supervised!({MetricStore, %{config | metrics: [metric]}})
+
+      # Simulate multiple consecutive export failures.
+      Bypass.expect(bypass, "POST", "/v1/metrics", fn conn ->
+        Plug.Conn.resp(conn, 500, "Internal Server Error")
+      end)
+
+      for i <- 1..5 do
+        MetricStore.write_metric(@name, metric, i, tags)
+        capture_log(fn -> MetricStore.export_sync(@name) end)
+      end
+
+      Bypass.down(bypass)
+      Bypass.up(bypass)
+
+      # Next successful export should contain exactly one data point with the
+      # cumulative sum, not five points.
+      Bypass.expect_once(bypass, "POST", "/v1/metrics", fn conn ->
+        {:ok, body, conn} = Plug.Conn.read_body(conn)
+        metrics = ExportMetricsServiceRequest.decode(body)
+
+        assert [%{scope_metrics: [%{metrics: [metric]}]}] = metrics.resource_metrics
+        assert {:sum, %{data_points: [point]}} = metric.data
+        assert {:as_int, 15} = point.value
+
+        Plug.Conn.resp(conn, 200, "")
+      end)
+
+      assert :ok = MetricStore.export_sync(@name)
     end
   end
 end