diff options
Diffstat (limited to 'lib/kernel/src')
25 files changed, 2078 insertions, 1789 deletions
diff --git a/lib/kernel/src/Makefile b/lib/kernel/src/Makefile index 7db05ebc90..43b776f37e 100644 --- a/lib/kernel/src/Makefile +++ b/lib/kernel/src/Makefile @@ -118,6 +118,8 @@ MODULES = \ logger_h_common \ logger_filters \ logger_formatter \ + logger_olp \ + logger_proxy \ logger_server \ logger_simple_h \ logger_sup \ @@ -150,7 +152,7 @@ INTERNAL_HRL_FILES= application_master.hrl disk_log.hrl \ inet_dns.hrl inet_res.hrl \ inet_boot.hrl inet_config.hrl inet_int.hrl \ inet_dns_record_adts.hrl \ - logger_internal.hrl logger_h_common.hrl + logger_internal.hrl logger_olp.hrl logger_h_common.hrl ERL_FILES= $(MODULES:%=%.erl) @@ -278,6 +280,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_olp.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/erl_epmd.erl b/lib/kernel/src/erl_epmd.erl index b7e8868911..7a14e2635c 100644 --- a/lib/kernel/src/erl_epmd.erl +++ b/lib/kernel/src/erl_epmd.erl @@ -77,8 +77,8 @@ stop() -> %% -spec port_please(Name, Host) -> {ok, Port, Version} | noport when - Name :: string(), - Host :: inet:ip_address(), + Name :: atom() | string(), + Host :: atom() | string() | inet:ip_address(), Port :: non_neg_integer(), Version :: non_neg_integer(). @@ -86,8 +86,8 @@ port_please(Node, Host) -> port_please(Node, Host, infinity). -spec port_please(Name, Host, Timeout) -> {ok, Port, Version} | noport when - Name :: string(), - Host :: inet:ip_address(), + Name :: atom() | string(), + Host :: atom() | string() | inet:ip_address(), Timeout :: non_neg_integer() | infinity, Port :: non_neg_integer(), Version :: non_neg_integer(). diff --git a/lib/kernel/src/inet.erl b/lib/kernel/src/inet.erl index 5dd68dc285..9f22eb6aaa 100644 --- a/lib/kernel/src/inet.erl +++ b/lib/kernel/src/inet.erl @@ -154,6 +154,15 @@ 'running' | 'multicast' | 'loopback']} | {'hwaddr', ether_address()}. +-type getifaddrs_ifopts() :: + [Ifopt :: {flags, Flags :: [up | broadcast | loopback | + pointtopoint | running | multicast]} | + {addr, Addr :: ip_address()} | + {netmask, Netmask :: ip_address()} | + {broadaddr, Broadaddr :: ip_address()} | + {dstaddr, Dstaddr :: ip_address()} | + {hwaddr, Hwaddr :: [byte()]}]. + -type address_family() :: 'inet' | 'inet6' | 'local'. -type socket_protocol() :: 'tcp' | 'udp' | 'sctp'. -type socket_type() :: 'stream' | 'dgram' | 'seqpacket'. @@ -321,32 +330,32 @@ getopts(Socket, Opts) -> Other end. --spec getifaddrs(Socket :: socket()) -> - {'ok', [string()]} | {'error', posix()}. - +-spec getifaddrs( + [Option :: {netns, Namespace :: file:filename_all()}] + | socket()) -> + {'ok', [{Ifname :: string(), + Ifopts :: getifaddrs_ifopts()}]} + | {'error', posix()}. +getifaddrs(Opts) when is_list(Opts) -> + withsocket(fun(S) -> prim_inet:getifaddrs(S) end, Opts); getifaddrs(Socket) -> prim_inet:getifaddrs(Socket). --spec getifaddrs() -> {ok, Iflist} | {error, posix()} when - Iflist :: [{Ifname,[Ifopt]}], - Ifname :: string(), - Ifopt :: {flags,[Flag]} | {addr,Addr} | {netmask,Netmask} - | {broadaddr,Broadaddr} | {dstaddr,Dstaddr} - | {hwaddr,Hwaddr}, - Flag :: up | broadcast | loopback | pointtopoint - | running | multicast, - Addr :: ip_address(), - Netmask :: ip_address(), - Broadaddr :: ip_address(), - Dstaddr :: ip_address(), - Hwaddr :: [byte()]. - +-spec getifaddrs() -> + {'ok', [{Ifname :: string(), + Ifopts :: getifaddrs_ifopts()}]} + | {'error', posix()}. getifaddrs() -> withsocket(fun(S) -> prim_inet:getifaddrs(S) end). --spec getiflist(Socket :: socket()) -> - {'ok', [string()]} | {'error', posix()}. +-spec getiflist( + [Option :: {netns, Namespace :: file:filename_all()}] + | socket()) -> + {'ok', [string()]} | {'error', posix()}. + +getiflist(Opts) when is_list(Opts) -> + withsocket(fun(S) -> prim_inet:getiflist(S) end, Opts); getiflist(Socket) -> prim_inet:getiflist(Socket). @@ -363,11 +372,19 @@ getiflist() -> ifget(Socket, Name, Opts) -> prim_inet:ifget(Socket, Name, Opts). --spec ifget(Name :: string() | atom(), Opts :: [if_getopt()]) -> +-spec ifget( + Name :: string() | atom(), + Opts :: [if_getopt() | + {netns, Namespace :: file:filename_all()}]) -> {'ok', [if_getopt_result()]} | {'error', posix()}. ifget(Name, Opts) -> - withsocket(fun(S) -> prim_inet:ifget(S, Name, Opts) end). + {NSOpts,IFOpts} = + lists:partition( + fun ({netns,_}) -> true; + (_) -> false + end, Opts), + withsocket(fun(S) -> prim_inet:ifget(S, Name, IFOpts) end, NSOpts). -spec ifset(Socket :: socket(), Name :: string() | atom(), @@ -377,11 +394,19 @@ ifget(Name, Opts) -> ifset(Socket, Name, Opts) -> prim_inet:ifset(Socket, Name, Opts). --spec ifset(Name :: string() | atom(), Opts :: [if_setopt()]) -> +-spec ifset( + Name :: string() | atom(), + Opts :: [if_setopt() | + {netns, Namespace :: file:filename_all()}]) -> 'ok' | {'error', posix()}. ifset(Name, Opts) -> - withsocket(fun(S) -> prim_inet:ifset(S, Name, Opts) end). + {NSOpts,IFOpts} = + lists:partition( + fun ({netns,_}) -> true; + (_) -> false + end, Opts), + withsocket(fun(S) -> prim_inet:ifset(S, Name, IFOpts) end, NSOpts). -spec getif() -> {'ok', [{ip_address(), ip_address() | 'undefined', ip_address()}]} | @@ -391,10 +416,14 @@ getif() -> withsocket(fun(S) -> getif(S) end). %% backwards compatible getif --spec getif(Socket :: socket()) -> +-spec getif( + [Option :: {netns, Namespace :: file:filename_all()}] + | socket()) -> {'ok', [{ip_address(), ip_address() | 'undefined', ip_address()}]} | {'error', posix()}. +getif(Opts) when is_list(Opts) -> + withsocket(fun(S) -> getif(S) end, Opts); getif(Socket) -> case prim_inet:getiflist(Socket) of {ok, IfList} -> @@ -415,7 +444,10 @@ getif(Socket) -> end. withsocket(Fun) -> - case inet_udp:open(0,[]) of + withsocket(Fun, []). +%% +withsocket(Fun, Opts) -> + case inet_udp:open(0, Opts) of {ok,Socket} -> Res = Fun(Socket), inet_udp:close(Socket), diff --git a/lib/kernel/src/inet_int.hrl b/lib/kernel/src/inet_int.hrl index c8e09d18ad..f6525d7261 100644 --- a/lib/kernel/src/inet_int.hrl +++ b/lib/kernel/src/inet_int.hrl @@ -162,6 +162,7 @@ -define(INET_OPT_PKTOPTIONS, 45). -define(INET_OPT_TTL, 46). -define(INET_OPT_RECVTTL, 47). +-define(TCP_OPT_NOPUSH, 48). % Specific SCTP options: separate range: -define(SCTP_OPT_RTOINFO, 100). -define(SCTP_OPT_ASSOCINFO, 101). diff --git a/lib/kernel/src/inet_tcp_dist.erl b/lib/kernel/src/inet_tcp_dist.erl index d1701afdaa..c37212b0f9 100644 --- a/lib/kernel/src/inet_tcp_dist.erl +++ b/lib/kernel/src/inet_tcp_dist.erl @@ -450,7 +450,7 @@ get_tcp_address(Driver, Socket) -> get_address_resolver(EpmdModule) -> case erlang:function_exported(EpmdModule, address_please, 3) of true -> {EpmdModule, address_please}; - _ -> {inet, getaddr} + _ -> {erl_epmd, address_please} end. %% ------------------------------------------------------------ diff --git a/lib/kernel/src/kernel.app.src b/lib/kernel/src/kernel.app.src index 4933eae76f..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, @@ -146,6 +148,6 @@ {logger_sasl_compatible, false} ]}, {mod, {kernel, []}}, - {runtime_dependencies, ["erts-10.0", "stdlib-3.5", "sasl-3.0"]} + {runtime_dependencies, ["erts-10.1", "stdlib-3.5", "sasl-3.0"]} ] }. diff --git a/lib/kernel/src/kernel.appup.src b/lib/kernel/src/kernel.appup.src index 305a1c788c..ccf0a82ced 100644 --- a/lib/kernel/src/kernel.appup.src +++ b/lib/kernel/src/kernel.appup.src @@ -16,13 +16,42 @@ %% limitations under the License. %% %% %CopyrightEnd% +%% +%% We allow upgrade from, and downgrade to all previous +%% versions from the following OTP releases: +%% - OTP 20 +%% - OTP 21 +%% +%% We also allow upgrade from, and downgrade to all +%% versions that have branched off from the above +%% stated previous versions. +%% {"%VSN%", - %% Up from - max one major revision back - [{<<"5\\.3(\\.[0-9]+)*">>,[restart_new_emulator]}, % OTP-20.0 - {<<"5\\.4(\\.[0-9]+)*">>,[restart_new_emulator]}, % OTP-20.1+ - {<<"6\\.0(\\.[0-9]+)*">>,[restart_new_emulator]}], % OTP-21 - %% Down to - max one major revision back - [{<<"5\\.3(\\.[0-9]+)*">>,[restart_new_emulator]}, % OTP-20.0 - {<<"5\\.4(\\.[0-9]+)*">>,[restart_new_emulator]}, % OTP-20.1+ - {<<"6\\.0(\\.[0-9]+)*">>,[restart_new_emulator]}] % OTP-21 -}. + [{<<"^5\\.3$">>,[restart_new_emulator]}, + {<<"^5\\.3\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^5\\.3\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^5\\.4$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.2(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.3(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^6\\.0$">>,[restart_new_emulator]}, + {<<"^6\\.0\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^6\\.0\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^6\\.1$">>,[restart_new_emulator]}, + {<<"^6\\.1\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^6\\.1\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}], + [{<<"^5\\.3$">>,[restart_new_emulator]}, + {<<"^5\\.3\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^5\\.3\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^5\\.4$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.2(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^5\\.4\\.3(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^6\\.0$">>,[restart_new_emulator]}, + {<<"^6\\.0\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^6\\.0\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}, + {<<"^6\\.1$">>,[restart_new_emulator]}, + {<<"^6\\.1\\.0(?:\\.[0-9]+)+$">>,[restart_new_emulator]}, + {<<"^6\\.1\\.1(?:\\.[0-9]+)*$">>,[restart_new_emulator]}]}. diff --git a/lib/kernel/src/logger.erl b/lib/kernel/src/logger.erl index 752dd8d493..abdd9a9ceb 100644 --- a/lib/kernel/src/logger.erl +++ b/lib/kernel/src/logger.erl @@ -43,10 +43,14 @@ get_module_level/0, get_module_level/1, set_primary_config/1, set_primary_config/2, set_handler_config/2, set_handler_config/3, - update_primary_config/1, update_handler_config/2, + set_proxy_config/1, + update_primary_config/1, + update_handler_config/2, update_handler_config/3, + update_proxy_config/1, update_formatter_config/2, update_formatter_config/3, get_primary_config/0, get_handler_config/1, get_handler_config/0, get_handler_ids/0, get_config/0, + get_proxy_config/0, add_handlers/1]). %% Private configuration @@ -121,6 +125,18 @@ {filters,log | stop,[{filter_id(),filter()}]} | {module_level,level(),[module()]}]. +-type olp_config() :: #{sync_mode_qlen => non_neg_integer(), + drop_mode_qlen => pos_integer(), + flush_qlen => pos_integer(), + burst_limit_enable => boolean(), + burst_limit_max_count => pos_integer(), + burst_limit_window_time => pos_integer(), + overload_kill_enable => boolean(), + overload_kill_qlen => pos_integer(), + overload_kill_mem_size => pos_integer(), + overload_kill_restart_after => + non_neg_integer() | infinity}. + -export_type([log_event/0, level/0, report/0, @@ -136,7 +152,8 @@ filter_arg/0, filter_return/0, config_handler/0, - formatter_config/0]). + formatter_config/0, + olp_config/0]). %%%----------------------------------------------------------------- %%% API @@ -389,6 +406,7 @@ set_primary_config(Key,Value) -> set_primary_config(Config) -> logger_server:set_config(primary,Config). + -spec set_handler_config(HandlerId,level,Level) -> Return when HandlerId :: handler_id(), Level :: level() | all | none, @@ -418,17 +436,50 @@ set_handler_config(HandlerId,Key,Value) -> set_handler_config(HandlerId,Config) -> logger_server:set_config(HandlerId,Config). +-spec set_proxy_config(Config) -> ok | {error,term()} when + Config :: olp_config(). +set_proxy_config(Config) -> + logger_server:set_config(proxy,Config). + -spec update_primary_config(Config) -> ok | {error,term()} when Config :: primary_config(). update_primary_config(Config) -> logger_server:update_config(primary,Config). +-spec update_handler_config(HandlerId,level,Level) -> Return when + HandlerId :: handler_id(), + Level :: level() | all | none, + Return :: ok | {error,term()}; + (HandlerId,filter_default,FilterDefault) -> Return when + HandlerId :: handler_id(), + FilterDefault :: log | stop, + Return :: ok | {error,term()}; + (HandlerId,filters,Filters) -> Return when + HandlerId :: handler_id(), + Filters :: [{filter_id(),filter()}], + Return :: ok | {error,term()}; + (HandlerId,formatter,Formatter) -> Return when + HandlerId :: handler_id(), + Formatter :: {module(), formatter_config()}, + Return :: ok | {error,term()}; + (HandlerId,config,Config) -> Return when + HandlerId :: handler_id(), + Config :: term(), + Return :: ok | {error,term()}. +update_handler_config(HandlerId,Key,Value) -> + logger_server:update_config(HandlerId,Key,Value). + -spec update_handler_config(HandlerId,Config) -> ok | {error,term()} when HandlerId :: handler_id(), Config :: handler_config(). update_handler_config(HandlerId,Config) -> logger_server:update_config(HandlerId,Config). +-spec update_proxy_config(Config) -> ok | {error,term()} when + Config :: olp_config(). +update_proxy_config(Config) -> + logger_server:update_config(proxy,Config). + -spec get_primary_config() -> Config when Config :: primary_config(). get_primary_config() -> @@ -439,7 +490,14 @@ get_primary_config() -> HandlerId :: handler_id(), Config :: handler_config(). get_handler_config(HandlerId) -> - logger_config:get(?LOGGER_TABLE,HandlerId). + case logger_config:get(?LOGGER_TABLE,HandlerId) of + {ok,#{module:=Module}=Config} -> + {ok,try Module:filter_config(Config) + catch _:_ -> Config + end}; + Error -> + Error + end. -spec get_handler_config() -> [Config] when Config :: handler_config(). @@ -455,6 +513,12 @@ get_handler_ids() -> {ok,#{handlers:=HandlerIds}} = logger_config:get(?LOGGER_TABLE,primary), HandlerIds. +-spec get_proxy_config() -> Config when + Config :: olp_config(). +get_proxy_config() -> + {ok,Config} = logger_config:get(?LOGGER_TABLE,proxy), + Config. + -spec update_formatter_config(HandlerId,FormatterConfig) -> ok | {error,term()} when HandlerId :: handler_id(), @@ -575,10 +639,12 @@ unset_process_metadata() -> -spec get_config() -> #{primary=>primary_config(), handlers=>[handler_config()], + proxy=>olp_config(), module_levels=>[{module(),level() | all | none}]}. get_config() -> #{primary=>get_primary_config(), handlers=>get_handler_config(), + proxy=>get_proxy_config(), module_levels=>lists:keysort(1,get_module_level())}. -spec internal_init_logger() -> ok | {error,term()}. @@ -641,6 +707,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 set_proxy_config(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) -> @@ -698,6 +775,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). @@ -753,6 +832,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; @@ -849,30 +935,30 @@ log_allowed(Location,Level,Msg,Meta0) when is_map(Meta0) -> maps:merge(Location,maps:merge(proc_meta(),Meta0))), 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); _ -> - do_log_allowed(Level,Msg,Meta) - end. + ok + end, + do_log_allowed(Level,Msg,Meta,tid()). -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). @@ -882,7 +968,7 @@ log_remote(Node,Level,Msg,Meta) -> log_remote(Node,{log,Level,Msg,Meta}). log_remote(Node,Request) -> - {logger,Node} ! Request, + logger_proxy:log({remote,Node,Request}), ok. add_default_metadata(Meta) -> diff --git a/lib/kernel/src/logger_config.erl b/lib/kernel/src/logger_config.erl index 6bfe658552..5024d20cfe 100644 --- a/lib/kernel/src/logger_config.erl +++ b/lib/kernel/src/logger_config.erl @@ -31,7 +31,9 @@ -include("logger_internal.hrl"). new(Name) -> - _ = ets:new(Name,[set,protected,named_table,{write_concurrency,true}]), + _ = ets:new(Name,[set,protected,named_table, + {read_concurrency,true}, + {write_concurrency,true}]), ets:whereis(Name). delete(Tid,Id) -> @@ -64,6 +66,8 @@ get(Tid,What) -> case ets:lookup(Tid,table_key(What)) of [{_,_,Config}] -> {ok,Config}; + [{_,Config}] when What=:=proxy -> + {ok,Config}; [] -> {error,{not_found,What}} end. @@ -77,10 +81,15 @@ get(Tid,What,Level) -> [Data] -> {ok,Data} end. +create(Tid,proxy,Config) -> + ets:insert(Tid,{table_key(proxy),Config}); create(Tid,What,Config) -> LevelInt = level_to_int(maps:get(level,Config)), ets:insert(Tid,{table_key(What),LevelInt,Config}). +set(Tid,proxy,Config) -> + ets:insert(Tid,{table_key(proxy),Config}), + ok; set(Tid,What,Config) -> LevelInt = level_to_int(maps:get(level,Config)), %% Should do this only if the level has actually changed. Possibly @@ -146,5 +155,6 @@ int_to_level(?LOG_ALL) -> all. %%%----------------------------------------------------------------- %%% Internal +table_key(proxy) -> ?PROXY_KEY; table_key(primary) -> ?PRIMARY_KEY; table_key(HandlerId) -> {?HANDLER_KEY,HandlerId}. diff --git a/lib/kernel/src/logger_disk_log_h.erl b/lib/kernel/src/logger_disk_log_h.erl index a8f141f135..47b39da900 100644 --- a/lib/kernel/src/logger_disk_log_h.erl +++ b/lib/kernel/src/logger_disk_log_h.erl @@ -19,183 +19,117 @@ %% -module(logger_disk_log_h). --behaviour(gen_server). - -include("logger.hrl"). -include("logger_internal.hrl"). -include("logger_h_common.hrl"). %%% API --export([start_link/3, info/1, filesync/1, reset/1]). +-export([filesync/1]). -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). +%% logger_h_common callbacks +-export([init/2, check_config/4, reset_state/2, + filesync/3, write/4, handle_info/3, terminate/3]). %% logger callbacks --export([log/2, adding_handler/1, removing_handler/1, changing_config/2]). - -%% handler internal --export([log_handler_info/4]). +-export([log/2, adding_handler/1, removing_handler/1, changing_config/3, + filter_config/1]). %%%=================================================================== %%% API %%%=================================================================== %%%----------------------------------------------------------------- -%%% Start a disk_log handler process and link to caller. -%%% This function is called by the kernel supervisor when this -%%% handler process gets added (as a result of calling add/3). --spec start_link(Name, Config, HandlerState) -> {ok,Pid} | {error,Reason} when - Name :: atom(), - Config :: logger:handler_config(), - HandlerState :: map(), - Pid :: pid(), - Reason :: term(). - -start_link(Name, Config, HandlerState) -> - proc_lib:start_link(?MODULE,init,[[Name,Config,HandlerState]]). - -%%%----------------------------------------------------------------- %%% -spec filesync(Name) -> ok | {error,Reason} when Name :: atom(), Reason :: handler_busy | {badarg,term()}. -filesync(Name) when is_atom(Name) -> - try - gen_server:call(?name_to_reg_name(?MODULE,Name), - disk_log_sync, ?DEFAULT_CALL_TIMEOUT) - catch - _:{timeout,_} -> {error,handler_busy} - end; filesync(Name) -> - {error,{badarg,{filesync,[Name]}}}. + logger_h_common:filesync(?MODULE,Name). + +%%%=================================================================== +%%% logger callbacks +%%%=================================================================== %%%----------------------------------------------------------------- -%%% --spec info(Name) -> Info | {error,Reason} when - Name :: atom(), - Info :: term(), - Reason :: handler_busy | {badarg,term()}. +%%% Handler being added +adding_handler(Config) -> + logger_h_common:adding_handler(Config). + +%%%----------------------------------------------------------------- +%%% Updating handler config +changing_config(SetOrUpdate, OldConfig, NewConfig) -> + logger_h_common:changing_config(SetOrUpdate, OldConfig, NewConfig). -info(Name) when is_atom(Name) -> - try - gen_server:call(?name_to_reg_name(?MODULE,Name), - info, ?DEFAULT_CALL_TIMEOUT) - catch - _:{timeout,_} -> {error,handler_busy} - end; -info(Name) -> - {error,{badarg,{info,[Name]}}}. +%%%----------------------------------------------------------------- +%%% Handler being removed +removing_handler(Config) -> + logger_h_common:removing_handler(Config). %%%----------------------------------------------------------------- -%%% --spec reset(Name) -> ok | {error,Reason} when - Name :: atom(), - Reason :: handler_busy | {badarg,term()}. +%%% Log a string or report +-spec log(LogEvent, Config) -> ok when + LogEvent :: logger:log_event(), + Config :: logger:handler_config(). -reset(Name) when is_atom(Name) -> - try - gen_server:call(?name_to_reg_name(?MODULE,Name), - reset, ?DEFAULT_CALL_TIMEOUT) - catch - _:{timeout,_} -> {error,handler_busy} - end; -reset(Name) -> - {error,{badarg,{reset,[Name]}}}. +log(LogEvent, Config) -> + logger_h_common:log(LogEvent, Config). +%%%----------------------------------------------------------------- +%%% Remove internal fields from configuration +filter_config(Config) -> + logger_h_common:filter_config(Config). %%%=================================================================== -%%% logger callbacks +%%% logger_h_common callbacks %%%=================================================================== - -%%%----------------------------------------------------------------- -%%% Handler being added -adding_handler(#{id:=Name}=Config) -> - case check_config(adding, Config) of - {ok, Config1} -> - %% create initial handler state by merging defaults with config - HConfig = maps:get(config, Config1, #{}), - HState = maps:merge(get_init_state(), HConfig), - case logger_h_common:overload_levels_ok(HState) of - true -> - start(Name, Config1, HState); - false -> - #{sync_mode_qlen := SMQL, - drop_mode_qlen := DMQL, - flush_qlen := FQL} = HState, - {error,{invalid_levels,{SMQL,DMQL,FQL}}} - end; +init(Name, #{file:=File,type:=Type,max_no_bytes:=MNB,max_no_files:=MNF}) -> + case open_disk_log(Name, File, Type, MNB, MNF) of + ok -> + {ok,#{log_opts => #{file => File, + type => Type, + max_no_bytes => MNB, + max_no_files => MNF}, + prev_log_result => ok, + prev_sync_result => ok, + prev_disk_log_info => undefined}}; Error -> Error end. -%%%----------------------------------------------------------------- -%%% Updating handler config -changing_config(OldConfig = #{id:=Name, config:=OldHConfig}, - NewConfig = #{id:=Name, config:=NewHConfig}) -> - #{type:=Type, file:=File, max_no_files:=MaxFs, - max_no_bytes:=MaxBytes} = OldHConfig, - case NewHConfig of - #{type:=Type, file:=File, max_no_files:=MaxFs, - max_no_bytes:=MaxBytes} -> - changing_config1(OldConfig, NewConfig); - _ -> - {error,{illegal_config_change,OldConfig,NewConfig}} - end; -changing_config(OldConfig, NewConfig) -> - {error,{illegal_config_change,OldConfig,NewConfig}}. - -changing_config1(OldConfig=#{config:=OldHConfig}, NewConfig) -> - case check_config(changing, NewConfig) of - {ok,NewConfig1 = #{config:=NewHConfig}} -> - #{handler_pid:=HPid, - mode_tab:=ModeTab} = OldHConfig, - NewHConfig1 = NewHConfig#{handler_pid=>HPid, - mode_tab=>ModeTab}, - NewConfig2 = NewConfig1#{config=>NewHConfig1}, - try gen_server:call(HPid, {change_config,OldConfig,NewConfig2}, - ?DEFAULT_CALL_TIMEOUT) of - ok -> {ok,NewConfig2}; - HError -> HError - catch - _:{timeout,_} -> {error,handler_busy} - end; - Error -> - Error +check_config(Name,set,undefined,HConfig0) -> + HConfig=merge_default_logopts(Name,maps:merge(get_default_config(),HConfig0)), + check_config(HConfig); +check_config(_Name,SetOrUpdate,OldHConfig,NewHConfig0) -> + WriteOnce = maps:with([type,file,max_no_files,max_no_bytes],OldHConfig), + Default = + case SetOrUpdate of + set -> + %% Do not reset write-once fields to defaults + maps:merge(get_default_config(),WriteOnce); + update -> + OldHConfig + end, + + NewHConfig = maps:merge(Default,NewHConfig0), + + %% Fail if write-once fields are changed + case maps:with([type,file,max_no_files,max_no_bytes],NewHConfig) of + WriteOnce -> + check_config(NewHConfig); + Other -> + {Old,New} = logger_server:diff_maps(WriteOnce,Other), + {error,{illegal_config_change,?MODULE,Old,New}} end. -check_config(adding, #{id:=Name}=Config) -> - %% merge handler specific config data - HConfig = merge_default_logopts(Name, maps:get(config, Config, #{})), +check_config(HConfig) -> case check_h_config(maps:to_list(HConfig)) of ok -> - {ok,Config#{config=>HConfig}}; - Error -> - Error - end; -check_config(changing, Config) -> - HConfig = maps:get(config, Config, #{}), - case check_h_config(maps:to_list(HConfig)) of - ok -> {ok,Config}; - Error -> Error + {ok,HConfig}; + {error,{Key,Value}} -> + {error,{invalid_config,?MODULE,#{Key=>Value}}} end. -merge_default_logopts(Name, HConfig) -> - Type = maps:get(type, HConfig, wrap), - {DefaultNoFiles,DefaultNoBytes} = - case Type of - halt -> {undefined,infinity}; - _wrap -> {10,1048576} - end, - {ok,Dir} = file:get_cwd(), - Defaults = #{file => filename:join(Dir,Name), - max_no_files => DefaultNoFiles, - max_no_bytes => DefaultNoBytes, - type => Type}, - maps:merge(Defaults, HConfig). - check_h_config([{file,File}|Config]) when is_list(File) -> check_h_config(Config); check_h_config([{max_no_files,undefined}|Config]) -> @@ -208,447 +142,59 @@ check_h_config([{max_no_bytes,N}|Config]) when is_integer(N), N>0 -> check_h_config(Config); check_h_config([{type,Type}|Config]) when Type==wrap; Type==halt -> check_h_config(Config); -check_h_config([Other | Config]) -> - case logger_h_common:check_common_config(Other) of - valid -> - check_h_config(Config); - invalid -> - {error,{invalid_config,?MODULE,Other}} - end; +check_h_config([Other | _]) -> + {error,Other}; check_h_config([]) -> ok. -%%%----------------------------------------------------------------- -%%% Handler being removed -removing_handler(#{id:=Name}) -> - stop(Name). - -%%%----------------------------------------------------------------- -%%% Log a string or report --spec log(LogEvent, Config) -> ok when - LogEvent :: logger:log_event(), - Config :: logger:handler_config(). +get_default_config() -> + #{}. -log(LogEvent, Config = #{id := Name, - config := #{handler_pid := HPid, - mode_tab := ModeTab}}) -> - %% if the handler has crashed, we must drop this event - %% and hope the handler restarts so we can try again - true = is_process_alive(HPid), - Bin = logger_h_common:log_to_binary(LogEvent, Config), - logger_h_common:call_cast_or_drop(Name, HPid, ModeTab, Bin). +merge_default_logopts(Name, HConfig) -> + Type = maps:get(type, HConfig, wrap), + {DefaultNoFiles,DefaultNoBytes} = + case Type of + halt -> {undefined,infinity}; + _wrap -> {10,1048576} + end, + {ok,Dir} = file:get_cwd(), + Defaults = #{file => filename:join(Dir,Name), + max_no_files => DefaultNoFiles, + max_no_bytes => DefaultNoBytes, + type => Type}, + maps:merge(Defaults, HConfig). -%%%=================================================================== -%%% gen_server callbacks -%%%=================================================================== +filesync(Name,_Mode,State) -> + Result = ?disk_log_sync(Name), + maybe_notify_error(Name, filesync, Result, prev_sync_result, State). -init([Name, - Config = #{config := HConfig = #{file:=File, - type:=Type, - max_no_bytes:=MNB, - max_no_files:=MNF}}, - State = #{dl_sync_int := DLSyncInt}]) -> - - RegName = ?name_to_reg_name(?MODULE,Name), - register(RegName, self()), - process_flag(trap_exit, true), - process_flag(message_queue_data, off_heap), - - ?init_test_hooks(), - ?start_observation(Name), - - LogOpts = #{file=>File, type=>Type, max_no_bytes=>MNB, max_no_files=>MNF}, - case open_disk_log(Name, File, Type, MNB, MNF) of - ok -> - try ets:new(Name, [public]) of - ModeTab -> - ?set_mode(ModeTab, async), - T0 = ?timestamp(), - State1 = - ?merge_with_stats(State#{ - id => Name, - mode_tab => ModeTab, - mode => async, - dl_sync => DLSyncInt, - log_opts => LogOpts, - last_qlen => 0, - last_log_ts => T0, - burst_win_ts => T0, - burst_msg_count => 0, - last_op => sync, - prev_log_result => ok, - prev_sync_result => ok, - prev_disk_log_info => undefined}), - Config1 = - Config#{config => HConfig#{handler_pid => self(), - mode_tab => ModeTab}}, - proc_lib:init_ack({ok,self(),Config1}), - gen_server:cast(self(), repeated_disk_log_sync), - case logger_h_common:unset_restart_flag(Name, ?MODULE) of - true -> - %% inform about restart - gen_server:cast(self(), {log_handler_info, - "Handler ~p restarted", - [Name]}); - false -> - %% initial start - ok - end, - gen_server:enter_loop(?MODULE, [], State1) - catch - _:Error -> - unregister(RegName), - logger_h_common:error_notify({open_disk_log,Name,Error}), - proc_lib:init_ack(Error) - end; - Error -> - unregister(RegName), - logger_h_common:error_notify({open_disk_log,Name,Error}), - proc_lib:init_ack(Error) - end. +write(Name, Mode, Bin, State) -> + Result = ?disk_log_write(Name, Mode, Bin), + maybe_notify_error(Name, log, Result, prev_log_result, State). -%% This is the synchronous log event. -handle_call({log, Bin}, _From, State) -> - {Result,State1} = do_log(Bin, call, State), - %% Result == ok | dropped - {reply, Result, State1}; - -handle_call(disk_log_sync, _From, State = #{id := Name}) -> - State1 = #{prev_sync_result := Result} = disk_log_sync(Name, State), - {reply, Result, State1}; - -handle_call({change_config,_OldConfig,NewConfig}, _From, - State = #{filesync_repeat_interval := FSyncInt0}) -> - HConfig = maps:get(config, NewConfig, #{}), - State1 = #{sync_mode_qlen := SMQL, - drop_mode_qlen := DMQL, - flush_qlen := FQL} = maps:merge(State, HConfig), - case logger_h_common:overload_levels_ok(State1) of - true -> - _ = - case maps:get(filesync_repeat_interval, HConfig, undefined) of - undefined -> - ok; - no_repeat -> - _ = logger_h_common:cancel_timer(maps:get(rep_sync_tref, - State, - undefined)); - FSyncInt0 -> - ok; - _FSyncInt1 -> - _ = logger_h_common:cancel_timer(maps:get(rep_sync_tref, - State, - undefined)), - _ = gen_server:cast(self(), repeated_disk_log_sync) - end, - {reply, ok, State1}; - false -> - {reply, {error,{invalid_levels,{SMQL,DMQL,FQL}}}, State} - end; - -handle_call(info, _From, State) -> - {reply, State, State}; - -handle_call(reset, _From, State) -> - State1 = ?merge_with_stats(State), - {reply, ok, State1#{last_qlen => 0, - last_log_ts => ?timestamp(), - prev_log_result => ok, - prev_sync_result => ok, - prev_disk_log_info => undefined}}; - -handle_call(stop, _From, State) -> - {stop, {shutdown,stopped}, ok, State}. - - -%% This is the asynchronous log event. -handle_cast({log, Bin}, State) -> - {_,State1} = do_log(Bin, cast, State), - {noreply, State1}; - -handle_cast({log_handler_info, Format, Args}, State = #{id:=Name}) -> - log_handler_info(Name, Format, Args, State), - {noreply, State}; - -%% If FILESYNC_REPEAT_INTERVAL is set to a millisec value, this -%% clause gets called repeatedly by the handler. In order to -%% guarantee that a filesync *always* happens after the last log -%% event, the repeat operation must be active! -handle_cast(repeated_disk_log_sync, - State = #{id := Name, - filesync_repeat_interval := FSyncInt, - last_op := LastOp}) -> - State1 = - if is_integer(FSyncInt) -> - %% only do filesync if something has been - %% written since last time we checked - NewState = if LastOp == sync -> - State; - true -> - disk_log_sync(Name, State) - end, - {ok,TRef} = - timer:apply_after(FSyncInt, gen_server,cast, - [self(),repeated_disk_log_sync]), - NewState#{rep_sync_tref => TRef, last_op => sync}; - true -> - State - end, - {noreply,State1}. +reset_state(_Name, State) -> + State#{prev_log_result => ok, + prev_sync_result => ok, + prev_disk_log_info => undefined}. %% The disk log owner must handle status messages from disk_log. -handle_info({disk_log, _Node, _Log, {wrap,_NoLostItems}}, State) -> - {noreply, State}; -handle_info({disk_log, _Node, Log, Info = {truncated,_NoLostItems}}, - State = #{id := Name, prev_disk_log_info := PrevInfo}) -> - error_notify_new(Info, PrevInfo, {disk_log,Name,Log,Info}), - {noreply, State#{prev_disk_log_info => Info}}; -handle_info({disk_log, _Node, Log, Info = {blocked_log,_Items}}, - State = #{id := Name, prev_disk_log_info := PrevInfo}) -> - error_notify_new(Info, PrevInfo, {disk_log,Name,Log,Info}), - {noreply, State#{prev_disk_log_info => Info}}; -handle_info({disk_log, _Node, Log, full}, - State = #{id := Name, prev_disk_log_info := PrevInfo}) -> - error_notify_new(full, PrevInfo, {disk_log,Name,Log,full}), - {noreply, State#{prev_disk_log_info => full}}; -handle_info({disk_log, _Node, Log, Info = {error_status,_Status}}, - State = #{id := Name, prev_disk_log_info := PrevInfo}) -> - error_notify_new(Info, PrevInfo, {disk_log,Name,Log,Info}), - {noreply, State#{prev_disk_log_info => Info}}; - -handle_info({'EXIT',_Pid,_Why}, State = #{id := _Name}) -> - {noreply, State}; - -handle_info(_, State) -> - {noreply, State}. - -terminate(Reason, State = #{id := Name}) -> - _ = logger_h_common:cancel_timer(maps:get(rep_sync_tref, State, - undefined)), +handle_info(Name, {disk_log, _Node, Log, Info={truncated,_NoLostItems}}, State) -> + maybe_notify_status(Name, Log, Info, prev_disk_log_info, State); +handle_info(Name, {disk_log, _Node, Log, Info = {blocked_log,_Items}}, State) -> + maybe_notify_status(Name, Log, Info, prev_disk_log_info, State); +handle_info(Name, {disk_log, _Node, Log, Info = full}, State) -> + maybe_notify_status(Name, Log, Info, prev_disk_log_info, State); +handle_info(Name, {disk_log, _Node, Log, Info = {error_status,_Status}}, State) -> + maybe_notify_status(Name, Log, Info, prev_disk_log_info, State); +handle_info(_, _, State) -> + State. + +terminate(Name, _Reason, _State) -> _ = close_disk_log(Name, normal), - ok = logger_h_common:stop_or_restart(Name, Reason, State), - unregister(?name_to_reg_name(?MODULE, Name)), ok. -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - %%%----------------------------------------------------------------- %%% Internal functions - -%%%----------------------------------------------------------------- -%%% -get_init_state() -> - #{sync_mode_qlen => ?SYNC_MODE_QLEN, - drop_mode_qlen => ?DROP_MODE_QLEN, - flush_qlen => ?FLUSH_QLEN, - burst_limit_enable => ?BURST_LIMIT_ENABLE, - burst_limit_max_count => ?BURST_LIMIT_MAX_COUNT, - burst_limit_window_time => ?BURST_LIMIT_WINDOW_TIME, - overload_kill_enable => ?OVERLOAD_KILL_ENABLE, - overload_kill_qlen => ?OVERLOAD_KILL_QLEN, - overload_kill_mem_size => ?OVERLOAD_KILL_MEM_SIZE, - overload_kill_restart_after => ?OVERLOAD_KILL_RESTART_AFTER, - dl_sync_int => ?CONTROLLER_SYNC_INTERVAL, - filesync_ok_qlen => ?FILESYNC_OK_QLEN, - filesync_repeat_interval => ?FILESYNC_REPEAT_INTERVAL}. - -%%%----------------------------------------------------------------- -%%% Add a disk_log handler to the logger. -%%% This starts a dedicated handler process which should always -%%% exist if the handler is registered with logger (and should not -%%% exist if the handler is not registered). -%%% -%%% Config is the logger:handler_config() map. Handler specific parameters -%%% should be provided with a sub map associated with a key named -%%% 'config', e.g: -%%% -%%% Config = #{config => #{sync_mode_qlen => 50} -%%% -%%% The 'config' sub map will also contain parameters for configuring -%%% the disk_log: -%%% -%%% Config = #{config => #{file => file:filename(), -%%% max_no_bytes => integer(), -%%% max_no_files => integer(), -%%% type => wrap | halt}}. -%%% -%%% If type == halt, then max_no_files is ignored. -%%% -%%% The disk_log handler process is linked to logger_sup, which is -%%% part of the kernel application's supervision tree. -start(Name, Config, HandlerState) -> - LoggerDLH = - #{id => Name, - start => {?MODULE, start_link, [Name,Config,HandlerState]}, - restart => temporary, - shutdown => 2000, - type => worker, - modules => [?MODULE]}, - case supervisor:start_child(logger_sup, LoggerDLH) of - {ok,Pid,Config1} -> - ok = logger_handler_watcher:register_handler(Name,Pid), - {ok,Config1}; - Error -> - Error - end. - -%%%----------------------------------------------------------------- -%%% Stop and remove the handler. -stop(Name) -> - case whereis(?name_to_reg_name(?MODULE,Name)) of - undefined -> - ok; - Pid -> - %% We don't want to do supervisor:terminate_child here - %% since we need to distinguish this explicit stop from a - %% system termination in order to avoid circular attempts - %% at removing the handler (implying deadlocks and - %% timeouts). - %% And we don't need to do supervisor:delete_child, since - %% the restart type is temporary, which means that the - %% child specification is automatically removed from the - %% supervisor when the process dies. - _ = gen_server:call(Pid, stop), - ok - end. - -%%%----------------------------------------------------------------- -%%% Logging and overload control. --define(update_dl_sync(C, Interval), - if C == 0 -> Interval; - true -> C-1 end). - -%% check for overload between every event (and set Mode to async, -%% sync or drop accordingly), but never flush the whole mailbox -%% before LogWindowSize events have been handled -do_log(Bin, CallOrCast, State = #{id:=Name, mode := Mode0}) -> - T1 = ?timestamp(), - - %% check if the handler 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} = logger_h_common:check_load(State), - - if (Mode1 == drop) andalso (Mode0 =/= drop) -> - log_handler_info(Name, "Handler ~p switched to drop mode", - [Name], State); - (Mode0 == drop) andalso ((Mode1 == async) orelse (Mode1 == sync)) -> - log_handler_info(Name, "Handler ~p switched to ~w mode", - [Name,Mode1], State); - true -> - ok - end, - - %% kill the handler if it can't keep up with the load - logger_h_common:kill_if_choked(Name, QLen, Mem, ?MODULE, State), - - if Mode1 == flush -> - flush(Name, QLen, T1, State1); - true -> - write(Name, Mode1, T1, Bin, CallOrCast, State1) - end. - -%% this function is called by do_log/3 after an overload check -%% has been performed, where QLen > FlushQLen -flush(Name, _QLen0, T1, State=#{last_log_ts := _T0, mode_tab := ModeTab}) -> - %% flush messages in the mailbox (a limited number in - %% order to not cause long delays) - NewFlushed = logger_h_common:flush_log_events(?FLUSH_MAX_N), - - %% write info in log about flushed messages - log_handler_info(Name, "Handler ~p flushed ~w log events", - [Name,NewFlushed], State), - - %% 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}), - - %% Add 1 for the current log event - ?observe(Name,{flushed,NewFlushed+1}), - - State1 = ?update_max_time(?diff_time(T1,_T0),State), - {dropped,?update_other(flushed,FLUSHED,NewFlushed, - State1#{mode => ?set_mode(ModeTab,async), - last_qlen => 0, - last_log_ts => T1})}. - -%% this function is called to write to disk_log -write(Name, Mode, T1, Bin, _CallOrCast, - State = #{mode_tab := ModeTab, - dl_sync := DLSync, - dl_sync_int := DLSyncInt, - last_qlen := LastQLen, - last_log_ts := T0}) -> - %% check if we need to limit the number of writes - %% during a burst of log events - {DoWrite,BurstWinT,BurstMsgCount} = logger_h_common:limit_burst(State), - - %% only send a synhrounous event to the disk_log process - %% every DLSyncInt time, to give the handler time between - %% writes so it can keep up with incoming messages - {Status,LastQLen1,State1} = - if DoWrite, DLSync == 0 -> - ?observe(Name,{_CallOrCast,1}), - NewState = disk_log_write(Name, Bin, State), - {ok, element(2,process_info(self(),message_queue_len)), - NewState}; - DoWrite -> - ?observe(Name,{_CallOrCast,1}), - NewState = disk_log_write(Name, Bin, State), - {ok, LastQLen, NewState}; - not DoWrite -> - ?observe(Name,{flushed,1}), - {dropped, LastQLen, State} - end, - - %% Check if the time since the previous log event is long enough - - %% and the queue length small enough - to assume the mailbox has - %% been emptied, and if so, do filesync operation and reset mode to - %% async. Note that this is the best we can do to detect an idle - %% handler without setting a timer after each log call/cast. If the - %% time between two consecutive log events is fast and no new - %% event comes in after the last one, idle state won't be detected! - Time = ?diff_time(T1,T0), - {Mode1,BurstMsgCount1,State2} = - if (LastQLen1 < ?FILESYNC_OK_QLEN) andalso - (Time > ?IDLE_DETECT_TIME_USEC) -> - {?change_mode(ModeTab,Mode,async), 0, disk_log_sync(Name,State1)}; - true -> - {Mode, BurstMsgCount,State1} - end, - - State3 = - ?update_calls_or_casts(_CallOrCast,1,State2), - State4 = - ?update_max_time(Time, - State3#{mode => Mode1, - last_qlen := LastQLen1, - last_log_ts => T1, - burst_win_ts => BurstWinT, - burst_msg_count => BurstMsgCount1, - dl_sync => ?update_dl_sync(DLSync,DLSyncInt)}), - {Status,State4}. - - -log_handler_info(Name, Format, Args, State) -> - Config = - case logger:get_handler_config(Name) of - {ok,Conf} -> Conf; - _ -> #{formatter=>{?DEFAULT_FORMATTER,?DEFAULT_FORMAT_CONFIG}} - end, - Meta = #{time=>erlang:system_time(microsecond)}, - Bin = logger_h_common:log_to_binary(#{level => notice, - msg => {Format,Args}, - meta => Meta}, Config), - _ = disk_log_write(Name, Bin, State), - ok. - - open_disk_log(Name, File, Type, MaxNoBytes, MaxNoFiles) -> case filelib:ensure_dir(File) of ok -> @@ -681,43 +227,26 @@ close_disk_log(Name, _) -> _ = disk_log:lclose(Name), ok. -disk_log_write(Name, Bin, State) -> - case ?disk_log_blog(Name, Bin) of - ok -> - State#{prev_log_result => ok, last_op => write}; - LogError -> - _ = case maps:get(prev_log_result, State) of - LogError -> - %% don't report same error twice - ok; - _ -> - LogOpts = maps:get(log_opts, State), - logger_h_common:error_notify({Name,log, - LogOpts, - LogError}) - end, - State#{prev_log_result => LogError} - end. - -disk_log_sync(Name, State) -> - case ?disk_log_sync(Name) of - ok -> - State#{prev_sync_result => ok, last_op => sync}; - SyncError -> - _ = case maps:get(prev_sync_result, State) of - SyncError -> - %% don't report same error twice - ok; - _ -> - LogOpts = maps:get(log_opts, State), - logger_h_common:error_notify({Name,filesync, - LogOpts, - SyncError}) - end, - State#{prev_sync_result => SyncError} - end. +disk_log_write(Name, sync, Bin) -> + disk_log:blog(Name, Bin); +disk_log_write(Name, async, Bin) -> + disk_log:balog(Name, Bin). + +%%%----------------------------------------------------------------- +%%% Print error messages, but don't repeat the same message +maybe_notify_error(Name, Op, Result, Key, #{log_opts:=LogOpts}=State) -> + {Result,error_notify_new({Name, Op, LogOpts, Result}, Result, Key, State)}. -error_notify_new(Info,Info, _Term) -> +maybe_notify_status(Name, Log, Info, Key, State) -> + error_notify_new({disk_log, Name, Log, Info}, Info, Key, State). + +error_notify_new(Term, What, Key, State) -> + error_notify_new(What, maps:get(Key,State), Term), + State#{Key => What}. + +error_notify_new(ok,_Prev,_Term) -> + ok; +error_notify_new(Same,Same,_Term) -> ok; -error_notify_new(_Info0,_Info1, Term) -> +error_notify_new(_New,_Prev,Term) -> logger_h_common:error_notify(Term). diff --git a/lib/kernel/src/logger_h_common.erl b/lib/kernel/src/logger_h_common.erl index 38ac7d8ffc..e69f6de38d 100644 --- a/lib/kernel/src/logger_h_common.erl +++ b/lib/kernel/src/logger_h_common.erl @@ -18,26 +18,345 @@ %% %CopyrightEnd% %% -module(logger_h_common). +-behaviour(gen_server). -include("logger_h_common.hrl"). -include("logger_internal.hrl"). --export([log_to_binary/2, - check_common_config/1, - call_cast_or_drop/4, - check_load/1, - limit_burst/1, - kill_if_choked/5, - flush_log_events/0, - flush_log_events/1, - handler_exit/2, - set_restart_flag/2, - unset_restart_flag/2, - cancel_timer/1, - stop_or_restart/3, - overload_levels_ok/1, - error_notify/1, - info_notify/1]). +%% API +-export([filesync/2]). + +%% logger_olp callbacks +-export([init/1, handle_load/2, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3, notify/2, reset_state/1]). + +%% logger callbacks +-export([log/2, adding_handler/1, removing_handler/1, changing_config/3, + filter_config/1]). + +%% Library functions for handlers +-export([error_notify/1]). + +-define(OLP_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(COMMON_KEYS,[filesync_repeat_interval]). + +-define(READ_ONLY_KEYS,[olp]). + +%%%----------------------------------------------------------------- +%%% API + +%% This function is called by the logger_sup supervisor +filesync(Module, Name) -> + call(Module, Name, filesync). + +%%%----------------------------------------------------------------- +%%% Handler being added +adding_handler(#{id:=Name,module:=Module}=Config) -> + HConfig0 = maps:get(config, Config, #{}), + HandlerConfig0 = maps:without(?OLP_KEYS++?COMMON_KEYS,HConfig0), + case Module:check_config(Name,set,undefined,HandlerConfig0) of + {ok,HandlerConfig} -> + ModifiedCommon = maps:with(?COMMON_KEYS,HandlerConfig), + CommonConfig0 = maps:with(?COMMON_KEYS,HConfig0), + CommonConfig = maps:merge( + maps:merge(get_default_config(), CommonConfig0), + ModifiedCommon), + case check_config(CommonConfig) of + ok -> + HConfig = maps:merge(CommonConfig,HandlerConfig), + OlpOpts = maps:with(?OLP_KEYS,HConfig0), + start(OlpOpts, Config#{config => HConfig}); + {error,Faulty} -> + {error,{invalid_config,Module,Faulty}} + end; + Error -> + Error + end. + +%%%----------------------------------------------------------------- +%%% Handler being removed +removing_handler(#{id:=Name, module:=Module, config:=#{olp:=Olp}}) -> + case whereis(?name_to_reg_name(Module,Name)) of + undefined -> + ok; + _Pid -> + %% We don't want to do supervisor:terminate_child here + %% since we need to distinguish this explicit stop from a + %% system termination in order to avoid circular attempts + %% at removing the handler (implying deadlocks and + %% timeouts). + %% And we don't need to do supervisor:delete_child, since + %% the restart type is temporary, which means that the + %% child specification is automatically removed from the + %% supervisor when the process dies. + _ = logger_olp:stop(Olp), + ok + end. + +%%%----------------------------------------------------------------- +%%% Updating handler config +changing_config(SetOrUpdate, + #{id:=Name,config:=OldHConfig,module:=Module}, + NewConfig0) -> + NewHConfig0 = maps:get(config, NewConfig0, #{}), + NoHandlerKeys = ?OLP_KEYS++?COMMON_KEYS++?READ_ONLY_KEYS, + OldHandlerConfig = maps:without(NoHandlerKeys,OldHConfig), + NewHandlerConfig0 = maps:without(NoHandlerKeys,NewHConfig0), + case Module:check_config(Name, SetOrUpdate, + OldHandlerConfig,NewHandlerConfig0) of + {ok, NewHandlerConfig} -> + ModifiedCommon = maps:with(?COMMON_KEYS,NewHandlerConfig), + NewCommonConfig0 = maps:with(?COMMON_KEYS,NewHConfig0), + OldCommonConfig = maps:with(?COMMON_KEYS,OldHConfig), + CommonDefault = + case SetOrUpdate of + set -> + get_default_config(); + update -> + OldCommonConfig + end, + NewCommonConfig = maps:merge( + maps:merge(CommonDefault,NewCommonConfig0), + ModifiedCommon), + case check_config(NewCommonConfig) of + ok -> + OlpDefault = + case SetOrUpdate of + set -> + logger_olp:get_default_opts(); + update -> + maps:with(?OLP_KEYS,OldHConfig) + end, + Olp = maps:get(olp,OldHConfig), + NewOlpOpts = maps:merge(OlpDefault, + maps:with(?OLP_KEYS,NewHConfig0)), + case logger_olp:set_opts(Olp,NewOlpOpts) of + ok -> + maybe_set_repeated_filesync(Olp,OldCommonConfig, + NewCommonConfig), + ReadOnly = maps:with(?READ_ONLY_KEYS,OldHConfig), + NewHConfig = + maps:merge( + maps:merge( + maps:merge(NewCommonConfig,NewHandlerConfig), + ReadOnly), + NewOlpOpts), + NewConfig = NewConfig0#{config=>NewHConfig}, + {ok,NewConfig}; + Error -> + Error + end; + {error,Faulty} -> + {error,{invalid_config,Module,Faulty}} + end; + Error -> + Error + end. + +%%%----------------------------------------------------------------- +%%% Log a string or report +-spec log(LogEvent, Config) -> ok when + LogEvent :: logger:log_event(), + Config :: logger:handler_config(). + +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 = is_process_alive(logger_olp:get_pid(Olp)), + Bin = log_to_binary(LogEvent, Config), + logger_olp:load(Olp,Bin). + +%%%----------------------------------------------------------------- +%%% Remove internal fields from configuration +filter_config(#{config:=HConfig}=Config) -> + Config#{config=>maps:without(?READ_ONLY_KEYS,HConfig)}. + +%%%----------------------------------------------------------------- +%%% Start the handler process +%%% +%%% The process must always exist if the handler is registered with +%%% logger (and must not exist if the handler is not registered). +%%% +%%% The handler process is linked to logger_sup, which is part of the +%%% kernel application's supervision tree. +start(OlpOpts0, #{id := Name, module:=Module, config:=HConfig} = Config0) -> + RegName = ?name_to_reg_name(Module,Name), + ChildSpec = + #{id => Name, + start => {logger_olp, start_link, [RegName,?MODULE, + Config0, OlpOpts0]}, + restart => temporary, + shutdown => 2000, + type => worker, + modules => [?MODULE]}, + case supervisor:start_child(logger_sup, ChildSpec) of + {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}; + Error -> + Error + end. + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +init(#{id := Name, module := Module, config := HConfig}) -> + process_flag(trap_exit, true), + + ?init_test_hooks(), + + case Module:init(Name, HConfig) of + {ok,HState} -> + %% Storing common config in state to avoid copying + %% (sending) the config data for each log message + CommonConfig = maps:with(?COMMON_KEYS,HConfig), + State = CommonConfig#{id => Name, + module => Module, + ctrl_sync_count => + ?CONTROLLER_SYNC_INTERVAL, + last_op => sync, + handler_state => HState}, + State1 = set_repeated_filesync(State), + {ok,State1}; + Error -> + Error + end. + +%% This is the log event. +handle_load(Bin, #{id:=Name, + module:=Module, + handler_state:=HandlerState, + ctrl_sync_count := CtrlSync}=State) -> + if CtrlSync==0 -> + {_,HS1} = Module:write(Name, sync, Bin, HandlerState), + State#{handler_state => HS1, + ctrl_sync_count => ?CONTROLLER_SYNC_INTERVAL, + last_op=>write}; + true -> + {_,HS1} = Module:write(Name, async, Bin, HandlerState), + State#{handler_state => HS1, + ctrl_sync_count => CtrlSync-1, + last_op=>write} + end. + +handle_call(filesync, _From, State = #{id := Name, + module := Module, + handler_state := HandlerState}) -> + {Result,HandlerState1} = Module:filesync(Name,sync,HandlerState), + {reply, Result, State#{handler_state=>HandlerState1, last_op=>sync}}. + +%% If FILESYNC_REPEAT_INTERVAL is set to a millisec value, this +%% clause gets called repeatedly by the handler. In order to +%% guarantee that a filesync *always* happens after the last log +%% event, the repeat operation must be active! +handle_cast(repeated_filesync,State = #{filesync_repeat_interval := no_repeat}) -> + %% This clause handles a race condition which may occur when + %% config changes filesync_repeat_interval from an integer value + %% to no_repeat. + {noreply,State}; +handle_cast(repeated_filesync, + State = #{id := Name, + module := Module, + handler_state := HandlerState, + last_op := LastOp}) -> + State1 = + if LastOp == sync -> + State; + true -> + {_,HS} = Module:filesync(Name, async, HandlerState), + State#{handler_state => HS, last_op => sync} + end, + {noreply,set_repeated_filesync(State1)}; + +handle_cast({set_repeated_filesync,FSyncInt},State) -> + State1 = State#{filesync_repeat_interval=>FSyncInt}, + State2 = set_repeated_filesync(cancel_repeated_filesync(State1)), + {noreply, State2}. + +handle_info(Info, #{id := Name, module := Module, + handler_state := HandlerState} = State) -> + {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), + do_terminate(Reason,State), + ConfigResult = logger:get_handler_config(Name), + case ConfigResult of + {ok,#{module:=Module}=HConfig0} -> + spawn(fun() -> logger:remove_handler(Name) end), + HConfig = try Module:filter_config(HConfig0) + catch _:_ -> HConfig0 + end, + {ok,fun() -> logger:add_handler(Name,Module,HConfig) end}; + Error -> + error_notify({Name,restart_impossible,Error}), + Error + end; +terminate(Reason, State) -> + do_terminate(Reason, State). + +do_terminate(Reason, State = #{id := Name, + module := Module, + handler_state := HandlerState}) -> + _ = cancel_repeated_filesync(State), + _ = Module:terminate(Name, Reason, HandlerState), + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +reset_state(#{id:=Name, module:=Module, handler_state:=HandlerState} = State) -> + State#{handler_state=>Module:reset_state(Name, HandlerState)}. + +%%%----------------------------------------------------------------- +%%% Internal functions +call(Module, Name, Op) when is_atom(Name) -> + case logger_olp:call(?name_to_reg_name(Module,Name), Op) of + {error,busy} -> {error,handler_busy}; + Other -> Other + end; +call(_, Name, Op) -> + {error,{badarg,{Op,[Name]}}}. + +notify({mode_change,Mode0,Mode1},#{id:=Name}=State) -> + log_handler_info(Name,"Handler ~p switched from ~p to ~p mode", + [Name,Mode0,Mode1], State); +notify({flushed,Flushed},#{id:=Name}=State) -> + log_handler_info(Name, "Handler ~p flushed ~w log events", + [Name,Flushed], State); +notify(restart,#{id:=Name}=State) -> + log_handler_info(Name, "Handler ~p restarted", [Name], State); +notify(idle,#{id:=Name,module:=Module,handler_state:=HandlerState}=State) -> + {_,HS} = Module:filesync(Name,async,HandlerState), + State#{handler_state=>HS, last_op=>sync}. + +log_handler_info(Name, Format, Args, #{module:=Module, + handler_state:=HandlerState}=State) -> + Config = + case logger:get_handler_config(Name) of + {ok,Conf} -> Conf; + _ -> #{formatter=>{?DEFAULT_FORMATTER,?DEFAULT_FORMAT_CONFIG}} + end, + Meta = #{time=>erlang:system_time(microsecond)}, + Bin = log_to_binary(#{level => notice, + msg => {Format,Args}, + meta => Meta}, Config), + {_,HS} = Module:write(Name, async, Bin, HandlerState), + State#{handler_state=>HS, last_op=>write}. %%%----------------------------------------------------------------- %%% Convert log data on any form to binary @@ -94,243 +413,44 @@ string_to_binary(String) -> throw(Error) end. - %%%----------------------------------------------------------------- %%% Check that the configuration term is valid -check_common_config({mode_tab,_Tid}) -> - valid; -check_common_config({handler_pid,Pid}) when is_pid(Pid) -> - valid; - -check_common_config({sync_mode_qlen,N}) when is_integer(N) -> - valid; -check_common_config({drop_mode_qlen,N}) when is_integer(N) -> - valid; -check_common_config({flush_qlen,N}) when is_integer(N) -> - valid; - -check_common_config({burst_limit_enable,Bool}) when Bool == true; - Bool == false -> - valid; -check_common_config({burst_limit_max_count,N}) when is_integer(N) -> - valid; -check_common_config({burst_limit_window_time,N}) when is_integer(N) -> - valid; - -check_common_config({overload_kill_enable,Bool}) when Bool == true; - Bool == false -> - valid; -check_common_config({overload_kill_qlen,N}) when is_integer(N) -> - valid; -check_common_config({overload_kill_mem_size,N}) when is_integer(N) -> - valid; -check_common_config({overload_kill_restart_after,NorA}) when is_integer(NorA); - NorA == infinity -> - valid; - -check_common_config({filesync_repeat_interval,NorA}) when is_integer(NorA); - NorA == no_repeat -> - valid; -check_common_config(_) -> - invalid. - +check_config(Config) when is_map(Config) -> + check_common_config(maps:to_list(Config)). -%%%----------------------------------------------------------------- -%%% Overload Protection -call_cast_or_drop(_Name, HandlerPid, ModeTab, Bin) -> - %% If the handler process is getting overloaded, the log event - %% will be synchronous instead of asynchronous (slows down the - %% logging tempo of a process doing lots of logging. If the - %% handler is choked, drop mode is set and no event will be sent. - try ?get_mode(ModeTab) of - async -> - gen_server:cast(HandlerPid, {log,Bin}); - sync -> - try gen_server:call(HandlerPid, {log,Bin}, ?DEFAULT_CALL_TIMEOUT) of - %% if return value from call == dropped, the - %% message has been flushed by handler and should - %% therefore not be counted as dropped in stats - ok -> ok; - dropped -> ok - catch - _:{timeout,_} -> - ?observe(_Name,{dropped,1}) - end; - drop -> - ?observe(_Name,{dropped,1}) - catch - %% if the ETS table doesn't exist (maybe because of a - %% handler restart), we can only drop the event - _:_ -> ?observe(_Name,{dropped,1}) - end, +check_common_config([{filesync_repeat_interval,NorA}|Config]) + when is_integer(NorA); NorA == no_repeat -> + check_common_config(Config); +check_common_config([{Key,Value}|_]) -> + {error,#{Key=>Value}}; +check_common_config([]) -> ok. -handler_exit(_Name, Reason) -> - exit(Reason). +get_default_config() -> + #{filesync_repeat_interval => ?FILESYNC_REPEAT_INTERVAL}. -set_restart_flag(Name, Module) -> - Flag = list_to_atom(lists:concat([Module,"_",Name,"_restarting"])), - spawn(fun() -> - register(Flag, self()), - timer:sleep(infinity) - end), - ok. +set_repeated_filesync(#{filesync_repeat_interval:=FSyncInt} = State) + when is_integer(FSyncInt) -> + {ok,TRef} = timer:apply_after(FSyncInt, gen_server, cast, + [self(),repeated_filesync]), + State#{rep_sync_tref=>TRef}; +set_repeated_filesync(State) -> + State. -unset_restart_flag(Name, Module) -> - Flag = list_to_atom(lists:concat([Module,"_",Name,"_restarting"])), - case whereis(Flag) of - undefined -> - false; - Pid -> - exit(Pid, kill), - true +cancel_repeated_filesync(State) -> + case maps:take(rep_sync_tref,State) of + {TRef,State1} -> + _ = timer:cancel(TRef), + State1; + error -> + State end. - -check_load(State = #{id:=_Name, mode_tab := ModeTab, mode := Mode, - sync_mode_qlen := SyncModeQLen, - drop_mode_qlen := DropModeQLen, - flush_qlen := FlushQLen}) -> - {_,Mem} = process_info(self(), memory), - ?observe(_Name,{max_mem,Mem}), - {_,QLen} = process_info(self(), message_queue_len), - ?observe(_Name,{max_qlen,QLen}), - %% When the handler process gets scheduled in, it's impossible - %% to predict the QLen. We could jump "up" arbitrarily from say - %% async to sync, async to drop, sync to flush, etc. However, when - %% the handler process manages the log events (without flushing), - %% one after the other, we will move "down" from drop to sync and - %% from sync to async. This way we don't risk getting stuck in - %% drop or sync mode with an empty mailbox. - {Mode1,_NewDrops,_NewFlushes} = - if - QLen >= FlushQLen -> - {flush, 0,1}; - QLen >= DropModeQLen -> - %% Note that drop mode will force log events to - %% be dropped on the client side (never sent get to - %% the handler). - IncDrops = if Mode == drop -> 0; true -> 1 end, - {?change_mode(ModeTab, Mode, drop), IncDrops,0}; - QLen >= SyncModeQLen -> - {?change_mode(ModeTab, Mode, sync), 0,0}; - true -> - {?change_mode(ModeTab, Mode, async), 0,0} - end, - State1 = ?update_other(drops,DROPS,_NewDrops,State), - {Mode1, QLen, Mem, - ?update_other(flushes,FLUSHES,_NewFlushes, - State1#{last_qlen => QLen})}. - -limit_burst(#{burst_limit_enable := false}) -> - {true,0,0}; -limit_burst(#{burst_win_ts := BurstWinT0, - burst_msg_count := BurstMsgCount, - burst_limit_window_time := BurstLimitWinTime, - burst_limit_max_count := BurstLimitMaxCnt}) -> - if (BurstMsgCount >= BurstLimitMaxCnt) -> - %% the limit for allowed messages has been reached - BurstWinT1 = ?timestamp(), - case ?diff_time(BurstWinT1,BurstWinT0) of - BurstCheckTime when BurstCheckTime < (BurstLimitWinTime*1000) -> - %% we're still within the burst time frame - {false,BurstWinT0,BurstMsgCount}; - _BurstCheckTime -> - %% burst time frame passed, reset counters - {true,BurstWinT1,0} - end; - true -> - %% the limit for allowed messages not yet reached - {true,BurstWinT0,BurstMsgCount+1} - end. - -kill_if_choked(Name, QLen, Mem, HandlerMod, - State = #{overload_kill_enable := KillIfOL, - overload_kill_qlen := OLKillQLen, - overload_kill_mem_size := OLKillMem}) -> - if KillIfOL andalso - ((QLen > OLKillQLen) orelse (Mem > OLKillMem)) -> - HandlerMod:log_handler_info(Name, - "Handler ~p overloaded and stopping", - [Name], State), - set_restart_flag(Name, HandlerMod), - handler_exit(Name, {shutdown,{overloaded,Name,QLen,Mem}}); - true -> - ok - end. - -flush_log_events() -> - flush_log_events(-1). - -flush_log_events(Limit) -> - process_flag(priority, high), - Flushed = flush_log_events(0, Limit), - process_flag(priority, normal), - Flushed. - -flush_log_events(Limit, Limit) -> - Limit; -flush_log_events(N, Limit) -> - %% flush log events but leave other events, such as - %% filesync, info and change_config, so that these - %% have a chance to be processed even under heavy load - receive - {'$gen_cast',{log,_}} -> - flush_log_events(N+1, Limit); - {'$gen_call',{Pid,MRef},{log,_}} -> - Pid ! {MRef, dropped}, - flush_log_events(N+1, Limit) - after - 0 -> N - end. - -cancel_timer(TRef) when is_atom(TRef) -> ok; -cancel_timer(TRef) -> timer:cancel(TRef). - - -stop_or_restart(Name, {shutdown,Reason={overloaded,_Name,_QLen,_Mem}}, - #{overload_kill_restart_after := RestartAfter}) -> - %% If we're terminating because of an overload situation (see - %% logger_h_common:kill_if_choked/4), we need to remove the handler - %% and set a restart timer. A separate process must perform this - %% in order to avoid deadlock. - HandlerPid = self(), - ConfigResult = logger:get_handler_config(Name), - RemoveAndRestart = - fun() -> - MRef = erlang:monitor(process, HandlerPid), - receive - {'DOWN',MRef,_,_,_} -> - ok - after 30000 -> - error_notify(Reason), - exit(HandlerPid, kill) - end, - case ConfigResult of - {ok,#{module:=HMod}=HConfig} when is_integer(RestartAfter) -> - _ = logger:remove_handler(Name), - _ = timer:apply_after(RestartAfter, logger, add_handler, - [Name,HMod,HConfig]); - {ok,_} -> - _ = logger:remove_handler(Name); - {error,CfgReason} when is_integer(RestartAfter) -> - error_notify({Name,restart_impossible,CfgReason}); - {error,_} -> - ok - end - end, - spawn(RemoveAndRestart), - ok; -stop_or_restart(_Name, _Reason, _State) -> - ok. - -overload_levels_ok(HandlerConfig) -> - SMQL = maps:get(sync_mode_qlen, HandlerConfig, ?SYNC_MODE_QLEN), - DMQL = maps:get(drop_mode_qlen, HandlerConfig, ?DROP_MODE_QLEN), - FQL = maps:get(flush_qlen, HandlerConfig, ?FLUSH_QLEN), - (DMQL > 1) andalso (SMQL =< DMQL) andalso (DMQL =< FQL). - error_notify(Term) -> ?internal_log(error, Term). -info_notify(Term) -> - ?internal_log(info, Term). +maybe_set_repeated_filesync(_Olp, + #{filesync_repeat_interval:=FSyncInt}, + #{filesync_repeat_interval:=FSyncInt}) -> + ok; +maybe_set_repeated_filesync(Olp,_,#{filesync_repeat_interval:=FSyncInt}) -> + logger_olp:cast(Olp,{set_repeated_filesync,FSyncInt}). diff --git a/lib/kernel/src/logger_h_common.hrl b/lib/kernel/src/logger_h_common.hrl index e0a7b6e3ca..004a61d9d9 100644 --- a/lib/kernel/src/logger_h_common.hrl +++ b/lib/kernel/src/logger_h_common.hrl @@ -1,50 +1,22 @@ - -%%%----------------------------------------------------------------- -%%% Overload protection configuration - -%%! *** NOTE *** -%%! It's important that: -%%! SYNC_MODE_QLEN =< DROP_MODE_QLEN =< FLUSH_QLEN -%%! and that DROP_MODE_QLEN >= 2. -%%! Otherwise the handler could end up in drop mode with no new -%%! log requests to process. This would cause all future requests -%%! to be dropped (no switch to async mode would ever take place). - -%% This specifies the message_queue_len value where the log -%% requests switch from asynchronous casts to synchronous calls. --define(SYNC_MODE_QLEN, 10). -%% Above this message_queue_len, log requests will be dropped, -%% i.e. no log requests get sent to the handler process. --define(DROP_MODE_QLEN, 200). -%% Above this message_queue_len, the handler process will flush -%% its mailbox and only leave this number of messages in it. --define(FLUSH_QLEN, 1000). - -%% Never flush more than this number of messages in one go, -%% or the handler will be unresponsive for seconds (keep this -%% number as large as possible or the mailbox could grow large). --define(FLUSH_MAX_N, 5000). - -%% BURST_LIMIT_MAX_COUNT is the max number of log requests allowed -%% to be written within a BURST_LIMIT_WINDOW_TIME time frame. --define(BURST_LIMIT_ENABLE, true). --define(BURST_LIMIT_MAX_COUNT, 500). --define(BURST_LIMIT_WINDOW_TIME, 1000). - -%% This enables/disables the feature to automatically get the -%% handler terminated if it gets too loaded (and can't keep up). --define(OVERLOAD_KILL_ENABLE, false). -%% If the message_queue_len goes above this size even after -%% flushing has been performed, the handler is terminated. --define(OVERLOAD_KILL_QLEN, 20000). -%% If the memory usage exceeds this level --define(OVERLOAD_KILL_MEM_SIZE, 3000000). - -%% This is the default time that the handler will wait before -%% restarting and accepting new requests. The value 'infinity' -%% disables restarts. --define(OVERLOAD_KILL_RESTART_AFTER, 5000). -%%-define(OVERLOAD_KILL_RESTART_AFTER, infinity). +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 1997-2015. 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% +%% %% The handler sends asynchronous write requests to the process %% controlling the i/o device, but every once in this interval @@ -65,12 +37,6 @@ -define(FILESYNC_REPEAT_INTERVAL, 5000). %%-define(FILESYNC_REPEAT_INTERVAL, no_repeat). -%% This is the time after last message received that we think/hope -%% that the handler has an empty mailbox (no new log request has -%% come in). --define(IDLE_DETECT_TIME_MSEC, 100). --define(IDLE_DETECT_TIME_USEC, 100000). - %% Default disk log option values -define(DISK_LOG_TYPE, wrap). -define(DISK_LOG_MAX_NO_FILES, 10). @@ -83,43 +49,6 @@ list_to_atom(lists:concat([MODULE,"_",Name]))). %%%----------------------------------------------------------------- -%%% Overload protection macros - --define(timestamp(), erlang:monotonic_time(microsecond)). - --define(get_mode(Tid), - case ets:lookup(Tid, mode) of - [{mode,M}] -> M; - _ -> async - end). - --define(set_mode(Tid, M), - begin ets:insert(Tid, {mode,M}), M end). - --define(change_mode(Tid, M0, M1), - if M0 == M1 -> - M0; - true -> - ets:insert(Tid, {mode,M1}), - M1 - end). - --define(min(X1, X2), - if X2 == undefined -> X1; - X2 < X1 -> X2; - true -> X1 - end). - --define(max(X1, X2), - if - X2 == undefined -> X1; - X2 > X1 -> X2; - true -> X1 - end). - --define(diff_time(OS_T1, OS_T0), OS_T1-OS_T0). - -%%%----------------------------------------------------------------- %%% The test hook macros make it possible to observe and manipulate %%% internal handler functionality. When enabled, these macros will %%% slow down execution and therefore should not be include in code @@ -137,7 +66,7 @@ ets:insert(?TEST_HOOKS_TAB, {internal_log,{logger,internal_log}}), ets:insert(?TEST_HOOKS_TAB, {file_write,ok}), ets:insert(?TEST_HOOKS_TAB, {file_datasync,ok}), - ets:insert(?TEST_HOOKS_TAB, {disk_log_blog,ok}), + ets:insert(?TEST_HOOKS_TAB, {disk_log_write,ok}), ets:insert(?TEST_HOOKS_TAB, {disk_log_sync,ok})). -define(set_internal_log(MOD_FUNC), @@ -150,7 +79,7 @@ ets:insert(?TEST_HOOKS_TAB, {internal_log,{logger,internal_log}}), ets:insert(?TEST_HOOKS_TAB, {file_write,ok}), ets:insert(?TEST_HOOKS_TAB, {file_datasync,ok}), - ets:insert(?TEST_HOOKS_TAB, {disk_log_blog,ok}), + ets:insert(?TEST_HOOKS_TAB, {disk_log_write,ok}), ets:insert(?TEST_HOOKS_TAB, {disk_log_sync,ok})). -define(internal_log(TYPE, TERM), @@ -171,11 +100,11 @@ [{_,ERROR}] -> ERROR catch _:_ -> file:datasync(DEVICE) end). - -define(disk_log_blog(LOG, DATA), - try ets:lookup(?TEST_HOOKS_TAB, disk_log_blog) of - [{_,ok}] -> disk_log:blog(LOG, DATA); + -define(disk_log_write(LOG, MODE, DATA), + try ets:lookup(?TEST_HOOKS_TAB, disk_log_write) of + [{_,ok}] -> disk_log_write(LOG, MODE, DATA); [{_,ERROR}] -> ERROR - catch _:_ -> disk_log:blog(LOG, DATA) end). + catch _:_ -> disk_log_write(LOG, MODE, DATA) end). -define(disk_log_sync(LOG), try ets:lookup(?TEST_HOOKS_TAB, disk_log_sync) of @@ -183,7 +112,6 @@ [{_,ERROR}] -> ERROR catch _:_ -> disk_log:sync(LOG) end). - -define(DEFAULT_CALL_TIMEOUT, 5000). -else. % DEFAULTS! -define(TEST_HOOKS_TAB, undefined). @@ -194,70 +122,6 @@ -define(internal_log(TYPE, TERM), logger:internal_log(TYPE, TERM)). -define(file_write(DEVICE, DATA), file:write(DEVICE, DATA)). -define(file_datasync(DEVICE), file:datasync(DEVICE)). - -define(disk_log_blog(LOG, DATA), disk_log:blog(LOG, DATA)). + -define(disk_log_write(LOG, MODE, DATA), disk_log_write(LOG, MODE, DATA)). -define(disk_log_sync(LOG), disk_log:sync(LOG)). - -define(DEFAULT_CALL_TIMEOUT, 10000). --endif. - -%%%----------------------------------------------------------------- -%%% These macros enable statistics counters in the state of the -%%% handler which is useful for analysing the overload protection -%%% behaviour. These counters should not be included in code to be -%%% officially released (as some counters will grow very large -%%% over time). - -%%-define(SAVE_STATS, true). --ifdef(SAVE_STATS). - -define(merge_with_stats(STATE), - STATE#{flushes => 0, flushed => 0, drops => 0, - casts => 0, calls => 0, - max_qlen => 0, max_time => 0}). - - -define(update_max_qlen(QLEN, STATE), - begin #{max_qlen := QLEN0} = STATE, - STATE#{max_qlen => ?max(QLEN0,QLEN)} end). - - -define(update_calls_or_casts(CALL_OR_CAST, INC, STATE), - case CALL_OR_CAST of - cast -> - #{casts := CASTS0} = STATE, - STATE#{casts => CASTS0+INC}; - call -> - #{calls := CALLS0} = STATE, - STATE#{calls => CALLS0+INC} - end). - - -define(update_max_time(TIME, STATE), - begin #{max_time := TIME0} = STATE, - STATE#{max_time => ?max(TIME0,TIME)} end). - - -define(update_other(OTHER, VAR, INCVAL, STATE), - begin #{OTHER := VAR} = STATE, - STATE#{OTHER => VAR+INCVAL} 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). --endif. - -%%%----------------------------------------------------------------- -%%% These macros enable callbacks that make it possible to analyse -%%% the overload protection behaviour from outside the handler -%%% process (including dropped requests on the client side). -%%% An external callback module (?OBSERVER_MOD) is required which -%%% is not part of the kernel application. For this reason, these -%%% callbacks should not be included in code to be officially released. - -%%-define(OBSERVER_MOD, logger_test). --ifdef(OBSERVER_MOD). - -define(start_observation(NAME), ?OBSERVER:start_observation(NAME)). - -define(observe(NAME,EVENT), ?OBSERVER:observe(NAME,EVENT)). - --else. % DEFAULT! - -define(start_observation(_NAME), ok). - -define(observe(_NAME,_EVENT), ok). -endif. -%%! <--- diff --git a/lib/kernel/src/logger_internal.hrl b/lib/kernel/src/logger_internal.hrl index d96a4ac78b..e53922e5d3 100644 --- a/lib/kernel/src/logger_internal.hrl +++ b/lib/kernel/src/logger_internal.hrl @@ -19,6 +19,7 @@ %% -include_lib("kernel/include/logger.hrl"). -define(LOGGER_TABLE,logger). +-define(PROXY_KEY,'$proxy_config$'). -define(PRIMARY_KEY,'$primary_config$'). -define(HANDLER_KEY,'$handler_config$'). -define(LOGGER_META_KEY,'$logger_metadata$'). @@ -40,12 +41,14 @@ -define(DEFAULT_LOGGER_CALL_TIMEOUT, infinity). --define(LOG_INTERNAL(Level,Report), +-define(LOG_INTERNAL(Level,Report),?DO_LOG_INTERNAL(Level,[Report])). +-define(LOG_INTERNAL(Level,Format,Args),?DO_LOG_INTERNAL(Level,[Format,Args])). +-define(DO_LOG_INTERNAL(Level,Data), case logger:allow(Level,?MODULE) of true -> %% Spawn this to avoid deadlocks - _ = spawn(logger,macro_log,[?LOCATION,Level,Report, - logger:add_default_metadata(#{})]), + _ = spawn(logger,macro_log,[?LOCATION,Level|Data]++ + [logger:add_default_metadata(#{})]), ok; false -> ok diff --git a/lib/kernel/src/logger_olp.erl b/lib/kernel/src/logger_olp.erl new file mode 100644 index 0000000000..009280a9c9 --- /dev/null +++ b/lib/kernel/src/logger_olp.erl @@ -0,0 +1,626 @@ +%% +%% %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_olp). +-behaviour(gen_server). + +-include("logger_olp.hrl"). +-include("logger_internal.hrl"). + +%% 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, 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(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() :: logger:olp_config(). + +%%%----------------------------------------------------------------- +%%% API + +-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 -> + proc_lib:start_link(?MODULE,init,[[Name,Module,Args,Options]]); + Error -> + Error + end. + +-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 + %% process causing much load). If the process is choked, drop mode + %% is set and no message is sent. + try ?get_mode(ModeRef) of + async -> + gen_server:cast(Pid, {'$olp_load',Msg}); + sync -> + case call(Pid, {'$olp_load',Msg}) of + ok -> + ok; + _Other -> + %% dropped or {error,busy} + ?observe(_Name,{dropped,1}), + ok + end; + drop -> + ?observe(_Name,{dropped,1}) + catch + %% if the ETS table doesn't exist (maybe because of a + %% process restart), we can only drop the event + _:_ -> ?observe(_Name,{dropped,1}) + 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. + +-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}). + +-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, + flush_qlen => ?FLUSH_QLEN, + burst_limit_enable => ?BURST_LIMIT_ENABLE, + burst_limit_max_count => ?BURST_LIMIT_MAX_COUNT, + burst_limit_window_time => ?BURST_LIMIT_WINDOW_TIME, + overload_kill_enable => ?OVERLOAD_KILL_ENABLE, + overload_kill_qlen => ?OVERLOAD_KILL_QLEN, + overload_kill_mem_size => ?OVERLOAD_KILL_MEM_SIZE, + overload_kill_restart_after => ?OVERLOAD_KILL_RESTART_AFTER}. + +-spec restart(fun(() -> any())) -> ok. +restart(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 +%%%=================================================================== + +init([Name,Module,Args,Options]) -> + register(Name, self()), + process_flag(message_queue_data, off_heap), + + ?start_observation(Name), + + 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(),OlpRef}), + %% Storing options in state to avoid copying + %% (sending) the option data with each message + State0 = ?merge_with_stats( + Options#{id => Name, + idle=> true, + module => Module, + mode_ref => ModeRef, + mode => async, + last_qlen => 0, + last_load_ts => T0, + burst_win_ts => T0, + burst_msg_count => 0, + cb_state => CBState}), + State = reset_restart_flag(State0), + 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 + catch + _:Error -> + unregister(Name), + proc_lib:init_ack(Error) + end. + +%% This is the synchronous load event. +handle_call({'$olp_load', Msg}, _From, State) -> + {Result,State1} = do_load(Msg, call, State#{idle=>false}), + %% Result == ok | dropped + reply_return(Result,State1); + +handle_call(get_ref,_From,#{id:=Name,mode_ref:=ModeRef}=State) -> + reply_return({Name,self(),ModeRef},State); + +handle_call({set_opts,Opts0},_From,State) -> + Opts = maps:merge(maps:with(?OPT_KEYS,State),Opts0), + case check_opts(Opts) of + ok -> + reply_return(ok, maps:merge(State,Opts)); + Error -> + reply_return(Error, State) + end; + +handle_call(get_opts,_From,State) -> + reply_return(maps:with(?OPT_KEYS,State), State); + +handle_call(info, _From, State) -> + reply_return(State, State); + +handle_call(reset, _From, #{module:=Module,cb_state:=CBState}=State) -> + State1 = ?merge_with_stats(State), + CBState1 = try_callback_call(Module,reset_state,[CBState],CBState), + reply_return(ok, State1#{idle => true, + last_qlen => 0, + last_load_ts => ?timestamp(), + cb_state => CBState1}); + +handle_call(stop, _From, State) -> + {stop, {shutdown,stopped}, ok, State}; + +handle_call(Msg, From, #{module:=Module,cb_state:=CBState}=State) -> + case try_callback_call(Module,handle_call,[Msg, From, CBState]) of + {reply,Reply,CBState1} -> + reply_return(Reply,State#{cb_state=>CBState1}); + {noreply,CBState1} -> + noreply_return(State#{cb_state=>CBState1}); + {stop, Reason, Reply, CBState1} -> + {stop, Reason, Reply, State#{cb_state=>CBState1}}; + {stop, Reason, CBState1} -> + {stop, Reason, State#{cb_state=>CBState1}} + end. + +%% This is the asynchronous load event. +handle_cast({'$olp_load', Msg}, State) -> + {_Result,State1} = do_load(Msg, cast, State#{idle=>false}), + noreply_return(State1); + +handle_cast(Msg, #{module:=Module, cb_state:=CBState} = State) -> + case try_callback_call(Module,handle_cast,[Msg, CBState]) of + {noreply,CBState1} -> + noreply_return(State#{cb_state=>CBState1}); + {stop, Reason, CBState1} -> + {stop, Reason, State#{cb_state=>CBState1}} + end. + +handle_info(timeout, #{mode_ref:=_ModeRef, mode:=Mode} = State) -> + State1 = notify(idle,State), + State2 = maybe_notify_mode_change(async,State1), + {noreply, State2#{idle => true, + mode => ?change_mode(_ModeRef, Mode, async), + burst_msg_count => 0}}; +handle_info(Msg, #{module := Module, cb_state := CBState} = State) -> + case try_callback_call(Module,handle_info,[Msg, CBState]) of + {noreply,CBState1} -> + noreply_return(State#{cb_state=>CBState1}); + {stop, Reason, CBState1} -> + {stop, Reason, State#{cb_state=>CBState1}}; + {load,CBState1} -> + {_,State1} = do_load(Msg, cast, State#{idle=>false, + cb_state=>CBState1}), + noreply_return(State1) + end. + +terminate({shutdown,{overloaded,_QLen,_Mem}}, + #{id:=Name, module := Module, cb_state := CBState, + overload_kill_restart_after := RestartAfter} = State) -> + %% We're terminating because of an overload situation (see + %% kill_if_choked/3). + unregister(Name), %%!!!! to avoid error printout of callback crashed on stop + 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]), + ok; + _ -> + ok + end; +terminate(Reason, #{id:=Name, module:=Module, cb_state:=CBState}) -> + _ = try_callback_call(Module,terminate,[Reason,CBState],ok), + unregister(Name), + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + + +%%%----------------------------------------------------------------- +%%% 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) -> + try + gen_server:call(Server, Msg) + catch + _:{timeout,_} -> {error,busy} + end. + +-spec cast(olp_ref(),term()) -> ok. +cast({_Name, Pid, _ModeRef},Msg) -> + gen_server:cast(Pid, Msg). + +%% check for overload between every event (and set Mode to async, +%% sync or drop accordingly), but never flush the whole mailbox +%% 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,State2} = check_load(State1), + + %% kill the handler if it can't keep up with the load + kill_if_choked(QLen, Mem, State2), + + if Mode1 == flush -> + flush(T1, State2); + true -> + 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, 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), + + %% write info in log about flushed messages + State1=notify({flushed,NewFlushed},State), + + %% 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}), + + %% 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), + State4 = maybe_notify_mode_change(async,State3), + {dropped,?update_other(flushed,FLUSHED,NewFlushed, + State4#{mode => ?change_mode(ModeRef,Mode,async), + last_qlen => QLen1, + last_load_ts => T1})}. + +%% this function is called to actually handle the message +handle_load(Mode, T1, Msg, _CallOrCast, + State = #{id := _Name, + module := Module, + cb_state := CBState, + last_qlen := LastQLen, + last_load_ts := _T0}) -> + %% check if we need to limit the number of writes + %% during a burst of log events + {DoWrite,State1} = limit_burst(State), + + {Result,LastQLen1,CBState1} = + if DoWrite -> + ?observe(_Name,{_CallOrCast,1}), + CBS = try_callback_call(Module,handle_load,[Msg,CBState]), + {ok,element(2, process_info(self(), message_queue_len)),CBS}; + true -> + ?observe(_Name,{flushed,1}), + {dropped,LastQLen,CBState} + end, + State2 = State1#{cb_state=>CBState1}, + + State3 = State2#{mode => Mode}, + State4 = ?update_calls_or_casts(_CallOrCast,1,State3), + State5 = ?update_max_qlen(LastQLen1,State4), + State6 = + ?update_max_time(?diff_time(T1,_T0), + State5#{last_qlen := LastQLen1, + last_load_ts => T1}), + State7 = case Result of + ok -> + S = ?update_freq(T1,State6), + ?update_other(writes,WRITES,1,S); + _ -> + State6 + end, + {Result,State7}. + + +%%%----------------------------------------------------------------- +%%% Check that the options are valid +check_opts(Options) when is_map(Options) -> + case do_check_opts(maps:to_list(Options)) of + ok -> + case overload_levels_ok(Options) of + true -> + ok; + false -> + Faulty = maps:with([sync_mode_qlen, + drop_mode_qlen, + flush_qlen],Options), + {error,{invalid_olp_levels,Faulty}} + end; + {error,Key,Value} -> + {error,{invalid_olp_config,#{Key=>Value}}} + end. + +do_check_opts([{sync_mode_qlen,N}|Options]) when is_integer(N) -> + do_check_opts(Options); +do_check_opts([{drop_mode_qlen,N}|Options]) when is_integer(N) -> + do_check_opts(Options); +do_check_opts([{flush_qlen,N}|Options]) when is_integer(N) -> + do_check_opts(Options); +do_check_opts([{burst_limit_enable,Bool}|Options]) when is_boolean(Bool) -> + do_check_opts(Options); +do_check_opts([{burst_limit_max_count,N}|Options]) when is_integer(N) -> + do_check_opts(Options); +do_check_opts([{burst_limit_window_time,N}|Options]) when is_integer(N) -> + do_check_opts(Options); +do_check_opts([{overload_kill_enable,Bool}|Options]) when is_boolean(Bool) -> + do_check_opts(Options); +do_check_opts([{overload_kill_qlen,N}|Options]) when is_integer(N) -> + do_check_opts(Options); +do_check_opts([{overload_kill_mem_size,N}|Options]) when is_integer(N) -> + do_check_opts(Options); +do_check_opts([{overload_kill_restart_after,NorA}|Options]) + when is_integer(NorA); NorA == infinity -> + do_check_opts(Options); +do_check_opts([{Key,Value}|_]) -> + {error,Key,Value}; +do_check_opts([]) -> + ok. + +set_restart_flag(#{id := Name, module := Module}) -> + Flag = list_to_atom(lists:concat([Module,"_",Name,"_restarting"])), + spawn(fun() -> + register(Flag, self()), + timer:sleep(infinity) + end), + ok. + +reset_restart_flag(#{id := Name, module := Module} = State) -> + Flag = list_to_atom(lists:concat([Module,"_",Name,"_restarting"])), + case whereis(Flag) of + undefined -> + State; + Pid -> + exit(Pid, kill), + notify(restart,State) + end. + +check_load(State = #{id:=_Name, mode_ref := ModeRef, mode := Mode, + sync_mode_qlen := SyncModeQLen, + drop_mode_qlen := DropModeQLen, + flush_qlen := FlushQLen}) -> + {_,Mem} = process_info(self(), memory), + ?observe(_Name,{max_mem,Mem}), + {_,QLen} = process_info(self(), message_queue_len), + ?observe(_Name,{max_qlen,QLen}), + %% When the handler process gets scheduled in, it's impossible + %% to predict the QLen. We could jump "up" arbitrarily from say + %% async to sync, async to drop, sync to flush, etc. However, when + %% the handler process manages the log events (without flushing), + %% one after the other, we will move "down" from drop to sync and + %% from sync to async. This way we don't risk getting stuck in + %% drop or sync mode with an empty mailbox. + {Mode1,_NewDrops,_NewFlushes} = + if + QLen >= FlushQLen -> + {flush, 0,1}; + QLen >= DropModeQLen -> + %% Note that drop mode will force load messages to + %% be dropped on the client side (never sent to + %% the olp process). + IncDrops = if Mode == drop -> 0; true -> 1 end, + {?change_mode(ModeRef, Mode, drop), IncDrops,0}; + QLen >= SyncModeQLen -> + {?change_mode(ModeRef, Mode, sync), 0,0}; + true -> + {?change_mode(ModeRef, Mode, async), 0,0} + end, + State1 = ?update_other(drops,DROPS,_NewDrops,State), + State2 = ?update_max_qlen(QLen,State1), + State3 = maybe_notify_mode_change(Mode1,State2), + {Mode1, QLen, Mem, + ?update_other(flushes,FLUSHES,_NewFlushes, + State3#{last_qlen => QLen})}. + +limit_burst(#{burst_limit_enable := false}=State) -> + {true,State}; +limit_burst(#{burst_win_ts := BurstWinT0, + burst_msg_count := BurstMsgCount, + burst_limit_window_time := BurstLimitWinTime, + burst_limit_max_count := BurstLimitMaxCnt} = State) -> + if (BurstMsgCount >= BurstLimitMaxCnt) -> + %% the limit for allowed messages has been reached + BurstWinT1 = ?timestamp(), + case ?diff_time(BurstWinT1,BurstWinT0) of + BurstCheckTime when BurstCheckTime < (BurstLimitWinTime*1000) -> + %% we're still within the burst time frame + {false,?update_other(burst_drops,BURSTS,1,State)}; + _BurstCheckTime -> + %% burst time frame passed, reset counters + {true,State#{burst_win_ts => BurstWinT1, + burst_msg_count => 0}} + end; + true -> + %% the limit for allowed messages not yet reached + {true,State#{burst_win_ts => BurstWinT0, + burst_msg_count => BurstMsgCount+1}} + end. + +kill_if_choked(QLen, Mem, #{overload_kill_enable := KillIfOL, + overload_kill_qlen := OLKillQLen, + overload_kill_mem_size := OLKillMem}) -> + if KillIfOL andalso + ((QLen > OLKillQLen) orelse (Mem > OLKillMem)) -> + exit({shutdown,{overloaded,QLen,Mem}}); + true -> + ok + end. + +flush_load(Limit) -> + process_flag(priority, high), + Flushed = flush_load(0, Limit), + process_flag(priority, normal), + Flushed. + +flush_load(Limit, Limit) -> + Limit; +flush_load(N, Limit) -> + %% flush log events but leave other events, such as info, reset + %% and stop, so that these have a chance to be processed even + %% under heavy load + receive + {'$gen_cast',{'$olp_load',_}} -> + flush_load(N+1, Limit); + {'$gen_call',{Pid,MRef},{'$olp_load',_}} -> + Pid ! {MRef, dropped}, + flush_load(N+1, Limit); + {log,_,_,_,_} -> + flush_load(N+1, Limit); + {log,_,_,_} -> + flush_load(N+1, Limit) + after + 0 -> N + end. + +overload_levels_ok(Options) -> + SMQL = maps:get(sync_mode_qlen, Options, ?SYNC_MODE_QLEN), + DMQL = maps:get(drop_mode_qlen, Options, ?DROP_MODE_QLEN), + FQL = maps:get(flush_qlen, Options, ?FLUSH_QLEN), + (DMQL > 1) andalso (SMQL =< DMQL) andalso (DMQL =< FQL). + +maybe_notify_mode_change(drop,#{mode:=Mode0}=State) + when Mode0=/=drop -> + notify({mode_change,Mode0,drop},State); +maybe_notify_mode_change(Mode1,#{mode:=drop}=State) + when Mode1==async; Mode1==sync -> + notify({mode_change,drop,Mode1},State); +maybe_notify_mode_change(_,State) -> + State. + +notify(Note,#{module:=Module,cb_state:=CBState}=State) -> + CBState1 = try_callback_call(Module,notify,[Note,CBState],CBState), + State#{cb_state=>CBState1}. + +try_callback_call(Module, Function, Args) -> + try_callback_call(Module, Function, Args, '$no_default_return'). + +try_callback_call(Module, Function, Args, DefRet) -> + try apply(Module, Function, Args) + catch + throw:R -> R; + error:undef:S when DefRet=/='$no_default_return' -> + case S of + [{Module,Function,Args,_}|_] -> + DefRet; + _ -> + erlang:raise(error,undef,S) + end + end. + +noreply_return(#{idle:=true}=State) -> + {noreply,State}; +noreply_return(#{idle:=false}=State) -> + {noreply,State,?IDLE_DETECT_TIME}. + +reply_return(Reply,#{idle:=true}=State) -> + {reply,Reply,State}; +reply_return(Reply,#{idle:=false}=State) -> + {reply,Reply,State,?IDLE_DETECT_TIME}. diff --git a/lib/kernel/src/logger_olp.hrl b/lib/kernel/src/logger_olp.hrl new file mode 100644 index 0000000000..9b4f5ebf27 --- /dev/null +++ b/lib/kernel/src/logger_olp.hrl @@ -0,0 +1,180 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 1997-2015. 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% +%% + +%%%----------------------------------------------------------------- +%%% Overload protection configuration + +%%! *** NOTE *** +%%! It's important that: +%%! SYNC_MODE_QLEN =< DROP_MODE_QLEN =< FLUSH_QLEN +%%! and that DROP_MODE_QLEN >= 2. +%%! Otherwise the process could end up in drop mode with no new +%%! log requests to process. This would cause all future requests +%%! to be dropped (no switch to async mode would ever take place). + +%% This specifies the message_queue_len value where the log +%% requests switch from asynchronous casts to synchronous calls. +-define(SYNC_MODE_QLEN, 10). +%% Above this message_queue_len, log requests will be dropped, +%% i.e. no log requests get sent to the process. +-define(DROP_MODE_QLEN, 200). +%% Above this message_queue_len, the process will flush its mailbox +%% and only leave this number of messages in it. +-define(FLUSH_QLEN, 1000). + +%% Never flush more than this number of messages in one go, or the +%% process will be unresponsive for seconds (keep this number as large +%% as possible or the mailbox could grow large). +-define(FLUSH_MAX_N, 5000). + +%% BURST_LIMIT_MAX_COUNT is the max number of log requests allowed +%% to be written within a BURST_LIMIT_WINDOW_TIME time frame. +-define(BURST_LIMIT_ENABLE, true). +-define(BURST_LIMIT_MAX_COUNT, 500). +-define(BURST_LIMIT_WINDOW_TIME, 1000). + +%% This enables/disables the feature to automatically terminate the +%% process if it gets too loaded (and can't keep up). +-define(OVERLOAD_KILL_ENABLE, false). +%% If the message_queue_len goes above this size even after +%% flushing has been performed, the process is terminated. +-define(OVERLOAD_KILL_QLEN, 20000). +%% If the memory usage exceeds this level, the process is terminated. +-define(OVERLOAD_KILL_MEM_SIZE, 3000000). + +%% This is the default time to wait before restarting and accepting +%% new requests. The value 'infinity' disables restarts. +-define(OVERLOAD_KILL_RESTART_AFTER, 5000). + +%% This is the time in milliseconds after last load message received +%% that we notify the callback about being idle. +-define(IDLE_DETECT_TIME, 100). + +%%%----------------------------------------------------------------- +%%% Overload protection macros + +-define(timestamp(), erlang:monotonic_time(microsecond)). + +-define(get_mode(Tid), + case ets:lookup(Tid, mode) of + [{mode,M}] -> M; + _ -> async + end). + +-define(set_mode(Tid, M), + begin ets:insert(Tid, {mode,M}), M end). + +-define(change_mode(Tid, M0, M1), + if M0 == M1 -> + M0; + true -> + ets:insert(Tid, {mode,M1}), + M1 + end). + +-define(max(X1, X2), + if + X2 == undefined -> X1; + X2 > X1 -> X2; + true -> X1 + end). + +-define(diff_time(OS_T1, OS_T0), OS_T1-OS_T0). + +%%%----------------------------------------------------------------- +%%% These macros enable statistics counters in the state of the +%%% process, which is useful for analysing the overload protection +%%% behaviour. These counters should not be included in code to be +%%% officially released (as some counters will grow very large over +%%% time). + +%% -define(SAVE_STATS, true). +-ifdef(SAVE_STATS). + -define(merge_with_stats(STATE), + 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, + STATE#{max_qlen => ?max(QLEN0,QLEN)} end). + + -define(update_calls_or_casts(CALL_OR_CAST, INC, STATE), + case CALL_OR_CAST of + cast -> + #{casts := CASTS0} = STATE, + STATE#{casts => CASTS0+INC}; + call -> + #{calls := CALLS0} = STATE, + STATE#{calls => CALLS0+INC} + end). + + -define(update_max_time(TIME, STATE), + begin #{max_time := TIME0} = STATE, + STATE#{max_time => ?max(TIME0,TIME)} end). + + -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. + +%%%----------------------------------------------------------------- +%%% These macros enable callbacks that make it possible to analyse the +%%% overload protection behaviour from outside the process (including +%%% dropped requests on the client side). An external callback module +%%% (?OBSERVER_MOD) is required which is not part of the kernel +%%% application. For this reason, these callbacks should not be +%%% included in code to be officially released. + +%%-define(OBSERVER_MOD, logger_test). +-ifdef(OBSERVER_MOD). + -define(start_observation(NAME), ?OBSERVER:start_observation(NAME)). + -define(observe(NAME,EVENT), ?OBSERVER:observe(NAME,EVENT)). + +-else. % DEFAULT! + -define(start_observation(_NAME), ok). + -define(observe(_NAME,_EVENT), ok). +-endif. diff --git a/lib/kernel/src/logger_proxy.erl b/lib/kernel/src/logger_proxy.erl new file mode 100644 index 0000000000..24b293805c --- /dev/null +++ b/lib/kernel/src/logger_proxy.erl @@ -0,0 +1,165 @@ +%% +%% %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/1, child_spec/0, get_default_config/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(RemoteLog) -> ok when + 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(RemoteLog) -> + Olp = persistent_term:get(?MODULE), + 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). + logger_olp:start_link(?SERVER,?MODULE,[],logger:get_proxy_config()). + +%% 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]}. + +get_default_config() -> + OlpDefault = logger_olp:get_default_opts(), + OlpDefault#{sync_mode_qlen=>500, + drop_mode_qlen=>1000, + flush_qlen=>5000, + burst_limit_enable=>false}. + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +init([]) -> + process_flag(trap_exit, true), + _ = erlang:system_flag(system_logger,self()), + persistent_term:put(?MODULE,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,Mode1},State) -> + _ = if Mode1=:=drop -> % entering drop mode + erlang:system_flag(system_logger,undefined); + Mode0=:=drop -> % leaving drop mode + erlang:system_flag(system_logger,self()); + true -> + ok + end, + ?LOG_INTERNAL(notice,"~w switched from ~w to ~w mode",[?MODULE,Mode0,Mode1]), + State; +notify({flushed,Flushed},State) -> + ?LOG_INTERNAL(notice, "~w flushed ~w log events",[?MODULE,Flushed]), + State; +notify(restart,State) -> + ?LOG_INTERNAL(notice, "~w restarted", [?MODULE]), + 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 a1d40f1123..722246e82c 100644 --- a/lib/kernel/src/logger_server.erl +++ b/lib/kernel/src/logger_server.erl @@ -22,14 +22,17 @@ -behaviour(gen_server). %% API --export([start_link/0, - add_handler/3, remove_handler/1, +-export([start_link/0, add_handler/3, remove_handler/1, add_filter/2, remove_filter/2, set_module_level/2, unset_module_level/0, unset_module_level/1, cache_module_level/1, - set_config/2, set_config/3, update_config/2, + set_config/2, set_config/3, + update_config/2, update_config/3, update_formatter_config/2]). +%% Helper +-export([diff_maps/2]). + %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). @@ -39,7 +42,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 @@ -105,12 +108,25 @@ cache_module_level(Module) -> gen_server:cast(?SERVER,{cache_module_level,Module}). set_config(Owner,Key,Value) -> - update_config(Owner,#{Key=>Value}). + case sanity_check(Owner,Key,Value) of + ok -> + call({change_config,set,Owner,Key,Value}); + Error -> + Error + end. set_config(Owner,Config) -> case sanity_check(Owner,Config) of ok -> - call({set_config,Owner,Config}); + call({change_config,set,Owner,Config}); + Error -> + Error + end. + +update_config(Owner,Key,Value) -> + case sanity_check(Owner,Key,Value) of + ok -> + call({change_config,update,Owner,Key,Value}); Error -> Error end. @@ -118,7 +134,7 @@ set_config(Owner,Config) -> update_config(Owner, Config) -> case sanity_check(Owner,Config) of ok -> - call({update_config,Owner,Config}); + call({change_config,update,Owner,Config}); Error -> Error end. @@ -138,6 +154,8 @@ init([]) -> process_flag(trap_exit, true), put(?LOGGER_SERVER_TAG,true), Tid = logger_config:new(?LOGGER_TABLE), + %% Store initial proxy config. logger_proxy reads config from here at startup. + logger_config:create(Tid,proxy,logger_proxy:get_default_config()), PrimaryConfig = maps:merge(default_config(primary), #{handlers=>[simple]}), logger_config:create(Tid,primary,PrimaryConfig), @@ -204,46 +222,90 @@ handle_call({add_filter,Id,Filter}, _From,#state{tid=Tid}=State) -> handle_call({remove_filter,Id,FilterId}, _From, #state{tid=Tid}=State) -> Reply = do_remove_filter(Tid,Id,FilterId), {reply,Reply,State}; -handle_call({update_config,primary,NewConfig}, _From, #state{tid=Tid}=State) -> +handle_call({change_config,SetOrUpd,proxy,Config0},_From,#state{tid=Tid}=State) -> + Default = + case SetOrUpd of + set -> + logger_proxy:get_default_config(); + update -> + {ok,OldConfig} = logger_config:get(Tid,proxy), + OldConfig + end, + Config = maps:merge(Default,Config0), + Reply = + case logger_olp:set_opts(logger_proxy,Config) of + ok -> + logger_config:set(Tid,proxy,Config); + Error -> + Error + end, + {reply,Reply,State}; +handle_call({change_config,SetOrUpd,primary,Config0}, _From, + #state{tid=Tid}=State) -> + {ok,#{handlers:=Handlers}=OldConfig} = logger_config:get(Tid,primary), + Default = + case SetOrUpd of + set -> default_config(primary); + update -> OldConfig + end, + Config = maps:merge(Default,Config0), + Reply = logger_config:set(Tid,primary,Config#{handlers=>Handlers}), + {reply,Reply,State}; +handle_call({change_config,_SetOrUpd,primary,Key,Value}, _From, + #state{tid=Tid}=State) -> {ok,OldConfig} = logger_config:get(Tid,primary), - Config = maps:merge(OldConfig,NewConfig), - {reply,logger_config:set(Tid,primary,Config),State}; -handle_call({update_config,HandlerId,NewConfig}, From, #state{tid=Tid}=State) -> + Reply = logger_config:set(Tid,primary,OldConfig#{Key=>Value}), + {reply,Reply,State}; +handle_call({change_config,SetOrUpd,HandlerId,Config0}, From, + #state{tid=Tid}=State) -> case logger_config:get(Tid,HandlerId) of {ok,#{module:=Module}=OldConfig} -> - Config = maps:merge(OldConfig,NewConfig), - call_h_async( - fun() -> - call_h(Module,changing_config,[OldConfig,Config], - {ok,Config}) - end, - fun({ok,Config1}) -> - logger_config:set(Tid,HandlerId,Config1); - (Error) -> - Error - end,From,State); - Error -> - {reply,Error,State} + Default = + case SetOrUpd of + set -> default_config(HandlerId,Module); + update -> OldConfig + end, + Config = maps:merge(Default,Config0), + case check_config_change(OldConfig,Config) of + ok -> + call_h_async( + fun() -> + call_h(Module,changing_config, + [SetOrUpd,OldConfig,Config], + {ok,Config}) + end, + fun({ok,Config1}) -> + logger_config:set(Tid,HandlerId,Config1); + (Error) -> + Error + end,From,State); + Error -> + {reply,Error,State} + end; + _ -> + {reply,{error,{not_found,HandlerId}},State} end; -handle_call({set_config,primary,Config0}, _From, #state{tid=Tid}=State) -> - Config = maps:merge(default_config(primary),Config0), - {ok,#{handlers:=Handlers}} = logger_config:get(Tid,primary), - Reply = logger_config:set(Tid,primary,Config#{handlers=>Handlers}), - {reply,Reply,State}; -handle_call({set_config,HandlerId,Config0}, From, #state{tid=Tid}=State) -> +handle_call({change_config,SetOrUpd,HandlerId,Key,Value}, From, + #state{tid=Tid}=State) -> case logger_config:get(Tid,HandlerId) of {ok,#{module:=Module}=OldConfig} -> - Config = maps:merge(default_config(HandlerId,Module),Config0), - call_h_async( - fun() -> - call_h(Module,changing_config,[OldConfig,Config], - {ok,Config}) - end, - fun({ok,Config1}) -> - logger_config:set(Tid,HandlerId,Config1); - (Error) -> - Error - end,From,State); + Config = OldConfig#{Key=>Value}, + case check_config_change(OldConfig,Config) of + ok -> + call_h_async( + fun() -> + call_h(Module,changing_config, + [SetOrUpd,OldConfig,Config], + {ok,Config}) + end, + fun({ok,Config1}) -> + logger_config:set(Tid,HandlerId,Config1); + (Error) -> + Error + end,From,State); + Error -> + {reply,Error,State} + end; _ -> {reply,{error,{not_found,HandlerId}},State} end; @@ -314,18 +376,19 @@ 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 Action == add_handler; Action == remove_handler; Action == add_filter; Action == remove_filter; - Action == update_config; Action == set_config -> + Action == change_config -> {error,{attempting_syncronous_call_to_self,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} -> @@ -370,11 +433,13 @@ default_config(Id,Module) -> sanity_check(Owner,Key,Value) -> sanity_check_1(Owner,[{Key,Value}]). -sanity_check(HandlerId,Config) when is_map(Config) -> - sanity_check_1(HandlerId,maps:to_list(Config)); +sanity_check(Owner,Config) when is_map(Config) -> + sanity_check_1(Owner,maps:to_list(Config)); sanity_check(_,Config) -> {error,{invalid_config,Config}}. +sanity_check_1(proxy,_Config) -> + ok; % Details are checked by logger_olp:set_opts/2 sanity_check_1(Owner,Config) when is_list(Config) -> try Type = get_type(Owner), @@ -458,6 +523,15 @@ check_formatter({Mod,Config}) -> check_formatter(Formatter) -> throw({invalid_formatter,Formatter}). +%% When changing configuration for a handler, the id and module fields +%% can not be changed. +check_config_change(#{id:=Id,module:=Module},#{id:=Id,module:=Module}) -> + ok; +check_config_change(OldConfig,NewConfig) -> + {Old,New} = logger_server:diff_maps(maps:with([id,module],OldConfig), + maps:with([id,module],NewConfig)), + {error,{illegal_config_change,Old,New}}. + call_h(Module, Function, Args, DefRet) -> %% Not calling code:ensure_loaded + erlang:function_exported here, %% since in some rare terminal cases, the code_server might not @@ -466,6 +540,11 @@ call_h(Module, Function, Args, DefRet) -> catch C:R:S -> case {C,R,S} of + {error,undef,[{Module,Function=changing_config,Args,_}|_]} + when length(Args)=:=3 -> + %% Backwards compatible call, if changing_config/3 + %% did not exist. + call_h(Module, Function, tl(Args), DefRet); {error,undef,[{Module,Function,Args,_}|_]} -> DefRet; _ -> @@ -525,3 +604,14 @@ call_h_reply(Unexpected,State) -> {process,?SERVER}, {message,Unexpected}]), {noreply,State}. + +%% Return two maps containing only the fields that differ. +diff_maps(M1,M2) -> + diffs(lists:sort(maps:to_list(M1)),lists:sort(maps:to_list(M2)),#{},#{}). + +diffs([H|T1],[H|T2],D1,D2) -> + diffs(T1,T2,D1,D2); +diffs([{K,V1}|T1],[{K,V2}|T2],D1,D2) -> + diffs(T1,T2,D1#{K=>V1},D2#{K=>V2}); +diffs([],[],D1,D2) -> + {D1,D2}. diff --git a/lib/kernel/src/logger_simple_h.erl b/lib/kernel/src/logger_simple_h.erl index 8b51dd8569..fe181722f3 100644 --- a/lib/kernel/src/logger_simple_h.erl +++ b/lib/kernel/src/logger_simple_h.erl @@ -50,7 +50,6 @@ removing_handler(#{id:=simple}) -> ok; Pid -> Ref = erlang:monitor(process,Pid), - unlink(Pid), Pid ! stop, receive {'DOWN',Ref,process,Pid,_} -> ok @@ -99,7 +98,11 @@ loop(Buffer) -> replay_buffer(Buffer); _ -> ok - end; + end, + %% Before stopping, we unlink the logger process to avoid + %% an unexpected EXIT message + unlink(whereis(logger)), + ok; {log,#{msg:=_,meta:=#{time:=_}}=Log} -> do_log(Log), loop(update_buffer(Buffer,Log)); diff --git a/lib/kernel/src/logger_std_h.erl b/lib/kernel/src/logger_std_h.erl index 66fa6b6ab6..0669164bb6 100644 --- a/lib/kernel/src/logger_std_h.erl +++ b/lib/kernel/src/logger_std_h.erl @@ -19,8 +19,6 @@ %% -module(logger_std_h). --behaviour(gen_server). - -include("logger.hrl"). -include("logger_internal.hrl"). -include("logger_h_common.hrl"). @@ -28,185 +26,64 @@ -include_lib("kernel/include/file.hrl"). %% API --export([start_link/3, info/1, filesync/1, reset/1]). +-export([filesync/1]). -%% gen_server and proc_lib callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). +%% logger_h_common callbacks +-export([init/2, check_config/4, reset_state/2, + filesync/3, write/4, handle_info/3, terminate/3]). %% logger callbacks --export([log/2, adding_handler/1, removing_handler/1, changing_config/2]). +-export([log/2, adding_handler/1, removing_handler/1, changing_config/3, + filter_config/1]). -%% handler internal --export([log_handler_info/4]). +-define(DEFAULT_CALL_TIMEOUT, 5000). %%%=================================================================== %%% API %%%=================================================================== %%%----------------------------------------------------------------- -%%% Start a standard handler process and link to caller. -%%% This function is called by the kernel supervisor when this -%%% handler process gets added --spec start_link(Name, Config, HandlerState) -> {ok,Pid} | {error,Reason} when - Name :: atom(), - Config :: logger:handler_config(), - HandlerState :: map(), - Pid :: pid(), - Reason :: term(). - -start_link(Name, Config, HandlerState) -> - proc_lib:start_link(?MODULE,init,[[Name,Config,HandlerState]]). - -%%%----------------------------------------------------------------- %%% -spec filesync(Name) -> ok | {error,Reason} when Name :: atom(), Reason :: handler_busy | {badarg,term()}. -filesync(Name) when is_atom(Name) -> - try - gen_server:call(?name_to_reg_name(?MODULE,Name), - filesync, ?DEFAULT_CALL_TIMEOUT) - catch - _:{timeout,_} -> {error,handler_busy} - end; filesync(Name) -> - {error,{badarg,{filesync,[Name]}}}. - -%%%----------------------------------------------------------------- -%%% --spec info(Name) -> Info | {error,Reason} when - Name :: atom(), - Info :: term(), - Reason :: handler_busy | {badarg,term()}. - -info(Name) when is_atom(Name) -> - try - gen_server:call(?name_to_reg_name(?MODULE,Name), - info, ?DEFAULT_CALL_TIMEOUT) - catch - _:{timeout,_} -> {error,handler_busy} - end; -info(Name) -> - {error,{badarg,{info,[Name]}}}. - -%%%----------------------------------------------------------------- -%%% --spec reset(Name) -> ok | {error,Reason} when - Name :: atom(), - Reason :: handler_busy | {badarg,term()}. - -reset(Name) when is_atom(Name) -> - try - gen_server:call(?name_to_reg_name(?MODULE,Name), - reset, ?DEFAULT_CALL_TIMEOUT) - catch - _:{timeout,_} -> {error,handler_busy} - end; -reset(Name) -> - {error,{badarg,{reset,[Name]}}}. - + logger_h_common:filesync(?MODULE,Name). %%%=================================================================== -%%% logger callbacks +%%% logger callbacks - just forward to logger_h_common %%%=================================================================== %%%----------------------------------------------------------------- %%% Handler being added -adding_handler(#{id:=Name}=Config) -> - case check_config(adding, Config) of - {ok, Config1} -> - %% create initial handler state by merging defaults with config - HConfig = maps:get(config, Config1, #{}), - HState = maps:merge(get_init_state(), HConfig), - case logger_h_common:overload_levels_ok(HState) of - true -> - start(Name, Config1, HState); - false -> - #{sync_mode_qlen := SMQL, - drop_mode_qlen := DMQL, - flush_qlen := FQL} = HState, - {error,{invalid_levels,{SMQL,DMQL,FQL}}} - end; - Error -> - Error - end. +-spec adding_handler(Config) -> {ok,Config} | {error,Reason} when + Config :: logger:handler_config(), + Reason :: term(). + +adding_handler(Config) -> + logger_h_common:adding_handler(Config). %%%----------------------------------------------------------------- %%% Updating handler config -changing_config(OldConfig=#{id:=Name, config:=OldHConfig}, - NewConfig=#{id:=Name}) -> - #{type:=Type, handler_pid:=HPid, mode_tab:=ModeTab} = OldHConfig, - NewHConfig = maps:get(config, NewConfig, #{}), - case maps:get(type, NewHConfig, Type) of - Type -> - NewHConfig1 = NewHConfig#{type=>Type, - handler_pid=>HPid, - mode_tab=>ModeTab}, - changing_config1(HPid, OldConfig, - NewConfig#{config=>NewHConfig1}); - _ -> - {error,{illegal_config_change,OldConfig,NewConfig}} - end; -changing_config(OldConfig, NewConfig) -> - {error,{illegal_config_change,OldConfig,NewConfig}}. - -changing_config1(HPid, OldConfig, NewConfig) -> - case check_config(changing, NewConfig) of - Result = {ok,NewConfig1} -> - try gen_server:call(HPid, {change_config,OldConfig,NewConfig1}, - ?DEFAULT_CALL_TIMEOUT) of - ok -> Result; - HError -> HError - catch - _:{timeout,_} -> {error,handler_busy} - end; - Error -> - Error - end. - -check_config(adding, Config) -> - %% Merge in defaults on handler level - HConfig0 = maps:get(config, Config, #{}), - HConfig = maps:merge(#{type => standard_io}, - HConfig0), - case check_h_config(maps:to_list(HConfig)) of - ok -> - {ok,Config#{config=>HConfig}}; - Error -> - Error - end; -check_config(changing, Config) -> - HConfig = maps:get(config, Config, #{}), - case check_h_config(maps:to_list(HConfig)) of - ok -> {ok,Config}; - Error -> Error - end. - -check_h_config([{type,Type} | Config]) when Type == standard_io; - Type == standard_error -> - check_h_config(Config); -check_h_config([{type,{file,File}} | Config]) when is_list(File) -> - check_h_config(Config); -check_h_config([{type,{file,File,Modes}} | Config]) when is_list(File), - is_list(Modes) -> - check_h_config(Config); -check_h_config([Other | Config]) -> - case logger_h_common:check_common_config(Other) of - valid -> - check_h_config(Config); - invalid -> - {error,{invalid_config,?MODULE,Other}} - end; -check_h_config([]) -> - ok. +-spec changing_config(SetOrUpdate, OldConfig, NewConfig) -> + {ok,Config} | {error,Reason} when + SetOrUpdate :: set | update, + OldConfig :: logger:handler_config(), + NewConfig :: logger:handler_config(), + Config :: logger:handler_config(), + Reason :: term(). +changing_config(SetOrUpdate, OldConfig, NewConfig) -> + logger_h_common:changing_config(SetOrUpdate, OldConfig, NewConfig). %%%----------------------------------------------------------------- %%% Handler being removed -removing_handler(#{id:=Name}) -> - stop(Name). +-spec removing_handler(Config) -> ok when + Config :: logger:handler_config(). + +removing_handler(Config) -> + logger_h_common:removing_handler(Config). %%%----------------------------------------------------------------- %%% Log a string or report @@ -214,192 +91,103 @@ removing_handler(#{id:=Name}) -> LogEvent :: logger:log_event(), Config :: logger:handler_config(). -log(LogEvent, Config = #{id := Name, - config := #{handler_pid := HPid, - mode_tab := ModeTab}}) -> - %% if the handler has crashed, we must drop this event - %% and hope the handler restarts so we can try again - true = is_process_alive(HPid), - Bin = logger_h_common:log_to_binary(LogEvent, Config), - logger_h_common:call_cast_or_drop(Name, HPid, ModeTab, Bin). +log(LogEvent, Config) -> + logger_h_common:log(LogEvent, Config). -%%%=================================================================== -%%% gen_server callbacks -%%%=================================================================== +%%%----------------------------------------------------------------- +%%% Remove internal fields from configuration +-spec filter_config(Config) -> Config when + Config :: logger:handler_config(). -init([Name, Config = #{config := HConfig}, - State0 = #{type := Type, file_ctrl_sync_int := FileCtrlSyncInt}]) -> - RegName = ?name_to_reg_name(?MODULE,Name), - register(RegName, self()), - process_flag(trap_exit, true), - process_flag(message_queue_data, off_heap), +filter_config(Config) -> + logger_h_common:filter_config(Config). - ?init_test_hooks(), - ?start_observation(Name), - - case do_init(Name, Type) of - {ok,InitState} -> - try ets:new(Name, [public]) of - ModeTab -> - ?set_mode(ModeTab, async), - State = maps:merge(State0, InitState), - T0 = ?timestamp(), - State1 = - ?merge_with_stats(State#{ - mode_tab => ModeTab, - mode => async, - file_ctrl_sync => FileCtrlSyncInt, - last_qlen => 0, - last_log_ts => T0, - last_op => sync, - burst_win_ts => T0, - burst_msg_count => 0}), - Config1 = - Config#{config => HConfig#{handler_pid => self(), - mode_tab => ModeTab}}, - proc_lib:init_ack({ok,self(),Config1}), - gen_server:cast(self(), repeated_filesync), - gen_server:enter_loop(?MODULE, [], State1) - catch - _:Error -> - unregister(RegName), - logger_h_common:error_notify({init_handler,Name,Error}), - proc_lib:init_ack(Error) - end; - Error -> - unregister(RegName), - logger_h_common:error_notify({init_handler,Name,Error}), - proc_lib:init_ack(Error) - end. - -do_init(Name, Type) -> +%%%=================================================================== +%%% logger_h_common callbacks +%%%=================================================================== +init(Name, #{type := Type}) -> case open_log_file(Name, Type) of {ok,FileCtrlPid} -> - case logger_h_common:unset_restart_flag(Name, ?MODULE) of - true -> - %% inform about restart - gen_server:cast(self(), {log_handler_info, - "Handler ~p restarted", - [Name]}); - false -> - %% initial start - ok - end, - {ok,#{id=>Name,type=>Type,file_ctrl_pid=>FileCtrlPid}}; + {ok,#{type=>Type,file_ctrl_pid=>FileCtrlPid}}; Error -> Error end. -%% This is the synchronous log event. -handle_call({log, Bin}, _From, State) -> - {Result,State1} = do_log(Bin, call, State), - %% Result == ok | dropped - {reply,Result, State1}; - -handle_call(filesync, _From, State = #{type := Type, - file_ctrl_pid := FileCtrlPid}) -> - if is_atom(Type) -> - {reply, ok, State}; - true -> - {reply, file_ctrl_filesync_sync(FileCtrlPid), State#{last_op=>sync}} - end; +check_config(_Name,set,undefined,NewHConfig) -> + check_config(maps:merge(get_default_config(),NewHConfig)); +check_config(_Name,SetOrUpdate,OldHConfig,NewHConfig0) -> + WriteOnce = maps:with([type],OldHConfig), + Default = + case SetOrUpdate of + set -> + %% Do not reset write-once fields to defaults + maps:merge(get_default_config(),WriteOnce); + update -> + OldHConfig + end, -handle_call({change_config,_OldConfig,NewConfig}, _From, - State = #{filesync_repeat_interval := FSyncInt0}) -> - HConfig = maps:get(config, NewConfig, #{}), - State1 = maps:merge(State, HConfig), - case logger_h_common:overload_levels_ok(State1) of - true -> - _ = - case maps:get(filesync_repeat_interval, HConfig, undefined) of - undefined -> - ok; - no_repeat -> - _ = logger_h_common:cancel_timer(maps:get(rep_sync_tref, - State, - undefined)); - FSyncInt0 -> - ok; - _FSyncInt1 -> - _ = logger_h_common:cancel_timer(maps:get(rep_sync_tref, - State, - undefined)), - gen_server:cast(self(), repeated_filesync) - end, - {reply, ok, State1}; - false -> - #{sync_mode_qlen := SMQL, - drop_mode_qlen := DMQL, - flush_qlen := FQL} = State1, - {reply, {error,{invalid_levels,{SMQL,DMQL,FQL}}}, State} - end; + NewHConfig = maps:merge(Default, NewHConfig0), -handle_call(info, _From, State) -> - {reply, State, State}; - -handle_call(reset, _From, State) -> - State1 = ?merge_with_stats(State), - {reply, ok, State1#{last_qlen => 0, - last_log_ts => ?timestamp()}}; - -handle_call(stop, _From, State) -> - {stop, {shutdown,stopped}, ok, State}. - -%% This is the asynchronous log event. -handle_cast({log, Bin}, State) -> - {_,State1} = do_log(Bin, cast, State), - {noreply, State1}; - -handle_cast({log_handler_info, Format, Args}, State = #{id:=Name}) -> - log_handler_info(Name, Format, Args, State), - {noreply, State}; - -%% If FILESYNC_REPEAT_INTERVAL is set to a millisec value, this -%% clause gets called repeatedly by the handler. In order to -%% guarantee that a filesync *always* happens after the last log -%% event, the repeat operation must be active! -handle_cast(repeated_filesync, - State = #{type := Type, - file_ctrl_pid := FileCtrlPid, - filesync_repeat_interval := FSyncInt, - last_op := LastOp}) -> - State1 = - if not is_atom(Type), is_integer(FSyncInt) -> - %% only do filesync if something has been - %% written since last time we checked - if LastOp == sync -> - ok; - true -> - file_ctrl_filesync_async(FileCtrlPid) - end, - {ok,TRef} = - timer:apply_after(FSyncInt, gen_server,cast, - [self(),repeated_filesync]), - State#{rep_sync_tref => TRef, last_op => sync}; - true -> - State - end, - {noreply,State1}. - -handle_info({'EXIT',Pid,Why}, State = #{id := Name, type := FileInfo}) -> - case maps:get(file_ctrl_pid, State, undefined) of - Pid -> - %% file error, terminate handler - logger_h_common:handler_exit(Name, - {error,{write_failed,FileInfo,Why}}); - _Other -> - %% ignore EXIT - ok - end, - {noreply, State}; + %% Fail if write-once fields are changed + case maps:with([type],NewHConfig) of + WriteOnce -> + check_config(NewHConfig); + Other -> + {error,{illegal_config_change,?MODULE,WriteOnce,Other}} + end. -handle_info(_Info, State) -> - {noreply, State}. +check_config(#{type:=Type}=HConfig) -> + case check_h_config(maps:to_list(HConfig)) of + ok when is_atom(Type) -> + {ok,HConfig#{filesync_repeat_interval=>no_repeat}}; + ok -> + {ok,HConfig}; + {error,{Key,Value}} -> + {error,{invalid_config,?MODULE,#{Key=>Value}}} + end. -terminate(Reason, State = #{id:=Name, file_ctrl_pid:=FWPid, - type:=_FileInfo}) -> - _ = logger_h_common:cancel_timer(maps:get(rep_sync_tref, State, - undefined)), +check_h_config([{type,Type} | Config]) when Type == standard_io; + Type == standard_error -> + check_h_config(Config); +check_h_config([{type,{file,File}} | Config]) when is_list(File) -> + check_h_config(Config); +check_h_config([{type,{file,File,Modes}} | Config]) when is_list(File), + is_list(Modes) -> + check_h_config(Config); +check_h_config([Other | _]) -> + {error,Other}; +check_h_config([]) -> + ok. + +get_default_config() -> + #{type => standard_io}. + +filesync(_Name, _Mode, #{type := Type}=State) when is_atom(Type) -> + {ok,State}; +filesync(_Name, async, #{file_ctrl_pid := FileCtrlPid} = State) -> + ok = file_ctrl_filesync_async(FileCtrlPid), + {ok,State}; +filesync(_Name, sync, #{file_ctrl_pid := FileCtrlPid} = State) -> + Result = file_ctrl_filesync_sync(FileCtrlPid), + {Result,State}. + +write(_Name, async, Bin, #{file_ctrl_pid:=FileCtrlPid} = State) -> + ok = file_write_async(FileCtrlPid, Bin), + {ok,State}; +write(_Name, sync, Bin, #{file_ctrl_pid:=FileCtrlPid} = State) -> + Result = file_write_sync(FileCtrlPid, Bin), + {Result,State}. + +reset_state(_Name, State) -> + State. + +handle_info(_Name, {'EXIT',Pid,Why}, #{type := FileInfo, file_ctrl_pid := Pid}) -> + %% file_ctrl_pid died, file error, terminate handler + exit({error,{write_failed,FileInfo,Why}}); +handle_info(_, _, State) -> + State. + +terminate(_Name, _Reason, #{file_ctrl_pid:=FWPid}) -> case is_process_alive(FWPid) of true -> unlink(FWPid), @@ -410,17 +198,12 @@ terminate(Reason, State = #{id:=Name, file_ctrl_pid:=FWPid, ok after ?DEFAULT_CALL_TIMEOUT -> - exit(FWPid, kill) + exit(FWPid, kill), + ok end; false -> ok - end, - ok = logger_h_common:stop_or_restart(Name, Reason, State), - unregister(?name_to_reg_name(?MODULE, Name)), - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. + end. %%%=================================================================== %%% Internal functions @@ -428,200 +211,6 @@ code_change(_OldVsn, State, _Extra) -> %%%----------------------------------------------------------------- %%% -get_init_state() -> - #{sync_mode_qlen => ?SYNC_MODE_QLEN, - drop_mode_qlen => ?DROP_MODE_QLEN, - flush_qlen => ?FLUSH_QLEN, - burst_limit_enable => ?BURST_LIMIT_ENABLE, - burst_limit_max_count => ?BURST_LIMIT_MAX_COUNT, - burst_limit_window_time => ?BURST_LIMIT_WINDOW_TIME, - overload_kill_enable => ?OVERLOAD_KILL_ENABLE, - overload_kill_qlen => ?OVERLOAD_KILL_QLEN, - overload_kill_mem_size => ?OVERLOAD_KILL_MEM_SIZE, - overload_kill_restart_after => ?OVERLOAD_KILL_RESTART_AFTER, - file_ctrl_sync_int => ?CONTROLLER_SYNC_INTERVAL, - filesync_ok_qlen => ?FILESYNC_OK_QLEN, - filesync_repeat_interval => ?FILESYNC_REPEAT_INTERVAL}. - -%%%----------------------------------------------------------------- -%%% Add a standard handler to the logger. -%%% This starts a dedicated handler process which should always -%%% exist if the handler is registered with logger (and should not -%%% exist if the handler is not registered). -%%% -%%% Handler specific config should be provided with a sub map associated -%%% with a key named 'config', e.g: -%%% -%%% Config = #{config => #{sync_mode_qlen => 50} -%%% -%%% The standard handler process is linked to logger_sup, which is -%%% part of the kernel application's supervision tree. -start(Name, Config, HandlerState) -> - LoggerStdH = - #{id => Name, - start => {?MODULE, start_link, [Name,Config,HandlerState]}, - restart => temporary, - shutdown => 2000, - type => worker, - modules => [?MODULE]}, - case supervisor:start_child(logger_sup, LoggerStdH) of - {ok,Pid,Config1} -> - ok = logger_handler_watcher:register_handler(Name,Pid), - {ok,Config1}; - Error -> - Error - end. - -%%%----------------------------------------------------------------- -%%% Stop and remove the handler. -stop(Name) -> - case whereis(?name_to_reg_name(?MODULE,Name)) of - undefined -> - ok; - Pid -> - %% We don't want to do supervisor:terminate_child here - %% since we need to distinguish this explicit stop from a - %% system termination in order to avoid circular attempts - %% at removing the handler (implying deadlocks and - %% timeouts). - %% And we don't need to do supervisor:delete_child, since - %% the restart type is temporary, which means that the - %% child specification is automatically removed from the - %% supervisor when the process dies. - _ = gen_server:call(Pid, stop), - ok - end. - -%%%----------------------------------------------------------------- -%%% Logging and overload control. --define(update_file_ctrl_sync(C, Interval), - if C == 0 -> Interval; - true -> C-1 end). - -%% check for overload between every event (and set Mode to async, -%% sync or drop accordingly), but never flush the whole mailbox -%% before LogWindowSize events have been handled -do_log(Bin, CallOrCast, State = #{id:=Name, mode:=Mode0}) -> - T1 = ?timestamp(), - - %% check if the handler 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} = logger_h_common:check_load(State), - - if (Mode1 == drop) andalso (Mode0 =/= drop) -> - log_handler_info(Name, "Handler ~p switched to drop mode", - [Name], State); - (Mode0 == drop) andalso ((Mode1 == async) orelse (Mode1 == sync)) -> - log_handler_info(Name, "Handler ~p switched to ~w mode", - [Name,Mode1], State); - true -> - ok - end, - - %% kill the handler if it can't keep up with the load - logger_h_common:kill_if_choked(Name, QLen, Mem, ?MODULE, State), - - if Mode1 == flush -> - flush(Name, QLen, T1, State1); - true -> - write(Name, Mode1, T1, Bin, CallOrCast, State1) - end. - -%% this clause is called by do_log/3 after an overload check -%% has been performed, where QLen > FlushQLen -flush(Name, _QLen0, T1, State=#{last_log_ts := _T0, mode_tab := ModeTab}) -> - %% flush messages in the mailbox (a limited number in - %% order to not cause long delays) - NewFlushed = logger_h_common:flush_log_events(?FLUSH_MAX_N), - - %% write info in log about flushed messages - log_handler_info(Name, "Handler ~p flushed ~w log events", - [Name,NewFlushed], State), - - %% 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}), - - %% Add 1 for the current log event - ?observe(Name,{flushed,NewFlushed+1}), - - State1 = ?update_max_time(?diff_time(T1,_T0),State), - {dropped,?update_other(flushed,FLUSHED,NewFlushed, - State1#{mode => ?set_mode(ModeTab,async), - last_qlen => 0, - last_log_ts => T1})}. - -%% this clause is called to write to file -write(_Name, Mode, T1, Bin, _CallOrCast, - State = #{mode_tab := ModeTab, - file_ctrl_pid := FileCtrlPid, - file_ctrl_sync := FileCtrlSync, - last_qlen := LastQLen, - last_log_ts := T0, - file_ctrl_sync_int := FileCtrlSyncInt}) -> - %% check if we need to limit the number of writes - %% during a burst of log events - {DoWrite,BurstWinT,BurstMsgCount} = logger_h_common:limit_burst(State), - - %% only send a synhrounous event to the file controller process - %% every FileCtrlSyncInt time, to give the handler time between - %% file writes so it can keep up with incoming messages - {Result,LastQLen1} = - if DoWrite, FileCtrlSync == 0 -> - ?observe(_Name,{_CallOrCast,1}), - file_write_sync(FileCtrlPid, Bin, false), - {ok,element(2, process_info(self(), message_queue_len))}; - DoWrite -> - ?observe(_Name,{_CallOrCast,1}), - file_write_async(FileCtrlPid, Bin), - {ok,LastQLen}; - not DoWrite -> - ?observe(_Name,{flushed,1}), - {dropped,LastQLen} - end, - - %% Check if the time since the previous log event is long enough - - %% and the queue length small enough - to assume the mailbox has - %% been emptied, and if so, do filesync operation and reset mode to - %% async. Note that this is the best we can do to detect an idle - %% handler without setting a timer after each log call/cast. If the - %% time between two consecutive log events is fast and no new - %% event comes in after the last one, idle state won't be detected! - Time = ?diff_time(T1,T0), - {Mode1,BurstMsgCount1} = - if (LastQLen1 < ?FILESYNC_OK_QLEN) andalso - (Time > ?IDLE_DETECT_TIME_USEC) -> - %% do filesync if necessary - case maps:get(type, State) of - Std when is_atom(Std) -> - ok; - _File -> - file_ctrl_filesync_async(FileCtrlPid) - end, - {?change_mode(ModeTab, Mode, async),0}; - true -> - {Mode,BurstMsgCount} - end, - State1 = - ?update_calls_or_casts(_CallOrCast,1,State), - State2 = - ?update_max_time(Time, - State1#{mode => Mode1, - last_qlen := LastQLen1, - last_log_ts => T1, - last_op => write, - burst_win_ts => BurstWinT, - burst_msg_count => BurstMsgCount1, - file_ctrl_sync => - ?update_file_ctrl_sync(FileCtrlSync, - FileCtrlSyncInt)}), - {Result,State2}. - open_log_file(HandlerName, FileInfo) -> case file_ctrl_start(HandlerName, FileInfo) of OK = {ok,_FileCtrlPid} -> OK; @@ -653,19 +242,6 @@ close_log_file(Fd) -> _ = file:close(Fd). -log_handler_info(Name, Format, Args, #{file_ctrl_pid := FileCtrlPid}) -> - Config = - case logger:get_handler_config(Name) of - {ok,Conf} -> Conf; - _ -> #{formatter=>{?DEFAULT_FORMATTER,?DEFAULT_FORMAT_CONFIG}} - end, - Meta = #{time=>erlang:system_time(microsecond)}, - Bin = logger_h_common:log_to_binary(#{level => notice, - msg => {Format,Args}, - meta => Meta}, Config), - _ = file_write_async(FileCtrlPid, Bin), - ok. - %%%----------------------------------------------------------------- %%% File control process @@ -692,24 +268,19 @@ file_write_async(Pid, Bin) -> Pid ! {log,Bin}, ok. -file_write_sync(Pid, Bin, FileSync) -> - case file_ctrl_call(Pid, {log,self(),Bin,FileSync}) of - {error,Reason} -> - {error,{write_failed,Bin,Reason}}; - Result -> - Result - end. +file_write_sync(Pid, Bin) -> + file_ctrl_call(Pid, {log,Bin}). file_ctrl_filesync_async(Pid) -> Pid ! filesync, ok. file_ctrl_filesync_sync(Pid) -> - file_ctrl_call(Pid, {filesync,self()}). + file_ctrl_call(Pid, filesync). file_ctrl_call(Pid, Msg) -> MRef = monitor(process, Pid), - Pid ! {Msg,MRef}, + Pid ! {Msg,{self(),MRef}}, receive {MRef,Result} -> demonitor(MRef, [flush]), @@ -727,74 +298,43 @@ file_ctrl_init(HandlerName, FileInfo, Starter) when is_tuple(FileInfo) -> case do_open_log_file(FileInfo) of {ok,Fd} -> Starter ! {self(),ok}, - file_ctrl_loop(Fd, file, FileName, false, ok, ok, HandlerName); + file_ctrl_loop(Fd, FileName, false, ok, ok, HandlerName); {error,Reason} -> Starter ! {self(),{error,{open_failed,FileName,Reason}}} end; file_ctrl_init(HandlerName, StdDev, Starter) -> Starter ! {self(),ok}, - file_ctrl_loop(StdDev, standard_io, StdDev, false, ok, ok, HandlerName). + file_ctrl_loop(StdDev, StdDev, false, ok, ok, HandlerName). -file_ctrl_loop(Fd, Type, DevName, Synced, +file_ctrl_loop(Fd, DevName, Synced, PrevWriteResult, PrevSyncResult, HandlerName) -> receive %% asynchronous event {log,Bin} -> - Result = if Type == file -> - write_to_dev(Fd, Bin, DevName, - PrevWriteResult, HandlerName); - true -> - io:put_chars(Fd, Bin) - end, - file_ctrl_loop(Fd, Type, DevName, false, + Fd1 = ensure(Fd, DevName), + Result = write_to_dev(Fd1, Bin, DevName, PrevWriteResult, HandlerName), + file_ctrl_loop(Fd1, DevName, false, Result, PrevSyncResult, HandlerName); %% synchronous event - {{log,From,Bin,FileSync},MRef} -> - if Type == file -> - %% check that file hasn't been deleted - CheckFile = - fun() -> {ok,_} = file:read_file_info(DevName) end, - spawn_link(CheckFile), - WResult = write_to_dev(Fd, Bin, DevName, - PrevWriteResult, HandlerName), - {Synced1,SResult} = - if not FileSync -> - {false,PrevSyncResult}; - true -> - case sync_dev(Fd, DevName, - PrevSyncResult, HandlerName) of - ok -> {true,ok}; - Error -> {false,Error} - end - end, - From ! {MRef,ok}, - file_ctrl_loop(Fd, Type, DevName, Synced1, - WResult, SResult, HandlerName); - true -> - _ = io:put_chars(Fd, Bin), - From ! {MRef,ok}, - file_ctrl_loop(Fd, Type, DevName, false, - ok, PrevSyncResult, HandlerName) - end; + {{log,Bin},{From,MRef}} -> + Fd1 = ensure(Fd, DevName), + Result = write_to_dev(Fd1, Bin, DevName, PrevWriteResult, HandlerName), + From ! {MRef,ok}, + file_ctrl_loop(Fd1, DevName, false, + Result, PrevSyncResult, HandlerName); - filesync when not Synced -> - Result = sync_dev(Fd, DevName, PrevSyncResult, HandlerName), - file_ctrl_loop(Fd, Type, DevName, true, + filesync -> + Fd1 = ensure(Fd, DevName), + Result = sync_dev(Fd1, DevName, Synced, PrevSyncResult, HandlerName), + file_ctrl_loop(Fd1, DevName, true, PrevWriteResult, Result, HandlerName); - filesync -> - file_ctrl_loop(Fd, Type, DevName, true, - PrevWriteResult, PrevSyncResult, HandlerName); - - {{filesync,From},MRef} -> - Result = if not Synced -> - sync_dev(Fd, DevName, PrevSyncResult, HandlerName); - true -> - ok - end, + {filesync,{From,MRef}} -> + Fd1 = ensure(Fd, DevName), + Result = sync_dev(Fd1, DevName, Synced, PrevSyncResult, HandlerName), From ! {MRef,ok}, - file_ctrl_loop(Fd, Type, DevName, true, + file_ctrl_loop(Fd1, DevName, true, PrevWriteResult, Result, HandlerName); stop -> @@ -802,27 +342,44 @@ file_ctrl_loop(Fd, Type, DevName, Synced, stopped end. -write_to_dev(Fd, Bin, FileName, PrevWriteResult, HandlerName) -> - case ?file_write(Fd, Bin) of - ok -> - ok; - PrevWriteResult -> - %% don't report same error twice - PrevWriteResult; - Error -> - logger_h_common:error_notify({HandlerName,write,FileName,Error}), - Error +%% In order to play well with tools like logrotate, we need to be able +%% to re-create the file if it has disappeared (e.g. if rotated by +%% logrotate) +ensure(Fd,DevName) when is_atom(DevName) -> + Fd; +ensure(Fd,FileName) -> + case file:read_file_info(FileName) of + {ok,_} -> + Fd; + _ -> + _ = file:close(Fd), + _ = file:close(Fd), % delayed_write cause close not to close + case do_open_log_file({file,FileName}) of + {ok,Fd1} -> + Fd1; + Error -> + exit({could_not_reopen_file,Error}) + end end. -sync_dev(Fd, DevName, PrevSyncResult, HandlerName) -> - case ?file_datasync(Fd) of - ok -> - ok; - PrevSyncResult -> - %% don't report same error twice - PrevSyncResult; - Error -> - logger_h_common:error_notify({HandlerName,filesync,DevName,Error}), - Error - end. +write_to_dev(DevName, Bin, _DevName, _PrevWriteResult, _HandlerName) + when is_atom(DevName) -> + io:put_chars(DevName, Bin); +write_to_dev(Fd, Bin, FileName, PrevWriteResult, HandlerName) -> + Result = ?file_write(Fd, Bin), + maybe_notify_error(write,Result,PrevWriteResult,FileName,HandlerName). +sync_dev(_Fd, _FileName, true, PrevSyncResult, _HandlerName) -> + PrevSyncResult; +sync_dev(Fd, FileName, false, PrevSyncResult, HandlerName) -> + Result = ?file_datasync(Fd), + maybe_notify_error(filesync,Result,PrevSyncResult,FileName,HandlerName). + +maybe_notify_error(_Op, ok, _PrevResult, _FileName, _HandlerName) -> + ok; +maybe_notify_error(_Op, PrevResult, PrevResult, _FileName, _HandlerName) -> + %% don't report same error twice + PrevResult; +maybe_notify_error(Op, Error, _PrevResult, FileName, HandlerName) -> + logger_h_common:error_notify({HandlerName,Op,FileName,Error}), + Error. 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/src/net_kernel.erl b/lib/kernel/src/net_kernel.erl index ef92f9f4d1..4915193196 100644 --- a/lib/kernel/src/net_kernel.erl +++ b/lib/kernel/src/net_kernel.erl @@ -369,11 +369,11 @@ do_auto_connect_1(Node, ConnId, From, State) -> end. do_auto_connect_2(Node, passive_cnct, From, State, ConnLookup) -> - case (catch erts_internal:new_connection(Node)) of - {Nr,_DHandle}=ConnId when is_integer(Nr) -> - do_auto_connect_2(Node, ConnId, From, State, ConnLookup); - - _Error -> + try erts_internal:new_connection(Node) of + ConnId -> + do_auto_connect_2(Node, ConnId, From, State, ConnLookup) + catch + _:_ -> error_logger:error_msg("~n** Cannot get connection id for node ~w~n", [Node]), {reply, false, State} @@ -406,7 +406,7 @@ do_auto_connect_2(Node, ConnId, From, State, ConnLookup) -> erts_internal:abort_connection(Node, ConnId), {reply, false, State}; _ -> - case setup(ConnLookup, Node,ConnId,normal,From,State) of + case setup(Node, ConnId, normal, From, State) of {ok, SetupPid} -> Owners = [{SetupPid, Node} | State#state.conn_owners], {noreply,State#state{conn_owners=Owners}}; @@ -430,8 +430,8 @@ do_explicit_connect([#connection{conn_id = ConnId}=Conn], _, _, ConnId, From, St do_explicit_connect([#barred_connection{}], Type, Node, ConnId, From , State) -> %% Barred connection only affects auto_connect, ignore it. do_explicit_connect([], Type, Node, ConnId, From , State); -do_explicit_connect(ConnLookup, Type, Node, ConnId, From , State) -> - case setup(ConnLookup, Node,ConnId,Type,From,State) of +do_explicit_connect(_ConnLookup, Type, Node, ConnId, From , State) -> + case setup(Node,ConnId,Type,From,State) of {ok, SetupPid} -> Owners = [{SetupPid, Node} | State#state.conn_owners], {noreply,State#state{conn_owners=Owners}}; @@ -440,18 +440,6 @@ do_explicit_connect(ConnLookup, Type, Node, ConnId, From , State) -> {reply, false, State} end. --define(ERTS_DIST_CON_ID_MASK, 16#ffffff). % also in external.h - -verify_new_conn_id([], {Nr,_DHandle}) - when (Nr band (bnot ?ERTS_DIST_CON_ID_MASK)) =:= 0 -> - true; -verify_new_conn_id([#connection{conn_id = {Old,_}}], {New,_}) - when New =:= ((Old+1) band ?ERTS_DIST_CON_ID_MASK) -> - true; -verify_new_conn_id(_, _) -> - false. - - %% ------------------------------------------------------------ %% handle_call. @@ -477,8 +465,8 @@ handle_call({connect, _, Node}, From, State) when Node =:= node() -> handle_call({connect, Type, Node}, From, State) -> verbose({connect, Type, Node}, 1, State), ConnLookup = ets:lookup(sys_dist, Node), - R = case (catch erts_internal:new_connection(Node)) of - {Nr,_DHandle}=ConnId when is_integer(Nr) -> + R = try erts_internal:new_connection(Node) of + ConnId -> R1 = do_explicit_connect(ConnLookup, Type, Node, ConnId, From, State), case R1 of {reply, true, _S} -> %% already connected @@ -488,9 +476,10 @@ handle_call({connect, Type, Node}, From, State) -> {reply, false, _S} -> %% connection refused erts_internal:abort_connection(Node, ConnId) end, - R1; + R1 - _Error -> + catch + _:_ -> error_logger:error_msg("~n** Cannot get connection id for node ~w~n", [Node]), {reply, false, State} @@ -708,9 +697,9 @@ terminate(_Reason, State) -> %% %% Asynchronous auto connect request %% -handle_info({auto_connect,Node, Nr, DHandle}, State) -> - verbose({auto_connect, Node, Nr, DHandle}, 1, State), - ConnId = {Nr, DHandle}, +handle_info({auto_connect,Node, DHandle}, State) -> + verbose({auto_connect, Node, DHandle}, 1, State), + ConnId = DHandle, NewState = case do_auto_connect_1(Node, ConnId, noreply, State) of {noreply, S} -> %% Pending connection @@ -804,8 +793,8 @@ handle_info({AcceptPid, {accept_pending,MyNode,Node,Address,Type}}, State) -> AcceptPid ! {self(), {accept_pending, already_pending}}, {noreply, State}; _ -> - case (catch erts_internal:new_connection(Node)) of - {Nr,_DHandle}=ConnId when is_integer(Nr) -> + try erts_internal:new_connection(Node) of + ConnId -> ets:insert(sys_dist, #connection{node = Node, conn_id = ConnId, state = pending, @@ -814,12 +803,13 @@ handle_info({AcceptPid, {accept_pending,MyNode,Node,Address,Type}}, State) -> type = Type}), AcceptPid ! {self(),{accept_pending,ok}}, Owners = [{AcceptPid,Node} | State#state.conn_owners], - {noreply, State#state{conn_owners = Owners}}; - - _ -> + {noreply, State#state{conn_owners = Owners}} + catch + _:_ -> error_logger:error_msg("~n** Cannot get connection id for node ~w~n", [Node]), - AcceptPid ! {self(),{accept_pending,nok_pending}} + AcceptPid ! {self(),{accept_pending,nok_pending}}, + {noreply, State} end end; @@ -1283,8 +1273,8 @@ spawn_func(_,{From,Tag},M,F,A,Gleader) -> %% Set up connection to a new node. %% ----------------------------------------------------------- -setup(ConnLookup, Node,ConnId,Type,From,State) -> - case setup_check(ConnLookup, Node, ConnId, State) of +setup(Node, ConnId, Type, From, State) -> + case setup_check(Node, State) of {ok, L} -> Mod = L#listen.module, LAddr = L#listen.address, @@ -1313,7 +1303,7 @@ setup(ConnLookup, Node,ConnId,Type,From,State) -> Error end. -setup_check(ConnLookup, Node, ConnId, State) -> +setup_check(Node, State) -> Allowed = State#state.allowed, case lists:member(Node, Allowed) of false when Allowed =/= [] -> @@ -1321,16 +1311,9 @@ setup_check(ConnLookup, Node, ConnId, State) -> "disallowed node ~w ** ~n", [Node]), {error, bad_node}; _ -> - case verify_new_conn_id(ConnLookup, ConnId) of - false -> - error_msg("** Connection attempt to ~w with " - "bad connection id ~w ** ~n", [Node, ConnId]), - {error, bad_conn_id}; - true -> - case select_mod(Node, State#state.listen) of - {ok, _L}=OK -> OK; - Error -> Error - end + case select_mod(Node, State#state.listen) of + {ok, _L}=OK -> OK; + Error -> Error end end. @@ -1450,7 +1433,7 @@ validate_hostname([$@|HostPart] = Host) -> end. valid_name_head(Head) -> - {ok, MP} = re:compile("^[0-9A-Za-z_\\-]*$", [unicode]), + {ok, MP} = re:compile("^[0-9A-Za-z_\\-]+$", [unicode]), case re:run(Head, MP) of {match, _} -> true; diff --git a/lib/kernel/src/seq_trace.erl b/lib/kernel/src/seq_trace.erl index 14fe21e9de..4f9d7b3e5c 100644 --- a/lib/kernel/src/seq_trace.erl +++ b/lib/kernel/src/seq_trace.erl @@ -98,7 +98,7 @@ print(Label, Term) -> -spec reset_trace() -> 'true'. reset_trace() -> - erlang:system_flag(1, 0). + erlang:system_flag(reset_seq_trace, true). %% reset_trace(Pid) -> % this might be a useful function too diff --git a/lib/kernel/src/standard_error.erl b/lib/kernel/src/standard_error.erl index 5d649e5f94..ef5b532960 100644 --- a/lib/kernel/src/standard_error.erl +++ b/lib/kernel/src/standard_error.erl @@ -27,7 +27,8 @@ -define(PROCNAME_SUP, standard_error_sup). %% Defines for control ops --define(CTRL_OP_GET_WINSIZE,100). +-define(ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER, 16#018b0900). +-define(CTRL_OP_GET_WINSIZE, (100 + ?ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER)). %% %% The basic server and start-up. diff --git a/lib/kernel/src/user.erl b/lib/kernel/src/user.erl index 872e63ab53..0c9e1ea303 100644 --- a/lib/kernel/src/user.erl +++ b/lib/kernel/src/user.erl @@ -28,7 +28,8 @@ -define(NAME, user). %% Defines for control ops --define(CTRL_OP_GET_WINSIZE,100). +-define(ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER, 16#018b0900). +-define(CTRL_OP_GET_WINSIZE, (100 + ?ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER)). %% %% The basic server and start-up. diff --git a/lib/kernel/src/user_drv.erl b/lib/kernel/src/user_drv.erl index 9f914aa222..08286dd476 100644 --- a/lib/kernel/src/user_drv.erl +++ b/lib/kernel/src/user_drv.erl @@ -32,9 +32,10 @@ -define(OP_BEEP,4). -define(OP_PUTC_SYNC,5). % Control op --define(CTRL_OP_GET_WINSIZE,100). --define(CTRL_OP_GET_UNICODE_STATE,101). --define(CTRL_OP_SET_UNICODE_STATE,102). +-define(ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER, 16#018b0900). +-define(CTRL_OP_GET_WINSIZE, (100 + ?ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER)). +-define(CTRL_OP_GET_UNICODE_STATE, (101 + ?ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER)). +-define(CTRL_OP_SET_UNICODE_STATE, (102 + ?ERTS_TTYSL_DRV_CONTROL_MAGIC_NUMBER)). %% start() %% start(ArgumentList) |