From 576ccb2b108d6c18c9b7ec30abb7b925ae30d776 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Mon, 20 Apr 2026 14:39:03 +0800 Subject: [PATCH] fix --- src/docker/docker_commands.erl | 388 +++++++++++++--------------- src/docker/docker_deployer.erl | 22 +- src/docker/docker_manager.erl | 17 +- src/efka_remote_agent.erl | 168 +----------- src/tests/docker_commands_tests.erl | 120 +++++---- 5 files changed, 267 insertions(+), 448 deletions(-) diff --git a/src/docker/docker_commands.erl b/src/docker/docker_commands.erl index 5448963..39149bf 100644 --- a/src/docker/docker_commands.erl +++ b/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 = [<>|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 -> - <>; +normalize_expose_port(#'PortExpose'{container_port = Port, protocol = Protocol}) -> + PortBin = integer_to_binary(Port), + ProtocolBin = to_binary(Protocol), + case ProtocolBin of + <<>> -> + <>; + <<"tcp">> -> + <>; _ -> - Port + <> 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 -> + <>; + false -> + <> 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. diff --git a/src/docker/docker_deployer.erl b/src/docker/docker_deployer.erl index 964315e..3381f73 100644 --- a/src/docker/docker_deployer.erl +++ b/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). \ No newline at end of file + efka_logger:write(Info). diff --git a/src/docker/docker_manager.erl b/src/docker/docker_manager.erl index 29186de..3d0d3e8 100644 --- a/src/docker/docker_manager.erl +++ b/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)}}; %% 处理容器关联的配置文件 diff --git a/src/efka_remote_agent.erl b/src/efka_remote_agent.erl index ad444c7..8d3e3c2 100644 --- a/src/efka_remote_agent.erl +++ b/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), - <>. - -volume_bind(#'VolumeBind'{host_path = HostPath, container_path = ContainerPath, read_only = true}) -> - HostPathBin = to_binary(HostPath), - ContainerPathBin = to_binary(ContainerPath), - <>; -volume_bind(#'VolumeBind'{host_path = HostPath, container_path = ContainerPath, read_only = ReadOnly}) -> - false = to_bool(ReadOnly), - HostPathBin = to_binary(HostPath), - ContainerPathBin = to_binary(ContainerPath), - <>. - -port_expose(#'PortExpose'{container_port = ContainerPort, protocol = Protocol}) -> - PortBin = integer_to_binary(ContainerPort), - ProtocolBin = to_binary(Protocol), - case ProtocolBin of - <<>> -> - PortBin; - <<"tcp">> -> - PortBin; - _ -> - <> - end. - -device_mapping(#'DeviceMapping'{host_path = HostPath, container_path = ContainerPath, cgroup_permissions = <<"rwm">>}) -> - HostPathBin = to_binary(HostPath), - ContainerPathBin = to_binary(ContainerPath), - <>; -device_mapping(#'DeviceMapping'{host_path = HostPath, container_path = ContainerPath, cgroup_permissions = Permissions}) -> - HostPathBin = to_binary(HostPath), - ContainerPathBin = to_binary(ContainerPath), - PermissionsBin = to_binary(Permissions), - <>. - -ulimit_spec(#'Ulimit'{soft = Soft, hard = Hard}) -> - SoftBin = integer_to_binary(Soft), - HardBin = integer_to_binary(Hard), - <>. - -tmpfs_mount(#'TmpfsMount'{path = Path, options = <<>>}) -> - to_binary(Path); -tmpfs_mount(#'TmpfsMount'{path = Path, options = Options}) -> - PathBin = to_binary(Path), - OptionsBin = to_binary(Options), - <>. - -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) -> diff --git a/src/tests/docker_commands_tests.erl b/src/tests/docker_commands_tests.erl index 8bd3fe8..5f9fa42 100644 --- a/src/tests/docker_commands_tests.erl +++ b/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).