diff options
| -rw-r--r-- | lib/kernel/src/Makefile | 3 | ||||
| -rw-r--r-- | lib/kernel/src/kernel.app.src | 2 | ||||
| -rw-r--r-- | lib/kernel/src/logger.erl | 51 | ||||
| -rw-r--r-- | lib/kernel/src/logger_h_common.erl | 19 | ||||
| -rw-r--r-- | lib/kernel/src/logger_h_common.hrl | 29 | ||||
| -rw-r--r-- | lib/kernel/src/logger_olp.erl | 213 | ||||
| -rw-r--r-- | lib/kernel/src/logger_proxy.erl | 152 | ||||
| -rw-r--r-- | lib/kernel/src/logger_server.erl | 20 | ||||
| -rw-r--r-- | lib/kernel/src/logger_sup.erl | 4 | ||||
| -rw-r--r-- | lib/kernel/test/logger.cover | 1 |
10 files changed, 392 insertions, 102 deletions
diff --git a/lib/kernel/src/Makefile b/lib/kernel/src/Makefile index c076726bf4..c3fee02334 100644 --- a/lib/kernel/src/Makefile +++ b/lib/kernel/src/Makefile @@ -119,6 +119,7 @@ MODULES = \ logger_filters \ logger_formatter \ logger_olp \ + logger_proxy \ logger_server \ logger_simple_h \ logger_sup \ @@ -280,6 +281,8 @@ $(EBIN)/logger_config.beam: logger_internal.hrl ../include/logger.hrl $(EBIN)/logger_disk_log_h.beam: logger_h_common.hrl logger_internal.hrl ../include/logger.hrl ../include/file.hrl $(EBIN)/logger_filters.beam: logger_internal.hrl ../include/logger.hrl $(EBIN)/logger_formatter.beam: logger_internal.hrl ../include/logger.hrl +$(EBIN)/logger_olp.beam: logger_h_common.hrl logger_internal.hrl +$(EBIN)/logger_proxy.beam: logger_internal.hrl $(EBIN)/logger_server.beam: logger_internal.hrl ../include/logger.hrl $(EBIN)/logger_simple_h.beam: logger_internal.hrl ../include/logger.hrl $(EBIN)/logger_std_h.beam: logger_h_common.hrl logger_internal.hrl ../include/logger.hrl ../include/file.hrl diff --git a/lib/kernel/src/kernel.app.src b/lib/kernel/src/kernel.app.src index fe073621c8..a1d9e8e215 100644 --- a/lib/kernel/src/kernel.app.src +++ b/lib/kernel/src/kernel.app.src @@ -68,6 +68,8 @@ logger_formatter, logger_h_common, logger_handler_watcher, + logger_olp, + logger_proxy, logger_server, logger_simple_h, logger_std_h, diff --git a/lib/kernel/src/logger.erl b/lib/kernel/src/logger.erl index 6762998d4f..1611d489e6 100644 --- a/lib/kernel/src/logger.erl +++ b/lib/kernel/src/logger.erl @@ -672,6 +672,17 @@ init_kernel_handlers(Env) -> %% This function is responsible for resolving the handler config %% and then starting the correct handlers. This is done after the %% kernel supervisor tree has been started as it needs the logger_sup. +add_handlers(kernel) -> + Env = get_logger_env(kernel), + case get_proxy_opts(Env) of + undefined -> + add_handlers(kernel,Env); + Opts -> + case logger_olp:set_opts(logger_proxy,Opts) of + ok -> add_handlers(kernel,Env); + {error, Reason} -> {error,{bad_proxy_config,Reason}} + end + end; add_handlers(App) when is_atom(App) -> add_handlers(App,get_logger_env(App)); add_handlers(HandlerConfig) -> @@ -729,6 +740,8 @@ check_logger_config(kernel,[{filters,_,_}|Env]) -> check_logger_config(kernel,Env); check_logger_config(kernel,[{module_level,_,_}|Env]) -> check_logger_config(kernel,Env); +check_logger_config(kernel,[{proxy,_}|Env]) -> + check_logger_config(kernel,Env); check_logger_config(_,Bad) -> throw(Bad). @@ -784,6 +797,13 @@ get_primary_filters(Env) -> _ -> throw({multiple_filters,Env}) end. +get_proxy_opts(Env) -> + case [P || P={proxy,_} <- Env] of + [{proxy,Opts}] -> Opts; + [] -> undefined; + _ -> throw({multiple_proxies,Env}) + end. + %% This function looks at the kernel logger environment %% and updates it so that the correct logger is configured init_default_config(Type,Env) when Type==standard_io; @@ -878,42 +898,43 @@ log_allowed(Location,Level,Msg,Meta0) when is_map(Meta0) -> %% (function or macro). Meta = add_default_metadata( maps:merge(Location,maps:merge(proc_meta(),Meta0))), + Tid = tid(), case node(maps:get(gl,Meta)) of Node when Node=/=node() -> - log_remote(Node,Level,Msg,Meta), - do_log_allowed(Level,Msg,Meta); + log_remote(Node,Level,Msg,Meta,Tid), + do_log_allowed(Level,Msg,Meta,Tid); _ -> - do_log_allowed(Level,Msg,Meta) + do_log_allowed(Level,Msg,Meta,Tid) end. -do_log_allowed(Level,{Format,Args}=Msg,Meta) +do_log_allowed(Level,{Format,Args}=Msg,Meta,Tid) when ?IS_LEVEL(Level), is_list(Format), is_list(Args), is_map(Meta) -> - logger_backend:log_allowed(#{level=>Level,msg=>Msg,meta=>Meta},tid()); -do_log_allowed(Level,Report,Meta) + logger_backend:log_allowed(#{level=>Level,msg=>Msg,meta=>Meta},Tid); +do_log_allowed(Level,Report,Meta,Tid) when ?IS_LEVEL(Level), ?IS_REPORT(Report), is_map(Meta) -> logger_backend:log_allowed(#{level=>Level,msg=>{report,Report},meta=>Meta}, - tid()); -do_log_allowed(Level,String,Meta) + Tid); +do_log_allowed(Level,String,Meta,Tid) when ?IS_LEVEL(Level), ?IS_STRING(String), is_map(Meta) -> logger_backend:log_allowed(#{level=>Level,msg=>{string,String},meta=>Meta}, - tid()). + Tid). tid() -> ets:whereis(?LOGGER_TABLE). -log_remote(Node,Level,{Format,Args},Meta) -> - log_remote(Node,{log,Level,Format,Args,Meta}); -log_remote(Node,Level,Msg,Meta) -> - log_remote(Node,{log,Level,Msg,Meta}). +log_remote(Node,Level,{Format,Args},Meta,Tid) -> + log_remote(Node,{log,Level,Format,Args,Meta},Tid); +log_remote(Node,Level,Msg,Meta,Tid) -> + log_remote(Node,{log,Level,Msg,Meta},Tid). -log_remote(Node,Request) -> - {logger,Node} ! Request, +log_remote(Node,Request,Tid) -> + logger_proxy:log(logger_server:get_proxy_ref(Tid),{remote,Node,Request}), ok. add_default_metadata(Meta) -> diff --git a/lib/kernel/src/logger_h_common.erl b/lib/kernel/src/logger_h_common.erl index dd8ace8249..6f55c5997d 100644 --- a/lib/kernel/src/logger_h_common.erl +++ b/lib/kernel/src/logger_h_common.erl @@ -178,7 +178,7 @@ changing_config(SetOrUpdate, log(LogEvent, Config = #{config := #{olp:=Olp}}) -> %% if the handler has crashed, we must drop this event %% and hope the handler restarts so we can try again - true = logger_olp:is_alive(Olp), + true = is_process_alive(logger_olp:get_pid(Olp)), Bin = log_to_binary(LogEvent, Config), logger_olp:load(Olp,Bin). @@ -206,8 +206,9 @@ start(OlpOpts0, #{id := Name, module:=Module, config:=HConfig} = Config0) -> type => worker, modules => [?MODULE]}, case supervisor:start_child(logger_sup, ChildSpec) of - {ok,Pid,{Olp,OlpOpts}} -> + {ok,Pid,Olp} -> ok = logger_handler_watcher:register_handler(Name,Pid), + OlpOpts = logger_olp:get_opts(Olp), {ok,Config0#{config=>(maps:merge(HConfig,OlpOpts))#{olp=>Olp}}}; {error,{Reason,Ch}} when is_tuple(Ch), element(1,Ch)==child -> {error,Reason}; @@ -246,14 +247,14 @@ handle_load(Bin, #{id:=Name, ctrl_sync_count := CtrlSync}=State) -> if CtrlSync==0 -> {_,HS1} = Module:write(Name, sync, Bin, HandlerState), - {ok,State#{handler_state => HS1, - ctrl_sync_count => ?CONTROLLER_SYNC_INTERVAL, - last_op=>write}}; + State#{handler_state => HS1, + ctrl_sync_count => ?CONTROLLER_SYNC_INTERVAL, + last_op=>write}; true -> {_,HS1} = Module:write(Name, async, Bin, HandlerState), - {ok,State#{handler_state => HS1, - ctrl_sync_count => CtrlSync-1, - last_op=>write}} + State#{handler_state => HS1, + ctrl_sync_count => CtrlSync-1, + last_op=>write} end. handle_call(filesync, _From, State = #{id := Name, @@ -295,7 +296,7 @@ handle_info(Info, #{id := Name, module := Module, {noreply,State#{handler_state => Module:handle_info(Name,Info,HandlerState)}}. terminate(overloaded=Reason, #{id:=Name}=State) -> - log_handler_info(Name, "Handler ~p overloaded and stopping", [Name], State), + _ = log_handler_info(Name,"Handler ~p overloaded and stopping",[Name],State), do_terminate(Reason,State), ConfigResult = logger:get_handler_config(Name), case ConfigResult of diff --git a/lib/kernel/src/logger_h_common.hrl b/lib/kernel/src/logger_h_common.hrl index 261b0a6246..f2c2dc2a4e 100644 --- a/lib/kernel/src/logger_h_common.hrl +++ b/lib/kernel/src/logger_h_common.hrl @@ -206,12 +206,16 @@ %%% officially released (as some counters will grow very large %%% over time). -%%-define(SAVE_STATS, true). +%% -define(SAVE_STATS, true). -ifdef(SAVE_STATS). -define(merge_with_stats(STATE), - STATE#{flushes => 0, flushed => 0, drops => 0, - burst_drops => 0, casts => 0, calls => 0, - max_qlen => 0, max_time => 0}). + begin + TIME = ?timestamp(), + STATE#{start => TIME, time => {TIME,0}, + flushes => 0, flushed => 0, drops => 0, + burst_drops => 0, casts => 0, calls => 0, + writes => 0, max_qlen => 0, max_time => 0, + freq => {TIME,0,0}} end). -define(update_max_qlen(QLEN, STATE), begin #{max_qlen := QLEN0} = STATE, @@ -234,13 +238,28 @@ -define(update_other(OTHER, VAR, INCVAL, STATE), begin #{OTHER := VAR} = STATE, STATE#{OTHER => VAR+INCVAL} end). - + + -define(update_freq(TIME,STATE), + begin + case STATE of + #{freq := {START, 49, _}} -> + STATE#{freq => {TIME, 0, trunc(1000000*50/(?diff_time(TIME,START)))}}; + #{freq := {START, N, FREQ}} -> + STATE#{freq => {START, N+1, FREQ}} + end end). + + -define(update_time(TIME,STATE), + begin #{start := START} = STATE, + STATE#{time => {TIME,trunc((?diff_time(TIME,START))/1000000)}} end). + -else. % DEFAULT! -define(merge_with_stats(STATE), STATE). -define(update_max_qlen(_QLEN, STATE), STATE). -define(update_calls_or_casts(_CALL_OR_CAST, _INC, STATE), STATE). -define(update_max_time(_TIME, STATE), STATE). -define(update_other(_OTHER, _VAR, _INCVAL, STATE), STATE). + -define(update_freq(_TIME, STATE), STATE). + -define(update_time(_TIME, STATE), STATE). -endif. %%%----------------------------------------------------------------- diff --git a/lib/kernel/src/logger_olp.erl b/lib/kernel/src/logger_olp.erl index 6b76c78c73..7c6a1a8547 100644 --- a/lib/kernel/src/logger_olp.erl +++ b/lib/kernel/src/logger_olp.erl @@ -25,47 +25,62 @@ %% API -export([start_link/4, load/2, info/1, reset/1, stop/1, restart/1, - set_opts/2, get_opts/1, get_default_opts/0, is_alive/1, - call/2, cast/2]). + set_opts/2, get_opts/1, get_default_opts/0, get_pid/1, + call/2, cast/2, get_ref/0, get_ref/1]). %% gen_server and proc_lib callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -%%%----------------------------------------------------------------- -%% -define(CONFIG_KEYS,[sync_mode_qlen, -%% drop_mode_qlen, -%% flush_qlen, -%% burst_limit_enable, -%% burst_limit_max_count, -%% burst_limit_window_time, -%% overload_kill_enable, -%% overload_kill_qlen, -%% overload_kill_mem_size, -%% overload_kill_restart_after]). +-define(OPT_KEYS,[sync_mode_qlen, + drop_mode_qlen, + flush_qlen, + burst_limit_enable, + burst_limit_max_count, + burst_limit_window_time, + overload_kill_enable, + overload_kill_qlen, + overload_kill_mem_size, + overload_kill_restart_after]). + +-export_type([olp_ref/0, options/0]). + +-opaque olp_ref() :: {atom(),pid(),ets:tid()}. + +-type options() :: #{sync_mode_qlen => integer(), + drop_mode_qlen => integer(), + flush_qlen => integer(), + burst_limit_enable => boolean(), + burst_limit_max_count => integer(), + burst_limit_window_time => integer(), + overload_kill_enable => boolean(), + overload_kill_qlen => integer(), + overload_kill_mem_size => integer(), + overload_kill_restart_after => integer()}. %%%----------------------------------------------------------------- %%% API -%-spec start_link(Name,Module,Args,Options) -> {ok,Pid,Olp} | {error,Reason}. +-spec start_link(Name,Module,Args,Options) -> {ok,Pid,Olp} | {error,Reason} when + Name :: atom(), + Module :: module(), + Args :: term(), + Options :: options(), + Pid :: pid(), + Olp :: olp_ref(), + Reason :: term(). start_link(Name,Module,Args,Options0) when is_map(Options0) -> Options = maps:merge(get_default_opts(),Options0), case check_opts(Options) of ok -> - case proc_lib:start_link(?MODULE,init, - [[Name,Module,Args,Options]]) of - {ok,Pid,Olp} -> - {ok,Pid,{Olp,Options}}; - Error -> - Error - end; + proc_lib:start_link(?MODULE,init,[[Name,Module,Args,Options]]); Error -> Error end. -is_alive({_Name,Pid,_ModeRef}) -> - is_process_alive(Pid). - +-spec load(Olp, Msg) -> ok when + Olp :: olp_ref(), + Msg :: term(). load({_Name,Pid,ModeRef},Msg) -> %% If the process is getting overloaded, the message will be %% synchronous instead of asynchronous (slows down the tempo of a @@ -92,22 +107,36 @@ load({_Name,Pid,ModeRef},Msg) -> end, ok. +-spec info(Olp) -> map() | {error, busy} when + Olp :: atom() | pid() | olp_ref(). info(Olp) -> call(Olp, info). +-spec reset(Olp) -> ok | {error, busy} when + Olp :: atom() | pid() | olp_ref(). reset(Olp) -> call(Olp, reset). +-spec stop(Olp) -> ok when + Olp :: atom() | pid() | olp_ref(). stop({_Name,Pid,_ModRef}) -> + stop(Pid); +stop(Pid) -> _ = gen_server:call(Pid, stop), ok. -set_opts({_Name,Pid,_ModRef}, Opts) -> - gen_server:call(Pid, {set_opts,Opts}). +-spec set_opts(Olp, Opts) -> ok | {error,term()} | {error, busy} when + Olp :: atom() | pid() | olp_ref(), + Opts :: options(). +set_opts(Olp, Opts) -> + call(Olp, {set_opts,Opts}). -get_opts({_Name,Pid,_ModRef}) -> - gen_server:call(Pid, get_opts). +-spec get_opts(Olp) -> options() | {error, busy} when + Olp :: atom() | pid() | olp_ref(). +get_opts(Olp) -> + call(Olp, get_opts). +-spec get_default_opts() -> options(). get_default_opts() -> #{sync_mode_qlen => ?SYNC_MODE_QLEN, drop_mode_qlen => ?DROP_MODE_QLEN, @@ -120,11 +149,30 @@ get_default_opts() -> overload_kill_mem_size => ?OVERLOAD_KILL_MEM_SIZE, overload_kill_restart_after => ?OVERLOAD_KILL_RESTART_AFTER}. +-spec restart(fun(() -> any())) -> ok. restart(Fun) -> - erlang:display(restarting), - erlang:display(_ = Fun()), + Result = + try Fun() + catch C:R:S -> + {error,{restart_failed,Fun,C,R,S}} + end, + ?LOG_INTERNAL(debug,[{logger_olp,restart}, + {result,Result}]), ok. +-spec get_ref() -> olp_ref(). +get_ref() -> + get(olp_ref). + +-spec get_ref(PidOrName) -> olp_ref() | {error, busy} when + PidOrName :: pid() | atom(). +get_ref(PidOrName) -> + call(PidOrName,get_ref). + +-spec get_pid(olp_ref()) -> pid(). +get_pid({_Name,Pid,_ModeRef}) -> + Pid. + %%%=================================================================== %%% gen_server callbacks %%%=================================================================== @@ -136,13 +184,15 @@ init([Name,Module,Args,Options]) -> ?init_test_hooks(), ?start_observation(Name), - try Module:init(Args) of - {ok,CBState} -> - try ets:new(Name, [public]) of - ModeRef -> + try ets:new(Name, [public]) of + ModeRef -> + OlpRef = {Name,self(),ModeRef}, + put(olp_ref,OlpRef), + try Module:init(Args) of + {ok,CBState} -> ?set_mode(ModeRef, async), T0 = ?timestamp(), - proc_lib:init_ack({ok,self(),{Name,self(),ModeRef}}), + proc_lib:init_ack({ok,self(),OlpRef}), %% Storing options in state to avoid copying %% (sending) the option data with each message State0 = ?merge_with_stats( @@ -156,15 +206,17 @@ init([Name,Module,Args,Options]) -> burst_msg_count => 0, cb_state => CBState}), State = reset_restart_flag(State0), - gen_server:enter_loop(?MODULE, [], State) + gen_server:enter_loop(?MODULE, [], State); + Error -> + _ = ets:delete(ModeRef), + unregister(Name), + proc_lib:init_ack(Error) catch _:Error -> + _ = ets:delete(ModeRef), unregister(Name), proc_lib:init_ack(Error) - end; - Error -> - unregister(Name), - proc_lib:init_ack(Error) + end catch _:Error -> unregister(Name), @@ -177,8 +229,11 @@ handle_call({'$olp_load', Msg}, _From, State) -> %% Result == ok | dropped {reply,Result, State1}; +handle_call(get_ref,_From,#{id:=Name,mode_ref:=ModeRef}=State) -> + {reply,{Name,self(),ModeRef},State}; + handle_call({set_opts,Opts0},_From,State) -> - Opts = maps:merge(get_default_opts(),Opts0), + Opts = maps:merge(maps:with(?OPT_KEYS,State),Opts0), case check_opts(Opts) of ok -> {reply, ok, maps:merge(State,Opts)}; @@ -186,6 +241,9 @@ handle_call({set_opts,Opts0},_From,State) -> {reply, Error, State} end; +handle_call(get_opts,_From,State) -> + {reply, maps:with(?OPT_KEYS,State), State}; + handle_call(info, _From, State) -> {reply, State, State}; @@ -214,7 +272,6 @@ handle_call(Msg, From, #{module:=Module,cb_state:=CBState}=State) -> %% This is the asynchronous load event. handle_cast({'$olp_load', Msg}, State) -> {_Result,State1} = do_load(Msg, cast, State), - %% Result == ok | dropped {noreply,State1}; handle_cast(Msg, #{module:=Module, cb_state:=CBState} = State) -> @@ -230,7 +287,10 @@ handle_info(Msg, #{module := Module, cb_state := CBState} = State) -> {noreply,CBState1} -> {noreply,State#{cb_state=>CBState1}}; {noreply,CBState1,Timeout} -> - {noreply,State#{cb_state=>CBState1},Timeout} + {noreply,State#{cb_state=>CBState1},Timeout}; + {load,CBState1} -> + {_,State1} = do_load(Msg, cast, State#{cb_state=>CBState1}), + {noreply,State1} end. terminate({shutdown,{overloaded,_QLen,_Mem}}, @@ -242,12 +302,11 @@ terminate({shutdown,{overloaded,_QLen,_Mem}}, case try_callback_call(Module,terminate,[overloaded,CBState],ok) of {ok,Fun} when is_function(Fun,0), is_integer(RestartAfter) -> set_restart_flag(State), - timer:apply_after(RestartAfter,?MODULE,restart,[Fun]), + _ = timer:apply_after(RestartAfter,?MODULE,restart,[Fun]), ok; _ -> ok - end, - ok; + end; terminate(Reason, #{id:=Name, module:=Module, cb_state:=CBState}) -> _ = try_callback_call(Module,terminate,[Reason,CBState],ok), unregister(Name), @@ -259,6 +318,8 @@ code_change(_OldVsn, State, _Extra) -> %%%----------------------------------------------------------------- %%% Internal functions +-spec call(Olp, term()) -> term() | {error,busy} when + Olp :: atom() | pid() | olp_ref(). call({_Name, Pid, _ModeRef},Msg) -> call(Pid, Msg); call(Server, Msg) -> @@ -268,6 +329,7 @@ call(Server, Msg) -> _:{timeout,_} -> {error,busy} end. +-spec cast(olp_ref(),term()) -> ok. cast({_Name, Pid, _ModeRef},Msg) -> gen_server:cast(Pid, Msg). @@ -276,26 +338,27 @@ cast({_Name, Pid, _ModeRef},Msg) -> %% before LogWindowSize events have been handled do_load(Msg, CallOrCast, State) -> T1 = ?timestamp(), + State1 = ?update_time(T1,State), %% check if the process is getting overloaded, or if it's %% recovering from overload (the check must be done for each %% event to react quickly to large bursts of events and %% to ensure that the handler can never end up in drop mode %% with an empty mailbox, which would stop operation) - {Mode1,QLen,Mem,State1} = check_load(State), + {Mode1,QLen,Mem,State2} = check_load(State1), %% kill the handler if it can't keep up with the load - kill_if_choked(QLen, Mem, State1), + kill_if_choked(QLen, Mem, State2), if Mode1 == flush -> - flush(T1, State1); + flush(T1, State2); true -> - handle_load(Mode1, T1, Msg, CallOrCast, State1) + handle_load(Mode1, T1, Msg, CallOrCast, State2) end. %% this function is called by do_load/3 after an overload check %% has been performed, where QLen > FlushQLen -flush(T1, State=#{id := _Name, last_load_ts := T0, mode_ref := ModeRef}) -> +flush(T1, State=#{id := _Name, mode := Mode, last_load_ts := T0, mode_ref := ModeRef}) -> %% flush load messages in the mailbox (a limited number in order %% to not cause long delays) NewFlushed = flush_load(?FLUSH_MAX_N), @@ -306,17 +369,18 @@ flush(T1, State=#{id := _Name, last_load_ts := T0, mode_ref := ModeRef}) -> %% because of the receive loop when flushing messages, the %% handler will be scheduled out often and the mailbox could %% grow very large, so we'd better check the queue again here - {_,_QLen1} = process_info(self(), message_queue_len), - ?observe(_Name,{max_qlen,_QLen1}), + {_,QLen1} = process_info(self(), message_queue_len), + ?observe(_Name,{max_qlen,QLen1}), %% Add 1 for the current log event ?observe(_Name,{flushed,NewFlushed+1}), State2 = ?update_max_time(?diff_time(T1,T0),State1), - State3 = ?update_max_qlen(_QLen1,State2), + State3 = ?update_max_qlen(QLen1,State2), + State4 = maybe_notify_mode_change(async,QLen1,State3), {dropped,?update_other(flushed,FLUSHED,NewFlushed, - State3#{mode => ?set_mode(ModeRef,async), - last_qlen => 0, + State4#{mode => ?change_mode(ModeRef,Mode,async), + last_qlen => QLen1, last_load_ts => T1})}. %% this function is called to actually handle the message @@ -334,7 +398,7 @@ handle_load(Mode, T1, Msg, _CallOrCast, {Result,LastQLen1,CBState1} = if DoWrite -> ?observe(_Name,{_CallOrCast,1}), - {ok,CBS} = try_callback_call(Module,handle_load,[Msg,CBState]), + CBS = try_callback_call(Module,handle_load,[Msg,CBState]), {ok,element(2, process_info(self(), message_queue_len)),CBS}; true -> ?observe(_Name,{flushed,1}), @@ -353,9 +417,10 @@ handle_load(Mode, T1, Msg, _CallOrCast, State3 = if (LastQLen1 < ?FILESYNC_OK_QLEN) andalso (Time > ?IDLE_DETECT_TIME_USEC) -> - S = notify(idle,State2), - S#{mode => ?change_mode(ModeRef, Mode, async), - burst_msg_count => 0}; + S1 = notify(idle,State2), + S2 = maybe_notify_mode_change(async,LastQLen1,S1), + S2#{mode => ?change_mode(ModeRef, Mode, async), + burst_msg_count => 0}; true -> State2#{mode => Mode} end, @@ -365,7 +430,14 @@ handle_load(Mode, T1, Msg, _CallOrCast, ?update_max_time(Time, State5#{last_qlen := LastQLen1, last_load_ts => T1}), - {Result,State6}. + State7 = case Result of + ok -> + S = ?update_freq(T1,State6), + ?update_other(writes,WRITES,1,S); + _ -> + State6 + end, + {Result,State7}. %%%----------------------------------------------------------------- @@ -452,7 +524,7 @@ check_load(State = #{id:=_Name, mode_ref := ModeRef, mode := Mode, QLen >= DropModeQLen -> %% Note that drop mode will force load messages to %% be dropped on the client side (never sent to - %% the handler). + %% the olp process). IncDrops = if Mode == drop -> 0; true -> 1 end, {?change_mode(ModeRef, Mode, drop), IncDrops,0}; QLen >= SyncModeQLen -> @@ -461,10 +533,11 @@ check_load(State = #{id:=_Name, mode_ref := ModeRef, mode := Mode, {?change_mode(ModeRef, Mode, async), 0,0} end, State1 = ?update_other(drops,DROPS,_NewDrops,State), - State2 = maybe_notify_mode_change(Mode1,State1), + State2 = ?update_max_qlen(QLen,State1), + State3 = maybe_notify_mode_change(Mode1,QLen,State2), {Mode1, QLen, Mem, ?update_other(flushes,FLUSHES,_NewFlushes, - State2#{last_qlen => QLen})}. + State3#{last_qlen => QLen})}. limit_burst(#{burst_limit_enable := false}=State) -> {true,State}; @@ -517,7 +590,11 @@ flush_load(N, Limit) -> flush_load(N+1, Limit); {'$gen_call',{Pid,MRef},{'$olp_load',_}} -> Pid ! {MRef, dropped}, - flush_load(N+1, Limit) + flush_load(N+1, Limit); + {log,_,_,_,_} -> + flush_load(N+1, Limit); + {log,_,_,_} -> + flush_load(N+1, Limit) after 0 -> N end. @@ -528,13 +605,13 @@ overload_levels_ok(Options) -> FQL = maps:get(flush_qlen, Options, ?FLUSH_QLEN), (DMQL > 1) andalso (SMQL =< DMQL) andalso (DMQL =< FQL). -maybe_notify_mode_change(drop,#{mode:=Mode0}=State) +maybe_notify_mode_change(drop,_QLen,#{mode:=Mode0}=State) when Mode0=/=drop -> notify({mode_change,Mode0,drop},State); -maybe_notify_mode_change(Mode1,#{mode:=drop}=State) +maybe_notify_mode_change(Mode1,_QLen,#{mode:=drop}=State) when Mode1==async; Mode1==sync -> notify({mode_change,drop,Mode1},State); -maybe_notify_mode_change(_,State) -> +maybe_notify_mode_change(_,_,State) -> State. notify(Note,#{module:=Module,cb_state:=CBState}=State) -> diff --git a/lib/kernel/src/logger_proxy.erl b/lib/kernel/src/logger_proxy.erl new file mode 100644 index 0000000000..f89891bff0 --- /dev/null +++ b/lib/kernel/src/logger_proxy.erl @@ -0,0 +1,152 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017-2018. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(logger_proxy). + +%% API +-export([start_link/0, restart/0, log/2, child_spec/0]). + +%% logger_olp callbacks +-export([init/1, handle_load/2, handle_info/2, terminate/2, + notify/2]). + +-include("logger_internal.hrl"). + +-define(SERVER,?MODULE). + +%%%----------------------------------------------------------------- +%%% API +-spec log(Olp, RemoteLog) -> ok when + Olp :: logger_olp:olp_ref(), + RemoteLog :: {remote,node(),LogEvent}, + LogEvent :: {log,Level,Format,Args,Meta} | + {log,Level,StringOrReport,Meta}, + Level :: logger:level(), + Format :: io:format(), + Args :: list(term()), + StringOrReport :: unicode:chardata() | logger:report(), + Meta :: logger:metadata(). +log(Olp, RemoteLog) -> + case logger_olp:get_pid(Olp) =:= self() of + true -> + %% This happens when the log event comes from the + %% emulator, and the group leader is on a remote node. + _ = handle_load(RemoteLog, no_state), + ok; + false -> + logger_olp:load(Olp, RemoteLog) + end. + +%% Called by supervisor +-spec start_link() -> {ok,pid(),logger_olp:olp_ref()} | {error,term()}. +start_link() -> + %% Notice that sync_mode is only used when logging to remote node, + %% i.e. when the log/2 API function is called. + %% + %% When receiving log events from the emulator or from a remote + %% node, the log event is sent as a message to this process, and + %% thus received directly in handle_info/2. This means that the + %% mode (async/sync/drop) is not read before the message is + %% sent. Thus sync mode is never entered, and drop mode is + %% implemented by setting the system_logger flag to undefined (see + %% notify/2) + %% + %% Burst limit is disabled, since this is only a proxy and we + %% don't want to limit bursts twice (here and in the handler). + Opts = #{sync_mode_qlen=>500, + drop_mode_qlen=>1000, + flush_qlen=>5000, + burst_limit_enable=>false}, + logger_olp:start_link(?SERVER,?MODULE,[],Opts). + +%% Fun used for restarting this process after it has been killed due +%% to overload (must set overload_kill_enable=>true in opts) +restart() -> + case supervisor:start_child(logger_sup, child_spec()) of + {ok,_Pid,Olp} -> + {ok,Olp}; + {error,{Reason,Ch}} when is_tuple(Ch), element(1,Ch)==child -> + {error,Reason}; + Error -> + Error + end. + +%% Called internally and by logger_sup +child_spec() -> + Name = ?SERVER, + #{id => Name, + start => {?MODULE, start_link, []}, + restart => temporary, + shutdown => 2000, + type => worker, + modules => [?MODULE]}. + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +init([]) -> + process_flag(trap_exit, true), + _ = erlang:system_flag(system_logger,self()), + logger_server:set_proxy_ref(logger_olp:get_ref()), + {ok,no_state}. + +%% Log event to send to the node where the group leader of it's client resides +handle_load({remote,Node,Log},State) -> + %% If the connection is overloaded (send_nosuspend returns false), + %% we drop the message. + _ = erlang:send_nosuspend({?SERVER,Node},Log), + State; +%% Log event to log on this node +handle_load({log,Level,Format,Args,Meta},State) -> + try_log([Level,Format,Args,Meta]), + State; +handle_load({log,Level,Report,Meta},State) -> + try_log([Level,Report,Meta]), + State. + +%% Log event sent to this process e.g. from the emulator - it is really load +handle_info(Log,State) when is_tuple(Log), element(1,Log)==log -> + {load,State}. + +terminate(overloaded, _State) -> + _ = erlang:system_flag(system_logger,undefined), + {ok,fun ?MODULE:restart/0}; +terminate(_Reason, _State) -> + _ = erlang:system_flag(system_logger,whereis(logger)), + ok. + +notify({mode_change,_Mode0,drop},State) -> + _ = erlang:system_flag(system_logger,undefined), + State; +notify({mode_change,drop,_Mode1},State) -> + _ = erlang:system_flag(system_logger,self()), + State; +notify(_Note,State) -> + State. + +%%%----------------------------------------------------------------- +%%% Internal functions +try_log(Args) -> + try apply(logger,log,Args) + catch C:R:S -> + ?LOG_INTERNAL(debug,[{?MODULE,log_failed}, + {log,Args}, + {reason,{C,R,S}}]) + end. diff --git a/lib/kernel/src/logger_server.erl b/lib/kernel/src/logger_server.erl index b7735dbcf7..c58edf51f8 100644 --- a/lib/kernel/src/logger_server.erl +++ b/lib/kernel/src/logger_server.erl @@ -22,7 +22,7 @@ -behaviour(gen_server). %% API --export([start_link/0, +-export([start_link/0, set_proxy_ref/1, get_proxy_ref/1, add_handler/3, remove_handler/1, add_filter/2, remove_filter/2, set_module_level/2, unset_module_level/0, @@ -43,7 +43,7 @@ -define(SERVER, logger). -define(LOGGER_SERVER_TAG, '$logger_cb_process'). --record(state, {tid, async_req, async_req_queue}). +-record(state, {tid, async_req, async_req_queue, remote_logger}). %%%=================================================================== %%% API @@ -52,6 +52,14 @@ start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). +-spec set_proxy_ref(logger_olp:olp_ref()) -> ok. +set_proxy_ref(ProxyRef) -> + call({set_proxy_ref,ProxyRef}). + +-spec get_proxy_ref(ets:tid()) -> logger_olp:olp_ref(). +get_proxy_ref(Tid) -> + ets:lookup_element(Tid,proxy_ref,2). + add_handler(Id,Module,Config0) -> try {check_id(Id),check_mod(Module)} of {ok,ok} -> @@ -311,7 +319,10 @@ handle_call({set_module_level,Modules,Level}, _From, #state{tid=Tid}=State) -> {reply,Reply,State}; handle_call({unset_module_level,Modules}, _From, #state{tid=Tid}=State) -> Reply = logger_config:unset_module_level(Tid,Modules), - {reply,Reply,State}. + {reply,Reply,State}; +handle_call({set_proxy_ref,ProxyRef},_From,#state{tid=Tid}=State) -> + true = ets:insert(Tid,{proxy_ref,ProxyRef}), + {reply,ok,State}. handle_cast({async_req_reply,_Ref,_Reply} = Reply,State) -> call_h_reply(Reply,State); @@ -357,7 +368,7 @@ terminate(_Reason, _State) -> %%%=================================================================== %%% Internal functions %%%=================================================================== -call(Request) -> +call(Request) when is_tuple(Request) -> Action = element(1,Request), case get(?LOGGER_SERVER_TAG) of true when @@ -369,6 +380,7 @@ call(Request) -> gen_server:call(?SERVER,Request,?DEFAULT_LOGGER_CALL_TIMEOUT) end. + do_add_filter(Tid,Id,{FId,_} = Filter) -> case logger_config:get(Tid,Id) of {ok,Config} -> diff --git a/lib/kernel/src/logger_sup.erl b/lib/kernel/src/logger_sup.erl index 3d6f482e20..9ea8558a16 100644 --- a/lib/kernel/src/logger_sup.erl +++ b/lib/kernel/src/logger_sup.erl @@ -50,7 +50,9 @@ init([]) -> start => {logger_handler_watcher, start_link, []}, shutdown => brutal_kill}, - {ok, {SupFlags, [Watcher]}}. + Proxy = logger_proxy:child_spec(), + + {ok, {SupFlags, [Watcher,Proxy]}}. %%%=================================================================== %%% Internal functions diff --git a/lib/kernel/test/logger.cover b/lib/kernel/test/logger.cover index a9ef81903d..9691aa295e 100644 --- a/lib/kernel/test/logger.cover +++ b/lib/kernel/test/logger.cover @@ -9,6 +9,7 @@ logger_handler_watcher, logger_h_common, logger_olp, + logger_proxy, logger_server, logger_simple_h, logger_std_h, |