fix docker_deployer

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>>),
-
+    try
-    case docker_commands:check_container_exist(ContainerName) of
+        ok = ensure_container_absent(TaskId, ContainerName),
-        true ->
+        {ok, _Image} = ensure_image_ready(TaskId, Image0),
-            trace_log(TaskId, <<"info">>, <<"本地容器已经存在:"/utf8, ContainerName/binary>>),
+        {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 ->
+        Class:Reason:Stacktrace ->
-            Image = normalize_image(Image0),
+            Error = iolist_to_binary(io_lib:format("deploy crashed: ~p:~p ~p", [Class, Reason, Stacktrace])),
-
+            trace_log(TaskId, <<"error">>, Error),
-            trace_log(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
+            trace_log(TaskId, <<"error">>, <<"任务失败"/utf8>>),
-            PullResult = case docker_commands:check_image_exist(Image) of
+            efka_task_reporter:close(TaskId, ?TASK_FAIL)
-                             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
     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.