beam-telemetry · robacourt · Jun 29, 2026
@@ -1,5 +1,24 @@
 # Changelog
 
+## Unreleased
+
+### Fixes
+
+  * Bound the distribution (histogram) sample buffer so it no longer grows without limit when
+    the `/metrics` endpoint is scraped infrequently or never. The registry now aggregates
+    buffered samples on its own — once `:flush_threshold` samples have accumulated (default
+    `10_000`) and at most every `:max_flush_interval_ms` (default `60_000`) as a fallback —
+    in addition to aggregating on scrape. Set either option to `:infinity` to restore the
+    previous scrape-only behaviour. Fixes
+    [#41](https://github.com/beam-telemetry/telemetry_metrics_prometheus_core/issues/41) and
+    [#52](https://github.com/beam-telemetry/telemetry_metrics_prometheus_core/issues/52).
+
+### Changes
+
+  * Aggregation now runs inside the registry process, serializing it with the automatic
+    flushes and with concurrent scrapes. This removes a pre-existing race in the
+    read-modify-write of the cumulative aggregates table when scrapes overlapped.
+
 ## v1.2.1
 
 ### Changes

@@ -22,10 +22,24 @@ defmodule TelemetryMetricsPrometheus.Core do
         Supervisor.start_link(children, opts)
       end
 
-  Note that aggregations for distributions (histogram) only occur at scrape time.
-  These aggregations only have to process events that have occurred since the last
-  scrape, so it's recommended at this time to keep an eye on scrape durations if
-  you're reporting a large number of distributions or you have a high tag cardinality.
+  Distribution (histogram) samples are buffered as they are observed and folded into their
+  histogram buckets during aggregation. Aggregation happens at scrape time and, to keep the
+  buffer bounded regardless of how often (or whether) you scrape, automatically once
+  `:flush_threshold` samples have been buffered and at most every `:max_flush_interval_ms` as
+  a fallback. Each aggregation only has to process events buffered since the last one, so it's
+  recommended to keep an eye on scrape durations if you're reporting a large number of
+  distributions or you have a high tag cardinality. See `start_link/1` for tuning these
+  options.
+
+  > #### Scrape the endpoint, or rely on the auto-flush {: .warning}
+  >
+  > Before these flush options existed, a reporter whose `/metrics` endpoint was never scraped
+  > would buffer distribution samples forever, growing unbounded (see issues
+  > [#41](https://github.com/beam-telemetry/telemetry_metrics_prometheus_core/issues/41) and
+  > [#52](https://github.com/beam-telemetry/telemetry_metrics_prometheus_core/issues/52)). The
+  > defaults now bound that buffer even when nothing scrapes, but distributions are still only
+  > *exported* on scrape — point a Prometheus-compatible scraper at the endpoint to actually
+  > collect them.
 
   ## Telemetry.Metrics to Prometheus Equivalents
 
@@ -197,6 +211,14 @@ defmodule TelemetryMetricsPrometheus.Core do
     tree from proceeding until all Telemetry event handlers are initialized. This is
     useful if subsequent supervision tree children emit events on start up and you don't
     want to miss those events due to an async start. Defaults to `true`.
+  * `:flush_threshold` - the number of buffered distribution samples that triggers an
+    automatic aggregation, bounding the buffer's memory use independently of scrape frequency.
+    Counted lock-free across all distributions; a burst over the threshold enqueues a single
+    flush. Set to `:infinity` to disable size-based flushing. Defaults to `10_000`.
+  * `:max_flush_interval_ms` - fallback interval, in milliseconds, after which buffered
+    distribution samples are aggregated even if `:flush_threshold` has not been reached. This
+    drains low-volume distributions and keeps the exported histograms fresh. Set to
+    `:infinity` to disable. Defaults to `60_000`.
   """
   @spec start_link(prometheus_options()) :: GenServer.on_start()
   def start_link(options) do
@@ -211,11 +233,14 @@ defmodule TelemetryMetricsPrometheus.Core do
   """
   @spec scrape(name :: atom()) :: String.t()
   def scrape(name \\ :prometheus_metrics) do
+    # Aggregate inside the registry process so the read-modify-write of the cumulative
+    # aggregates table is serialized with the periodic/threshold flushes (and with any other
+    # concurrent scrape). The read-only export below can safely run in the caller.
+    :ok = Registry.aggregate(name)
+
     config = Registry.config(name)
     metrics = Registry.metrics(name)
 
-    :ok = Aggregator.aggregate(metrics, config.aggregates_table_id, config.dist_table_id)
-
     Aggregator.get_time_series(config.aggregates_table_id)
     |> Exporter.export(metrics)
   end

@@ -17,6 +17,20 @@ defmodule TelemetryMetricsPrometheus.Core.Distribution do
   """
   @type buckets :: [number(), ...]
 
+  @typedoc """
+  Trigger that bounds the size of the buffered-sample (`:duplicate_bag`) table.
+
+  `staged` counts buffered samples, `flush_pending` de-bounces the signal so a burst over the
+  threshold enqueues a single flush, and `registry` is the process that performs the
+  (serialized) aggregation. `flush_threshold` is `:infinity` to disable size-based flushing.
+  """
+  @type flush_ctx :: %{
+          registry: pid(),
+          staged: :atomics.atomics_ref(),
+          flush_pending: :atomics.atomics_ref(),
+          flush_threshold: pos_integer() | :infinity
+        }
+
   @type config :: %{
           keep: Metrics.keep(),
           measurement: Metrics.measurement(),
@@ -26,13 +40,19 @@ defmodule TelemetryMetricsPrometheus.Core.Distribution do
           tags: Metrics.tags(),
           tag_values_fun: Metrics.tag_values(),
           type: :histogram,
-          unit: Metrics.unit()
+          unit: Metrics.unit(),
+          flush_ctx: flush_ctx()
         }
 
-  @spec register(metric :: Metrics.Distribution.t(), table_id :: atom(), owner :: pid()) ::
+  @spec register(
+          metric :: Metrics.Distribution.t(),
+          table_id :: atom(),
+          owner :: pid(),
+          flush_ctx :: flush_ctx()
+        ) ::
           {:ok, :telemetry.handler_id()} | {:error, :already_exists}
 
-  def register(metric, table_id, owner) do
+  def register(metric, table_id, owner, flush_ctx \\ disabled_flush_ctx()) do
     handler_id = EventHandler.handler_id(metric.name, owner)
 
     with :ok <-
@@ -49,7 +69,8 @@ defmodule TelemetryMetricsPrometheus.Core.Distribution do
                tags: metric.tags,
                tag_values_fun: metric.tag_values,
                type: :histogram,
-               unit: metric.unit
+               unit: metric.unit,
+               flush_ctx: flush_ctx
              }
            ) do
       {:ok, handler_id}
@@ -74,10 +95,37 @@ defmodule TelemetryMetricsPrometheus.Core.Distribution do
          labels <- Map.take(mapped_values, config.tags) do
       true = :ets.insert(config.table, {config.name, {labels, measurement}})
 
-      :ok
+      maybe_request_flush(config.flush_ctx)
     else
       false -> :ok
       error -> EventHandler.handle_event_error(error, config)
     end
   end
+
+  # Used when a distribution is registered outside the registry (e.g. in tests): no registry to
+  # flush to, so size-based flushing is disabled and the dist table is drained on scrape only.
+  @spec disabled_flush_ctx() :: flush_ctx()
+  defp disabled_flush_ctx do
+    %{
+      registry: self(),
+      staged: :atomics.new(1, []),
+      flush_pending: :atomics.new(1, []),
+      flush_threshold: :infinity
+    }
+  end
+
+  # Lock-free, on the hot path: bump the staged-sample counter and, only when it crosses the
+  # threshold, ask the registry to flush. `compare_exchange/4` guarantees a single in-flight
+  # request per drain cycle, so a burst over the threshold never floods the registry mailbox.
+  @spec maybe_request_flush(flush_ctx()) :: :ok
+  defp maybe_request_flush(%{flush_threshold: :infinity}), do: :ok
+
+  defp maybe_request_flush(%{flush_threshold: threshold} = ctx) do
+    if :atomics.add_get(ctx.staged, 1, 1) >= threshold and
+         :atomics.compare_exchange(ctx.flush_pending, 1, 0, 1) == :ok do
+      send(ctx.registry, :flush)
+    end
+
+    :ok
+  end
 end
@@ -5,7 +5,7 @@ defmodule TelemetryMetricsPrometheus.Core.Registry do
   require Logger
 
   alias Telemetry.Metrics
-  alias TelemetryMetricsPrometheus.Core.{Counter, Distribution, LastValue, Sum}
+  alias TelemetryMetricsPrometheus.Core.{Aggregator, Counter, Distribution, LastValue, Sum}
 
   @type name :: atom()
   @type metric_exists_error() :: {:error, :already_exists, Metrics.t()}
@@ -32,11 +32,28 @@ defmodule TelemetryMetricsPrometheus.Core.Registry do
 
     Process.flag(:trap_exit, true)
 
-    state = %{
-      config: %{aggregates_table_id: aggregates_table_id, dist_table_id: dist_table_id},
-      metrics: []
+    # Distribution (histogram) samples are buffered in the dist table and only folded into
+    # the aggregates table when aggregated. To keep that buffer bounded even when the
+    # `/metrics` endpoint is scraped infrequently (or never), the registry aggregates on its
+    # own: as soon as `flush_threshold` samples have been buffered (tracked lock-free in the
+    # event handlers via `staged`), and at most every `max_flush_interval_ms` as a fallback
+    # for low-volume metrics. `flush_pending` ensures a burst over the threshold enqueues a
+    # single flush rather than one per event. Set either option to `:infinity` to disable it.
+    flush_threshold = normalize_limit(Keyword.get(opts, :flush_threshold, 10_000))
+    max_flush_interval_ms = normalize_limit(Keyword.get(opts, :max_flush_interval_ms, 60_000))
+
+    config = %{
+      aggregates_table_id: aggregates_table_id,
+      dist_table_id: dist_table_id,
+      flush_threshold: flush_threshold,
+      staged: :atomics.new(1, []),
+      flush_pending: :atomics.new(1, [])
     }
 
+    state = %{config: config, metrics: [], max_flush_interval_ms: max_flush_interval_ms}
+
+    schedule_flush_timer(max_flush_interval_ms)
+
     if start_async do
       send(self(), {:setup, opts})
 
@@ -48,6 +65,13 @@ defmodule TelemetryMetricsPrometheus.Core.Registry do
     end
   end
 
+  defp normalize_limit(:infinity), do: :infinity
+  defp normalize_limit(nil), do: :infinity
+  defp normalize_limit(n) when is_integer(n) and n > 0, do: n
+
+  defp schedule_flush_timer(:infinity), do: :ok
+  defp schedule_flush_timer(interval), do: Process.send_after(self(), :flush_timer, interval)
+
   @spec register(Metrics.t(), atom()) ::
           :ok | metric_exists_error() | unsupported_metric_type_error()
   def register(metric, name \\ __MODULE__) do
@@ -136,13 +160,38 @@ defmodule TelemetryMetricsPrometheus.Core.Registry do
     GenServer.call(name, :get_metrics)
   end
 
+  @doc """
+  Aggregate buffered distribution samples into the aggregates table.
+
+  Runs inside the registry process so it is serialized with the periodic/threshold flushes,
+  making the read-modify-write of the cumulative aggregates table race-free even when several
+  scrapes (or a scrape and a flush) happen concurrently.
+  """
+  @spec aggregate(name()) :: :ok
+  def aggregate(name) do
+    GenServer.call(name, :aggregate)
+  end
+
   @impl true
   def handle_info({:setup, opts}, state) do
     registered = setup_registry(opts, state.config)
 
     {:noreply, %{state | metrics: registered}}
   end
 
+  # Threshold-triggered flush from the event handlers: drain only, no timer involved.
+  def handle_info(:flush, state) do
+    {:noreply, do_aggregate(state)}
+  end
+
+  # Fallback-interval flush: drain, then re-arm exactly one successor timer so the periodic
+  # chain never multiplies (threshold flushes deliberately do not reschedule).
+  def handle_info(:flush_timer, state) do
+    state = do_aggregate(state)
+    schedule_flush_timer(state.max_flush_interval_ms)
+    {:noreply, state}
+  end
+
   def handle_call(:get_config, _from, state) do
     {:reply, state.config, state}
   end
@@ -152,6 +201,10 @@ defmodule TelemetryMetricsPrometheus.Core.Registry do
     {:reply, metrics, state}
   end
 
+  def handle_call(:aggregate, _from, state) do
+    {:reply, :ok, do_aggregate(state)}
+  end
+
   @impl true
   @spec handle_call({:register, Metrics.t()}, GenServer.from(), map()) ::
           {:reply, :ok, map()}
@@ -171,6 +224,18 @@ defmodule TelemetryMetricsPrometheus.Core.Registry do
          do: :ok
   end
 
+  # Drains the distribution buffer into the aggregates table and re-arms the flush trigger.
+  # The order matters: aggregate (which `:ets.take`s the buffered samples) first, then reset
+  # the staged counter, then clear `flush_pending` so the next over-threshold event can
+  # enqueue a fresh flush.
+  defp do_aggregate(%{config: config, metrics: metrics} = state) do
+    definitions = Enum.map(metrics, &elem(&1, 0))
+    :ok = Aggregator.aggregate(definitions, config.aggregates_table_id, config.dist_table_id)
+    :atomics.put(config.staged, 1, 0)
+    :atomics.put(config.flush_pending, 1, 0)
+    state
+  end
+
   defp setup_registry(opts, config) do
     opts
     |> Keyword.get(:metrics, [])
@@ -233,7 +298,14 @@ defmodule TelemetryMetricsPrometheus.Core.Registry do
   defp register_metric(%Metrics.Distribution{} = metric, config) do
     validate_distribution_buckets!(metric)
 
-    case Distribution.register(metric, config.dist_table_id, self()) do
+    flush_ctx = %{
+      registry: self(),
+      staged: config.staged,
+      flush_pending: config.flush_pending,
+      flush_threshold: config.flush_threshold
+    }
+
+    case Distribution.register(metric, config.dist_table_id, self(), flush_ctx) do
       {:ok, handler_id} ->
         reporter_options =
           Keyword.update!(