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

fix docker_client

Browse Source
This commit is contained in:
anlicheng 2026-04-23 17:38:17 +08:00
parent 660ab164dd
commit 3aae06577a
4 changed files with 74 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

182
src/docker/docker_client.erl
View File

@ -3,37 +3,20 @@
%%% Short-lived Docker API client process.
%%%
%%% Each request owns one process and one gun Unix socket connection.
%%% This keeps long-running stream calls isolated while preserving the
%%% existing open/request/close lifecycle.
%%% The process exits after the Docker API response finishes.
%%% @end
%%%-------------------------------------------------------------------
-module(docker_client).
-behaviour(gen_server).
%% API
-export([request/4, start_stream/5]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-define(DOCKER_SOCKET, "/var/run/docker.sock").
-define(RESPONSE_TIMEOUT, 5000).
-define(BODY_TIMEOUT, 10000).
-define(STREAM_BODY_TIMEOUT, 30000).
-define(CLIENT_TIMEOUT, 60000).
-record(state, {
mode :: request | stream,
owner :: pid(),
owner_ref :: reference() | undefined,
ref :: reference(),
method :: string(),
path :: string(),
body :: binary(),
headers :: list()
}).
%%%===================================================================
%%% API
%%%===================================================================
@ -41,11 +24,9 @@
-spec request(Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) ->
{ok, StatusCode :: integer(), RespHeaders :: proplists:proplist(), RespBody :: binary()} | {error, any()}.
request(Method, Path, Body, Headers) when is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
Ref = make_ref(),
Owner = self(),
case gen_server:start(?MODULE, {request, Owner, Ref, Method, Path, Body, Headers}, []) of
{ok, Pid} ->
MRef = erlang:monitor(process, Pid),
Ref = make_ref(),
{Pid, MRef} = spawn_monitor(fun() -> request_worker(Owner, Ref, Method, Path, Body, Headers) end),
receive
{docker_client, Ref, {result, Result}} ->
erlang:demonitor(MRef, [flush]),
@ -56,83 +37,39 @@ request(Method, Path, Body, Headers) when is_list(Method), is_list(Path), is_bin
exit(Pid, shutdown),
erlang:demonitor(MRef, [flush]),
{error, timeout}
end;
{error, Reason} ->
{error, Reason}
end.
-spec start_stream(Owner :: pid(), Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) ->
{ok, Ref :: reference(), Pid :: pid()} | {error, any()}.
{ok, Ref :: reference(), Pid :: pid(), MRef :: reference()}.
start_stream(Owner, Method, Path, Body, Headers)
when is_pid(Owner), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
Ref = make_ref(),
case gen_server:start(?MODULE, {stream, Owner, Ref, Method, Path, Body, Headers}, []) of
{ok, Pid} ->
{ok, Ref, Pid};
{error, Reason} ->
{error, Reason}
end.
{Pid, MRef} = spawn_monitor(fun() -> stream_worker(Owner, Ref, Method, Path, Body, Headers) end),
{ok, Ref, Pid, MRef}.
%%%===================================================================
%%% gen_server callbacks
%%% Worker functions
%%%===================================================================
-spec init(term()) -> {ok, #state{}}.
init({Mode, Owner, Ref, Method, Path, Body, Headers})
when (Mode =:= request orelse Mode =:= stream),
is_pid(Owner), is_reference(Ref), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
-spec request_worker(pid(), reference(), string(), string(), binary(), list()) -> ok.
request_worker(Owner, Ref, Method, Path, Body, Headers) ->
Result = do_request(Method, Path, Body, Headers),
send_owner(Owner, Ref, {result, Result}).
-spec stream_worker(pid(), reference(), string(), string(), binary(), list()) -> ok.
stream_worker(Owner, Ref, Method, Path, Body, Headers) ->
OwnerRef = erlang:monitor(process, Owner),
erlang:send_after(0, self(), execute),
{ok, #state{
mode = Mode,
owner = Owner,
owner_ref = OwnerRef,
ref = Ref,
method = Method,
path = Path,
body = Body,
headers = Headers
}}.
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, ok, #state{}}.
handle_call(_Request, _From, State) ->
{reply, ok, State}.
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast(_Request, State) ->
{noreply, State}.
-spec handle_info(term(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}.
handle_info(execute, State = #state{mode = request}) ->
Result = do_request(State),
send_owner(State, {result, Result}),
{stop, normal, State};
handle_info(execute, State = #state{mode = stream}) ->
_Result = do_stream(State),
{stop, normal, State};
handle_info({'DOWN', OwnerRef, process, _Pid, Reason}, State = #state{owner_ref = OwnerRef}) ->
{stop, {owner_down, Reason}, State};
handle_info(_Info, State) ->
{noreply, State}.
-spec terminate(term(), #state{}) -> ok.
terminate(_Reason, #state{owner_ref = undefined}) ->
ok;
terminate(_Reason, #state{owner_ref = OwnerRef}) ->
_Result = do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers),
erlang:demonitor(OwnerRef, [flush]),
ok.
-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec do_request(#state{}) ->
-spec do_request(string(), string(), binary(), list()) ->
{ok, integer(), proplists:proplist(), binary()} | {error, any()}.
do_request(#state{method = Method, path = Path, body = Body, headers = Headers}) ->
do_request(Method, Path, Body, Headers) ->
case open_connection() of
{ok, ConnPid} ->
try
@ -145,19 +82,20 @@ do_request(#state{method = Method, path = Path, body = Body, headers = Headers})
{error, Reason}
end.
-spec do_stream(#state{}) -> ok | {error, binary()}.
do_stream(State = #state{method = Method, path = Path, body = Body, headers = Headers}) ->
-spec do_stream(pid(), reference(), reference(), string(), string(), binary(), list()) ->
ok | {error, binary()}.
do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers) ->
case open_connection() of
{ok, ConnPid} ->
try
StreamRef = gun:request(ConnPid, Method, Path, Headers, Body),
receive_stream_response(State, ConnPid, StreamRef)
receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef)
after
close_connection(ConnPid)
end;
{error, Reason0} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason}
end.
@ -198,97 +136,97 @@ receive_body(ConnPid, StreamRef, Status, Headers, Acc) ->
{error, timeout}
end.
-spec receive_stream_response(#state{}, pid(), reference()) -> ok | {error, binary()}.
receive_stream_response(State, ConnPid, StreamRef) ->
-spec receive_stream_response(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}.
receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
receive
{gun_response, ConnPid, StreamRef, nofin, Status, Headers} when Status >= 200, Status < 300 ->
send_owner(State, {response, Status, Headers}),
receive_stream_body(State, ConnPid, StreamRef);
send_owner(Owner, Ref, {response, Status, Headers}),
receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef);
{gun_response, ConnPid, StreamRef, fin, Status, Headers} when Status >= 200, Status < 300 ->
send_owner(State, {response, Status, Headers}),
send_owner(State, done),
send_owner(Owner, Ref, {response, Status, Headers}),
send_owner(Owner, Ref, done),
ok;
{gun_response, ConnPid, StreamRef, nofin, Status, _Headers} ->
receive_stream_error_body(State, ConnPid, StreamRef, Status, <<>>);
receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, <<>>);
{gun_response, ConnPid, StreamRef, fin, Status, _Headers} ->
Reason = http_status_error(Status, <<>>),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_error, ConnPid, StreamRef, Reason0} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_error, ConnPid, Reason0} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_down, ConnPid, _, Reason0, _} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{'DOWN', OwnerRef, process, _Pid, Reason0} when OwnerRef =:= State#state.owner_ref ->
{'DOWN', OwnerRef, process, _Pid, Reason0} ->
{error, format_error({owner_down, Reason0})}
after ?RESPONSE_TIMEOUT ->
send_owner(State, {error, <<"处理超时"/utf8>>}),
send_owner(Owner, Ref, {error, <<"处理超时"/utf8>>}),
{error, <<"timeout">>}
end.
-spec receive_stream_body(#state{}, pid(), reference()) -> ok | {error, binary()}.
receive_stream_body(State, ConnPid, StreamRef) ->
-spec receive_stream_body(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}.
receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
receive
{gun_data, ConnPid, StreamRef, fin, Data} ->
maybe_send_stream_data(State, Data),
send_owner(State, done),
maybe_send_stream_data(Owner, Ref, Data),
send_owner(Owner, Ref, done),
ok;
{gun_data, ConnPid, StreamRef, nofin, Data} ->
maybe_send_stream_data(State, Data),
receive_stream_body(State, ConnPid, StreamRef);
maybe_send_stream_data(Owner, Ref, Data),
receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef);
{gun_error, ConnPid, StreamRef, Reason0} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_error, ConnPid, Reason0} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_down, ConnPid, _, Reason0, _} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{'DOWN', OwnerRef, process, _Pid, Reason0} when OwnerRef =:= State#state.owner_ref ->
{'DOWN', OwnerRef, process, _Pid, Reason0} ->
{error, format_error({owner_down, Reason0})}
after ?STREAM_BODY_TIMEOUT ->
send_owner(State, {error, <<"timeout">>}),
send_owner(Owner, Ref, {error, <<"timeout">>}),
{error, <<"timeout">>}
end.
-spec receive_stream_error_body(#state{}, pid(), reference(), integer(), iodata()) ->
-spec receive_stream_error_body(pid(), reference(), reference(), pid(), reference(), integer(), iodata()) ->
{error, binary()}.
receive_stream_error_body(State, ConnPid, StreamRef, Status, Acc) ->
receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, Acc) ->
receive
{gun_data, ConnPid, StreamRef, fin, Data} ->
Reason = http_status_error(Status, iolist_to_binary([Acc, Data])),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_data, ConnPid, StreamRef, nofin, Data} ->
receive_stream_error_body(State, ConnPid, StreamRef, Status, [Acc, Data]);
receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, [Acc, Data]);
{gun_error, ConnPid, StreamRef, Reason0} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_error, ConnPid, Reason0} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{gun_down, ConnPid, _, Reason0, _} ->
Reason = format_error(Reason0),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason};
{'DOWN', OwnerRef, process, _Pid, Reason0} when OwnerRef =:= State#state.owner_ref ->
{'DOWN', OwnerRef, process, _Pid, Reason0} ->
{error, format_error({owner_down, Reason0})}
after ?BODY_TIMEOUT ->
Reason = http_status_error(Status, iolist_to_binary(Acc)),
send_owner(State, {error, Reason}),
send_owner(Owner, Ref, {error, Reason}),
{error, Reason}
end.
@ -312,18 +250,18 @@ close_connection(ConnPid) ->
catch gun:close(ConnPid),
ok.
-spec send_owner(#state{}, term()) -> ok.
send_owner(#state{owner = Owner, ref = Ref}, Event) ->
-spec send_owner(pid(), reference(), term()) -> ok.
send_owner(Owner, Ref, Event) ->
Owner ! {docker_client, Ref, Event},
ok.
-spec maybe_send_stream_data(#state{}, iodata()) -> ok.
maybe_send_stream_data(State, Data) ->
-spec maybe_send_stream_data(pid(), reference(), iodata()) -> ok.
maybe_send_stream_data(Owner, Ref, Data) ->
case iolist_to_binary(Data) of
<<>> ->
ok;
DataBin ->
send_owner(State, {data, DataBin})
send_owner(Owner, Ref, {data, DataBin})
end.
-spec http_status_error(integer(), binary()) -> binary().

2
src/docker/docker_commands.erl
View File

@ -16,7 +16,7 @@
start_container/1, stop_container/1, stop_container/2, remove_container/1, remove_container/3, kill_container/1, kill_container/2,
get_containers/0]).
-spec pull_image(Image :: binary()) -> {ok, Ref :: reference(), Pid :: pid()} | {error, Reason :: any()}.
-spec pull_image(Image :: binary()) -> {ok, Ref :: reference(), Pid :: pid(), MRef :: reference()} | {error, Reason :: any()}.
pull_image(Image) when is_binary(Image) ->
Url = lists:flatten(io_lib:format("/images/create?fromImage=~s", [binary_to_list(Image)])),
docker_client:start_stream(self(), "POST", Url, <<>>, []).

9
src/docker/docker_deployer.erl
View File

@ -126,8 +126,8 @@ ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
report_task_event(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
report_task_event(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
case docker_commands:pull_image(Image) of
{ok, Ref, Pid} ->
await_pull_image(TaskId, Ref, Pid),
{ok, Ref, Pid, MRef} ->
await_pull_image(TaskId, Ref, Pid, MRef),
{ok, Image};
{error, Reason} when is_binary(Reason) ->
throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>});
@ -136,11 +136,6 @@ ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
throw({deploy_error, Error})
end.
-spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid()) -> ok.
await_pull_image(TaskId, Ref, Pid) when is_integer(TaskId), is_reference(Ref), is_pid(Pid) ->
MRef = erlang:monitor(process, Pid),
await_pull_image(TaskId, Ref, Pid, MRef).
-spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok.
await_pull_image(TaskId, Ref, Pid, MRef) ->
receive

7
src/tests/docker_commands_tests.erl
View File

@ -67,12 +67,7 @@ test_all() ->
-spec test_pull() -> ok.
test_pull() ->
{ok, Ref, Pid} = docker_commands:pull_image(?TEST_IMAGE),
await_pull(Ref, Pid).
-spec await_pull(reference(), pid()) -> ok.
await_pull(Ref, Pid) when is_reference(Ref), is_pid(Pid) ->
MRef = erlang:monitor(process, Pid),
{ok, Ref, Pid, MRef} = docker_commands:pull_image(?TEST_IMAGE),
await_pull(Ref, Pid, MRef).
-spec await_pull(reference(), pid(), reference()) -> ok.