切换日志系统
This commit is contained in:
parent
caf14634c3
commit
9f1075347e
@ -31,39 +31,31 @@
|
||||
|
||||
]},
|
||||
|
||||
%% 系统日志配置，系统日志为lager, 支持日志按日期自动分割
|
||||
{lager, [
|
||||
{colored, true},
|
||||
%% Whether to write a crash log, and where. Undefined means no crash logger.
|
||||
{crash_log, "trade_hub.crash.log"},
|
||||
%% Maximum size in bytes of events in the crash log - defaults to 65536
|
||||
{crash_log_msg_size, 65536},
|
||||
%% Maximum size of the crash log in bytes, before its rotated, set
|
||||
%% to 0 to disable rotation - default is 0
|
||||
{crash_log_size, 10485760},
|
||||
%% What time to rotate the crash log - default is no time
|
||||
%% rotation. See the README for a description of this format.
|
||||
{crash_log_date, "$D0"},
|
||||
%% Number of rotated crash logs to keep, 0 means keep only the
|
||||
%% current one - default is 0
|
||||
{crash_log_count, 5},
|
||||
%% Whether to redirect error_logger messages into lager - defaults to true
|
||||
{error_logger_redirect, true},
|
||||
%% 系统日志配置，使用 OTP logger
|
||||
{kernel, [
|
||||
%% 设置 Logger 的 primary log level
|
||||
{logger_level, debug},
|
||||
{logger, [
|
||||
{handler, default, logger_std_h,
|
||||
#{
|
||||
level => debug,
|
||||
formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
|
||||
}
|
||||
},
|
||||
|
||||
%% How big the gen_event mailbox can get before it is switched into sync mode
|
||||
{async_threshold, 20},
|
||||
%% Switch back to async mode, when gen_event mailbox size decrease from `async_threshold'
|
||||
%% to async_threshold - async_threshold_window
|
||||
{async_threshold_window, 5},
|
||||
|
||||
{handlers, [
|
||||
%% debug | info | warning | error, 日志级别
|
||||
{lager_console_backend, debug},
|
||||
{lager_file_backend, [{file, "debug.log"}, {level, debug}, {size, 314572800}]},
|
||||
{lager_file_backend, [{file, "notice.log"}, {level, notice}, {size, 314572800}]},
|
||||
{lager_file_backend, [{file, "error.log"}, {level, error}, {size, 314572800}]},
|
||||
{lager_file_backend, [{file, "info.log"}, {level, info}, {size, 314572800}]}
|
||||
]}
|
||||
{handler, disk, logger_disk_log_h,
|
||||
#{
|
||||
level => debug,
|
||||
config => #{
|
||||
file => "log/debug.log",
|
||||
max_no_files => 10,
|
||||
max_no_bytes => 524288000
|
||||
},
|
||||
formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
|
||||
}
|
||||
}
|
||||
|
||||
]}
|
||||
]}
|
||||
|
||||
].
|
||||
@ -1,11 +1,9 @@
|
||||
{erl_opts, [debug_info]}.
|
||||
{erl_opts, [debug_info, {i, "include"}]}.
|
||||
{deps, [
|
||||
{sync, ".*", {git, "https://github.com/rustyio/sync.git", {branch, "master"}}},
|
||||
{jiffy, ".*", {git, "https://github.com/davisp/jiffy.git", {tag, "1.1.2"}}},
|
||||
{cowboy, ".*", {git, "https://github.com/ninenines/cowboy.git", {tag, "2.10.0"}}},
|
||||
{gun, ".*", {git, "https://github.com/ninenines/gun.git", {tag, "2.2.0"}}},
|
||||
{parse_trans, ".*", {git, "https://github.com/uwiger/parse_trans", {tag, "3.0.0"}}},
|
||||
{lager, ".*", {git,"https://github.com/erlang-lager/lager.git", {tag, "3.9.2"}}}
|
||||
{gun, ".*", {git, "https://github.com/ninenines/gun.git", {tag, "2.2.0"}}}
|
||||
]}.
|
||||
|
||||
{relx, [{release, {efka, "0.1.0"},
|
||||
@ -43,6 +41,4 @@
|
||||
{pc, {git, "https://github.com/blt/port_compiler.git", {tag, "v1.15.0"}}}
|
||||
]}.
|
||||
|
||||
{erl_opts, [{parse_transform,lager_transform}]}.
|
||||
|
||||
{rebar_packages_cdn, "https://hexpm.upyun.com"}.
|
||||
@ -14,10 +14,10 @@
|
||||
|
||||
init(Req0, Opts) ->
|
||||
Method = binary_to_list(cowboy_req:method(Req0)),
|
||||
lager:debug("[upload_channel] method is: ~p", [Method]),
|
||||
logger:debug("[upload_channel] method is: ~p", [Method]),
|
||||
|
||||
Headers = cowboy_req:headers(Req0),
|
||||
lager:debug("headers is: ~p", [Headers]),
|
||||
logger:debug("headers is: ~p", [Headers]),
|
||||
case maps:find(<<"content-type">>, Headers) of
|
||||
{ok, <<"application/octet-stream">>} ->
|
||||
Filename = maps:get(<<"x-filename">>, Headers),
|
||||
@ -36,7 +36,7 @@ init(Req0, Opts) ->
|
||||
{ok, Req2, Opts}
|
||||
end;
|
||||
{ok, ContentType} ->
|
||||
lager:debug("[upload_channel] unexpect content-type: ~p", [ContentType]),
|
||||
logger:debug("[upload_channel] unexpect content-type: ~p", [ContentType]),
|
||||
Req = cowboy_req:reply(400, #{
|
||||
<<"Content-Type">> => <<"text/html;charset=utf-8">>
|
||||
}, <<"Expected application/octet-stream">>, Req0),
|
||||
|
||||
@ -36,7 +36,7 @@ init(Req, Opts) ->
|
||||
{cowboy_websocket, Req, Opts}.
|
||||
|
||||
websocket_init(_State) ->
|
||||
lager:debug("[ws_channel] get a new connection"),
|
||||
logger:debug("[ws_channel] get a new connection"),
|
||||
%% 初始状态为true
|
||||
{ok, #state{}}.
|
||||
|
||||
@ -45,11 +45,11 @@ websocket_handle(ping, State) ->
|
||||
|
||||
websocket_handle({text, Data}, State) ->
|
||||
Request = jiffy:decode(Data, [return_maps]),
|
||||
lager:debug("[ws_channle] get request: ~p", [Request]),
|
||||
logger:debug("[ws_channle] get request: ~p", [Request]),
|
||||
handle_request(Request, State);
|
||||
|
||||
websocket_handle(Info, State) ->
|
||||
lager:error("[ws_channel] get a unknown message: ~p, channel will closed", [Info]),
|
||||
logger:error("[ws_channel] get a unknown message: ~p, channel will closed", [Info]),
|
||||
{ok, State}.
|
||||
|
||||
%% 订阅的消息
|
||||
@ -59,13 +59,13 @@ websocket_info({topic_broadcast, Topic, Content}, State = #state{}) ->
|
||||
<<"params">> => #{<<"topic">> => Topic, <<"content">> => Content}
|
||||
}, [force_utf8])),
|
||||
|
||||
lager:debug("[ws_channel] will publish topic: ~p, message: ~p", [Topic, Req]),
|
||||
logger:debug("[ws_channel] will publish topic: ~p, message: ~p", [Topic, Req]),
|
||||
|
||||
{reply, {text, Req}, State};
|
||||
|
||||
%% service进程关闭
|
||||
websocket_info({'DOWN', _Ref, process, ServicePid, Reason}, State = #state{service_pid = ServicePid}) ->
|
||||
lager:debug("[ws_channel] container_pid: ~p, exited: ~p", [ServicePid, Reason]),
|
||||
logger:debug("[ws_channel] container_pid: ~p, exited: ~p", [ServicePid, Reason]),
|
||||
{stop, State#state{service_pid = undefined}};
|
||||
|
||||
%% stream进程关闭
|
||||
@ -105,12 +105,12 @@ websocket_info({stream_reply, StreamPid, Reply}, State = #state{stream_map = Str
|
||||
|
||||
%% 处理关闭信号
|
||||
websocket_info({stop, Reason}, State) ->
|
||||
lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
|
||||
logger:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
|
||||
{stop, State};
|
||||
|
||||
%% 处理其他未知消息
|
||||
websocket_info(Info, State) ->
|
||||
lager:debug("[ws_channel] channel get unknown info: ~p", [Info]),
|
||||
logger:debug("[ws_channel] channel get unknown info: ~p", [Info]),
|
||||
{ok, State}.
|
||||
|
||||
%% 进程关闭事件
|
||||
@ -121,7 +121,7 @@ terminate(Reason, _Req, State = #state{service_id = ServiceId, is_registered = I
|
||||
false ->
|
||||
ok
|
||||
end,
|
||||
lager:debug("[ws_channel] channel close with reason: ~p, state is: ~p", [Reason, State]),
|
||||
logger:debug("[ws_channel] channel close with reason: ~p, state is: ~p", [Reason, State]),
|
||||
ok.
|
||||
|
||||
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
|
||||
@ -150,7 +150,7 @@ handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> :=
|
||||
|
||||
{reply, {text, Reply}, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}};
|
||||
{error, Error} ->
|
||||
lager:warning("[ws_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]),
|
||||
logger:warning("[ws_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]),
|
||||
{stop, State}
|
||||
end;
|
||||
|
||||
@ -200,7 +200,7 @@ handle_request(#{<<"method">> := <<"metric_data">>,
|
||||
{ok, Metric} ->
|
||||
efka_service:metric_data(ServicePid, RouteKey, Metric);
|
||||
error ->
|
||||
lager:debug("[ws_channel] metric_data get invalid metric: ~p", Metric0)
|
||||
logger:debug("[ws_channel] metric_data get invalid metric: ~p", [Metric0])
|
||||
end,
|
||||
{ok, State}.
|
||||
|
||||
|
||||
@ -462,5 +462,5 @@ build_extra_hosts(Config) ->
|
||||
|
||||
-spec display_options(Options :: map()) -> no_return().
|
||||
display_options(Options) when is_map(Options) ->
|
||||
lager:debug("deploy options: ~p", [jiffy:encode(Options, [force_utf8])]),
|
||||
lists:foreach(fun({K, V}) -> lager:debug("~p => ~p", [K, V]) end, maps:to_list(Options)).
|
||||
logger:debug("deploy options: ~p", [jiffy:encode(Options, [force_utf8])]),
|
||||
lists:foreach(fun({K, V}) -> logger:debug("~p => ~p", [K, V]) end, maps:to_list(Options)).
|
||||
|
||||
@ -94,7 +94,7 @@ handle_info({timeout, _, attach_docker_events}, State = #state{port = undefined}
|
||||
end;
|
||||
handle_info({Port, {data, {eol, BinLine}}}, State = #state{port = Port}) ->
|
||||
Event = catch jiffy:decode(BinLine, [return_maps]),
|
||||
lager:debug("event: ~p", [Event]),
|
||||
logger:debug("event: ~p", [Event]),
|
||||
handle_event(Event, State),
|
||||
{noreply, State};
|
||||
|
||||
@ -105,7 +105,7 @@ handle_info({'DOWN', MRef, process, _Pid, _Reason}, State = #state{monitors = Mo
|
||||
|
||||
%% Port退出的时候，尝试重启
|
||||
handle_info({'EXIT', Port, Reason}, State = #state{port = Port}) ->
|
||||
lager:warning("[efka_docker_events] exit with reason: ~p", [Reason]),
|
||||
logger:warning("[efka_docker_events] exit with reason: ~p", [Reason]),
|
||||
try_attach_events(5000),
|
||||
{noreply, State#state{port = undefined}}.
|
||||
|
||||
|
||||
@ -95,7 +95,7 @@ handle_call({deploy, TaskId, Config = #{<<"container_name">> := ContainerName}},
|
||||
%% 创建目录
|
||||
{ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName),
|
||||
{ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Config),
|
||||
lager:debug("[docker_manager] start deploy task_id: ~p, config: ~p", [TaskId, Config]),
|
||||
logger:debug("[docker_manager] start deploy task_id: ~p, config: ~p", [TaskId, Config]),
|
||||
{reply, ok, State#state{task_map = maps:put(TaskPid, TaskId, TaskMap)}};
|
||||
|
||||
%% 处理容器关联的配置文件
|
||||
@ -106,10 +106,10 @@ handle_call({config_container, ContainerName, Config}, _From, State = #state{roo
|
||||
ConfigFile = docker_helper:get_config_file(ContainerDir),
|
||||
case file:write_file(ConfigFile, Config, [write, binary]) of
|
||||
ok ->
|
||||
lager:warning("[docker_manager] write config file: ~p success", [ConfigFile]),
|
||||
logger:warning("[docker_manager] write config file: ~p success", [ConfigFile]),
|
||||
{reply, ok, State};
|
||||
{error, Reason} ->
|
||||
lager:warning("[docker_manager] write config file: ~p, get error: ~p", [ConfigFile, Reason]),
|
||||
logger:warning("[docker_manager] write config file: ~p, get error: ~p", [ConfigFile, Reason]),
|
||||
{reply, {error, <<"write config failed">>}, State}
|
||||
end;
|
||||
error ->
|
||||
@ -187,13 +187,13 @@ handle_info({'DOWN', _Ref, process, TaskPid, Reason}, State = #state{task_map =
|
||||
{TaskId, NTaskMap} ->
|
||||
case Reason of
|
||||
normal ->
|
||||
lager:debug("[docker_manager] task_id: ~p, exit normal", [TaskId]),
|
||||
logger:debug("[docker_manager] task_id: ~p, exit normal", [TaskId]),
|
||||
ok;
|
||||
Error0 ->
|
||||
Error = iolist_to_binary(io_lib:format("~p", [Error0])),
|
||||
efka_remote_agent:task_event_stream(TaskId, <<"error">>, <<"任务失败: "/utf8, Error/binary>>),
|
||||
efka_remote_agent:close_task_event_stream(TaskId, <<"task exited">>),
|
||||
lager:notice("[docker_manager] task_id: ~p, exit with error: ~p", [TaskId, Error]),
|
||||
logger:notice("[docker_manager] task_id: ~p, exit with error: ~p", [TaskId, Error]),
|
||||
ok
|
||||
end,
|
||||
{noreply, State#state{task_map = NTaskMap}}
|
||||
|
||||
@ -7,8 +7,6 @@
|
||||
[
|
||||
sync,
|
||||
jiffy,
|
||||
parse_trans,
|
||||
lager,
|
||||
cowboy,
|
||||
ranch,
|
||||
crypto,
|
||||
|
||||
@ -45,7 +45,7 @@ start_http_server() ->
|
||||
],
|
||||
{ok, Pid} = cowboy:start_clear(ws_listener, TransOpts, #{env => #{dispatch => Dispatcher}}),
|
||||
|
||||
lager:debug("[efka_app] websocket server start at: ~p, pid is: ~p", [Port, Pid]).
|
||||
logger:debug("[efka_app] websocket server start at: ~p, pid is: ~p", [Port, Pid]).
|
||||
|
||||
ensure_upload_dir() ->
|
||||
{ok, UploadDir} = application:get_env(efka, upload_dir),
|
||||
|
||||
@ -103,7 +103,7 @@ handle_event(cast, {metric_data, RouteKey, Metric}, _, State) ->
|
||||
{keep_state, State};
|
||||
|
||||
handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
|
||||
lager:debug("[efka_remote_agent] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
|
||||
logger:debug("[efka_remote_agent] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
|
||||
EventPacket = message_codec:encode(?MESSAGE_EVENT_STREAM, #task_event_stream{
|
||||
task_id = TaskId,
|
||||
type = Type,
|
||||
@ -163,7 +163,7 @@ handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{tra
|
||||
efka_transport:auth_request(TransportPid, AuthBin),
|
||||
{next_state, ?STATE_AUTH, State};
|
||||
{error, Reason} ->
|
||||
lager:debug("[efka_remote_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
|
||||
logger:debug("[efka_remote_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
|
||||
efka_transport:stop(TransportPid),
|
||||
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
|
||||
end;
|
||||
@ -173,26 +173,26 @@ handle_event(info, {auth_reply, Reply}, ?STATE_AUTH, State = #state{transport_pi
|
||||
{ok, #auth_reply{code = Code, payload = Message}} ->
|
||||
case Code of
|
||||
0 ->
|
||||
lager:debug("[efka_remote_agent] auth success, message: ~p", [Message]),
|
||||
logger:debug("[efka_remote_agent] auth success, message: ~p", [Message]),
|
||||
{next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]};
|
||||
1 ->
|
||||
%% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent
|
||||
%% socket的连接状态需要维持
|
||||
lager:debug("[efka_remote_agent] auth denied, message: ~p", [Message]),
|
||||
logger:debug("[efka_remote_agent] auth denied, message: ~p", [Message]),
|
||||
{next_state, ?STATE_RESTRICTED, State};
|
||||
2 ->
|
||||
% 其他类型的错误，需要间隔时间重试
|
||||
lager:debug("[efka_remote_agent] auth failed, message: ~p", [Message]),
|
||||
logger:debug("[efka_remote_agent] auth failed, message: ~p", [Message]),
|
||||
efka_transport:stop(TransportPid),
|
||||
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
|
||||
_ ->
|
||||
% 其他类型的错误，需要间隔时间重试
|
||||
lager:debug("[efka_remote_agent] auth failed, invalid message"),
|
||||
logger:debug("[efka_remote_agent] auth failed, invalid message"),
|
||||
efka_transport:stop(TransportPid),
|
||||
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
|
||||
end;
|
||||
{error, Reason} ->
|
||||
lager:debug("[efka_remote_agent] auth_request failed, error: ~p", [Reason]),
|
||||
logger:debug("[efka_remote_agent] auth_request failed, error: ~p", [Reason]),
|
||||
efka_transport:stop(TransportPid),
|
||||
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
|
||||
end;
|
||||
@ -296,7 +296,7 @@ handle_event(info, {server_rpc, PacketId, #jsonrpc_request{method = <<"config_co
|
||||
%% 处理task_log
|
||||
%handle_event(info, {server_async_call, PacketId, <<?PUSH_TASK_LOG:8, TaskLogBin/binary>>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
|
||||
% #fetch_task_log{task_id = TaskId} = message_pb:decode_msg(TaskLogBin, fetch_task_log),
|
||||
% lager:debug("[efka_remote_agent] get task_log request: ~p", [TaskId]),
|
||||
% logger:debug("[efka_remote_agent] get task_log request: ~p", [TaskId]),
|
||||
% {ok, Logs} = efka_inetd_task_log:get_logs(TaskId),
|
||||
% Reply = case length(Logs) > 0 of
|
||||
% true ->
|
||||
@ -327,14 +327,14 @@ handle_event(info, {server_cast, #command{command_type = ?COMMAND_AUTH, command
|
||||
|
||||
%% 处理Pub/Sub机制
|
||||
handle_event(info, {server_cast, #pub{topic = Topic, qos = Qos, content = Content}}, ?STATE_ACTIVATED, State) ->
|
||||
lager:debug("[efka_remote_agent] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
|
||||
logger:debug("[efka_remote_agent] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
|
||||
%% 消息发送到订阅系统
|
||||
efka_subscription:publish(Topic, Qos, Content),
|
||||
{keep_state, State};
|
||||
|
||||
%% transport进程退出
|
||||
handle_event(info, {'DOWN', MRef, process, TransportPid, Reason}, _, State = #state{transport_ref = MRef}) ->
|
||||
lager:debug("[efka_remote_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
|
||||
logger:debug("[efka_remote_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
|
||||
erlang:start_timer(5000, self(), create_transport),
|
||||
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined, transport_ref = undefined}}.
|
||||
|
||||
|
||||
@ -69,7 +69,7 @@ start_link(Name, ServiceId) when is_atom(Name), is_binary(ServiceId) ->
|
||||
{stop, Reason :: term()} | ignore).
|
||||
init([ServiceId]) ->
|
||||
%% supervisor进程通过exit(ChildPid, shutdown)调用的时候，确保terminate函数被调用
|
||||
lager:debug("[efka_service] service_id: ~p, started", [ServiceId]),
|
||||
logger:debug("[efka_service] service_id: ~p, started", [ServiceId]),
|
||||
{ok, #state{service_id = ServiceId}}.
|
||||
|
||||
%% @private
|
||||
@ -87,7 +87,7 @@ handle_call({attach_channel, ChannelPid}, _From, State = #state{channel_pid = Ol
|
||||
case is_pid(OldChannelPid) andalso is_process_alive(OldChannelPid) of
|
||||
false ->
|
||||
erlang:monitor(process, ChannelPid),
|
||||
lager:debug("[efka_service] service_id: ~p, channel attched", [ServiceId]),
|
||||
logger:debug("[efka_service] service_id: ~p, channel attched", [ServiceId]),
|
||||
{reply, ok, State#state{channel_pid = ChannelPid}};
|
||||
true ->
|
||||
{reply, {error, <<"channel exists">>}, State}
|
||||
@ -103,7 +103,7 @@ handle_call(_Request, _From, State = #state{}) ->
|
||||
{noreply, NewState :: #state{}, timeout() | hibernate} |
|
||||
{stop, Reason :: term(), NewState :: #state{}}).
|
||||
handle_cast({metric_data, RouteKey, Metric}, State = #state{service_id = ServiceId}) ->
|
||||
lager:debug("[efka_service] metric_data service_id: ~p, route_key: ~p, metric data: ~p", [ServiceId, RouteKey, Metric]),
|
||||
logger:debug("[efka_service] metric_data service_id: ~p, route_key: ~p, metric data: ~p", [ServiceId, RouteKey, Metric]),
|
||||
efka_remote_agent:metric_data(RouteKey, Metric),
|
||||
{noreply, State};
|
||||
|
||||
@ -118,7 +118,7 @@ handle_cast(_Request, State = #state{}) ->
|
||||
{stop, Reason :: term(), NewState :: #state{}}).
|
||||
%% 处理channel进程的退出
|
||||
handle_info({'DOWN', _Ref, process, ChannelPid, Reason}, State = #state{channel_pid = ChannelPid, service_id = ServiceId}) ->
|
||||
lager:debug("[efka_service] service_id: ~p, channel exited: ~p", [ServiceId, Reason]),
|
||||
logger:debug("[efka_service] service_id: ~p, channel exited: ~p", [ServiceId, Reason]),
|
||||
{noreply, State#state{channel_pid = undefined}}.
|
||||
|
||||
%% @private
|
||||
@ -129,7 +129,7 @@ handle_info({'DOWN', _Ref, process, ChannelPid, Reason}, State = #state{channel_
|
||||
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
|
||||
State :: #state{}) -> term()).
|
||||
terminate(Reason, _State = #state{service_id = ServiceId}) ->
|
||||
lager:debug("[efka_service] service_id: ~p, terminate with reason: ~p", [ServiceId, Reason]),
|
||||
logger:debug("[efka_service] service_id: ~p, terminate with reason: ~p", [ServiceId, Reason]),
|
||||
ok.
|
||||
|
||||
%% @private
|
||||
|
||||
@ -113,7 +113,7 @@ handle_cast(finish, State = #state{parent_pid = ParentPid, io_device = IoDevice,
|
||||
handle_info({'DOWN', Ref, process, Pid, normal}, State = #state{ref = Ref, parent_pid = Pid}) ->
|
||||
{noreply, State};
|
||||
handle_info({'DOWN', Ref, process, Pid, Reason}, State = #state{ref = Ref, parent_pid = Pid, io_device = IoDevice, real_file = RealFile}) ->
|
||||
lager:debug("[efka_stream] ws_channel close with reason: ~p", [Reason]),
|
||||
logger:debug("[efka_stream] ws_channel close with reason: ~p", [Reason]),
|
||||
case IoDevice =:= undefined of
|
||||
true ->
|
||||
ok;
|
||||
|
||||
@ -113,7 +113,7 @@ handle_call(debug_info, _From, State = #state{subscribers = Subscribers, remaini
|
||||
%% 发布消息
|
||||
handle_cast({publish, Topic, Qos, Content}, State = #state{subscribers = Subscribers, remaining_messages = RemainingMessages}) ->
|
||||
MatchedSubscribers = match_subscribers(Subscribers, Topic),
|
||||
lager:debug("[efka_subscription] topic: ~p, content: ~p, match subscribers: ~p", [Topic, Content, MatchedSubscribers]),
|
||||
logger:debug("[efka_subscription] topic: ~p, content: ~p, match subscribers: ~p", [Topic, Content, MatchedSubscribers]),
|
||||
case length(MatchedSubscribers) > 0 of
|
||||
true ->
|
||||
broadcast(Topic, Content, MatchedSubscribers),
|
||||
@ -131,12 +131,12 @@ handle_cast({publish, Topic, Qos, Content}, State = #state{subscribers = Subscri
|
||||
{noreply, NewState :: #state{}, timeout() | hibernate} |
|
||||
{stop, Reason :: term(), NewState :: #state{}}).
|
||||
handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{subscribers = Subscribers}) ->
|
||||
lager:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
|
||||
logger:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
|
||||
NSubscribers = lists:filter(fun(#subscriber{subscriber_pid = Pid0}) -> SubscriberPid /= Pid0 end, Subscribers),
|
||||
{noreply, State#state{subscribers = NSubscribers}};
|
||||
|
||||
handle_info(Info, State = #state{}) ->
|
||||
lager:debug("[efka_subscription] get unknown info: ~p", [Info]),
|
||||
logger:debug("[efka_subscription] get unknown info: ~p", [Info]),
|
||||
{noreply, State}.
|
||||
|
||||
%% @private
|
||||
|
||||
@ -122,7 +122,7 @@ handle_cast({auth_request, AuthRequestBin}, State = #state{parent_pid = ParentPi
|
||||
ParentPid ! {auth_reply, {ok, Reply}},
|
||||
{noreply, State};
|
||||
{ssl, Socket, Info} ->
|
||||
lager:warning("[efka_transport] get invalid auth_reply: ~p", [Info]),
|
||||
logger:warning("[efka_transport] get invalid auth_reply: ~p", [Info]),
|
||||
ParentPid ! {auth_reply, {error, invalid_auth_reply}},
|
||||
{noreply, State}
|
||||
after 5000 ->
|
||||
@ -157,7 +157,7 @@ handle_info({ssl, Socket, <<?PACKET_REQUEST, PacketId:32, RPCRequestBin/binary>>
|
||||
{noreply, State};
|
||||
|
||||
handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) ->
|
||||
lager:debug("[efka_transport] ssl error: ~p", [Reason]),
|
||||
logger:debug("[efka_transport] ssl error: ~p", [Reason]),
|
||||
{stop, normal, State};
|
||||
|
||||
handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
|
||||
@ -168,7 +168,7 @@ handle_info({timeout, _, ping_ticker}, State) ->
|
||||
{noreply, State};
|
||||
|
||||
handle_info(Info, State = #state{}) ->
|
||||
lager:notice("[efka_transport] get unknown info: ~p", [Info]),
|
||||
logger:notice("[efka_transport] get unknown info: ~p", [Info]),
|
||||
{noreply, State}.
|
||||
|
||||
%% @private
|
||||
@ -179,7 +179,7 @@ handle_info(Info, State = #state{}) ->
|
||||
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
|
||||
State :: #state{}) -> term()).
|
||||
terminate(Reason, #state{}) ->
|
||||
lager:notice("[efka_transport] terminate with reason: ~p", [Reason]),
|
||||
logger:notice("[efka_transport] terminate with reason: ~p", [Reason]),
|
||||
ok.
|
||||
|
||||
%% @private
|
||||
|
||||
@ -14,14 +14,14 @@
|
||||
|
||||
test_pull() ->
|
||||
Image = <<"docker.1ms.run/library/nginx:latest">>,
|
||||
docker_commands:pull_image(Image, fun(Msg) -> lager:debug("msg is: ~p", [Msg]) end).
|
||||
docker_commands:pull_image(Image, fun(Msg) -> logger:debug("msg is: ~p", [Msg]) end).
|
||||
|
||||
test_commands() ->
|
||||
Id = <<"redpanda-console">>,
|
||||
StopRes = docker_commands:stop_container(Id),
|
||||
lager:debug("stop res: ~p", [StopRes]),
|
||||
logger:debug("stop res: ~p", [StopRes]),
|
||||
StartRes = docker_commands:start_container(Id),
|
||||
lager:debug("start res: ~p", [StartRes]).
|
||||
logger:debug("start res: ~p", [StartRes]).
|
||||
|
||||
test_create_container() ->
|
||||
M = #{
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user