Export BEAM + cgroup/OS host metrics for reconciliation

.changeset/per-shape-disk-usage-metric.md

@@ -0,0 +1,6 @@
+---
+"@core/electric-telemetry": patch
+"@core/sync-service": patch
+---
+
+Export in-app BEAM allocator (`vm.alloc.*`), cgroup v2 (`cgroup.*`), and per-process (`host.proc.beam.*`) metrics for reconciling VM-reported memory/CPU/disk against what the kernel charges. Also emit a new `electric.storage.dir.bytes` stack-level metric reporting on-disk size for the top-N largest shapes (tagged by shape handle), computed during the existing periodic disk-usage walk so it adds no extra filesystem traversal.

integration-tests/tests/otel-export.lux

@@ -101,6 +101,21 @@
   # Verify the presence of process.memory.total metric
   [invoke grep_otel_collector_output "Metric process\.memory\.total .*process_type=A_memory_hog.*Value=[0-9]+"]
 
+  # Verify BEAM allocator metrics (vm.alloc.*) from SystemMetrics. These come from
+  # :recon_alloc and are cross-platform; they are sampled every poll tick.
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.allocated .*Value=[0-9]+"]
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.used .*Value=[0-9]+"]
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.unused .*Value=[0-9]+"]
+  [invoke grep_otel_collector_output "Metric vm\.alloc\.carrier_usage .*Value=[.0-9]+"]
+
+  # Verify per-process /proc metrics (host.proc.beam.*) from SystemMetrics.
+  # These read /proc and are populated on Linux (the CI runner).
+  [invoke grep_otel_collector_output "Metric host\.proc\.beam\.vm_rss .*Value=[0-9]+"]
+
+  # Verify cgroup accounting metrics (cgroup.*) from SystemMetrics. Populated when a
+  # cgroup is detected — cgroup v2 on the CI runner.
+  [invoke grep_otel_collector_output "Metric cgroup\.memory\.current .*Value=[0-9]+"]
+
   # Verify that LSN metrics are exported
   [invoke grep_otel_collector_output "Metric electric\.postgres\.replication\.pg_wal_offset .*stack_id=single_stack"]
   [invoke grep_otel_collector_output "Metric electric\.postgres\.replication\.slot_confirmed_flush_lsn_lag .*stack_id=single_stack"]

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

@@ -86,7 +86,16 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
           :get_system_memory_usage
         ],
         &{__MODULE__, &1, [telemetry_opts]}
-      )
+      ) ++
+      [
+        # System metrics (vm.alloc.*, cgroup.*, host.*); each measurement no-ops
+        # where unsupported (no cgroup v2, non-Linux). The expensive
+        # vm.alloc.fragmentation.* breakdown is sampled by SystemMonitor on a
+        # one-minute timer instead of this poller.
+        {ElectricTelemetry.SystemMetrics, :recon_alloc_measurement, [telemetry_opts]},
+        {ElectricTelemetry.SystemMetrics.Cgroup, :measurement, [telemetry_opts]},
+        {ElectricTelemetry.SystemMetrics.Proc, :measurement, [telemetry_opts]}
+      ]
   end
 
   def metrics(telemetry_opts) do
@@ -117,6 +126,35 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
       last_value("vm.memory.processes_used", unit: :byte),
       last_value("vm.memory.system", unit: :byte),
       last_value("vm.memory.total", unit: :byte),
