fix

src/docker/docker_commands.erl

@@ -8,10 +8,11 @@
 %%%-------------------------------------------------------------------
 -module(docker_commands).
 -author("anlicheng").
+-include("message_pb.hrl").
 
 %% API
 -export([pull_image/2, check_image_exist/1]).
--export([create_container/3, check_container_exist/1, is_container_running/1,
+-export([create_container/2, check_container_exist/1, is_container_running/1,
     start_container/1, stop_container/1, stop_container/2, remove_container/1, remove_container/3, kill_container/1, kill_container/2,
     get_containers/0]).
 
@@ -31,18 +32,22 @@ check_image_exist(Image) when is_binary(Image) ->
             false
     end.
 
--spec create_container(ContainerName :: binary(), ContainerDir :: string(), Config :: map()) -> {ok, ContainerId :: binary()} | {error, Reason :: any()}.
-create_container(ContainerName, ContainerDir, Config) when is_binary(ContainerName), is_list(ContainerDir), is_map(Config) ->
+-spec create_container(ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) ->
+    {ok, ContainerId :: binary()} | {error, Reason :: any()}.
+create_container(ContainerDir, #'ContainerDeployParams'{
+    container_name = ContainerName,
+    spec = Spec0
+}) when is_binary(ContainerName), is_list(ContainerDir), is_record(Spec0, 'ContainerSpec') ->
     Url = lists:flatten(io_lib:format("/containers/create?name=~s", [binary_to_list(ContainerName)])),
     %% 挂载预留的目录，用来作为配置文件的存放
     ConfigFile = list_to_binary(docker_helper:get_config_file(ContainerDir)),
-
-    %% 增加自定义的用来放配置文件的目录
-    Volumes0 = maps:get(<<"volumes">>, Config, []),
-    Volumes = [<<ConfigFile/binary, ":/usr/local/etc/service.conf">>|Volumes0],
-    NewConfig = Config#{<<"volumes">> => Volumes},
-
-    Options = build_options(ContainerName, NewConfig),
+    ConfigVolume = #'VolumeBind'{
+        host_path = ConfigFile,
+        container_path = <<"/usr/local/etc/service.conf">>,
+        read_only = false
+    },
+    Spec = Spec0#'ContainerSpec'{volumes = [ConfigVolume | Spec0#'ContainerSpec'.volumes]},
+    Options = build_options(ContainerName, Spec),
     display_options(Options),
 
     Body = iolist_to_binary(jiffy:encode(Options, [force_utf8])),
@@ -234,36 +239,35 @@ inspect_container(ContainerId) when is_binary(ContainerId) ->
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 
 %% 构建最终 JSON Map
