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

fix container

Browse Source
This commit is contained in:
anlicheng 2025-09-12 17:47:24 +08:00
parent ee1c025cd5
commit 0165e52ccb
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
apps/efka/src/docker/efka_docker_container.erl
View File

@ -24,7 +24,7 @@
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-record(state, {
service_id :: binary(),
%service_id :: binary(),
container_id :: binary(),
%% 通道id信息
channel_pid :: pid() | undefined,
@ -179,14 +179,14 @@ handle_cast(_Request, State = #state{}) ->
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
%% 重启服务
handle_info({timeout, _, reboot_service}, State = #state{service_id = ServiceId, manifest = Manifest}) ->
handle_info({timeout, _, reboot_service}, State = #state{container_id = ContainerId, manifest = Manifest}) ->
case startup(Manifest, []) of
{ok, Port} ->
{os_pid, OSPid} = erlang:port_info(Port, os_pid),
lager:debug("[efka_service] service_id: ~p, reboot success, port: ~p, os_pid: ~p", [ServiceId, Port, OSPid]),
lager:debug("[efka_service] service_id: ~p, reboot success, port: ~p, os_pid: ~p", [ContainerId, Port, OSPid]),
{noreply, State#state{port = Port, os_pid = OSPid}};
{error, Reason} ->
lager:debug("[efka_service] service_id: ~p, boot_service get error: ~p", [ServiceId, Reason]),
lager:debug("[efka_service] service_id: ~p, boot_service get error: ~p", [ContainerId, Reason]),
try_reboot(),
{noreply, State}
end;
@ -202,24 +202,24 @@ handle_info({channel_reply, Ref, Reply}, State = #state{inflight = Inflight, cal
{noreply, State#state{inflight = NInflight, callbacks = trigger_callback(Ref, Callbacks)}}
end;
handle_info({Port, {data, Data}}, State = #state{service_id = ServiceId}) when is_port(Port) ->
lager:debug("[efka_service] service_id: ~p, port data: ~p", [ServiceId, Data]),
handle_info({Port, {data, Data}}, State = #state{container_id = ContainerId}) when is_port(Port) ->
lager:debug("[efka_service] service_id: ~p, port data: ~p", [ContainerId, Data]),
{noreply, State};
%% 处理port的消息, Port的被动关闭会触发；因此这个时候的Port和State.port的值是相等的
handle_info({Port, {exit_status, Code}}, State = #state{service_id = ServiceId}) when is_port(Port) ->
lager:debug("[efka_service] service_id: ~p, port: ~p, exit with code: ~p", [ServiceId, Port, Code]),
handle_info({Port, {exit_status, Code}}, State = #state{container_id = ContainerId}) when is_port(Port) ->
lager:debug("[efka_service] service_id: ~p, port: ~p, exit with code: ~p", [ContainerId, Port, Code]),
{noreply, State#state{port = undefined, os_pid = undefined}};
%% 处理port的退出消息
handle_info({'EXIT', Port, Reason}, State = #state{service_id = ServiceId}) when is_port(Port) ->
lager:debug("[efka_service] service_id: ~p, port: ~p, exit with reason: ~p", [ServiceId, Port, Reason]),
handle_info({'EXIT', Port, Reason}, State = #state{container_id = ContainerId}) when is_port(Port) ->
lager:debug("[efka_service] service_id: ~p, port: ~p, exit with reason: ~p", [ContainerId, Port, Reason]),
try_reboot(),
{noreply, State#state{port = undefined, os_pid = undefined}};
%% 处理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]),
handle_info({'DOWN', _Ref, process, ChannelPid, Reason}, State = #state{channel_pid = ChannelPid, container_id = ContainerId}) ->
lager:debug("[efka_service] service_id: ~p, channel exited: ~p", [ContainerId, Reason]),
{noreply, State#state{channel_pid = undefined, inflight = #{}}}.
%% @private
@ -229,10 +229,10 @@ handle_info({'DOWN', _Ref, process, ChannelPid, Reason}, State = #state{channel_
%% with Reason. The return value is ignored.
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
State :: #state{}) -> term()).
terminate(Reason, _State = #state{service_id = ServiceId, port = Port, os_pid = OSPid}) ->
terminate(Reason, _State = #state{container_id = ContainerId, port = Port, os_pid = OSPid}) ->
erlang:is_port(Port) andalso erlang:port_close(Port),
catch kill_os_pid(OSPid),
lager:debug("[efka_service] service_id: ~p, terminate with reason: ~p", [ServiceId, Reason]),
lager:debug("[efka_service] service_id: ~p, terminate with reason: ~p", [ContainerId, Reason]),
ok.
%% @private