+      # BEAM allocator metrics (emitted by ElectricTelemetry.SystemMetrics).
+      last_value("vm.alloc.allocated", unit: :byte),
+      last_value("vm.alloc.used", unit: :byte),
+      last_value("vm.alloc.unused", unit: :byte),
+      last_value("vm.alloc.carrier_usage"),
+      last_value("vm.alloc.fragmentation.unused", tags: [:allocator], unit: :byte),
+      # cgroup v2 accounting metrics (emitted by
+      # ElectricTelemetry.SystemMetrics.Cgroup).
+      last_value("cgroup.memory.current", unit: :byte),
+      last_value("cgroup.memory.anon", unit: :byte),
+      last_value("cgroup.memory.file", unit: :byte),
+      last_value("cgroup.memory.working_set", unit: :byte),
+      last_value("cgroup.memory.max", unit: :byte),
+      # PSI "full avg10" stall percentage (v2 only).
+      last_value("cgroup.memory.pressure.full.avg10"),
+      last_value("cgroup.cpu.usage_usec", unit: :microsecond),
+      last_value("cgroup.cpu.nr_throttled"),
+      last_value("cgroup.cpu.throttled_usec", unit: :microsecond),
+      last_value("cgroup.cpu.pressure.full.avg10"),
+      last_value("cgroup.io.rbytes", unit: :byte),
+      last_value("cgroup.io.wbytes", unit: :byte),
+      # per-process /proc metrics (emitted by
+      # ElectricTelemetry.SystemMetrics.Proc).
+      last_value("host.proc.beam.rss_anon", unit: :byte),
+      last_value("host.proc.beam.rss_file", unit: :byte),
+      last_value("host.proc.beam.rss_shmem", unit: :byte),
+      last_value("host.proc.beam.vm_rss", unit: :byte),
+      last_value("host.proc.beam.io.read_bytes", unit: :byte),
+      last_value("host.proc.beam.io.write_bytes", unit: :byte),
       sum("vm.monitor.long_message_queue.length", tags: [:process_type]),
       distribution("vm.monitor.long_schedule.timeout",
         tags: [:process_type],

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

@@ -4,6 +4,7 @@ defmodule ElectricTelemetry.DiskUsage do
   alias ElectricTelemetry.DiskUsage.Disk
 
   @default_update_period 60_000
+  @default_top_n 10
 
   def start_link(args) do
     {:ok, stack_id} = Keyword.fetch(args, :stack_id)
@@ -15,15 +16,29 @@ defmodule ElectricTelemetry.DiskUsage do
   end
 
   def current(stack_id) do
-    table = name(stack_id)
-
-    case :ets.lookup(table, :usage_bytes) do
-      [] -> :pending
-      [{:usage_bytes, _bytes, nil, _duration}] -> :pending
+    case lookup(stack_id, :usage_bytes) do
       [{:usage_bytes, bytes, %DateTime{}, duration}] -> {:ok, bytes, duration}
+      _ -> :pending
+    end
+  end
+
+  @doc """
+  Returns the top-N largest per-directory subtotals as a list of
+  `{dir_name, bytes}` tuples sorted descending by size, or `:pending` if no
+  measurement with grouping enabled has completed yet.
+  """
+  def current_dirs(stack_id) do
+    case lookup(stack_id, :top_dirs) do
+      [{:top_dirs, dirs}] -> {:ok, dirs}
+      _ -> :pending
     end
+  end
+
+  # The table doesn't exist until this stack's DiskUsage server has started.
+  defp lookup(stack_id, key) do
+    :ets.lookup(name(stack_id), key)
   rescue
-    ArgumentError -> :pending
+    ArgumentError -> []
   end
 
   @impl GenServer
@@ -47,9 +62,12 @@ defmodule ElectricTelemetry.DiskUsage do
         storage_dir: storage_dir,
         manual_refresh: Keyword.get(args, :manual_refresh, false),
         update_period: update_period,
+        group_depth: Keyword.get(args, :group_depth),
+        top_n: Keyword.get(args, :top_n, @default_top_n),
         usage_bytes: usage_bytes,
         updated_at: updated_at,
         measurement_duration: 0,
+        top_dirs: nil,
         timer: nil
       }
       |> ets_write()
@@ -77,32 +95,45 @@ defmodule ElectricTelemetry.DiskUsage do
   end
 
   defp read_disk_usage(state) do
