anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix logger

Browse Source
This commit is contained in:
anlicheng 2025-04-29 22:45:00 +08:00
parent d0531fd443
commit 8b9774132b
4 changed files with 21 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
apps/efka/src/efka_agent.erl
View File

@ -92,7 +92,7 @@ handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{tra
efka_transport:auth_request(TransportPid, 5000), efka_transport:auth_request(TransportPid, 5000),
{next_state, ?STATE_AUTH, State}; {next_state, ?STATE_AUTH, State};
{error, Reason} -> {error, Reason} ->
lager:debug("[efka_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]), efka_logger:debug("[efka_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
efka_transport:stop(TransportPid), efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State} {next_state, ?STATE_DENIED, State}
end; end;
@ -101,23 +101,23 @@ handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{tra
handle_event(info, {auth_reply, Reply}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) -> handle_event(info, {auth_reply, Reply}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
case Reply of case Reply of
{ok, #auth_reply{code = 1, message = Message, repository_url = RepositoryUrl}} -> {ok, #auth_reply{code = 1, message = Message, repository_url = RepositoryUrl}} ->
lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]), efka_logger:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
{next_state, ?STATE_ACTIVATED, State}; {next_state, ?STATE_ACTIVATED, State};
%% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent %% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent
%% socket的连接状态需要维持 %% socket的连接状态需要维持
{ok, #auth_reply{code = -1, message = Message}} -> {ok, #auth_reply{code = -1, message = Message}} ->
lager:debug("[efka_agent] auth denied, message: ~p", [Message]), efka_logger:debug("[efka_agent] auth denied, message: ~p", [Message]),
{next_state, ?STATE_RESTRICTED, State}; {next_state, ?STATE_RESTRICTED, State};
%% 其他类型的错误，需要间隔时间重试 %% 其他类型的错误，需要间隔时间重试
{ok, #auth_reply{code = -2, message = Message}} -> {ok, #auth_reply{code = -2, message = Message}} ->
lager:debug("[efka_agent] auth failed, message: ~p", [Message]), efka_logger:debug("[efka_agent] auth failed, message: ~p", [Message]),
efka_transport:stop(TransportPid), efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}; {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
{error, Reason} -> {error, Reason} ->
lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]), efka_logger:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
efka_transport:stop(TransportPid), efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}} {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
end; end;
@ -143,7 +143,7 @@ handle_event(info, {server_push_message, <<8:8, ActivatePush>>}, StateName, Stat
%% 收到需要回复的指令 %% 收到需要回复的指令
handle_event(info, {server_push_message, PacketId, <<16:8, Directive>>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) -> handle_event(info, {server_push_message, PacketId, <<16:8, Directive>>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
#topic_message{topic = Topic, content = Content} = message_pb:decode_msg(Directive, directive), #topic_message{topic = Topic, content = Content} = message_pb:decode_msg(Directive, directive),
lager:debug("[efka_agent] get directive with packet_id: ~p, to device_uuid: ~p, content: ~p", [PacketId, Topic, Content]), efka_logger:debug("[efka_agent] get directive with packet_id: ~p, to device_uuid: ~p, content: ~p", [PacketId, Topic, Content]),
%% 消息发送到订阅系统 %% 消息发送到订阅系统
case PacketId > 0 of case PacketId > 0 of
@ -158,7 +158,7 @@ handle_event(info, {server_push_message, PacketId, <<16:8, Directive>>}, ?STATE_
%% transport进程退出 %% transport进程退出
handle_event(info, {'EXIT', TransportPid, Reason}, _StateName, State = #state{transport_pid = TransportPid}) -> handle_event(info, {'EXIT', TransportPid, Reason}, _StateName, State = #state{transport_pid = TransportPid}) ->
lager:warning("[efka_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]), efka_logger:warning("[efka_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
erlang:start_timer(500000, self(), create_transport), erlang:start_timer(500000, self(), create_transport),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}; {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};

8
apps/efka/src/efka_subscription.erl
View File

@ -119,7 +119,7 @@ handle_cast({publish, Topic, Content}, State = #state{subscribers = Subscribers}
erlang:send(SubscriberPid, {topic_broadcast, Content}) erlang:send(SubscriberPid, {topic_broadcast, Content})
end, MatchedSubscribers), end, MatchedSubscribers),
lager:debug("[efka_subscription] topic: ~p, content: ~p, match subscribers: ~p", [Topic, Content, MatchedSubscribers]), efka_logger:debug("[efka_subscription] topic: ~p, content: ~p, match subscribers: ~p", [Topic, Content, MatchedSubscribers]),
{noreply, State}; {noreply, State};
%% 需要响应的推送 %% 需要响应的推送
@ -133,7 +133,7 @@ handle_cast({publish, PacketId, Topic, Content, CallbackFun}, State = #state{sub
[] -> [] ->
ok; ok;
[S = #subscriber{subscriber_pid = SubscriberPid} | _] -> [S = #subscriber{subscriber_pid = SubscriberPid} | _] ->
lager:debug("[efka_subscription] topic_sink topic: ~p, content: ~p, match subscriber: ~p", [Topic, Content, S]), efka_logger:debug("[efka_subscription] topic_sink topic: ~p, content: ~p, match subscriber: ~p", [Topic, Content, S]),
erlang:send(SubscriberPid, {topic_sink, Content}) erlang:send(SubscriberPid, {topic_sink, Content})
end, end,
@ -156,12 +156,12 @@ handle_cast({response, ShaUUID, Response}, State = #state{callbacks = Callbacks}
{noreply, NewState :: #state{}, timeout() | hibernate} | {noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}). {stop, Reason :: term(), NewState :: #state{}}).
handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{subscribers = Subscribers}) -> handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{subscribers = Subscribers}) ->
lager:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]), efka_logger:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
NSubscribers = lists:filter(fun(#subscriber{subscriber_pid = Pid0}) -> SubscriberPid /= Pid0 end, Subscribers), NSubscribers = lists:filter(fun(#subscriber{subscriber_pid = Pid0}) -> SubscriberPid /= Pid0 end, Subscribers),
{noreply, State#state{subscribers = NSubscribers}}; {noreply, State#state{subscribers = NSubscribers}};
handle_info(Info, State = #state{}) -> handle_info(Info, State = #state{}) ->
lager:debug("[efka_subscription] get unknown info: ~p", [Info]), efka_logger:debug("[efka_subscription] get unknown info: ~p", [Info]),
{noreply, State}. {noreply, State}.
%% @private %% @private

8
apps/efka/src/efka_transport.erl
View File

@ -148,13 +148,13 @@ handle_cast({response, PacketId, Response}, State = #state{socket = Socket}) ->
{stop, Reason :: term(), NewState :: #state{}}). {stop, Reason :: term(), NewState :: #state{}}).
%% 服务器主动推送的数据，有packetId的是要求返回的；为0的表示不需要返回值 %% 服务器主动推送的数据，有packetId的是要求返回的；为0的表示不需要返回值
handle_info({ssl, Socket, <<?PACKET_PUBLISH, 0:32, Msg/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) -> handle_info({ssl, Socket, <<?PACKET_PUBLISH, 0:32, Msg/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
lager:debug("[efka_agent] socket get message: ~p", [Msg]), efka_logger:debug("[efka_agent] socket get message: ~p", [Msg]),
ParentPid ! {server_push_message, Msg}, ParentPid ! {server_push_message, Msg},
{noreply, State}; {noreply, State};
%% 目前推送的消息包括: <<CommandType:8, Command/binary>>, <<16:8, Directive/binary>> %% 目前推送的消息包括: <<CommandType:8, Command/binary>>, <<16:8, Directive/binary>>
handle_info({ssl, Socket, <<?PACKET_PUBLISH, PacketId:32, Msg/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) -> handle_info({ssl, Socket, <<?PACKET_PUBLISH, PacketId:32, Msg/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
lager:debug("[efka_agent] socket get message: ~p", [Msg]), efka_logger:debug("[efka_agent] socket get message: ~p", [Msg]),
ParentPid ! {server_push_message, PacketId, Msg}, ParentPid ! {server_push_message, PacketId, Msg},
{noreply, State#state{}}; {noreply, State#state{}};
@ -165,7 +165,7 @@ handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
{stop, normal, State}; {stop, normal, State};
handle_info(Info, State = #state{}) -> handle_info(Info, State = #state{}) ->
lager:debug("[efka_transport] info: ~p", [Info]), efka_logger:debug("[efka_transport] info: ~p", [Info]),
{noreply, State}. {noreply, State}.
%% @private %% @private
@ -176,7 +176,7 @@ handle_info(Info, State = #state{}) ->
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
State :: #state{}) -> term()). State :: #state{}) -> term()).
terminate(Reason, #state{}) -> terminate(Reason, #state{}) ->
lager:debug("[efka_transport] terminate with reason: ~p", [Reason]), efka_logger:debug("[efka_transport] terminate with reason: ~p", [Reason]),
ok. ok.
%% @private %% @private

12
apps/efka/src/tcp_channel.erl
View File

@ -66,7 +66,7 @@ start_link(Transport, Sock) ->
{ok, proc_lib:spawn_link(?MODULE, init, [[Transport, Sock]])}. {ok, proc_lib:spawn_link(?MODULE, init, [[Transport, Sock]])}.
init([Transport, Sock]) -> init([Transport, Sock]) ->
lager:debug("[sdlan_channel] get a new connection: ~p", [Sock]), efka_logger:debug("[sdlan_channel] get a new connection: ~p", [Sock]),
case Transport:wait(Sock) of case Transport:wait(Sock) of
{ok, NewSock} -> {ok, NewSock} ->
Transport:setopts(Sock, [{active, true}]), Transport:setopts(Sock, [{active, true}]),
@ -85,15 +85,15 @@ handle_cast(_Msg, State) ->
%% 网络流量统计 %% 网络流量统计
handle_info({tcp, _Sock, <<Body/binary>>}, State = #state{is_registered = true}) -> handle_info({tcp, _Sock, <<Body/binary>>}, State = #state{is_registered = true}) ->
% #sdl_flows{forward_num = ForwardNum, p2p_num = P2PNum, inbound_num = InboundNum} = sdlan_pb:decode_msg(Body, sdl_flows), % #sdl_flows{forward_num = ForwardNum, p2p_num = P2PNum, inbound_num = InboundNum} = sdlan_pb:decode_msg(Body, sdl_flows),
lager:debug("[sdlan_channel] body: ~p", [Body]), efka_logger:debug("[sdlan_channel] body: ~p", [Body]),
{noreply, State}; {noreply, State};
handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) -> handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) ->
lager:notice("[sdlan_channel] tcp_error: ~p", [Reason]), efka_logger:notice("[sdlan_channel] tcp_error: ~p", [Reason]),
{stop, normal, State}; {stop, normal, State};
handle_info({tcp_closed, Sock}, State = #state{socket = Sock}) -> handle_info({tcp_closed, Sock}, State = #state{socket = Sock}) ->
lager:notice("[sdlan_channel] tcp_closed", []), efka_logger:notice("[sdlan_channel] tcp_closed", []),
{stop, normal, State}; {stop, normal, State};
%% 关闭当前通道 %% 关闭当前通道
@ -101,11 +101,11 @@ handle_info({stop, Reason}, State) ->
{stop, Reason, State}; {stop, Reason, State};
handle_info(Info, State) -> handle_info(Info, State) ->
lager:warning("[sdlan_channel] get a unknown message: ~p, channel will closed", [Info]), efka_logger:warning("[sdlan_channel] get a unknown message: ~p, channel will closed", [Info]),
{noreply, State}. {noreply, State}.
terminate(Reason, #state{}) -> terminate(Reason, #state{}) ->
lager:warning("[sdlan_channel] stop with reason: ~p", [Reason]), efka_logger:warning("[sdlan_channel] stop with reason: ~p", [Reason]),
ok. ok.
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->