-build_options(ContainerName, Config) when is_binary(ContainerName), is_map(Config) ->
+build_options(ContainerName, #'ContainerSpec'{} = Spec) when is_binary(ContainerName) ->
     %% 容器名称作为环境变量，注入到运行时环境中
-    Envs0 = maps:get(<<"envs">>, Config, []),
+    Envs0 = [to_binary(Env) || Env <- Spec#'ContainerSpec'.env],
     Envs = [<<"CONTAINER_NAME=", ContainerName/binary>>|Envs0],
     #{
-        <<"Image">> => maps:get(<<"image">>, Config, <<>>),
-        <<"Cmd">> => maps:get(<<"command">>, Config, []),
-        <<"Entrypoint">> => maps:get(<<"entrypoint">>, Config, []),
+        <<"Image">> => to_binary(Spec#'ContainerSpec'.image),
+        <<"Cmd">> => [to_binary(Command) || Command <- Spec#'ContainerSpec'.command],
+        <<"Entrypoint">> => [to_binary(Entrypoint) || Entrypoint <- Spec#'ContainerSpec'.entrypoint],
         <<"Env">> => Envs,
-        <<"Labels">> => maps:get(<<"labels">>, Config, #{}),
-        <<"Volumes">> => build_volumes(Config),
-        <<"User">> => maps:get(<<"user">>, Config, <<>>),
-        <<"WorkingDir">> => maps:get(<<"working_dir">>, Config, <<>>),
-        <<"Hostname">> => maps:get(<<"hostname">>, Config, <<>>),
-        <<"ExposedPorts">> => build_expose(Config),
-        <<"NetworkingConfig">> => build_networks(Config),
-        <<"Healthcheck">> => build_healthcheck(Config),
+        <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- Spec#'ContainerSpec'.labels]),
+        <<"Volumes">> => build_volumes(Spec),
+        <<"User">> => to_binary(Spec#'ContainerSpec'.user),
+        <<"WorkingDir">> => to_binary(Spec#'ContainerSpec'.working_dir),
+        <<"Hostname">> => to_binary(Spec#'ContainerSpec'.hostname),
+        <<"ExposedPorts">> => build_expose(Spec),
+        <<"NetworkingConfig">> => build_networks(Spec),
+        <<"Healthcheck">> => build_healthcheck(Spec),
         <<"HostConfig">> => fold_merge([
-            build_network_mode(Config),
-            build_binds(Config),
-            build_restart(Config),
-            build_privileged(Config),
-            build_cap_add_drop(Config),
-            build_devices(Config),
-            build_memory(Config),
-            build_cpu(Config),
-            build_ulimits(Config),
-            build_tmpfs(Config),
-            build_sysctls(Config),
-            build_extra_hosts(Config)
+            build_network_mode(Spec),
+            build_binds(Spec),
+            build_restart(Spec),
+            build_privileged(Spec),
+            build_cap_add_drop(Spec),
+            build_devices(Spec),
+            build_resources(Spec),
+            build_ulimits(Spec),
+            build_tmpfs(Spec),
+            build_sysctls(Spec),
+            build_extra_hosts(Spec)
         ])
     }.
 
@@ -272,221 +276,180 @@ fold_merge(List) ->
     lists:foldl(fun maps:merge/2, #{}, List).
 
 %% --- 构建子字段 ---
-build_expose(Config) ->
-    Ports = maps:get(<<"expose">>, Config, []),
+build_expose(#'ContainerSpec'{expose = Ports}) ->
     case Ports of
         [] -> #{};
         _ ->
             maps:from_list([{normalize_expose_port(P), #{}} || P <- Ports])
     end.
 
-build_volumes(Config) ->
-    Vols = maps:get(<<"volumes">>, Config, []),
+build_volumes(#'ContainerSpec'{volumes = Vols}) ->
     case Vols of
         [] ->
             #{};
         _ ->
-            maps:from_list(lists:map(fun(V) ->
-                [_Host, Cont | _Modes] = binary:split(V, <<":">>, [global]),
-                {Cont, #{}}
-                                     end, Vols))
+            maps:from_list([{to_binary(Cont), #{}} || #'VolumeBind'{container_path = Cont} <- Vols])
     end.
 
-build_binds(Config) ->
-    Vols = maps:get(<<"volumes">>, Config, []),
+build_binds(#'ContainerSpec'{volumes = Vols}) ->
     case Vols of
         [] ->
             #{};
         _ ->
-            #{<<"Binds">> => Vols}
+            #{<<"Binds">> => [volume_bind(Vol) || Vol <- Vols]}
     end.
 
-build_networks(Config) ->
-    Nets = maps:get(<<"networks">>, Config, []),
+build_networks(#'ContainerSpec'{networks = Nets}) ->
     case Nets of
         [] -> #{};
         _ ->
-            NetCfg = maps:from_list([{N, #{}} || N <- Nets]),
+            NetCfg = maps:from_list([{to_binary(N), #{}} || N <- Nets]),
             #{<<"EndpointsConfig">> => NetCfg}
     end.
 
-build_network_mode(Config) ->
-    NetworkMode = maps:get(<<"network_mode">>, Config, <<"bridge">>),
-    #{<<"NetworkMode">> => NetworkMode}.
+build_network_mode(#'ContainerSpec'{network_mode = <<>>}) ->
+    #{};
+build_network_mode(#'ContainerSpec'{network_mode = NetworkMode}) ->
+    #{<<"NetworkMode">> => to_binary(NetworkMode)}.
 
-parse_mem(Val) ->
-    case binary:last(Val) of
-        $m ->
-            N = binary:part(Val, {0, byte_size(Val)-1}),
-            list_to_integer(binary_to_list(N)) * 1024 * 1024;
-        $g ->
-            N = binary:part(Val, {0, byte_size(Val)-1}),
-            list_to_integer(binary_to_list(N)) * 1024 * 1024 * 1024;
-        _ ->
-            list_to_integer(binary_to_list(Val))
-    end.
+build_healthcheck(#'ContainerSpec'{healthcheck = undefined}) ->
+    #{};
+build_healthcheck(#'ContainerSpec'{healthcheck = #'Healthcheck'{
+    test = Test,
+    interval_ns = IntervalNs,
+    timeout_ns = TimeoutNs,
+    retries = Retries
+}}) ->
+    #{
+        <<"Test">> => [to_binary(Item) || Item <- Test],
+        <<"Interval">> => IntervalNs,
+        <<"Timeout">> => TimeoutNs,
+        <<"Retries">> => Retries
+    }.
 
-build_healthcheck(Config) ->
-    HC = maps:get(<<"healthcheck">>, Config, #{}),
-    case maps:size(HC) of
+build_restart(#'ContainerSpec'{restart = undefined}) ->
+    #{};
+build_restart(#'ContainerSpec'{restart = #'RestartPolicy'{
+    name = Name,
+    maximum_retry_count = RetryCount
+}}) ->
+    RestartPolicy = #{
+        <<"Name">> => to_binary(Name)
+    },
+    case RetryCount of
         0 ->
-            #{};
+            #{<<"RestartPolicy">> => RestartPolicy};
         _ ->
-            #{
-                <<"Test">> => maps:get(<<"test">>, HC, []),
-                <<"Interval">> => parse_duration(maps:get(<<"interval">>, HC, <<"0s">>)),
-                <<"Timeout">> => parse_duration(maps:get(<<"timeout">>, HC, <<"0s">>)),
-                <<"Retries">> => maps:get(<<"retries">>, HC, 0)
-            }
+            #{<<"RestartPolicy">> => RestartPolicy#{
+                <<"MaximumRetryCount">> => RetryCount
+            }}
     end.
 
-parse_duration(Bin) ->
-    case re:run(Bin, <<"^(\\d+)(ns|us|ms|s|m|h)?$">>, [{capture, all_but_first, binary}]) of
-        {match, [NumberBin, Unit]} ->
-            Number = binary_to_integer(NumberBin),
-            case Unit of
-                <<"ns">> ->
-                    Number;
-                <<"us">> ->
-                    Number * 1000;
-                <<"ms">> ->
-                    Number * 1000000;
-                <<"s">> ->
-                    Number * 1000000000;
-                <<"m">> ->
-                    Number * 60000000000;
-                <<"h">> ->
-                    Number * 3600000000000
-            end;
-        {match, [NumberBin]} ->
-            binary_to_integer(NumberBin)
-    end.
-
-%% --- 构建子字段 ---
-
-build_restart(Config) ->
-    case maps:get(<<"restart">>, Config, undefined) of
-        undefined ->
-            #{};
-        Policy ->
-            case binary:split(Policy, <<":">>) of
-                [Name, RetryCountBin] ->
-                    #{<<"RestartPolicy">> => #{
-                        <<"Name">> => Name,
-                        <<"MaximumRetryCount">> => binary_to_integer(RetryCountBin)
-                    }};
-                [Name] ->
-                    #{<<"RestartPolicy">> => #{<<"Name">> => Name}}
-            end
-    end.
-
-build_privileged(Config) ->
-    case maps:get(<<"privileged">>, Config, false) of
+build_privileged(#'ContainerSpec'{privileged = Privileged}) ->
+    case to_bool(Privileged) of
         true ->
             #{<<"Privileged">> => true};
         _ ->
             #{}
     end.
 
-build_cap_add_drop(Config) ->
-    Add = maps:get(<<"cap_add">>, Config, []),
-    Drop = maps:get(<<"cap_drop">>, Config, []),
+build_cap_add_drop(#'ContainerSpec'{cap_add = Add, cap_drop = Drop}) ->
     case {Add, Drop} of
         {[], []} ->
             #{};
         _ ->
-            #{<<"CapAdd">> => Add, <<"CapDrop">> => Drop}
+            #{
+                <<"CapAdd">> => [to_binary(Item) || Item <- Add],
+                <<"CapDrop">> => [to_binary(Item) || Item <- Drop]
+            }
     end.
 
-build_devices(Config) ->
-    Devs = maps:get(<<"devices">>, Config, []),
+build_devices(#'ContainerSpec'{devices = Devs}) ->
     case Devs of
         [] ->
             #{};
         _ ->
-            DevObjs = [#{<<"PathOnHost">> => H, <<"PathInContainer">> => C,
-                <<"CgroupPermissions">> => P}
-                || D <- Devs,
-                {H, C, P} <- [parse_device_mapping(D)]],
+            DevObjs = [#{
+                <<"PathOnHost">> => to_binary(HostPath),
+                <<"PathInContainer">> => to_binary(ContainerPath),
+                <<"CgroupPermissions">> => device_permissions(Permissions)
+            } || #'DeviceMapping'{
+                host_path = HostPath,
+                container_path = ContainerPath,
+                cgroup_permissions = Permissions
+            } <- Devs],
             #{<<"Devices">> => DevObjs}
     end.
 
-build_memory(Config) ->
-    Mem = maps:get(<<"mem_limit">>, Config, undefined),
-    MemRes = maps:get(<<"mem_reservation">>, Config, undefined),
-    HCfg = #{},
-    HCfg1 = if
-                Mem /= undefined ->
-                    maps:put(<<"Memory">>, parse_mem(Mem), HCfg);
-                true ->
-                    HCfg
-            end,
-    if
-        MemRes /= undefined ->
-            maps:put(<<"MemoryReservation">>, parse_mem(MemRes), HCfg1);
-        true ->
-            HCfg1
-    end.
-
-
-build_cpu(Config) ->
-    CPU = maps:get(<<"cpus">>, Config, undefined),
-    Shares = maps:get(<<"cpu_shares">>, Config, undefined),
-    HCfg = #{},
-    HCfg1 = if
-                CPU /= undefined ->
-                    maps:put(<<"NanoCpus">>, trunc(CPU * 1000000000), HCfg);
-                true ->
-                    HCfg
-            end,
-    if
-        Shares /= undefined ->
-            maps:put(<<"CpuShares">>, Shares, HCfg1);
-        true ->
-            HCfg1
-    end.
-
-build_ulimits(Config) ->
-    UL = maps:get(<<"ulimits">>, Config, #{}),
-    case maps:size(UL) of
+build_resources(#'ContainerSpec'{resources = undefined}) ->
+    #{};
+build_resources(#'ContainerSpec'{resources = #'ResourceLimits'{
+    memory_bytes = MemoryBytes,
+    memory_reservation_bytes = ReservationBytes,
+    nano_cpus = NanoCpus,
+    cpu_shares = CpuShares
+}}) ->
+    HostConfig0 = #{},
+    HostConfig1 = case MemoryBytes of
         0 ->
-            #{};
+            HostConfig0;
         _ ->
-            ULList = lists:map(fun({K, V}) ->
-                [S1, H1] = binary:split(V, <<":">>, []),
-                S = list_to_integer(binary_to_list(S1)),
-                H = list_to_integer(binary_to_list(H1)),
-                #{<<"Name">> => K, <<"Soft">> => S, <<"Hard">> => H}
-                               end, maps:to_list(UL)),
-
-            #{<<"Ulimits">> => ULList}
-    end.
-
-build_sysctls(Config) ->
-    SC = maps:get(<<"sysctls">>, Config, #{}),
-    case maps:size(SC) of
+            HostConfig0#{<<"Memory">> => MemoryBytes}
+    end,
+    HostConfig2 = case ReservationBytes of
         0 ->
-            #{};
+            HostConfig1;
         _ ->
-            #{<<"Sysctls">> => SC}
+            HostConfig1#{<<"MemoryReservation">> => ReservationBytes}
+    end,
+    HostConfig3 = case NanoCpus of
+        0 ->
+            HostConfig2;
+        _ ->
+            HostConfig2#{<<"NanoCpus">> => NanoCpus}
+    end,
+    case CpuShares of
+        0 ->
+            HostConfig3;
+        _ ->
+            HostConfig3#{<<"CpuShares">> => CpuShares}
     end.
 
-build_tmpfs(Config) ->
-    Tmp = maps:get(<<"tmpfs">>, Config, []),
-    case Tmp of
+build_ulimits(#'ContainerSpec'{ulimits = Ulimits}) ->
+    case Ulimits of
         [] ->
             #{};
         _ ->
-            #{<<"Tmpfs">> => maps:from_list([parse_tmpfs_mount(T) || T <- Tmp])}
+            #{<<"Ulimits">> => [#{
+                <<"Name">> => to_binary(Name),
+                <<"Soft">> => Soft,
+                <<"Hard">> => Hard
+            } || #'Ulimit'{name = Name, soft = Soft, hard = Hard} <- Ulimits]}
     end.
 
-build_extra_hosts(Config) ->
-    Hosts = maps:get(<<"extra_hosts">>, Config, []),
+build_sysctls(#'ContainerSpec'{sysctls = Sysctls}) ->
+    case Sysctls of
+        [] ->
+            #{};
+        _ ->
+            #{<<"Sysctls">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- Sysctls])}
+    end.
+
+build_tmpfs(#'ContainerSpec'{tmpfs = Tmpfs}) ->
+    case Tmpfs of
+        [] ->
+            #{};
+        _ ->
+            #{<<"Tmpfs">> => maps:from_list([{to_binary(Path), to_binary(Options)} ||
+                #'TmpfsMount'{path = Path, options = Options} <- Tmpfs])}
+    end.
+
+build_extra_hosts(#'ContainerSpec'{extra_hosts = Hosts}) ->
     case Hosts of
         [] ->
             #{};
         _ ->
-            #{<<"ExtraHosts">> => Hosts}
+            #{<<"ExtraHosts">> => [to_binary(Host) || Host <- Hosts]}
     end.
 
 -spec display_options(Options :: map()) -> no_return().
@@ -494,28 +457,31 @@ display_options(Options) when is_map(Options) ->
     logger:debug("deploy options: ~p", [jiffy:encode(Options, [force_utf8])]),
     lists:foreach(fun({K, V}) -> logger:debug("~p => ~p", [K, V]) end, maps:to_list(Options)).
 
-normalize_expose_port(Port) when is_binary(Port) ->
-    case binary:match(Port, <<"/">>) of
-        nomatch ->
-            <<Port/binary, "/tcp">>;
+normalize_expose_port(#'PortExpose'{container_port = Port, protocol = Protocol}) ->
+    PortBin = integer_to_binary(Port),
+    ProtocolBin = to_binary(Protocol),
+    case ProtocolBin of
+        <<>> ->
+            <<PortBin/binary, "/tcp">>;
+        <<"tcp">> ->
+            <<PortBin/binary, "/tcp">>;
         _ ->
-            Port
+            <<PortBin/binary, "/", ProtocolBin/binary>>
     end.
 
-parse_device_mapping(Device) when is_binary(Device) ->
-    case binary:split(Device, <<":">>, [global]) of
-        [HostPath, ContainerPath] ->
-            {HostPath, ContainerPath, <<"rwm">>};
-        [HostPath, ContainerPath, Permissions] ->
-            {HostPath, ContainerPath, Permissions}
-    end.
+device_permissions(<<>>) ->
+    <<"rwm">>;
+device_permissions(Permissions) ->
+    to_binary(Permissions).
 
-parse_tmpfs_mount(Tmpfs) when is_binary(Tmpfs) ->
-    case binary:split(Tmpfs, <<":">>) of
-        [Path] ->
-            {Path, <<>>};
-        [Path, Options] ->
-            {Path, Options}
+volume_bind(#'VolumeBind'{host_path = HostPath, container_path = ContainerPath, read_only = ReadOnly}) ->
+    HostPathBin = to_binary(HostPath),
+    ContainerPathBin = to_binary(ContainerPath),
+    case to_bool(ReadOnly) of
+        true ->
+            <<HostPathBin/binary, ":", ContainerPathBin/binary, ":ro">>;
+        false ->
+            <<HostPathBin/binary, ":", ContainerPathBin/binary>>
     end.
 
 build_stop_container_url(ContainerName, 0) ->
@@ -541,3 +507,17 @@ boolean_to_query_value(true) ->
     "true";
 boolean_to_query_value(false) ->
     "false".
+
+to_binary(Value) when is_binary(Value) ->
+    Value;
+to_binary(Value) when is_list(Value) ->
+    unicode:characters_to_binary(Value).
+
+to_bool(true) ->
+    true;
+to_bool(1) ->
+    true;
+to_bool(false) ->
+    false;
+to_bool(0) ->
+    false.

src/docker/docker_deployer.erl

@@ -8,6 +8,7 @@
 %%%-------------------------------------------------------------------
 -module(docker_deployer).
 -author("anlicheng").
+-include("message_pb.hrl").
 -dialyzer([{nowarn_function, normalize_image/1}]).
 
 %% API
@@ -22,9 +23,11 @@
 %%%===================================================================
 
 %% @doc Spawns the server and registers the local name (unique)
--spec(start_monitor(TaskId :: integer(), ContainerDir :: string(), Config :: map()) -> {ok, {Pid :: pid(), MRef :: reference()}}).
-start_monitor(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerDir), is_map(Config) ->
-    {ok, spawn_monitor(?MODULE, deploy, [TaskId, ContainerDir, Config])}.
+-spec(start_monitor(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) ->
+    {ok, {Pid :: pid(), MRef :: reference()}}).
+start_monitor(TaskId, ContainerDir, Params)
+  when is_integer(TaskId), is_list(ContainerDir), is_record(Params, 'ContainerDeployParams') ->
+    {ok, spawn_monitor(?MODULE, deploy, [TaskId, ContainerDir, Params])}.
 
 %%%===================================================================
 %%% Internal functions
@@ -40,10 +43,12 @@ start_monitor(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(Con
 %   "command": ["nginx", "-g", "daemon off;"],
 %   "restart": "always"
 %}
--spec deploy(TaskId :: integer(), ContainerDir :: string(), Config :: map()) -> no_return().
-deploy(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerDir), is_map(Config) ->
+-spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) -> no_return().
+deploy(TaskId, ContainerDir, Params = #'ContainerDeployParams'{
+    container_name = ContainerName,
+    spec = #'ContainerSpec'{image = Image0}
+}) when is_integer(TaskId), is_list(ContainerDir) ->
     %% 尝试拉取镜像
-    ContainerName = maps:get(<<"container_name">>, Config),
     trace_log(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
 
     case docker_commands:check_container_exist(ContainerName) of
@@ -51,7 +56,6 @@ deploy(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerD
             trace_log(TaskId, <<"info">>, <<"本地容器已经存在:"/utf8, ContainerName/binary>>),
             efka_remote_agent:close_task_event_stream(TaskId, ?TASK_FAIL);
         false ->
-            Image0 = maps:get(<<"image">>, Config),
             Image = normalize_image(Image0),
 
             trace_log(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
@@ -73,7 +77,7 @@ deploy(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerD
             case PullResult of
                 ok ->
                     trace_log(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
-                    case docker_commands:create_container(ContainerName, ContainerDir, Config) of
+                    case docker_commands:create_container(ContainerDir, Params) of
                         {ok, ContainerId} ->
                             %% 创建容器对应的配置文件
                             ConfigFile = docker_helper:get_config_file(ContainerDir),
@@ -114,4 +118,4 @@ normalize_image(Image) when is_binary(Image) ->
 trace_log(TaskId, Level, Msg) when is_integer(TaskId), is_binary(Level), is_binary(Msg) ->
     efka_remote_agent:task_event_stream(TaskId, Level, Msg),
     Info = iolist_to_binary([<<"task_id=">>, integer_to_binary(TaskId), <<" ">>, Level, <<" ">>, Msg]),
-    efka_logger:write(Info).
+    efka_logger:write(Info).

src/docker/docker_manager.erl

@@ -10,6 +10,7 @@
 %%%-------------------------------------------------------------------
 -module(docker_manager).
 -author("anlicheng").
+-include("message_pb.hrl").
 
 -behaviour(gen_server).
 
@@ -37,9 +38,9 @@
 get_containers() ->
     gen_server:call(?SERVER, get_containers).
 
--spec deploy(TaskId :: integer(), Config :: map()) -> ok | {error, Reason :: binary()}.
-deploy(TaskId, Config) when is_integer(TaskId), is_map(Config) ->
-    gen_server:call(?SERVER, {deploy, TaskId, Config}).
+-spec deploy(TaskId :: integer(), Params :: message_pb:'ContainerDeployParams'()) -> ok | {error, Reason :: binary()}.
+deploy(TaskId, Params) when is_integer(TaskId), is_record(Params, 'ContainerDeployParams') ->
+    gen_server:call(?SERVER, {deploy, TaskId, Params}).
 
 -spec config_container(ContainerName :: binary(), Config :: binary()) -> ok | {error, Reason :: binary()}.
 config_container(ContainerName, Config) when is_binary(ContainerName), is_binary(Config) ->
@@ -104,12 +105,14 @@ init([]) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_call({deploy, TaskId, Config = #{<<"container_name">> := ContainerName}}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) ->
+handle_call({deploy, TaskId, Params = #'ContainerDeployParams'{
+    container_name = ContainerName,
+    container_dir = ContainerDir0
+}}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) ->
     %% 创建目录
-    ContainerDir0 = maps:get(<<"container_dir">>, Config, <<>>),
     {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName, ContainerDir0),
-    {ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Config),
-    logger:debug("[docker_manager] start deploy task_id: ~p, config: ~p", [TaskId, Config]),
+    {ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params),
+    logger:debug("[docker_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]),
     {reply, ok, State#state{task_map = maps:put(TaskPid, TaskId, TaskMap)}};
 
 %% 处理容器关联的配置文件

src/efka_remote_agent.erl

@@ -254,8 +254,7 @@ handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {d
     task_id = TaskId,
     params = Params
 }}}}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    Config = container_deploy_config(Params),
-    case docker_manager:deploy(TaskId, Config) of
+    case docker_manager:deploy(TaskId, Params) of
         ok ->
             Packet = message_pb:encode_msg(#'ResponseFrame'{
                 packet_id = PacketId,
@@ -530,75 +529,6 @@ auth_packet() ->
 encode_rpc_payload(Payload) ->
     jiffy:encode(Payload, [force_utf8]).
 
--spec container_deploy_config(message_pb:'ContainerDeployParams'()) -> map().
-container_deploy_config(#'ContainerDeployParams'{
-    container_name = ContainerName,
-    container_dir = ContainerDir,
-    spec = #'ContainerSpec'{
-        image = Image,
-        command = Command,
-        entrypoint = Entrypoint,
-        env = Env,
-        labels = Labels,
-        volumes = Volumes,
-        user = User,
-        working_dir = WorkingDir,
-        hostname = Hostname,
-        expose = Expose,
-        networks = Networks,
-        network_mode = NetworkMode,
-        healthcheck = Healthcheck,
-        restart = Restart,
-        privileged = Privileged,
-        cap_add = CapAdd,
-        cap_drop = CapDrop,
-        devices = Devices,
-        resources = Resources,
-        ulimits = Ulimits,
-        tmpfs = Tmpfs,
-        sysctls = Sysctls,
-        extra_hosts = ExtraHosts
-    }
-}) ->
-    BaseConfig = #{
-        <<"container_name">> => to_binary(ContainerName),
-        <<"container_dir">> => to_binary(ContainerDir),
-        <<"image">> => to_binary(Image),
-        <<"command">> => [to_binary(Item) || Item <- Command],
-        <<"entrypoint">> => [to_binary(Item) || Item <- Entrypoint],
-        <<"envs">> => [to_binary(Item) || Item <- Env],
-        <<"labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- Labels]),
-        <<"volumes">> => [volume_bind(Volume) || Volume <- Volumes],
-        <<"user">> => to_binary(User),
-        <<"working_dir">> => to_binary(WorkingDir),
-        <<"hostname">> => to_binary(Hostname),
-        <<"expose">> => [port_expose(Port) || Port <- Expose],
-        <<"networks">> => [to_binary(Network) || Network <- Networks],
-        <<"network_mode">> => to_binary(NetworkMode),
-        <<"restart">> => restart_policy(Restart),
-        <<"privileged">> => to_bool(Privileged),
-        <<"cap_add">> => [to_binary(Item) || Item <- CapAdd],
-        <<"cap_drop">> => [to_binary(Item) || Item <- CapDrop],
-        <<"devices">> => [device_mapping(Device) || Device <- Devices],
-        <<"ulimits">> => maps:from_list([{to_binary(Name), ulimit_spec(Ulimit)} || Ulimit = #'Ulimit'{name = Name} <- Ulimits]),
-        <<"tmpfs">> => [tmpfs_mount(Mount) || Mount <- Tmpfs],
-        <<"sysctls">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- Sysctls]),
-        <<"extra_hosts">> => [to_binary(Host) || Host <- ExtraHosts]
-    },
-    HealthcheckConfig = case Healthcheck of
-        undefined ->
-            #{};
-        _ ->
-            #{<<"healthcheck">> => healthcheck_config(Healthcheck)}
-    end,
-    ResourceConfig = case Resources of
-        undefined ->
-            #{};
-        _ ->
-            resource_limits_config(Resources)
-    end,
-    maps:merge(BaseConfig, maps:merge(HealthcheckConfig, ResourceConfig)).
-
 container_target(#'ContainerRef'{name = Name, id = Id}) ->
     NameBin = to_binary(Name),
     IdBin = to_binary(Id),
@@ -610,102 +540,6 @@ container_target(#'ContainerRef'{name = Name, id = Id}) ->
             NameBin
     end.
 
-healthcheck_config(#'Healthcheck'{test = Test, interval_ns = IntervalNs, timeout_ns = TimeoutNs, retries = Retries}) ->
-    #{
-        <<"test">> => [to_binary(Item) || Item <- Test],
-        <<"interval">> => integer_to_binary(IntervalNs),
-        <<"timeout">> => integer_to_binary(TimeoutNs),
-        <<"retries">> => Retries
-    }.
-
-resource_limits_config(#'ResourceLimits'{
-    memory_bytes = MemoryBytes,
-    memory_reservation_bytes = ReservationBytes,
-    nano_cpus = NanoCpus,
-    cpu_shares = CpuShares
-}) ->
-    Config0 = #{},
-    Config1 = case MemoryBytes of
-        0 ->
-            Config0;
-        _ ->
-            Config0#{<<"mem_limit">> => integer_to_binary(MemoryBytes)}
-    end,
-    Config2 = case ReservationBytes of
-        0 ->
-            Config1;
-        _ ->
-            Config1#{<<"mem_reservation">> => integer_to_binary(ReservationBytes)}
-    end,
-    Config3 = case NanoCpus of
-        0 ->
-            Config2;
-        _ ->
-            Config2#{<<"cpus">> => nano_cpus_to_cpus(NanoCpus)}
-    end,
-    case CpuShares of
-        0 ->
-            Config3;
-        _ ->
-            Config3#{<<"cpu_shares">> => CpuShares}
-    end.
-
-restart_policy(#'RestartPolicy'{name = Name, maximum_retry_count = 0}) ->
-    to_binary(Name);
-restart_policy(#'RestartPolicy'{name = Name, maximum_retry_count = RetryCount}) ->
-    NameBin = to_binary(Name),
-    RetryCountBin = integer_to_binary(RetryCount),
-    <<NameBin/binary, ":", RetryCountBin/binary>>.
-
-volume_bind(#'VolumeBind'{host_path = HostPath, container_path = ContainerPath, read_only = true}) ->
-    HostPathBin = to_binary(HostPath),
-    ContainerPathBin = to_binary(ContainerPath),
-    <<HostPathBin/binary, ":", ContainerPathBin/binary, ":ro">>;
-volume_bind(#'VolumeBind'{host_path = HostPath, container_path = ContainerPath, read_only = ReadOnly}) ->
-    false = to_bool(ReadOnly),
-    HostPathBin = to_binary(HostPath),
-    ContainerPathBin = to_binary(ContainerPath),
-    <<HostPathBin/binary, ":", ContainerPathBin/binary>>.
-
-port_expose(#'PortExpose'{container_port = ContainerPort, protocol = Protocol}) ->
-    PortBin = integer_to_binary(ContainerPort),
-    ProtocolBin = to_binary(Protocol),
-    case ProtocolBin of
-        <<>> ->
-            PortBin;
-        <<"tcp">> ->
-            PortBin;
-        _ ->
-            <<PortBin/binary, "/", ProtocolBin/binary>>
-    end.
-
-device_mapping(#'DeviceMapping'{host_path = HostPath, container_path = ContainerPath, cgroup_permissions = <<"rwm">>}) ->
-    HostPathBin = to_binary(HostPath),
-    ContainerPathBin = to_binary(ContainerPath),
-    <<HostPathBin/binary, ":", ContainerPathBin/binary>>;
-device_mapping(#'DeviceMapping'{host_path = HostPath, container_path = ContainerPath, cgroup_permissions = Permissions}) ->
-    HostPathBin = to_binary(HostPath),
-    ContainerPathBin = to_binary(ContainerPath),
-    PermissionsBin = to_binary(Permissions),
-    <<HostPathBin/binary, ":", ContainerPathBin/binary, ":", PermissionsBin/binary>>.
-
-ulimit_spec(#'Ulimit'{soft = Soft, hard = Hard}) ->
-    SoftBin = integer_to_binary(Soft),
-    HardBin = integer_to_binary(Hard),
-    <<SoftBin/binary, ":", HardBin/binary>>.
-
-tmpfs_mount(#'TmpfsMount'{path = Path, options = <<>>}) ->
-    to_binary(Path);
-tmpfs_mount(#'TmpfsMount'{path = Path, options = Options}) ->
-    PathBin = to_binary(Path),
-    OptionsBin = to_binary(Options),
-    <<PathBin/binary, ":", OptionsBin/binary>>.
-
-nano_cpus_to_cpus(NanoCpus) when NanoCpus rem 1000000000 =:= 0 ->
-    NanoCpus div 1000000000;
-nano_cpus_to_cpus(NanoCpus) ->
-    NanoCpus / 1000000000.
-
 to_binary(Value) when is_binary(Value) ->
     Value;
 to_binary(Value) when is_list(Value) ->

src/tests/docker_commands_tests.erl

@@ -8,6 +8,7 @@
 %%%-------------------------------------------------------------------
 -module(docker_commands_tests).
 -author("anlicheng").
+-include("message_pb.hrl").
 
 %% API
 -export([test_pull/0, test_commands/0, test_create_container/0]).
@@ -24,81 +25,78 @@ test_commands() ->
     logger:debug("start res: ~p", [StartRes]).
 
 test_create_container() ->
-    M = #{
-        <<"image">> => <<"docker.1ms.run/library/nginx:latest">>,
-        <<"container_name">> => <<"my_nginx_new1">>,
-        <<"command">> => [
+    Params = #'ContainerDeployParams'{
+        container_name = <<"my_nginx_new1">>,
+        spec = #'ContainerSpec'{
+            image = <<"docker.1ms.run/library/nginx:latest">>,
+            command = [
             <<"nginx">>,
             <<"-g">>,
             <<"daemon off;">>
         ],
-        <<"entrypoint">> => [
+            entrypoint = [
             <<"/docker-entrypoint.sh">>
         ],
-        <<"envs">> => [
+            env = [
             <<"ENV1=val1">>,
             <<"ENV2=val2">>
         ],
-        <<"env_file">> => [
-            <<"./env.list">>
+            expose = [
+                #'PortExpose'{container_port = 80, protocol = <<"tcp">>},
+                #'PortExpose'{container_port = 443, protocol = <<"tcp">>}
         ],
-        <<"ports">> => [
-            <<"8080:80">>,
-            <<"443:443">>
+            volumes = [
+                #'VolumeBind'{host_path = <<"/host/data">>, container_path = <<"/data">>},
+                #'VolumeBind'{host_path = <<"/host/log">>, container_path = <<"/var/log">>}
         ],
-        <<"expose">> => [
-            <<"80">>,
-            <<"443">>
+            networks = [
+                <<"mynet">>
         ],
-        <<"volumes">> => [
-            <<"/host/data:/data">>,
-            <<"/host/log:/var/log">>
-        ],
-        <<"networks">> => [
-            <<"mynet">>
-        ],
-        <<"labels">> => #{
-            <<"role">> => <<"web">>,
-            <<"env">> => <<"prod">>
-        },
-        <<"restart">> => <<"always">>,
-        <<"user">> => <<"www-data">>,
-        <<"working_dir">> => <<"/app">>,
-        <<"hostname">> => <<"myhost">>,
-        <<"privileged">> => true,
-        <<"cap_add">> => [
-            <<"NET_ADMIN">>
-        ],
-        <<"cap_drop">> => [
-            <<"MKNOD">>
-        ],
-        <<"devices">> => [
-            <<"/dev/snd:/dev/snd">>
-        ],
-        <<"mem_limit">> => <<"512m">>,
-        <<"mem_reservation">> => <<"256m">>,
-        <<"cpu_shares">> => 512,
-        <<"cpus">> => 1.5,
-        <<"ulimits">> => #{
-            <<"nofile">> => <<"1024:2048">>
-        },
-        <<"sysctls">> => #{
-            <<"net.ipv4.ip_forward">> => <<"1">>
-        },
-        <<"tmpfs">> => [
-            <<"/tmp">>
-        ],
-        <<"extra_hosts">> => [
-            <<"host1:192.168.0.1">>
-        ],
-        <<"healthcheck">> => #{
-            <<"test">> => [
+            labels = [
+                {<<"role">>, <<"web">>},
+                {<<"env">>, <<"prod">>}
+            ],
+            restart = #'RestartPolicy'{name = <<"always">>},
+            user = <<"www-data">>,
+            working_dir = <<"/app">>,
+            hostname = <<"myhost">>,
+            privileged = true,
+            cap_add = [
+                <<"NET_ADMIN">>
+            ],
+            cap_drop = [
+                <<"MKNOD">>
+            ],
+            devices = [
+                #'DeviceMapping'{host_path = <<"/dev/snd">>, container_path = <<"/dev/snd">>}
+            ],
+            resources = #'ResourceLimits'{
+                memory_bytes = 512 * 1024 * 1024,
+                memory_reservation_bytes = 256 * 1024 * 1024,
+                cpu_shares = 512,
+                nano_cpus = 1500000000
+            },
+            ulimits = [
+                #'Ulimit'{name = <<"nofile">>, soft = 1024, hard = 2048}
+            ],
+            sysctls = [
+                {<<"net.ipv4.ip_forward">>, <<"1">>}
+            ],
+            tmpfs = [
+                #'TmpfsMount'{path = <<"/tmp">>}
+            ],
+            extra_hosts = [
+                <<"host1:192.168.0.1">>
+            ],
+            healthcheck = #'Healthcheck'{
+                test = [
                 <<"CMD-SHELL">>,
                 <<"curl -f http://localhost || exit 1">>
             ],
-            <<"interval">> => <<"30s">>,
-            <<"timeout">> => <<"10s">>,
-            <<"retries">> => 3
+                interval_ns = 30000000000,
+                timeout_ns = 10000000000,
+                retries = 3
+            }
         }
     },
-    docker_commands:create_container(<<"my_nginx_xx3">>, "/usr/local/code/efka/", M).
+    docker_commands:create_container("/usr/local/code/efka/", Params).