-    {duration, bytes} =
+    exclude = [usage_cache_file(state.storage_dir)]
+
+    {duration, {bytes, buckets}} =
       :timer.tc(
-        fn ->
-          Disk.recursive_usage(state.storage_dir, [usage_cache_file(state.storage_dir)])
-        end,
+        fn -> Disk.recursive_usage_grouped(state.storage_dir, exclude, state.group_depth) end,
         :millisecond
       )
 
-    updated_at = DateTime.utc_now()
-
-    %{state | usage_bytes: bytes, updated_at: updated_at, measurement_duration: duration}
+    %{
+      state
+      | usage_bytes: bytes,
+        updated_at: DateTime.utc_now(),
+        measurement_duration: duration,
+        top_dirs: state.group_depth && top_n(buckets, state.top_n)
+    }
     |> ets_write()
     |> save_usage!()
     |> cancel_timer()
     |> schedule_update()
   end
 
+  # The `n` largest `{name, bytes}` buckets, sorted descending by size. The
+  # full bucket map is materialized and sorted once per (~60s) walk before
+  # being trimmed; fine at ~10k shapes per stack — replace with a bounded
+  # min-heap if a stack ever holds hundreds of thousands.
+  defp top_n(buckets, n) do
+    buckets
+    |> Enum.sort_by(fn {_name, bytes} -> bytes end, :desc)
+    |> Enum.take(n)
+  end
+
   defp ets_write(state) do
-    %{
-      usage_bytes: usage_bytes,
-      updated_at: updated_at,
-      table: table,
-      measurement_duration: duration
-    } = state
+    :ets.insert(
+      state.table,
+      {:usage_bytes, state.usage_bytes, state.updated_at, state.measurement_duration}
+    )
+
+    if dirs = state.top_dirs, do: :ets.insert(state.table, {:top_dirs, dirs})
 
-    :ets.insert(table, {:usage_bytes, usage_bytes, updated_at, duration})
     state
   end
 

packages/electric-telemetry/lib/electric/telemetry/disk_usage/disk.ex

@@ -3,36 +3,89 @@ defmodule ElectricTelemetry.DiskUsage.Disk do
 
   Record.defrecord(:file_info, Record.extract(:file_info, from_lib: "kernel/include/file.hrl"))
 
+  @doc """
+  Recursively sum the size of all regular files under `path`, excluding any
+  paths listed in `exclude`.
+  """
   def recursive_usage(path, exclude) do
-    do_recursive_usage(path, MapSet.new(exclude), 0)
+    {total, _buckets} = recursive_usage_grouped(path, exclude, nil)
+    total
   end
 
-  def do_recursive_usage(path, exclude, acc) do
+  @doc """
+  Like `recursive_usage/2`, but in the same single traversal also returns a map
+  of per-directory subtotals bucketed at `group_depth` (`nil` disables
+  bucketing and yields an empty map).
+
+  The bucket map is keyed by the name (not full path) of each directory found
+  at exactly `group_depth` levels below `path` (the root `path` itself is depth
+  0), with the value being the recursive size of every regular file under that
+  directory. Excluded paths contribute to neither the total nor the buckets.
+  The buckets are a best-effort tally that only ever grows, while the total
+  keeps the legacy behaviour of resetting to 0 on an unreadable entry.
+
+  Returns `{total_bytes, %{dir_name => bytes}}`.
+  """
+  def recursive_usage_grouped(path, exclude, group_depth)
+      when is_nil(group_depth) or (is_integer(group_depth) and group_depth >= 0) do
+    # When grouping at the root (depth 0), the root's own basename is the bucket.
+    initial_key = if group_depth == 0, do: Path.basename(path), else: nil
+    walk(path, initial_key, MapSet.new(exclude), group_depth, 0, {0, %{}})
+  end
+
+  # `bucket_key` is the name of the ancestor directory sitting at `group_depth`,
+  # or `nil` until that depth is reached.
+  defp walk(path, bucket_key, exclude, group_depth, depth, {acc, buckets}) do
     case stat(path) do
       {:ok, file_info(size: size, type: :regular)} ->
         if MapSet.member?(exclude, path) do
