Добавлена оперативная и историческая статистика нод и узлов mnesia #22

2026-05-28 18:59:21 +03:00
parent 29fe0386be
commit f507d08e2c
16 changed files with 753 additions and 201 deletions
@@ -19,10 +19,10 @@
  review, report, banned_word,
  ticket, subscription,
  admin_audit, notification,
-  stats, schema_migration
+  stats, node_metric, schema_migration
 ]).

-define(DISC_TABLES, ?TABLES -- [session, verification, admin_session]).
+-define(DISC_TABLES, ?TABLES -- [session, verification, admin_session, node_metric]).
 -define(TABLE_WAIT_TIMEOUT, 5000).
 -define(CLEANUP_INTERVAL, 30000). % 30 секунд

@@ -31,6 +31,8 @@
 %% ===================================================================

 start_link() ->
+  % Счётчики для метрик (сессии, ws-соединения)
+  ets:new(eventhub_counters, [named_table, public, set, {write_concurrency, true}]),
  gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

 init_tables() ->
@@ -220,7 +222,8 @@ table_opts(stats) -> [{disc_copies, [node()]}, {attributes, record_info(fields,
 table_opts(schema_migration) -> [{disc_copies, [node()]}, {attributes, record_info(fields, schema_migration)}];
 table_opts(session) -> [{ram_copies, [node()]}, {attributes, record_info(fields, session)}];
 table_opts(verification) -> [{ram_copies, [node()]}, {attributes, record_info(fields, verification)}];
-table_opts(admin_session) -> [{ram_copies, [node()]}, {attributes, record_info(fields, admin_session)}].
+table_opts(admin_session) -> [{ram_copies, [node()]}, {attributes, record_info(fields, admin_session)}];
+table_opts(node_metric) -> [{ram_copies, [node()]}, {local_content, true}, {attributes, record_info(fields, node_metric)}].

 %% ===================================================================
 %% Индексы
@@ -21,12 +21,20 @@ init([]) ->
      shutdown => 5000,
      type     => worker,
      modules  => [infra_mnesia]},
+    #{id => pg,
+      start => {pg, start_link, []},
+      restart => permanent,
+      type => worker},
    #{id       => stats_collector,
      start    => {stats_collector, start_link, []},
      restart  => permanent,
      shutdown => 5000,
      type     => worker,
      modules  => [stats_collector]},
+    #{id => node_monitor,
+      start => {node_monitor, start_link, []},
+      restart => permanent,
+      type => worker},
    #{id       => archive_manager,
      start    => {archive_manager, start_link, []},
      restart  => permanent,
