From 303856344e47e35b796cf7bbb35e3ccc9b41cb83 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Mon, 20 Apr 2026 18:34:25 +0800 Subject: [PATCH] fix docker_deployer --- src/docker/docker_deployer.erl | 144 +++++++++++++++++++++------------ 1 file changed, 92 insertions(+), 52 deletions(-) diff --git a/src/docker/docker_deployer.erl b/src/docker/docker_deployer.erl index 4065884..b25fae7 100644 --- a/src/docker/docker_deployer.erl +++ b/src/docker/docker_deployer.erl @@ -48,60 +48,25 @@ deploy(TaskId, ContainerDir, Params = #'ContainerDeployParams'{ container_name = ContainerName, spec = #'ContainerSpec'{image = Image0} }) when is_integer(TaskId), is_list(ContainerDir) -> - %% 尝试拉取镜像 trace_log(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>), - - case docker_commands:check_container_exist(ContainerName) of - true -> - trace_log(TaskId, <<"info">>, <<"本地容器已经存在:"/utf8, ContainerName/binary>>), + try + ok = ensure_container_absent(TaskId, ContainerName), + {ok, _Image} = ensure_image_ready(TaskId, Image0), + {ok, ContainerId} = create_container_and_config(TaskId, ContainerDir, Params), + ShortContainerId = short_container_id(ContainerId), + trace_log(TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>), + trace_log(TaskId, <<"info">>, <<"任务完成"/utf8>>), + efka_task_reporter:close(TaskId, ?TASK_SUCCESS) + catch + throw:{deploy_error, Reason} -> + trace_log(TaskId, <<"error">>, Reason), + trace_log(TaskId, <<"error">>, <<"任务失败"/utf8>>), efka_task_reporter:close(TaskId, ?TASK_FAIL); - false -> - Image = normalize_image(Image0), - - trace_log(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>), - PullResult = case docker_commands:check_image_exist(Image) of - true -> - trace_log(TaskId, <<"info">>, <<"镜像本地已存在:"/utf8, Image/binary>>), - ok; - false -> - trace_log(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>), - CB = fun - ({message, M}) -> - trace_log(TaskId, <<"info">>, M); - ({error, Error}) -> - trace_log(TaskId, <<"error">>, Error) - end, - docker_commands:pull_image(Image, CB) - end, - - case PullResult of - ok -> - trace_log(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>), - case docker_commands:create_container(ContainerDir, Params) of - {ok, ContainerId} -> - %% 创建容器对应的配置文件 - ConfigFile = docker_helper:get_config_file(ContainerDir), - case file:open(ConfigFile, [write, exclusive]) of - {ok, FD} -> - ok = file:write(FD, <<>>), - file:close(FD); - {error, Reason} -> - Reason1 = list_to_binary(io_lib:format("~p", [Reason])), - trace_log(TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, Reason1/binary>>) - end, - ShortContainerId = binary:part(ContainerId, 1, 12), - trace_log(TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>), - trace_log(TaskId, <<"info">>, <<"任务完成"/utf8>>), - efka_task_reporter:close(TaskId, ?TASK_SUCCESS); - {error, Reason} -> - trace_log(TaskId, <<"error">>, <<"容器创建失败: "/utf8, Reason/binary>>), - trace_log(TaskId, <<"error">>, <<"任务失败"/utf8>>), - efka_task_reporter:close(TaskId, ?TASK_FAIL) - end; - {error, Reason} -> - trace_log(TaskId, <<"error">>, <<"镜像拉取失败: "/utf8, Reason/binary>>), - efka_task_reporter:close(TaskId, ?TASK_FAIL) - end + Class:Reason:Stacktrace -> + Error = iolist_to_binary(io_lib:format("deploy crashed: ~p:~p ~p", [Class, Reason, Stacktrace])), + trace_log(TaskId, <<"error">>, Error), + trace_log(TaskId, <<"error">>, <<"任务失败"/utf8>>), + efka_task_reporter:close(TaskId, ?TASK_FAIL) end. -spec normalize_image(binary()) -> binary(). @@ -119,3 +84,78 @@ trace_log(TaskId, Level, Msg) when is_integer(TaskId), is_binary(Level), is_bina efka_task_reporter:stream(TaskId, Level, Msg), Info = iolist_to_binary([<<"task_id=">>, integer_to_binary(TaskId), <<" ">>, Level, <<" ">>, Msg]), efka_logger:write(Info). + +-spec ensure_container_absent(TaskId :: integer(), ContainerName :: binary()) -> ok. +ensure_container_absent(TaskId, ContainerName) when is_integer(TaskId), is_binary(ContainerName) -> + trace_log(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>), + ok. + +-spec ensure_image_ready(TaskId :: integer(), Image0 :: binary()) -> {ok, binary()}. +ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) -> + Image = normalize_image(Image0), + trace_log(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>), + trace_log(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>), + CB = fun + ({message, M}) -> + trace_log(TaskId, <<"info">>, M); + ({error, Error}) -> + trace_log(TaskId, <<"error">>, Error) + end, + case docker_commands:pull_image(Image, CB) of + ok -> + {ok, Image}; + {error, Reason} when is_binary(Reason) -> + throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>}); + {error, Reason} -> + Error = iolist_to_binary(io_lib:format("镜像拉取失败: ~p", [Reason])), + throw({deploy_error, Error}) + end. + +-spec create_container_and_config(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) -> + {ok, binary()}. +create_container_and_config(TaskId, ContainerDir, Params) + when is_integer(TaskId), is_list(ContainerDir), is_record(Params, 'ContainerDeployParams') -> + case docker_commands:create_container(ContainerDir, Params) of + {ok, ContainerId} -> + ok = create_config_file(TaskId, ContainerDir), + {ok, ContainerId}; + {error, Reason} when is_binary(Reason) -> + throw({deploy_error, format_create_container_error(Reason)}); + {error, Reason} -> + Error = iolist_to_binary(io_lib:format("容器创建失败: ~p", [Reason])), + throw({deploy_error, Error}) + end. + +-spec create_config_file(TaskId :: integer(), ContainerDir :: string()) -> ok. +create_config_file(TaskId, ContainerDir) when is_integer(TaskId), is_list(ContainerDir) -> + ConfigFile = docker_helper:get_config_file(ContainerDir), + case file:open(ConfigFile, [write, exclusive]) of + {ok, FD} -> + ok = file:write(FD, <<>>), + file:close(FD), + ok; + {error, Reason} -> + ReasonBin = list_to_binary(io_lib:format("~p", [Reason])), + trace_log(TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, ReasonBin/binary>>), + ok + end. + +-spec format_create_container_error(binary()) -> binary(). +format_create_container_error(Reason) when is_binary(Reason) -> + case is_container_already_exists_error(Reason) of + true -> + <<"本地容器已经存在"/utf8>>; + false -> + <<"容器创建失败: "/utf8, Reason/binary>> + end. + +-spec is_container_already_exists_error(binary()) -> boolean(). +is_container_already_exists_error(Reason) when is_binary(Reason) -> + binary:match(Reason, <<"is already in use by container">>) =/= nomatch orelse + binary:match(Reason, <<"Conflict. The container name ">>) =/= nomatch. + +-spec short_container_id(binary()) -> binary(). +short_container_id(ContainerId) when is_binary(ContainerId), byte_size(ContainerId) >= 12 -> + binary:part(ContainerId, 0, 12); +short_container_id(ContainerId) when is_binary(ContainerId) -> + ContainerId.