-          acc
+          {acc, buckets}
         else
-          size + acc
+          {size + acc, add_to_bucket(buckets, bucket_key, size)}
         end
 
       {:ok, file_info(type: :directory)} ->
         case ls(path) do
           {:ok, files} ->
-            Enum.reduce(files, acc, &do_recursive_usage(Path.join(path, &1), exclude, &2))
+            Enum.reduce(files, {acc, buckets}, fn name, {acc, bucks} ->
+              # `name` is a charlist from :prim_file.list_dir/1 and sits at
+              # `depth + 1`. Once established, the bucket key propagates to all
+              # descendants.
+              child_bucket_key =
+                cond do
+                  not is_nil(bucket_key) -> bucket_key
+                  depth + 1 == group_depth -> List.to_string(name)
+                  true -> nil
+                end
+
+              walk(
+                Path.join(path, name),
+                child_bucket_key,
+                exclude,
+                group_depth,
+                depth + 1,
+                {acc, bucks}
+              )
+            end)
 
           {:error, _} ->
-            0
+            {0, buckets}
         end
 
       {:ok, _} ->
-        0
+        {0, buckets}
 
       {:error, _} ->
-        0
+        {0, buckets}
     end
   end
 
+  defp add_to_bucket(buckets, nil, _size), do: buckets
+
+  defp add_to_bucket(buckets, key, size) do
+    Map.update(buckets, key, size, &(&1 + size))
+  end
+
   defdelegate stat(path), to: :prim_file, as: :read_file_info
 
   defp ls(path) do

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

@@ -5,6 +5,9 @@ defmodule ElectricTelemetry.Opts do
       installation_id: [type: :string],
       stack_id: [type: :string],
       storage_dir: [type: :string],
+      # Directory depth (relative to the disk-usage walk root) at which to
+      # bucket per-directory disk usage subtotals; `nil` disables bucketing.
+      disk_usage_group_depth: [type: {:or, [:pos_integer, nil]}, default: nil],
       version: [type: :string, required: true],
       reporters: [
         type: :keyword_list,

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

@@ -67,9 +67,20 @@ defmodule ElectricTelemetry.StackTelemetry do
     ]
   end
 
+  # `:disk_usage_group_depth` is supplied by the caller since it depends on the
+  # deployment's storage layout (sync-service derives it from the real shape
+  # storage path, keeping this package layout-agnostic). When set, the disk walk
+  # additionally buckets per-directory subtotals at that depth — for Electric
+  # these are shape dirs, emitted as `electric.storage.dir.bytes` for the top-N
+  # largest. When `nil`, only the aggregate total is measured.
   defp disk_usage_child_specs(%{stack_id: stack_id} = opts) do
     if storage_dir = Map.get(opts, :storage_dir) do
-      [{ElectricTelemetry.DiskUsage, stack_id: stack_id, storage_dir: storage_dir}]
+      [
+        {ElectricTelemetry.DiskUsage,
+         stack_id: stack_id,
+         storage_dir: storage_dir,
+         group_depth: Map.get(opts, :disk_usage_group_depth)}
+      ]
     else
       []
     end
@@ -113,6 +124,7 @@ defmodule ElectricTelemetry.StackTelemetry do
       distribution("electric.storage.transaction_stored.replication_lag", unit: :millisecond),
       last_value("electric.storage.used.bytes", unit: :byte),
       distribution("electric.storage.used.measurement_duration", unit: :millisecond),
+      last_value("electric.storage.dir.bytes", unit: :byte, tags: [:stack_id, :shape]),
       counter("electric.postgres.replication.transaction_received.count"),
       sum("electric.postgres.replication.transaction_received.bytes", unit: :byte),
       sum("electric.storage.transaction_stored.bytes", unit: :byte),