%% @doc Log handler and formatter for the logger app %% %% It is optimized to use a fixed size ring buffer and %% return chucks of older logs first while storing new ones. %% It can be synched to discard old logs if they are not needed anymore. %% If the buffer is filled, oldest logs are dropped %% and a fake logger event is inserted to inform the user. %% @end -module(grisp_connect_logger_bin). -include_lib("kernel/include/logger.hrl"). % API -export([sync/2]). -export([chunk/2]). % Behaviour logger_handler callback functions -export([adding_handler/1]). -export([changing_config/3]). -export([removing_handler/1]). -export([log/2]). -export([filter_config/1]). % Behaviour logger_formater callback functions -export([format/2]). -export([check_config/4]). % logger_h_common Callbacks -export([init/2]). -export([config_changed/3]). -export([reset_state/2]). -export([filesync/3]). -export([write/4]). -export([handle_info/3]). -export([terminate/3]). % Internal Callbacks -export([queue_ctrl_init/1]). -export([queue_ctrl_loop/1]). %--- Macros -------------------------------------------------------------------- -define(CNT_KEY, {?MODULE, counters}). -define(CNT_N, 1). -define(CNT_SEQ, 1). -define(DEFAULT_CONFIG, #{ count_max => 10_000, % 10 MiB: bytes_max => 10 * 1_024 * 1_024 }). -define(DEFAULT_CALL_TIMEOUT, 5000). % FixMe: % Sending over ~30_000 bytes over WS breaks rtems I/O driver. % We want avoid to return chunks that are bigger then that. -define(MAX_LOG_BYTES, 30_000). %--- API ----------------------------------------------------------------------- sync(Seq, DroppedConfirmed) when is_integer(Seq), is_integer(DroppedConfirmed) -> case whereis(?MODULE) of undefined -> ?LOG_WARNING("grisp_connect log handler not setup properly"), ok; Pid -> call(Pid, {sync, Seq, DroppedConfirmed}) end. %% @doc %% The chunk will contain Count logs at maximum. %% while ensuring that the whole chunk is less then MaxBytes. chunk(MaxCount, MaxBytes) -> case whereis(?MODULE) of undefined -> ?LOG_WARNING("grisp_connect log handler not setup properly"), {[], 0}; Pid -> insert_drop_event(call(Pid, {chunk, MaxCount, MaxBytes})) end. %--- Behaviour logger_handler Callback Callbacks ------------------------------- adding_handler(Config) -> logger_h_common:adding_handler(Config). changing_config(SetOrUpdate, OldConfig, NewConfig) -> logger_h_common:changing_config(SetOrUpdate, OldConfig, NewConfig). removing_handler(Config) -> logger_h_common:removing_handler(Config). log(LogEvent, Config) -> logger_h_common:log(LogEvent, Config). filter_config(Config) -> logger_h_common:filter_config(Config). %--- Behaviour logger_formater Callback Functions ------------------------------ format(Event, Config) -> case Config of #{stdout := {Formatter, FormatterConfig}} -> catch io:format(Formatter:format(Event, FormatterConfig)); _Else -> ok end, Encoded = base64:encode(term_to_binary(Event)), EncodedSize = byte_size(Encoded), case EncodedSize > ?MAX_LOG_BYTES of true -> NewLog = log_discard(EncodedSize), base64:encode(term_to_binary(NewLog)); false -> Encoded end. check_config(_Name, _SetOrUpdate, _OldHConfig, NewHConfig0) -> {ok, NewHConfig0}. %--- logger_h_common Callbacks ------------------------------------------------- init(_Name, UserConfig) -> CleanConfig = maps:with(maps:keys(?DEFAULT_CONFIG), UserConfig), Counters = case persistent_term:get(?CNT_KEY, undefined) of undefined -> C = atomics:new(?CNT_N, [{signed, false}]), persistent_term:put(?CNT_KEY, C), C; C -> C end, Config = maps:merge(?DEFAULT_CONFIG, CleanConfig), {ok, #{pid => queue_ctrl_start(Config#{counters => Counters})}}. config_changed(_Name, NewHConfig, #{pid := Pid} = State) -> call(Pid, {config_changed, NewHConfig}), State. reset_state(_Name, State) -> State. filesync(_Name, _SyncAsync, State) -> {ok, State}. write(_Name, async, Bin, #{pid := Pid} = State) -> Pid ! {log, Bin}, {ok, State}; write(_Name, sync, Bin, #{pid := Pid} = State) -> Result = call(Pid, {log, Bin}), {Result, State}. handle_info(_, _, State) -> State. terminate(_Name, _Reason, _State) -> ok. %--- Internal ------------------------------------------------------------------ call(Pid, Msg) -> MRef = monitor(process, Pid), Pid ! {Msg, {self(), MRef}}, receive {MRef, Result} -> demonitor(MRef, [flush]), Result; {'DOWN', MRef, _Type, _Object, Reason} -> {error, Reason} after ?DEFAULT_CALL_TIMEOUT -> %% If this timeout triggers we will get a stray %% reply message in our mailbox eventually. %% That does not really matter though as it will %% end up in this module's handle_info and be ignored demonitor(MRef, [flush]), {error, {no_response, Pid}} end. queue_ctrl_start(Config) -> spawn_link(fun() -> ?MODULE:queue_ctrl_init(Config) end). queue_ctrl_init(#{counters := Counters} = Config) -> register(?MODULE, self()), ?MODULE:queue_ctrl_loop(#{ counters => Counters, binlog => grisp_connect_binlog:new(Config), dropped => 0 }). queue_ctrl_loop(S0) -> S2 = receive {log, Bin} -> insert(Bin, S0); {{log, Bin}, From} -> S1 = insert(Bin, S0), reply(From, ok), S1; {config_changed, _Config} -> S0; {{sync, Seq, DroppedConfirmed}, From} -> S1 = sync(Seq, DroppedConfirmed, S0), reply(From, ok), S1; {{chunk, Count, Bytes}, From} -> reply(From, peek(Count, Bytes, S0)), S0 end, ?MODULE:queue_ctrl_loop(S2). reply({From, MRef}, Msg) -> From ! {MRef, Msg}. insert(Event, #{counters := Counters, binlog := B0, dropped := D0} = State) -> Seq = atomics:add_get(Counters, ?CNT_SEQ, 1), {B1, D1} = grisp_connect_binlog:insert({Seq, Event}, B0), State#{binlog := B1, dropped := D0 + D1}. sync(Seq, DroppedConfirmed, #{binlog := B0, dropped := Dropped} = State) -> B1 = grisp_connect_binlog:truncate(Seq, B0), State#{binlog := B1, dropped := Dropped - DroppedConfirmed}. peek(Count, MaxBytes, #{binlog := B0, dropped := Dropped}) -> Events = grisp_connect_binlog:peek(Count, B0), {trim_to_size(Events, MaxBytes), Dropped}. trim_to_size(Events, MaxBytes) -> rec_trim_to_size(Events, MaxBytes, 0, []). rec_trim_to_size([], _, _, Acc) -> lists:reverse(Acc); rec_trim_to_size([{_Seq, Bin} = E | TL], MaxBytes, AccSize, Acc) -> NewSize = byte_size(Bin) + AccSize, case NewSize > MaxBytes of true -> lists:reverse(Acc); false -> rec_trim_to_size(TL, MaxBytes, NewSize, [E | Acc]) end. insert_drop_event({Events, 0}) -> {Events, 0}; insert_drop_event({Events, Dropped}) -> {[drop_event(Dropped) | Events], Dropped}. log_discard(Size) -> Meta = simulate_log_metadata(), Report = {report, #{ event => log_discarded, reason => too_long, size => Size} }, #{level => warning, meta => Meta, msg => Report}. drop_event(Count) -> % Create a fake logger event mimicking a real log event as much as possible Meta = simulate_log_metadata(), Report = {report, #{event => log_drop, count => Count}}, {null, format(#{level => warning, meta => Meta, msg => Report}, #{})}. simulate_log_metadata() -> % See internal function logger:add_default_metadata/2 DefaultMetadata = #{ pid => self(), gl => group_leader(), time => logger:timestamp() }, PrimaryConfig = logger:get_primary_config(), PrimaryMetadata = maps:get(logger_metadata, PrimaryConfig, #{}), % See internal function logger:log_allowed/4 ProcessMetadata = case logger:get_process_metadata() of ProcMeta when is_map(ProcMeta) -> ProcMeta; _ -> #{} end, lists:foldl( fun(M, MAcc) -> maps:merge(MAcc, M) end, DefaultMetadata, [?LOCATION, PrimaryMetadata, ProcessMetadata] ).