EventHub/EventHubBack
3
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bagofix
Test & Deploy to Snapdeploy / build-test-deploy (push) Failing after 24m46s
Details

Browse Source
This commit is contained in:
Алексей Сабилин
2026-06-05 23:20:03 +03:00
parent 2497fe951b
commit 17b2e617d7
41 changed files with 365 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/core_node_metric.erl
+3 -2
View File
@@ -8,8 +8,9 @@
%%%-------------------------------------------------------------------
-spec save(#node_metric{}) -> ok.
save(Metric) ->
mnesia:dirty_write(Metric),
ok.
try mnesia:dirty_write(Metric)
catch _:_ -> ok
end.
%%%-------------------------------------------------------------------
%%% @doc Возвращает метрики за период, отсортированные по времени.
src/core/core_subscription.erl
+18 -6
View File
@@ -269,15 +269,21 @@ apply_updates(Sub, Updates) ->
%%%-------------------------------------------------------------------
%%% @doc Разбор ISO8601 строки в формат `calendar:datetime()`.
%%% Поддерживает миллисекунды и таймзону.
%%% @end
%%%-------------------------------------------------------------------
-spec parse_iso_datetime(binary() | term()) -> calendar:datetime() | term().
parse_iso_datetime(Bin) when is_binary(Bin) ->
try
[DateStr, TimeStr] = string:split(Bin, "T"),
TimeStrNoZ = string:trim(TimeStr, trailing, "Z"),
[YearStr, MonthStr, DayStr] = string:split(DateStr, "-", all),
[HourStr, MinuteStr, SecondStr] = string:split(TimeStrNoZ, ":", all),
[DatePart, TimePart] = string:split(Bin, "T"),
[YearStr, MonthStr, DayStr] = string:split(DatePart, "-", all),
TimeNoZ = string:trim(TimePart, trailing, "Z"),
[HourStr, MinuteStr, SecondPart] = string:split(TimeNoZ, ":", all),
% удаляем миллисекунды, если они есть
SecondStr = case string:split(SecondPart, ".") of
[Sec, _] -> Sec;
_ -> SecondPart
end,
Year = binary_to_integer(list_to_binary(YearStr)),
Month = binary_to_integer(list_to_binary(MonthStr)),
Day = binary_to_integer(list_to_binary(DayStr)),
@@ -357,6 +363,7 @@ count_trial_subscriptions() ->
%%%-------------------------------------------------------------------
%%% @doc Платные подписки (trial_used = true), истекающие в течение Days дней.
%%% Принимает меры к преобразованию expires_at, если оно сохранено как строка.
%%% @end
%%%-------------------------------------------------------------------
-spec get_ending_paid_subscriptions(Days :: non_neg_integer()) -> [#subscription{}].
@@ -366,6 +373,11 @@ get_ending_paid_subscriptions(Days) ->
Match = #subscription{status = active, trial_used = true, _ = '_'},
ActivePaid = mnesia:dirty_match_object(Match),
lists:filter(fun(#subscription{expires_at = Exp}) ->
ExpSec = calendar:datetime_to_gregorian_seconds(Exp),
ExpDateTime = case Exp of
{{_,_,_}, {_,_,_}} -> Exp; % уже datetime
Bin when is_binary(Bin) -> parse_iso_datetime(Bin);
_ -> Exp
end,
ExpSec = calendar:datetime_to_gregorian_seconds(ExpDateTime),
ExpSec >= NowSec andalso ExpSec =< ThresholdSec
end, ActivePaid).
end, ActivePaid).
src/handlers/admin/admin_handler_admins.erl
+1 -1
View File
@@ -131,7 +131,7 @@ create_admin(Req) ->
case logic_admin:create_admin(Email, Password, Role) of
{ok, Admin} ->
% Аудит создания администратора
admin_utils:log_admin_action(SuperAdminId, <<"create_admin">>, <<"admin">>, Email, Req2),
admin_utils:log_admin_action(SuperAdminId, <<"create_admin">>, <<"admin">>, Email, <<>>,Req2),
handler_utils:send_json(Req2, 201, handler_utils:admin_to_json(Admin));
{error, email_exists} ->
handler_utils:send_error(Req2, 409, <<"Email already exists">>);
src/handlers/admin/admin_handler_admins_by_id.erl
+2 -2
View File
@@ -134,7 +134,7 @@ update_admin(Req) ->
Updates = convert_admin_fields(Updates0),
case logic_admin:update_admin(Id, Updates) of
{ok, Admin} ->
admin_utils:log_admin_action(SuperAdminId, <<"update_admin">>, <<"admin">>, Id, Req2),
admin_utils:log_admin_action(SuperAdminId, <<"update_admin">>, <<"admin">>, Id, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:admin_to_json(Admin));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Admin not found">>);
@@ -158,7 +158,7 @@ delete_admin(Req) ->
Id = cowboy_req:binding(id, Req1),
case logic_admin:delete_admin(Id) of
{ok, _} ->
admin_utils:log_admin_action(SuperAdminId, <<"delete_admin">>, <<"admin">>, Id, Req1),
admin_utils:log_admin_action(SuperAdminId, <<"delete_admin">>, <<"admin">>, Id, <<>>,Req1),
handler_utils:send_json(Req1, 200, #{status => <<"deleted">>});
{error, not_found} ->
handler_utils:send_error(Req1, 404, <<"Admin not found">>);
src/handlers/admin/admin_handler_banned_words.erl
+4 -4
View File
@@ -180,7 +180,7 @@ add_word(Req) ->
#{<<"word">> := Word} ->
case core_banned_words:add_banned_word(Word, AdminId) of
{ok, _} ->
admin_utils:log_admin_action(AdminId, <<"add">>, <<"banned_word">>, Word, Req2),
admin_utils:log_admin_action(AdminId, <<"add">>, <<"banned_word">>, Word, <<>>, Req2),
handler_utils:send_json(Req2, 201, #{status => <<"added">>});
{error, already_exists} ->
handler_utils:send_error(Req2, 409, <<"Word already exists">>);
@@ -205,7 +205,7 @@ update_word(Req) ->
case core_banned_words:update_banned_word(OldWord, NewWord) of
{ok, _Updated} ->
admin_utils:log_admin_action(AdminId, <<"update">>, <<"banned_word">>,
#{old_word => OldWord, new_word => NewWord}, Req2),
#{old_word => OldWord, new_word => NewWord}, <<>>, Req2),
handler_utils:send_json(Req2, 200, #{status => <<"updated">>});
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Word not found">>);
@@ -228,7 +228,7 @@ delete_word(Req) ->
Word = cowboy_req:binding(word, Req1),
case core_banned_words:remove_banned_word(Word) of
{ok, _} ->
admin_utils:log_admin_action(AdminId, <<"delete">>, <<"banned_word">>, Word, Req1),
admin_utils:log_admin_action(AdminId, <<"delete">>, <<"banned_word">>, Word, <<>>, Req1),
handler_utils:send_json(Req1, 200, #{status => <<"deleted">>});
{error, not_found} ->
handler_utils:send_error(Req1, 404, <<"Word not found">>);
@@ -260,7 +260,7 @@ batch_add_words(Req) ->
Added = length([R || R <- Results, maps:get(status, R) =:= <<"added">>]),
Skipped = length([R || R <- Results, maps:get(status, R) =:= <<"skipped">>]),
admin_utils:log_admin_action(AdminId, <<"batch_add">>, <<"banned_words">>,
#{count => length(Words), added => Added, skipped => Skipped}, Req2),
#{count => length(Words), added => Added, skipped => Skipped}, <<>>, Req2),
handler_utils:send_json(Req2, 200, #{
results => Results,
added => Added,
src/handlers/admin/admin_handler_calendar_by_id.erl
+2 -2
View File
@@ -114,7 +114,7 @@ update_calendar(Req) ->
Updates ->
case logic_calendar:admin_update(Id, Updates) of
{ok, Updated} ->
admin_utils:log_admin_action(AdminId, <<"update">>, <<"calendar">>, Id, Req2),
admin_utils:log_admin_action(AdminId, <<"update">>, <<"calendar">>, Id, <<>>, Req2),
Json = handler_utils:calendar_to_json(Updated),
handler_utils:send_json(Req2, 200, Json);
{error, not_found} ->
@@ -139,7 +139,7 @@ delete_calendar(Req) ->
Id = cowboy_req:binding(id, Req1),
case logic_calendar:admin_delete(Id) of
ok ->
admin_utils:log_admin_action(AdminId, <<"delete">>, <<"calendar">>, Id, Req1),
admin_utils:log_admin_action(AdminId, <<"delete">>, <<"calendar">>, Id, <<>>, Req1),
handler_utils:send_json(Req1, 200, #{});
{error, not_found} ->
handler_utils:send_error(Req1, 404, <<"Calendar not found">>)
src/handlers/admin/admin_handler_event_by_id.erl
+9 -6
View File
@@ -154,7 +154,7 @@ update_event(Req) ->
case logic_event:update_event_admin(EventId, UpdatesWithTypes) of
{ok, Event} ->
% Аудит обновления события
admin_utils:log_admin_action(AdminId, <<"update_event">>, <<"event">>, EventId, Req2),
admin_utils:log_admin_action(AdminId, <<"update_event">>, <<"event">>, EventId, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:event_to_json(Event));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Event not found">>);
@@ -177,7 +177,7 @@ delete_event(Req) ->
case logic_event:delete_event_admin(EventId) of
{ok, _} ->
% Аудит удаления события
admin_utils:log_admin_action(AdminId, <<"delete_event">>, <<"event">>, EventId, Req1),
admin_utils:log_admin_action(AdminId, <<"delete_event">>, <<"event">>, EventId, <<>>, Req1),
handler_utils:send_json(Req1, 200, #{status => <<"deleted">>});
{error, not_found} ->
handler_utils:send_error(Req1, 404, <<"Event not found">>);
@@ -218,9 +218,12 @@ convert_field({<<"tags">>, Val}) -> {tags, Val};
convert_field({<<"capacity">>, Val}) -> {capacity, Val};
convert_field({<<"online_link">>, Val}) -> {online_link, Val};
convert_field({<<"status">>, Val}) ->
try binary_to_existing_atom(Val, utf8) of
Atom -> {status, Atom}
catch
error:badarg -> {status, Val}
case Val of
<<"active">> -> {status, active};
<<"cancelled">> -> {status, cancelled};
<<"completed">> -> {status, completed};
<<"frozen">> -> {status, frozen};
<<"deleted">> -> {status, deleted};
_ -> {status, Val}
end;
convert_field(Other) -> Other.
src/handlers/admin/admin_handler_login.erl
+1 -1
View File
@@ -25,7 +25,7 @@ init(Req0, _State) ->
core_admin_session:create(UserId, RefreshToken),
core_admin:update_last_login(UserId),
% Аудит успешного входа
admin_utils:log_admin_action(UserId, <<"login">>, <<"admin">>, Email, Req1),
admin_utils:log_admin_action(UserId, <<"login">>, <<"admin">>, Email, <<>>, Req1),
Resp = #{
<<"token">> => Token,
<<"user">> => #{
src/handlers/admin/admin_handler_me.erl
+1 -1
View File
@@ -116,7 +116,7 @@ update_me(Req) ->
case logic_admin:update_admin(AdminId, Updates) of
{ok, Admin} ->
% Аудит обновления профиля
admin_utils:log_admin_action(AdminId, <<"update_me">>, <<"admin">>, AdminId, Req2),
admin_utils:log_admin_action(AdminId, <<"update_me">>, <<"admin">>, AdminId, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:admin_to_json(Admin));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Admin not found">>);
src/handlers/admin/admin_handler_moderation.erl
+8 -8
View File
@@ -101,7 +101,7 @@ apply_moderation(<<"user">>, Id, Action, Reason, Req, AdminId) ->
handle_calendar(Id, <<"freeze">>, Reason, Req, AdminId) ->
case core_calendar:freeze(Id, Reason) of
{ok, Calendar} ->
admin_utils:log_admin_action(AdminId, <<"freeze_calendar">>, <<"calendar">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"freeze_calendar">>, <<"calendar">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:calendar_to_json(Calendar));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"Calendar not found">>)
@@ -109,7 +109,7 @@ handle_calendar(Id, <<"freeze">>, Reason, Req, AdminId) ->
handle_calendar(Id, <<"unfreeze">>, Reason, Req, AdminId) ->
case core_calendar:unfreeze(Id, Reason) of
{ok, Calendar} ->
admin_utils:log_admin_action(AdminId, <<"unfreeze_calendar">>, <<"calendar">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"unfreeze_calendar">>, <<"calendar">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:calendar_to_json(Calendar));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"Calendar not found">>)
@@ -120,7 +120,7 @@ handle_calendar(_Id, _Action, _Reason, Req, _AdminId) ->
handle_event(Id, <<"freeze">>, Reason, Req, AdminId) ->
case core_event:freeze(Id, Reason) of
{ok, Event} ->
admin_utils:log_admin_action(AdminId, <<"freeze_event">>, <<"event">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"freeze_event">>, <<"event">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:event_to_json(Event));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"Event not found">>)
@@ -128,7 +128,7 @@ handle_event(Id, <<"freeze">>, Reason, Req, AdminId) ->
handle_event(Id, <<"unfreeze">>, Reason, Req, AdminId) ->
case core_event:unfreeze(Id, Reason) of
{ok, Event} ->
admin_utils:log_admin_action(AdminId, <<"unfreeze_event">>, <<"event">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"unfreeze_event">>, <<"event">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:event_to_json(Event));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"Event not found">>)
@@ -139,7 +139,7 @@ handle_event(_Id, _Action, _Reason, Req, _AdminId) ->
handle_review(Id, <<"hide">>, Reason, Req, AdminId) ->
case core_review:hide(Id, Reason) of
{ok, Review} ->
admin_utils:log_admin_action(AdminId, <<"hide_review">>, <<"review">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"hide_review">>, <<"review">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:review_to_json(Review));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"Review not found">>)
@@ -147,7 +147,7 @@ handle_review(Id, <<"hide">>, Reason, Req, AdminId) ->
handle_review(Id, <<"unhide">>, Reason, Req, AdminId) ->
case core_review:unhide(Id, Reason) of
{ok, Review} ->
admin_utils:log_admin_action(AdminId, <<"unhide_review">>, <<"review">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"unhide_review">>, <<"review">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:review_to_json(Review));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"Review not found">>)
@@ -158,7 +158,7 @@ handle_review(_Id, _Action, _Reason, Req, _AdminId) ->
handle_user(Id, <<"block">>, Reason, Req, AdminId) ->
case core_user:block(Id, Reason) of
{ok, User} ->
admin_utils:log_admin_action(AdminId, <<"block_user">>, <<"user">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"block_user">>, <<"user">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:user_to_json(User));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"User not found">>)
@@ -166,7 +166,7 @@ handle_user(Id, <<"block">>, Reason, Req, AdminId) ->
handle_user(Id, <<"unblock">>, Reason, Req, AdminId) ->
case core_user:unblock(Id, Reason) of
{ok, User} ->
admin_utils:log_admin_action(AdminId, <<"unblock_user">>, <<"user">>, #{id => Id, reason => Reason}, Req),
admin_utils:log_admin_action(AdminId, <<"unblock_user">>, <<"user">>, #{id => Id}, #{reason => Reason}, Req),
handler_utils:send_json(Req, 200, handler_utils:user_to_json(User));
{error, not_found} ->
handler_utils:send_error(Req, 404, <<"User not found">>)
src/handlers/admin/admin_handler_node_metrics.erl
+122 -109
View File
@@ -1,6 +1,21 @@
%%%-------------------------------------------------------------------
%%% @doc Административный обработчик исторических метрик узлов.
%%% GET /v1/admin/nodes/metrics – возвращает массив метрик за период.
%%%
%%% == Эндпоинт ==
%%% ```
%%% GET /v1/admin/nodes/metrics?from=ISO8601&to=ISO8601&node=NameOrAll
%%% ```
%%%
%%% == Параметры ==
%%% <ul>
%%% <li>`from` (обязательный) – начало временного диапазона в ISO8601.</li>
%%% <li>`to` (обязательный) – конец временного диапазона в ISO8601.</li>
%%% <li>`node` (опциональный) – имя узла Erlang или `<<"all">>` (по умолчанию).
%%% При `<<"all">>` возвращаются метрики всех узлов кластера.</li>
%%% </ul>
%%%
%%% == Ответ ==
%%% Массив объектов метрик. Каждый объект содержит поля, описанные в `metric_schema/0`.
%%% @end
%%%-------------------------------------------------------------------
-module(admin_handler_node_metrics).
@@ -10,115 +25,25 @@
-include("records.hrl").
%%% cowboy_handler callback
-spec init(cowboy_req:req(), any()) -> {ok, binary(), cowboy_req:req()}.
%%%-------------------------------------------------------------------
%%% @doc Инициализация Cowboy-обработчика.
%%% @end
%%%-------------------------------------------------------------------
-spec init(cowboy_req:req(), any()) -> {ok, cowboy_req:req(), any()}.
init(Req, _Opts) ->
case cowboy_req:method(Req) of
<<"GET">> -> get_metrics(Req);
_ -> handler_utils:send_error(Req, 405, <<"Method not allowed">>)
end.
%%% Swagger metadata
-spec trails() -> [map()].
trails() ->
[#{path => <<"/v1/admin/nodes/metrics">>,
method => <<"GET">>,
description => <<"Get historical node metrics">>,
tags => [<<"Node Metrics">>],
parameters => [
#{name => <<"node">>, in => <<"query">>, schema => #{type => string}, description => <<"Node name, default current">>},
#{name => <<"from">>, in => <<"query">>, required => true, schema => #{type => string, format => <<"date-time">>}},
#{name => <<"to">>, in => <<"query">>, required => true, schema => #{type => string, format => <<"date-time">>}}
],
responses => #{
200 => #{
description => <<"Array of node metrics">>,
content => #{<<"application/json">> => #{schema => #{
type => array,
items => metric_schema()
}}}
},
400 => #{description => <<"Missing required parameters">>},
403 => #{description => <<"Admin access required">>}
}}
].
metric_schema() ->
#{
type => object,
properties => #{
<<"timestamp">> => #{
type => string,
format => <<"date-time">>,
description => <<"Metric collection timestamp (ISO8601)">>
},
<<"node">> => #{
type => string,
description => <<"Erlang node name">>
},
<<"memory_total">> => #{
type => integer,
description => <<"Total memory allocated by the Erlang VM (bytes)">>
},
<<"memory_processes">> => #{
type => integer,
description => <<"Memory used by processes (bytes)">>
},
<<"memory_atom">> => #{
type => integer,
description => <<"Memory used by atoms (bytes)">>
},
<<"memory_binary">> => #{
type => integer,
description => <<"Memory used by binaries (bytes)">>
},
<<"memory_ets">> => #{
type => integer,
description => <<"Memory used by ETS tables (bytes)">>
},
<<"process_count">> => #{
type => integer,
description => <<"Number of live Erlang processes">>
},
<<"run_queue">> => #{
type => integer,
description => <<"Length of the scheduler run queue">>
},
<<"mnesia_commits">> => #{
type => integer,
description => <<"Number of successful Mnesia transactions since node start">>
},
<<"mnesia_failures">> => #{
type => integer,
description => <<"Number of failed Mnesia transactions since node start">>
},
<<"table_sizes">> => #{
type => object,
description => <<"Map of Mnesia table names to their sizes (number of records)">>,
additionalProperties => #{type => integer}
},
<<"active_sessions">> => #{
type => integer,
description => <<"Total number of active user and admin sessions">>
},
<<"ws_connections">> => #{
type => integer,
description => <<"Number of open WebSocket connections">>
},
<<"io_in">> => #{
type => integer,
description => <<"Bytes received on sockets since last collection (delta)">>
},
<<"io_out">> => #{
type => integer,
description => <<"Bytes sent over sockets since last collection (delta)">>
}
}
}.
%%% Internal functions
%% @doc Возвращает исторические метрики по заданному периоду и узлу.
%%%-------------------------------------------------------------------
%%% @doc Возвращает исторические метрики узлов.
%%%
%%% При `node=all` (по умолчанию) собирает метрики со всех узлов кластера
%%% через RPC-вызовы к удалённым узлам. Локальные метрики извлекаются
%%% напрямую.
%%% @end
%%%-------------------------------------------------------------------
-spec get_metrics(cowboy_req:req()) -> {ok, binary(), cowboy_req:req()}.
get_metrics(Req) ->
case handler_utils:auth_admin(Req) of
@@ -126,20 +51,49 @@ get_metrics(Req) ->
Qs = cowboy_req:parse_qs(Req1),
From = handler_utils:parse_datetime_qs(proplists:get_value(<<"from">>, Qs)),
To = handler_utils:parse_datetime_qs(proplists:get_value(<<"to">>, Qs)),
Node = proplists:get_value(<<"node">>, Qs, node()),
NodeParam = proplists:get_value(<<"node">>, Qs, <<"all">>),
case {From, To} of
{undefined, _} -> handler_utils:send_error(Req1, 400, <<"Missing from">>);
{_, undefined} -> handler_utils:send_error(Req1, 400, <<"Missing to">>);
{F, T} ->
Metrics = core_node_metric:get_by_timerange(F, T, Node),
Json = [metric_to_map(M) || M <- Metrics],
Metrics = case NodeParam of
<<"all">> ->
% Локальные метрики текущего узла
Local = core_node_metric:get_by_timerange(F, T, node()),
% Остальные узлы кластера
OtherNodes = mnesia:system_info(db_nodes) -- [node()],
io:format("[NODE_METRICS] Aggregating metrics from ~p (local) and remote nodes: ~p~n", [node(), OtherNodes]),
Remote = lists:flatmap(fun(N) ->
io:format("[NODE_METRICS] Requesting metrics from node ~p...~n", [N]),
try rpc:call(N, core_node_metric, get_by_timerange, [F, T, N], 5000) of
List when is_list(List) ->
io:format("[NODE_METRICS] Received ~p metrics from node ~p~n", [length(List), N]),
List;
Other ->
io:format("[NODE_METRICS] Unexpected response from node ~p: ~p~n", [N, Other]),
[]
catch
Class:Reason:Stacktrace ->
io:format("[NODE_METRICS] RPC error for node ~p: ~p:~p~nStacktrace: ~p~n", [N, Class, Reason, Stacktrace]),
[]
end
end, OtherNodes),
Local ++ Remote;
SpecificNode ->
core_node_metric:get_by_timerange(F, T, SpecificNode)
end,
Sorted = lists:sort(fun(A, B) -> A#node_metric.timestamp =< B#node_metric.timestamp end, Metrics),
Json = [metric_to_map(M) || M <- Sorted],
handler_utils:send_json(Req1, 200, Json)
end;
{error, Code, Msg, Req1} ->
handler_utils:send_error(Req1, Code, Msg)
end.
%% @private Преобразует запись метрики в JSON-совместимую карту.
%%%-------------------------------------------------------------------
%%% @doc Преобразует запись `#node_metric{}` в JSON-совместимый map.
%%% @end
%%%-------------------------------------------------------------------
-spec metric_to_map(#node_metric{}) -> map().
metric_to_map(M) ->
#{
@@ -150,6 +104,7 @@ metric_to_map(M) ->
<<"memory_atom">> => M#node_metric.memory_atom,
<<"memory_binary">> => M#node_metric.memory_binary,
<<"memory_ets">> => M#node_metric.memory_ets,
<<"memory_available">> => M#node_metric.memory_available,
<<"process_count">> => M#node_metric.process_count,
<<"run_queue">> => M#node_metric.run_queue,
<<"mnesia_commits">> => M#node_metric.mnesia_commits,
@@ -158,5 +113,63 @@ metric_to_map(M) ->
<<"active_sessions">> => M#node_metric.active_sessions,
<<"ws_connections">> => M#node_metric.ws_connections,
<<"io_in">> => M#node_metric.io_in,
<<"io_out">> => M#node_metric.io_out
<<"io_out">> => M#node_metric.io_out,
<<"cpu_utilization">> => M#node_metric.cpu_utilization
}.
%%%-------------------------------------------------------------------
%%% @doc Возвращает Swagger-трассы для модуля.
%%% @end
%%%-------------------------------------------------------------------
-spec trails() -> [map()].
trails() ->
[#{path => <<"/v1/admin/nodes/metrics">>,
method => <<"GET">>,
description => <<"Get historical node metrics (aggregated from all nodes by default)">>,
tags => [<<"Node Metrics">>],
parameters => [
#{name => <<"node">>, in => <<"query">>, schema => #{type => string},
description => <<"Node name or 'all' (default). When 'all', returns metrics from every cluster node.">>},
#{name => <<"from">>, in => <<"query">>, required => true,
schema => #{type => string, format => <<"date-time">>},
description => <<"Start timestamp in ISO8601 format (milliseconds are supported)">>},
#{name => <<"to">>, in => <<"query">>, required => true,
schema => #{type => string, format => <<"date-time">>},
description => <<"End timestamp in ISO8601 format (milliseconds are supported)">>}
],
responses => #{200 => #{description => <<"Array of node metrics">>,
content => #{<<"application/json">> => #{schema => #{
type => array,
items => metric_schema()
}}}}}}
].
%%%-------------------------------------------------------------------
%%% @doc Swagger-схема для одного объекта метрики.
%%% @end
%%%-------------------------------------------------------------------
-spec metric_schema() -> map().
metric_schema() ->
#{
type => object,
properties => #{
<<"timestamp">> => #{type => string, format => <<"date-time">>, description => <<"Timestamp of the metric sample (ISO8601)">>},
<<"node">> => #{type => string, description => <<"Erlang node name">>},
<<"memory_total">> => #{type => integer, description => <<"Total memory used by Erlang VM (bytes)">>},
<<"memory_processes">> => #{type => integer, description => <<"Memory used by processes (bytes)">>},
<<"memory_atom">> => #{type => integer, description => <<"Memory used for atoms (bytes)">>},
<<"memory_binary">> => #{type => integer, description => <<"Memory used for binary data (bytes)">>},
<<"memory_ets">> => #{type => integer, description => <<"Memory used by ETS tables (bytes)">>},
<<"memory_available">> => #{type => integer, description => <<"Available system memory (bytes), 0 if unavailable">>},
<<"process_count">> => #{type => integer, description => <<"Number of Erlang processes">>},
<<"run_queue">> => #{type => integer, description => <<"Run queue length (0 means idle)">>},
<<"mnesia_commits">> => #{type => integer, description => <<"Mnesia transaction commits since boot">>},
<<"mnesia_failures">> => #{type => integer, description => <<"Mnesia transaction failures since boot">>},
<<"table_sizes">> => #{type => object, description => <<"Map of Mnesia table names to record counts">>, additionalProperties => #{type => integer}},
<<"active_sessions">> => #{type => integer, description => <<"Number of active user/admin sessions">>},
<<"ws_connections">> => #{type => integer, description => <<"Number of active WebSocket connections">>},
<<"io_in">> => #{type => integer, description => <<"Delta of bytes received since last sample">>},
<<"io_out">> => #{type => integer, description => <<"Delta of bytes sent since last sample">>},
<<"cpu_utilization">> => #{type => number, format => float, description => <<"CPU utilization (0.0-100.0 percent)">>}
}
}.
src/handlers/admin/admin_handler_report_by_id.erl
+1 -1
View File
@@ -109,7 +109,7 @@ update_report(Req) ->
case logic_report:update_report_status(AdminId, ReportId, StatusAtom) of
{ok, Report} ->
% Аудит изменения статуса жалобы
admin_utils:log_admin_action(AdminId, <<"update_report_status">>, <<"report">>, ReportId, Req2),
admin_utils:log_admin_action(AdminId, <<"update_report_status">>, <<"report">>, ReportId, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:report_to_json(Report));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Report not found">>);
src/handlers/admin/admin_handler_reviews.erl
+1 -1
View File
@@ -132,7 +132,7 @@ bulk_update_reviews(Req) ->
true = is_list(Operations),
case logic_review:bulk_update_status(Operations, Reason) of
{ok, UpdatedCount} ->
admin_utils:log_admin_action(AdminId, <<"bulk_update_reviews">>, <<"review">>, UpdatedCount, Req2),
admin_utils:log_admin_action(AdminId, <<"bulk_update_reviews">>, <<"review">>, UpdatedCount, Reason, Req2),
handler_utils:send_json(Req2, 200, #{updated_count => UpdatedCount});
{error, ReasonErr} ->
handler_utils:send_error(Req2, 400, ReasonErr)
src/handlers/admin/admin_handler_reviews_by_id.erl
+1 -1
View File
@@ -113,7 +113,7 @@ update_review(Req) ->
case logic_review:update_review_admin(ReviewId, Converted) of
{ok, Review} ->
% Аудит изменения отзыва
admin_utils:log_admin_action(AdminId, <<"update_review">>, <<"review">>, ReviewId, Req2),
admin_utils:log_admin_action(AdminId, <<"update_review">>, <<"review">>, ReviewId, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:review_to_json(Review));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Review not found">>);
src/handlers/admin/admin_handler_subscriptions_by_id.erl
+2 -2
View File
@@ -120,7 +120,7 @@ update_subscription(Req) ->
Data when is_map(Data) ->
case core_subscription:update_subscription(Id, Data) of
{ok, Updated} ->
admin_utils:log_admin_action(AdminId, <<"update_subscription">>, <<"subscription">>, Id, Req2),
admin_utils:log_admin_action(AdminId, <<"update_subscription">>, <<"subscription">>, Id, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:subscription_to_json(Updated));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Subscription not found">>);
@@ -142,7 +142,7 @@ delete_subscription(Req) ->
Id = cowboy_req:binding(id, Req1),
case core_subscription:delete_subscription(Id) of
{ok, _} ->
admin_utils:log_admin_action(AdminId, <<"delete_subscription">>, <<"subscription">>, Id, Req1),
admin_utils:log_admin_action(AdminId, <<"delete_subscription">>, <<"subscription">>, Id, <<>>, Req1),
handler_utils:send_json(Req1, 200, #{status => <<"deleted">>});
{error, not_found} ->
handler_utils:send_error(Req1, 404, <<"Subscription not found">>);
src/handlers/admin/admin_handler_ticket_by_id.erl
+2 -2
View File
@@ -131,7 +131,7 @@ update_ticket(Req) ->
Result = apply_ticket_changes(AdminId, TicketId, Data),
case Result of
{ok, Ticket} ->
admin_utils:log_admin_action(AdminId, <<"update_ticket">>, <<"ticket">>, TicketId, Req2),
admin_utils:log_admin_action(AdminId, <<"update_ticket">>, <<"ticket">>, TicketId, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:ticket_to_json(Ticket));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Ticket not found">>);
@@ -157,7 +157,7 @@ delete_ticket(Req) ->
TicketId = cowboy_req:binding(id, Req1),
case logic_ticket:delete_ticket(AdminId, TicketId) of
{ok, _} ->
admin_utils:log_admin_action(AdminId, <<"delete_ticket">>, <<"ticket">>, TicketId, Req1),
admin_utils:log_admin_action(AdminId, <<"delete_ticket">>, <<"ticket">>, TicketId, <<>>, Req1),
handler_utils:send_json(Req1, 200, #{status => <<"deleted">>});
{error, not_found} ->
handler_utils:send_error(Req1, 404, <<"Ticket not found">>);
src/handlers/admin/admin_handler_tickets.erl
+1 -1
View File
@@ -122,7 +122,7 @@ update_ticket(Req) ->
Result = apply_ticket_changes(AdminId, TicketId, Data),
case Result of
{ok, Ticket} ->
admin_utils:log_admin_action(AdminId, <<"update_ticket">>, <<"ticket">>, TicketId, Req2),
admin_utils:log_admin_action(AdminId, <<"update_ticket">>, <<"ticket">>, TicketId, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:ticket_to_json(Ticket));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"Ticket not found">>);
src/handlers/admin/admin_handler_user_by_id.erl
+2 -2
View File
@@ -129,7 +129,7 @@ update_user(Req) ->
Converted = convert_user_fields(Updates),
case logic_user:update_user_admin(UserId, Converted) of
{ok, User} ->
admin_utils:log_admin_action(AdminId, <<"update_user">>, <<"user">>, UserId, Req2),
admin_utils:log_admin_action(AdminId, <<"update_user">>, <<"user">>, UserId, <<>>, Req2),
handler_utils:send_json(Req2, 200, handler_utils:user_to_json(User));
{error, not_found} ->
handler_utils:send_error(Req2, 404, <<"User not found">>);
@@ -151,7 +151,7 @@ delete_user(Req) ->
UserId = cowboy_req:binding(id, Req1),
case logic_user:delete_user_admin(UserId) of
{ok, _} ->
admin_utils:log_admin_action(AdminId, <<"delete_user">>, <<"user">>, UserId, Req1),
admin_utils:log_admin_action(AdminId, <<"delete_user">>, <<"user">>, UserId, <<>>, Req1),
handler_utils:send_json(Req1, 200, #{status => <<"deleted">>});
{error, not_found} ->
handler_utils:send_error(Req1, 404, <<"User not found">>);
src/handlers/admin/admin_ws_handler.erl
+5 -2
View File
@@ -63,7 +63,8 @@ websocket_init(State) ->
io:format("[ADMIN_WS] WebSocket initialized for admin ~s~n", [State#state.admin_id]),
pg:join(eventhub_admin_ws, self()),
% Подписываемся на метрики узлов
pg:join(node_metrics, self()),
%% pg:join(node_metrics, self()),
pg:join(global, node_metrics, self()),
{ok, State}.
%% @doc Обрабатывает входящие текстовые сообщения (subscribe/unsubscribe/ping).
@@ -143,5 +144,7 @@ metric_to_map(M) ->
<<"active_sessions">> => M#node_metric.active_sessions,
<<"ws_connections">> => M#node_metric.ws_connections,
<<"io_in">> => M#node_metric.io_in,
<<"io_out">> => M#node_metric.io_out
<<"io_out">> => M#node_metric.io_out,
<<"memory_available">> => M#node_metric.memory_available,
<<"cpu_utilization">> => M#node_metric.cpu_utilization
}.
src/infra/admin_utils.erl
+24 -4
View File
@@ -2,7 +2,7 @@
-include("records.hrl").
-export([is_admin/1, check_role/2, get_permissions/1, is_superadmin/1]).
-export([client_ip/1]). % ← IP
-export([log_admin_action/5]). % ← аудит действий администратора
-export([log_admin_action/6]). % ← аудит действий администратора
-export([ip_to_binary/1]). % ← преобразование IP
%% -- существующие функции -------------------------------------------------
@@ -71,14 +71,34 @@ ip_to_binary(_) ->
%% EntityType – тип сущности (напр. <<"admin">>, <<"banned_word">>)
%% EntityId – идентификатор сущности
%% Req – cowboy_req для извлечения IP
-spec log_admin_action(binary(), binary(), binary(), term(), cowboy_req:req()) -> ok.
log_admin_action(AdminId, Action, EntityType, EntityId, Req) ->
%%-spec log_admin_action(binary(), binary(), binary(), term(), cowboy_req:req()) -> ok.
%%log_admin_action(AdminId, Action, EntityType, EntityId, Req) ->
%% case core_admin:get_by_id(AdminId) of
%% {ok, Admin} ->
%% Email = Admin#admin.email,
%% Role = atom_to_binary(Admin#admin.role, utf8),
%% Ip = ip_to_binary(cowboy_req:peer(Req)),
%% core_admin_audit:log(AdminId, Email, Role, Action, EntityType, EntityId, Ip),
%% ok;
%% _ -> ok
%% end.
%% @doc Логирует действие администратора.
%% Параметры:
%% AdminId – ID администратора, выполнившего действие
%% Action – бинарная строка действия (напр. <<"create">>)
%% EntityType – тип сущности (напр. <<"admin">>, <<"banned_word">>)
%% EntityId – идентификатор сущности
%% Reason - причина
%% Req – cowboy_req для извлечения IP
-spec log_admin_action(binary(), binary(), binary(), term(), binary(), cowboy_req:req()) -> ok.
log_admin_action(AdminId, Action, EntityType, EntityId, Reason, Req) ->
case core_admin:get_by_id(AdminId) of
{ok, Admin} ->
Email = Admin#admin.email,
Role = atom_to_binary(Admin#admin.role, utf8),
Ip = ip_to_binary(cowboy_req:peer(Req)),
core_admin_audit:log(AdminId, Email, Role, Action, EntityType, EntityId, Ip),
core_admin_audit:log(AdminId, Email, Role, Action, EntityType, EntityId, Ip, Reason),
ok;
_ -> ok
end.
src/infra/infra_mnesia.erl
+17 -1
View File
@@ -62,6 +62,22 @@ handle_call(init_tables, _From, State) ->
ok = join_cluster(ExtraNodes)
end,
lists:foreach(fun create_table/1, ?TABLES),
% Принудительное создание node_metric на каждом узле
case mnesia:create_table(node_metric, [
{disc_copies, [node()]}, % хранить на диске
{local_content, true}, % не реплицировать
{attributes, record_info(fields, node_metric)}
]) of
{atomic, ok} -> ok;
{aborted, {already_exists, node_metric}} -> ok;
_ -> ok
end,
% ГАРАНТИРУЕМ, что узел имеет локальную копию node_metric (критично для присоединяющихся узлов)
case lists:member(node(), mnesia:table_info(node_metric, disc_copies) ++
mnesia:table_info(node_metric, ram_copies)) of
false -> mnesia:add_table_copy(node_metric, node(), disc_copies);
true -> ok
end,
ok = create_indices(),
ok = stats_collector:subscribe(),
ok = start_cleanup_timer(),
@@ -223,7 +239,7 @@ table_opts(schema_migration) -> [{disc_copies, [node()]}, {attributes, record_in
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(node_metric) -> [{ram_copies, [node()]}, {local_content, true}, {attributes, record_info(fields, node_metric)}].
table_opts(node_metric) -> [{disc_copies, [node()]}, {local_content, true}, {attributes, record_info(fields, node_metric)}].
%% ===================================================================
%% Индексы
src/infra/infra_sup.erl
+4
View File
@@ -25,6 +25,10 @@ init([]) ->
start => {pg, start_link, []},
restart => permanent,
type => worker},
#{id => pg_global,
start => {pg, start_link, [global]},
restart => permanent,
type => worker},
#{id => stats_collector,
start => {stats_collector, start_link, []},
restart => permanent,
src/infra/node_monitor.erl
+74 -3
View File
@@ -38,6 +38,7 @@ start_link() ->
-spec init(term()) -> {ok, map()}.
init([]) ->
mnesia:wait_for_tables([node_metric], 5000),
% Создаём ETS-таблицу для кеширования предыдущих значений io
ets:new(?IO_CACHE_TAB, [named_table, public, set, {keypos, 1}]),
% Запускаем периодический сбор метрик и очистку
@@ -58,7 +59,9 @@ handle_info(collect, State) ->
Metric = collect_metrics(),
core_node_metric:save(Metric),
% Рассылаем метрику всем подписчикам группы node_metrics
Members = pg:get_local_members(?METRICS_GROUP),
%% Members = pg:get_local_members(?METRICS_GROUP),
% Рассылаем метрику всем подписчикам глобальной группы node_metrics
Members = pg:get_members(global, node_metrics),
[Pid ! {node_metric, Metric} || Pid <- Members],
schedule_collect(),
{noreply, State};
@@ -88,6 +91,8 @@ collect_metrics() ->
Now = calendar:universal_time(),
Mem = erlang:memory(),
{IoIn, IoOut} = io_stats(),
Cpu = cpu_utilization(),
FreeMem = free_memory(),
#node_metric{
timestamp = Now,
node = node(),
@@ -104,7 +109,9 @@ collect_metrics() ->
active_sessions = session_count(),
ws_connections = ws_count(),
io_in = IoIn,
io_out = IoOut
io_out = IoOut,
memory_available = FreeMem,
cpu_utilization = Cpu
}.
%%%-------------------------------------------------------------------
@@ -165,7 +172,71 @@ io_stats() ->
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}.
DeltaIn = max(0, TotalIn - PrevIn),
DeltaOut = max(0, TotalOut - PrevOut),
{DeltaIn, DeltaOut}.
%%%-------------------------------------------------------------------
%%% @doc Возвращает объём свободной системной памяти (в байтах).
%%% Безопасно при отсутствии поддержки в ОС.
%%% @end
%%%-------------------------------------------------------------------
-spec free_memory() -> non_neg_integer().
free_memory() ->
try erlang:system_info(os_free_memory) of
Free when Free > 0 -> Free;
_ -> read_meminfo()
catch
_:_ -> read_meminfo()
end.
read_meminfo() ->
try
{ok, Binary} = file:read_file("/proc/meminfo"),
Content = binary_to_list(Binary),
case re:run(Content, "MemAvailable:\\s+(\\d+) kB", [{capture, [1], list}]) of
{match, [MemKbStr]} -> list_to_integer(MemKbStr) * 1024; % кБ -> байты
_ -> 0
end
catch _:_ -> 0
end.
%%%-------------------------------------------------------------------
%%% @doc Вычисляет утилизацию CPU (в процентах) за интервал сбора.
%%% Использует статистику /proc/stat.
%%% @end
%%%-------------------------------------------------------------------
-spec cpu_utilization() -> float().
cpu_utilization() ->
try
{ok, Binary} = file:read_file("/proc/stat"),
Content = binary_to_list(Binary),
{match, [TotalStr]} = re:run(Content, "^cpu\\s+(.*)$", [{capture, [1], list}, multiline]),
Tokens = string:tokens(TotalStr, " "),
[User, Nice, System, Idle, Iowait, Irq, Softirq, Steal | _] = lists:map(fun list_to_integer/1, Tokens),
Total = User + Nice + System + Idle + Iowait + Irq + Softirq + Steal,
IdleTotal = Idle + Iowait,
% Получаем предыдущие значения безопасным способом
{PrevIdle, PrevTotal} = case ets:lookup(?IO_CACHE_TAB, cpu_stats) of
[{cpu_stats, PI, PT}] -> {PI, PT};
[] -> {0, 0}
end,
ets:insert(?IO_CACHE_TAB, [{cpu_stats, IdleTotal, Total}]),
case PrevTotal of
0 -> 0.0; % первый замер, нет данных для расчёта
_ ->
DeltaTotal = Total - PrevTotal,
DeltaIdle = IdleTotal - PrevIdle,
if DeltaTotal > 0 ->
(1.0 - DeltaIdle / DeltaTotal) * 100.0;
true -> 0.0
end
end
catch
_:_ -> 0.0
end.
-spec get_prev_io() -> {non_neg_integer(), non_neg_integer()}.
get_prev_io() ->
src/logic/logic_search.erl
+4 -3
View File
@@ -105,9 +105,10 @@ filter_accessible_events(Events, UserId) ->
CanAccess = logic_calendar:can_access(UserId, Calendar),
case CanAccess of
false ->
io:format("Access denied for user ~p to calendar ~p (type: ~p, owner: ~p, status: ~p)~n",
[UserId, Calendar#calendar.id, Calendar#calendar.type,
Calendar#calendar.owner_id, Calendar#calendar.status]);
%% io:format("Access denied for user ~p to calendar ~p (type: ~p, owner: ~p, status: ~p)~n",
%% [UserId, Calendar#calendar.id, Calendar#calendar.type,
%% Calendar#calendar.owner_id, Calendar#calendar.status]);
false;
true -> ok
end,
CanAccess;