@@ -0,0 +1,194 @@
+%%%-------------------------------------------------------------------
+%%% @doc Мониторинг узла.
+%%%
+%%% Каждые 5 секунд собирает системные метрики (память, процессы,
+%%% транзакции Mnesia, размеры таблиц, сессии, WebSocket-соединения,
+%%% сетевой ввод/вывод) и сохраняет их в локальную таблицу
+%%% `node_metric`. Также рассылает метрики через `pg` в группу
+%%% `node_metrics` для WebSocket-клиентов. Раз в час автоматически
+%%% удаляет метрики старше 7 дней.
+%%%
+%%% Запускается как постоянный gen_server под супервизором
+%%% `infra_sup`.
+%%% @end
+%%%-------------------------------------------------------------------
+-module(node_monitor).
+-behaviour(gen_server).
+-include("records.hrl").
+-export([start_link/0]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2]).
+
+-define(COLLECT_INTERVAL_MS, 5000).   % 5 секунд
+-define(CLEANUP_INTERVAL_MS, 3600000). % 1 час
+-define(RETENTION_DAYS, 7).
+-define(METRICS_GROUP, node_metrics).  % группа pg для метрик
+-define(IO_CACHE_TAB, node_io_cache). % ETS для кеширования предыдущих значений io
+
+%%%-------------------------------------------------------------------
+%%% @doc Запускает gen_server мониторинга.
+%%% @end
+%%%-------------------------------------------------------------------
+-spec start_link() -> {ok, pid()} | {error, term()}.
+start_link() ->
+  gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callback'и
+%%%===================================================================
+
+-spec init(term()) -> {ok, map()}.
+init([]) ->
+  % Создаём ETS-таблицу для кеширования предыдущих значений io
+  ets:new(?IO_CACHE_TAB, [named_table, public, set, {keypos, 1}]),
+  % Запускаем периодический сбор метрик и очистку
+  schedule_collect(),
+  schedule_cleanup(),
+  {ok, #{}}.
+
+-spec handle_call(term(), {pid(), term()}, map()) -> {reply, ok, map()}.
+handle_call(_Req, _From, State) ->
+  {reply, ok, State}.
+
+-spec handle_cast(term(), map()) -> {noreply, map()}.
+handle_cast(_Msg, State) ->
+  {noreply, State}.
+
+-spec handle_info(term(), map()) -> {noreply, map()}.
+handle_info(collect, State) ->
+  Metric = collect_metrics(),
+  core_node_metric:save(Metric),
+  % Рассылаем метрику всем подписчикам группы node_metrics
+  Members = pg:get_local_members(?METRICS_GROUP),
+  [Pid ! {node_metric, Metric} || Pid <- Members],
+  schedule_collect(),
+  {noreply, State};
+handle_info(cleanup, State) ->
+  % Удаляем метрики старше RETENTION_DAYS дней
+  Seconds = ?RETENTION_DAYS * 86400,
+  core_node_metric:delete_old(Seconds),
+  schedule_cleanup(),
+  {noreply, State};
+handle_info(_Info, State) ->
+  {noreply, State}.
+
+%%%===================================================================
+%%% Внутренние функции
+%%%===================================================================
+
+-spec schedule_collect() -> reference().
+schedule_collect() ->
+  erlang:send_after(?COLLECT_INTERVAL_MS, self(), collect).
+
+-spec schedule_cleanup() -> reference().
+schedule_cleanup() ->
+  erlang:send_after(?CLEANUP_INTERVAL_MS, self(), cleanup).
+
+-spec collect_metrics() -> #node_metric{}.
+collect_metrics() ->
+  Now = calendar:universal_time(),
+  Mem = erlang:memory(),
+  {IoIn, IoOut} = io_stats(),
+  #node_metric{
+    timestamp        = Now,
+    node             = node(),
+    memory_total     = proplists:get_value(total, Mem, 0),
+    memory_processes = proplists:get_value(processes, Mem, 0),
+    memory_atom      = proplists:get_value(atom, Mem, 0),
+    memory_binary    = proplists:get_value(binary, Mem, 0),
+    memory_ets       = proplists:get_value(ets, Mem, 0),
+    process_count    = erlang:system_info(process_count),
+    run_queue        = safe_run_queue(),
+    mnesia_commits   = mnesia:system_info(transaction_commits),
+    mnesia_failures  = mnesia:system_info(transaction_failures),
+    table_sizes      = table_sizes(),
+    active_sessions  = session_count(),
+    ws_connections   = ws_count(),
+    io_in            = IoIn,
+    io_out           = IoOut
+  }.
+
+%%%-------------------------------------------------------------------
+%%% @doc Возвращает карту `#{TableName => Size}` для всех таблиц Mnesia.
+%%% @end
+%%%-------------------------------------------------------------------
+-spec table_sizes() -> map().
+table_sizes() ->
+  Tables = mnesia:system_info(tables),
+  maps:from_list([{Tab, mnesia:table_info(Tab, size)} || Tab <- Tables]).
+
+%%%-------------------------------------------------------------------
+%%% @doc Возвращает количество активных сессий (пользовательских + админских).
+%%% Использует длину ETS-таблиц. Безопасно при отсутствии таблиц.
+%%% @end
+%%%-------------------------------------------------------------------
+-spec session_count() -> non_neg_integer().
+session_count() ->
+  core_counters:get(active_sessions) + core_counters:get(admin_active_sessions).
+
+%%%-------------------------------------------------------------------
+%%% @doc Возвращает количество открытых WebSocket-соединений.
+%%% Подсчитывает процессы-обработчики ws_handler и admin_ws_handler.
+%%% Безопасно при отсутствии супервизоров.
+%%% @end
+%%%-------------------------------------------------------------------
+-spec ws_count() -> non_neg_integer().
+ws_count() ->
+  lists:sum([
+    core_counters:get(ws_connections),
+    core_counters:get(admin_ws_connections)
+  ]).
+
+%%%-------------------------------------------------------------------
+%%% @doc Безопасное получение длины run queue.
+%%% Если `erlang:statistics/1` возвращает неожиданное значение,
+%%% возвращается 0.
+%%% @end
+%%%-------------------------------------------------------------------
+-spec safe_run_queue() -> non_neg_integer().
+safe_run_queue() ->
+  try erlang:statistics(run_queue) of
+    {RunQueue, _} when is_integer(RunQueue) -> RunQueue;
+    _ -> 0
+  catch
+    _:_ -> 0
+  end.
+
+%%%-------------------------------------------------------------------
+%%% @doc Вычисляет дельту входящего/исходящего трафика (байт) за интервал
+%%% сбора метрик. Использует `inet:getstat/1` для всех открытых сокетов.
+%%% @end
+%%%-------------------------------------------------------------------
+-spec io_stats() -> {non_neg_integer(), non_neg_integer()}.
+io_stats() ->
+  Ports = erlang:ports(),
+  TotalIn = lists:sum([recv_oct(Port) || Port <- Ports]),
+  TotalOut = lists:sum([send_oct(Port) || Port <- Ports]),
+  {PrevIn, PrevOut} = get_prev_io(),
+  ets:insert(?IO_CACHE_TAB, [{in, TotalIn}, {out, TotalOut}]),
+  {TotalIn - PrevIn, TotalOut - PrevOut}.
+
+-spec get_prev_io() -> {non_neg_integer(), non_neg_integer()}.
+get_prev_io() ->
+  In = case ets:lookup(?IO_CACHE_TAB, in) of
+         [{in, V1}] -> V1;
+         [] -> 0
+       end,
+  Out = case ets:lookup(?IO_CACHE_TAB, out) of
+          [{out, V2}] -> V2;
+          [] -> 0
+        end,
+  {In, Out}.
+
+-spec recv_oct(port()) -> non_neg_integer().
+recv_oct(Port) ->
+  case inet:getstat(Port, [recv_oct]) of
+    {ok, [{recv_oct, V}]} -> V;
+    _ -> 0
+  end.
+
+-spec send_oct(port()) -> non_neg_integer().
+send_oct(Port) ->
+  case inet:getstat(Port, [send_oct]) of
+    {ok, [{send_oct, V}]} -> V;
+    _ -> 0
+  end.