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

fix

Browse Source
This commit is contained in:
anlicheng 2025-10-09 15:05:45 +08:00
parent da726556dd
commit d9614ae209
1 changed files with 16 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
apps/efka/src/docker/docker_deployer.erl
View File

@ -41,36 +41,35 @@ start_monitor(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(Con
deploy(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerDir), is_map(Config) -> deploy(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerDir), is_map(Config) ->
%% 尝试拉取镜像 %% 尝试拉取镜像
ContainerName = maps:get(<<"container_name">>, Config), ContainerName = maps:get(<<"container_name">>, Config),
log(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>), trace_log(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
ContainerName = <<>>,
case docker_commands:check_container_exist(ContainerName) of case docker_commands:check_container_exist(ContainerName) of
true -> true ->
log(TaskId, <<"info">>, <<"本地容器已经存在:"/utf8, ContainerName/binary>>), trace_log(TaskId, <<"info">>, <<"本地容器已经存在:"/utf8, ContainerName/binary>>),
efka_remote_agent:close_task_event_stream(TaskId); efka_remote_agent:close_task_event_stream(TaskId);
false -> false ->
Image0 = maps:get(<<"image">>, Config), Image0 = maps:get(<<"image">>, Config),
Image = normalize_image(Image0), Image = normalize_image(Image0),
log(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>), trace_log(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
PullResult = case docker_commands:check_image_exist(Image) of PullResult = case docker_commands:check_image_exist(Image) of
true -> true ->
log(TaskId, <<"info">>, <<"镜像本地已存在:"/utf8, Image/binary>>), trace_log(TaskId, <<"info">>, <<"镜像本地已存在:"/utf8, Image/binary>>),
ok; ok;
false -> false ->
log(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>), trace_log(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
CB = fun CB = fun
({message, M}) -> ({message, M}) ->
log(TaskId, <<"info">>, M); trace_log(TaskId, <<"info">>, M);
({error, Error}) -> ({error, Error}) ->
log(TaskId, <<"error">>, Error) trace_log(TaskId, <<"error">>, Error)
end, end,
docker_commands:pull_image(Image, CB) docker_commands:pull_image(Image, CB)
end, end,
case PullResult of case PullResult of
ok -> ok ->
log(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>), trace_log(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
case docker_commands:create_container(ContainerName, ContainerDir, Config) of case docker_commands:create_container(ContainerName, ContainerDir, Config) of
{ok, ContainerId} -> {ok, ContainerId} ->
%% 创建容器对应的配置文件 %% 创建容器对应的配置文件
@ -81,19 +80,19 @@ deploy(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerD
file:close(FD); file:close(FD);
{error, Reason} -> {error, Reason} ->
Reason1 = list_to_binary(io_lib:format("~p", [Reason])), Reason1 = list_to_binary(io_lib:format("~p", [Reason])),
log(TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, Reason1/binary>>) trace_log(TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, Reason1/binary>>)
end, end,
ShortContainerId = binary:part(ContainerId, 1, 12), ShortContainerId = binary:part(ContainerId, 1, 12),
log(TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>), trace_log(TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>),
log(TaskId, <<"info">>, <<"任务完成"/utf8>>), trace_log(TaskId, <<"info">>, <<"任务完成"/utf8>>),
efka_remote_agent:close_task_event_stream(TaskId); efka_remote_agent:close_task_event_stream(TaskId);
{error, Reason} -> {error, Reason} ->
log(TaskId, <<"error">>, <<"容器创建失败: "/utf8, Reason/binary>>), trace_log(TaskId, <<"error">>, <<"容器创建失败: "/utf8, Reason/binary>>),
log(TaskId, <<"error">>, <<"任务失败"/utf8>>), trace_log(TaskId, <<"error">>, <<"任务失败"/utf8>>),
efka_remote_agent:close_task_event_stream(TaskId) efka_remote_agent:close_task_event_stream(TaskId)
end; end;
{error, Reason} -> {error, Reason} ->
log(TaskId, <<"error">>, <<"镜像拉取失败: "/utf8, Reason/binary>>), trace_log(TaskId, <<"error">>, <<"镜像拉取失败: "/utf8, Reason/binary>>),
efka_remote_agent:close_task_event_stream(TaskId) efka_remote_agent:close_task_event_stream(TaskId)
end end
end. end.
@ -108,8 +107,8 @@ normalize_image(Image) when is_binary(Image) ->
end, end,
iolist_to_binary(lists:join(<<"/">>, PrefixParts ++ [NormalizedLast])). iolist_to_binary(lists:join(<<"/">>, PrefixParts ++ [NormalizedLast])).
-spec log(TaskId :: integer(), Level :: binary(), Msg :: binary()) -> no_return(). -spec trace_log(TaskId :: integer(), Level :: binary(), Msg :: binary()) -> no_return().
log(TaskId, Level, Msg) when is_integer(TaskId), is_binary(Level), is_binary(Msg) -> trace_log(TaskId, Level, Msg) when is_integer(TaskId), is_binary(Level), is_binary(Msg) ->
efka_remote_agent:task_event_stream(TaskId, Level, Msg), efka_remote_agent:task_event_stream(TaskId, Level, Msg),
Info = iolist_to_binary([<<"task_id=">>, integer_to_binary(TaskId), <<" ">>, Level, <<" ">>, Msg]), Info = iolist_to_binary([<<"task_id=">>, integer_to_binary(TaskId), <<" ">>, Level, <<" ">>, Msg]),
efka_logger:write(Info). efka_logger:write(Info).