iot_cloud/src/docker/docker_container_builder.erl
2026-04-22 10:26:38 +08:00

510 lines
21 KiB
Erlang

%%%-------------------------------------------------------------------
%%% @author
%%% @copyright (C) 2026, <COMPANY>
%%% @doc
%%% ContainerRequest protobuf builder helpers.
%%% @end
%%%-------------------------------------------------------------------
-module(docker_container_builder).
-include("message_pb.hrl").
-export([list_request/0, config_request/2, deploy_request/2, start_request/1, stop_request/1, kill_request/1, remove_request/1]).
-spec list_request() -> message_pb:'ContainerRequest'().
list_request() ->
#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = true}}}.
-spec config_request(ContainerName :: binary(), ConfigJson :: binary()) -> message_pb:'ContainerRequest'().
config_request(ContainerName, ConfigJson) when is_binary(ContainerName), is_binary(ConfigJson) ->
#'ContainerRequest'{action = {config, #'ContainerRequest.Config'{
target = container_ref(ContainerName),
config = ConfigJson
}}}.
-spec deploy_request(TaskId :: integer(), Config :: map()) ->
{ok, message_pb:'ContainerRequest'()} | {error, binary()}.
deploy_request(TaskId, Config) when is_integer(TaskId), is_map(Config), TaskId >= 0 ->
try
ensure_supported_deploy_config(Config),
validate_deploy_config(Config),
Params = build_container_deploy_params(Config),
{ok, #'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{task_id = TaskId, params = Params}}}}
catch
throw:{error, Reason} ->
{error, Reason}
end.
-spec start_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
start_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = container_ref(ContainerName)}}}.
-spec stop_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
stop_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{
target = container_ref(ContainerName),
timeout_seconds = 0
}}}.
-spec kill_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
kill_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{
target = container_ref(ContainerName),
signal = <<>>
}}}.
-spec remove_request(ContainerName :: binary()) -> message_pb:'ContainerRequest'().
remove_request(ContainerName) when is_binary(ContainerName) ->
#'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{
target = container_ref(ContainerName),
force = false,
remove_volumes = false
}}}.
-spec container_ref(ContainerName :: binary()) -> message_pb:'ContainerRef'().
container_ref(ContainerName) when is_binary(ContainerName) ->
#'ContainerRef'{name = ContainerName}.
-spec ensure_supported_deploy_config(Config :: map()) -> ok.
ensure_supported_deploy_config(Config) when is_map(Config) ->
UnsupportedKeys = [Key || Key <- [<<"ports">>], maps:is_key(Key, Config)],
case UnsupportedKeys of
[] ->
ok;
_ ->
Unsupported = iolist_to_binary(lists:join(<<", ">>, UnsupportedKeys)),
throw({error, <<"unsupported container config keys: ", Unsupported/binary>>})
end.
-spec validate_deploy_config(Config :: map()) -> ok.
validate_deploy_config(Config) when is_map(Config) ->
Required = [
{<<"image">>, binary},
{<<"container_name">>, binary},
{<<"command">>, {list, binary}},
{<<"restart">>, binary}
],
Optional = [
{<<"privileged">>, boolean},
{<<"entrypoint">>, {list, binary}},
{<<"envs">>, {list, binary}},
{<<"ports">>, {list, binary}},
{<<"expose">>, {list, binary}},
{<<"volumes">>, {list, binary}},
{<<"networks">>, {list, binary}},
{<<"labels">>, {map, {binary, binary}}},
{<<"user">>, binary},
{<<"working_dir">>, binary},
{<<"hostname">>, binary},
{<<"container_dir">>, binary},
{<<"network_mode">>, binary},
{<<"cap_add">>, {list, binary}},
{<<"cap_drop">>, {list, binary}},
{<<"devices">>, {list, binary}},
{<<"mem_limit">>, binary},
{<<"mem_reservation">>, binary},
{<<"cpu_shares">>, integer},
{<<"cpus">>, number},
{<<"ulimits">>, {map, {binary, binary}}},
{<<"sysctls">>, {map, {binary, binary}}},
{<<"tmpfs">>, {list, binary}},
{<<"extra_hosts">>, {list, binary}},
{<<"healthcheck">>, {map, {binary, any}}}
],
Errors = check_required(Config, Required) ++ check_optional(Config, Optional),
case Errors of
[] ->
ok;
_ ->
throw({error, iolist_to_binary(lists:join(<<"|||">>, Errors))})
end.
-spec check_required(map(), list()) -> [binary()].
check_required(Config, Fields) ->
lists:foldl(
fun({Key, Type}, ErrAcc) ->
case maps:get(Key, Config, undefined) of
undefined ->
[iolist_to_binary(io_lib:format("miss requied parameter: ~p", [Key])) | ErrAcc];
Value ->
case check_type(Value, Type) of
true ->
ErrAcc;
false ->
[iolist_to_binary(io_lib:format("required parameter: ~p, type must be: ~ts", [Key, type_name(Type)])) | ErrAcc]
end
end
end,
[], Fields).
-spec check_optional(map(), list()) -> [binary()].
check_optional(Config, Fields) ->
lists:foldl(
fun({Key, Type}, ErrAcc) ->
case maps:get(Key, Config, undefined) of
undefined ->
ErrAcc;
Value ->
case check_type(Value, Type) of
true ->
ErrAcc;
false ->
[iolist_to_binary(io_lib:format("optional parameter: ~p, type must be: ~ts", [Key, type_name(Type)])) | ErrAcc]
end
end
end,
[], Fields).
-spec type_name(tuple() | atom()) -> binary().
type_name(binary) ->
<<"string">>;
type_name(integer) ->
<<"integer">>;
type_name(number) ->
<<"number">>;
type_name(list) ->
<<"list">>;
type_name({list, binary}) ->
<<"list of string">>;
type_name({list, number}) ->
<<"list of number">>;
type_name({list, integer}) ->
<<"list of integer">>;
type_name(map) ->
<<"map">>;
type_name({map, {binary, binary}}) ->
<<"map of string:string">>;
type_name({map, {binary, any}}) ->
<<"map of string:any">>;
type_name(boolean) ->
<<"boolean">>.
-spec check_type(Value :: any(), any()) -> boolean().
check_type(Value, binary) ->
is_binary(Value);
check_type(Value, integer) ->
is_integer(Value);
check_type(Value, number) ->
is_number(Value);
check_type(Value, list) when is_list(Value) ->
true;
check_type(Value, {list, binary}) when is_list(Value) ->
lists:all(fun(E) -> is_binary(E) end, Value);
check_type(Value, {list, number}) when is_list(Value) ->
lists:all(fun(E) -> is_number(E) end, Value);
check_type(Value, {list, integer}) when is_list(Value) ->
lists:all(fun(E) -> is_integer(E) end, Value);
check_type(Value, map) when is_map(Value) ->
true;
check_type(Value, {map, {binary, binary}}) when is_map(Value) ->
lists:all(fun({K, V}) -> is_binary(K) andalso is_binary(V) end, maps:to_list(Value));
check_type(Value, {map, {binary, any}}) when is_map(Value) ->
lists:all(fun({K, _}) -> is_binary(K) end, maps:to_list(Value));
check_type(Value, boolean) ->
is_boolean(Value);
check_type(_, _) ->
false.
-spec build_container_deploy_params(Config :: map()) -> message_pb:'ContainerDeployParams'().
build_container_deploy_params(Config) when is_map(Config) ->
ContainerName = maps:get(<<"container_name">>, Config),
ContainerDir = maps:get(<<"container_dir">>, Config, <<>>),
Create = build_docker_create_options(Config),
#'ContainerDeployParams'{
container_name = ContainerName,
container_dir = ContainerDir,
create = Create
}.
-spec build_docker_create_options(Config :: map()) -> message_pb:'DockerCreateOptions'().
build_docker_create_options(Config) when is_map(Config) ->
#'DockerCreateOptions'{
config = build_docker_container_config(Config),
host_config = build_docker_host_config(Config),
networking_config = build_docker_networking_config(Config)
}.
-spec build_docker_container_config(Config :: map()) -> message_pb:'DockerContainerConfig'().
build_docker_container_config(Config) when is_map(Config) ->
Volumes = build_container_volumes(maps:get(<<"volumes">>, Config, [])),
ExposedPorts = build_exposed_ports(maps:get(<<"expose">>, Config, [])),
Healthcheck = build_healthcheck(maps:get(<<"healthcheck">>, Config, undefined)),
#'DockerContainerConfig'{
image = maps:get(<<"image">>, Config),
cmd = maps:get(<<"command">>, Config),
entrypoint = maps:get(<<"entrypoint">>, Config, []),
env = maps:get(<<"envs">>, Config, []),
labels = maps:to_list(maps:get(<<"labels">>, Config, #{})),
volumes = Volumes,
user = maps:get(<<"user">>, Config, <<>>),
working_dir = maps:get(<<"working_dir">>, Config, <<>>),
hostname = maps:get(<<"hostname">>, Config, <<>>),
exposed_ports = ExposedPorts,
healthcheck = Healthcheck
}.
-spec build_docker_host_config(Config :: map()) -> message_pb:'DockerHostConfig'().
build_docker_host_config(Config) when is_map(Config) ->
Binds = build_host_binds(maps:get(<<"volumes">>, Config, [])),
RestartPolicy = build_restart_policy(maps:get(<<"restart">>, Config)),
Devices = build_device_mappings(maps:get(<<"devices">>, Config, [])),
Ulimits = build_ulimits(maps:get(<<"ulimits">>, Config, #{})),
Tmpfs = build_tmpfs_options(maps:get(<<"tmpfs">>, Config, [])),
Memory = default_uint64(parse_optional_size_bytes(maps:get(<<"mem_limit">>, Config, undefined), <<"mem_limit">>)),
MemoryReservation = default_uint64(parse_optional_size_bytes(maps:get(<<"mem_reservation">>, Config, undefined), <<"mem_reservation">>)),
NanoCpus = default_uint64(parse_optional_nano_cpus(maps:get(<<"cpus">>, Config, undefined))),
CpuShares = default_uint64(maps:get(<<"cpu_shares">>, Config, undefined)),
#'DockerHostConfig'{
binds = Binds,
network_mode = maps:get(<<"network_mode">>, Config, <<>>),
restart_policy = RestartPolicy,
privileged = maps:get(<<"privileged">>, Config, false),
cap_add = maps:get(<<"cap_add">>, Config, []),
cap_drop = maps:get(<<"cap_drop">>, Config, []),
devices = Devices,
memory = Memory,
memory_reservation = MemoryReservation,
nano_cpus = NanoCpus,
cpu_shares = CpuShares,
ulimits = Ulimits,
tmpfs = Tmpfs,
sysctls = maps:to_list(maps:get(<<"sysctls">>, Config, #{})),
extra_hosts = maps:get(<<"extra_hosts">>, Config, [])
}.
-spec build_docker_networking_config(Config :: map()) -> message_pb:'DockerNetworkingConfig'().
build_docker_networking_config(Config) when is_map(Config) ->
Networks = maps:get(<<"networks">>, Config, []),
#'DockerNetworkingConfig'{
endpoints = [#'DockerNetworkEndpoint'{name = Network} || Network <- Networks]
}.
-spec build_restart_policy(binary()) -> message_pb:'RestartPolicy'().
build_restart_policy(Restart0) when is_binary(Restart0) ->
case binary:split(Restart0, <<":">>) of
[Name, RetryCountBin] ->
#'RestartPolicy'{name = Name, maximum_retry_count = parse_uint32(RetryCountBin, <<"restart">>)};
[Name] ->
#'RestartPolicy'{name = Name, maximum_retry_count = 0}
end.
-spec build_healthcheck(undefined | map()) -> undefined | message_pb:'Healthcheck'().
build_healthcheck(undefined) ->
undefined;
build_healthcheck(Healthcheck) when is_map(Healthcheck) ->
#'Healthcheck'{
test = maps:get(<<"test">>, Healthcheck, []),
interval_ns = parse_duration_ns(maps:get(<<"interval">>, Healthcheck, <<"0s">>), <<"healthcheck.interval">>),
timeout_ns = parse_duration_ns(maps:get(<<"timeout">>, Healthcheck, <<"0s">>), <<"healthcheck.timeout">>),
retries = maps:get(<<"retries">>, Healthcheck, 0)
}.
-spec default_uint64(undefined | non_neg_integer()) -> non_neg_integer().
default_uint64(undefined) ->
0;
default_uint64(Value) when is_integer(Value), Value >= 0 ->
Value.
-spec parse_optional_nano_cpus(undefined | number()) -> undefined | non_neg_integer().
parse_optional_nano_cpus(undefined) ->
undefined;
parse_optional_nano_cpus(Cpus) when is_integer(Cpus), Cpus >= 0 ->
Cpus * 1000000000;
parse_optional_nano_cpus(Cpus) when is_float(Cpus), Cpus >= 0 ->
trunc(Cpus * 1000000000).
-spec build_container_volumes([binary()]) -> [binary()].
build_container_volumes(VolumeSpecs) when is_list(VolumeSpecs) ->
[ContainerPath || VolumeSpec <- VolumeSpecs, {_HostPath, ContainerPath, _ReadOnly} <- [parse_volume_spec(VolumeSpec)]].
-spec build_host_binds([binary()]) -> [binary()].
build_host_binds(VolumeSpecs) when is_list(VolumeSpecs) ->
[volume_bind(HostPath, ContainerPath, ReadOnly) ||
VolumeSpec <- VolumeSpecs,
{HostPath, ContainerPath, ReadOnly} <- [parse_volume_spec(VolumeSpec)]].
-spec parse_volume_spec(binary()) -> {binary(), binary(), boolean()}.
parse_volume_spec(VolumeSpec) when is_binary(VolumeSpec) ->
case binary:split(VolumeSpec, <<":">>, [global]) of
[HostPath, ContainerPath] when HostPath =/= <<>>, ContainerPath =/= <<>> ->
{HostPath, ContainerPath, false};
[HostPath, ContainerPath | Modes] when HostPath =/= <<>>, ContainerPath =/= <<>> ->
{HostPath, ContainerPath, lists:member(<<"ro">>, Modes)};
_ ->
throw({error, <<"invalid volume binding">>})
end.
-spec volume_bind(binary(), binary(), boolean()) -> binary().
volume_bind(HostPath, ContainerPath, true) when is_binary(HostPath), is_binary(ContainerPath) ->
<<HostPath/binary, ":", ContainerPath/binary, ":ro">>;
volume_bind(HostPath, ContainerPath, false) when is_binary(HostPath), is_binary(ContainerPath) ->
<<HostPath/binary, ":", ContainerPath/binary>>.
-spec build_exposed_ports([binary()]) -> [message_pb:'DockerExposedPort'()].
build_exposed_ports(ExposeSpecs) when is_list(ExposeSpecs) ->
[build_exposed_port(ExposeSpec) || ExposeSpec <- ExposeSpecs].
-spec build_exposed_port(binary()) -> message_pb:'DockerExposedPort'().
build_exposed_port(ExposeSpec) when is_binary(ExposeSpec) ->
case binary:split(ExposeSpec, <<"/">>) of
[PortBin] ->
#'DockerExposedPort'{container_port = parse_uint32(PortBin, <<"expose">>), protocol = <<"tcp">>};
[PortBin, Protocol] ->
#'DockerExposedPort'{container_port = parse_uint32(PortBin, <<"expose">>), protocol = Protocol}
end.
-spec build_device_mappings([binary()]) -> [message_pb:'DeviceMapping'()].
build_device_mappings(DeviceSpecs) when is_list(DeviceSpecs) ->
[build_device_mapping(DeviceSpec) || DeviceSpec <- DeviceSpecs].
-spec build_device_mapping(binary()) -> message_pb:'DeviceMapping'().
build_device_mapping(DeviceSpec) when is_binary(DeviceSpec) ->
case binary:split(DeviceSpec, <<":">>, [global]) of
[HostPath, ContainerPath] when HostPath =/= <<>>, ContainerPath =/= <<>> ->
#'DeviceMapping'{path_on_host = HostPath, path_in_container = ContainerPath, cgroup_permissions = <<"rwm">>};
[HostPath, ContainerPath, Permissions] when HostPath =/= <<>>, ContainerPath =/= <<>>, Permissions =/= <<>> ->
#'DeviceMapping'{path_on_host = HostPath, path_in_container = ContainerPath, cgroup_permissions = Permissions};
_ ->
throw({error, <<"invalid device mapping">>})
end.
-spec build_ulimits(map()) -> [message_pb:'Ulimit'()].
build_ulimits(Ulimits) when is_map(Ulimits) ->
[build_ulimit(Name, Value) || {Name, Value} <- maps:to_list(Ulimits)].
-spec build_ulimit(binary(), binary()) -> message_pb:'Ulimit'().
build_ulimit(Name, Value) when is_binary(Name), is_binary(Value) ->
case binary:split(Value, <<":">>) of
[SoftBin, HardBin] ->
#'Ulimit'{name = Name, soft = parse_uint64(SoftBin, <<"ulimits.soft">>), hard = parse_uint64(HardBin, <<"ulimits.hard">>)};
[LimitBin] ->
Limit = parse_uint64(LimitBin, <<"ulimits.limit">>),
#'Ulimit'{name = Name, soft = Limit, hard = Limit}
end.
-spec build_tmpfs_options([binary()]) -> [{binary(), binary()}].
build_tmpfs_options(TmpfsSpecs) when is_list(TmpfsSpecs) ->
[build_tmpfs_option(TmpfsSpec) || TmpfsSpec <- TmpfsSpecs].
-spec build_tmpfs_option(binary()) -> {binary(), binary()}.
build_tmpfs_option(TmpfsSpec) when is_binary(TmpfsSpec) ->
case binary:split(TmpfsSpec, <<":">>) of
[Path] when Path =/= <<>> ->
{Path, <<>>};
[Path, Options] when Path =/= <<>> ->
{Path, Options};
_ ->
throw({error, <<"invalid tmpfs mount">>})
end.
-spec parse_optional_size_bytes(undefined | binary(), binary()) -> undefined | non_neg_integer().
parse_optional_size_bytes(undefined, _Field) ->
undefined;
parse_optional_size_bytes(Value, Field) when is_binary(Value) ->
parse_size_bytes(Value, Field).
-spec parse_duration_ns(binary() | integer(), binary()) -> non_neg_integer().
parse_duration_ns(Value, _Field) when is_integer(Value), Value >= 0 ->
Value;
parse_duration_ns(Value, Field) when is_binary(Value) ->
parse_scaled_uint64(Value, Field, #{
<<"ns">> => 1,
<<"us">> => 1000,
<<"ms">> => 1000000,
<<"s">> => 1000000000,
<<"m">> => 60000000000,
<<"h">> => 3600000000000,
<<>> => 1000000000
}).
-spec parse_size_bytes(binary(), binary()) -> non_neg_integer().
parse_size_bytes(Value, Field) when is_binary(Value) ->
parse_scaled_uint64(Value, Field, #{
<<"b">> => 1,
<<"k">> => 1024,
<<"kb">> => 1024,
<<"ki">> => 1024,
<<"kib">> => 1024,
<<"m">> => 1048576,
<<"mb">> => 1048576,
<<"mi">> => 1048576,
<<"mib">> => 1048576,
<<"g">> => 1073741824,
<<"gb">> => 1073741824,
<<"gi">> => 1073741824,
<<"gib">> => 1073741824,
<<"t">> => 1099511627776,
<<"tb">> => 1099511627776,
<<"ti">> => 1099511627776,
<<"tib">> => 1099511627776,
<<>> => 1
}).
-spec parse_scaled_uint64(binary(), binary(), map()) -> non_neg_integer().
parse_scaled_uint64(Value0, Field, Multipliers) when is_binary(Value0), is_binary(Field), is_map(Multipliers) ->
Value = trim_binary(Value0),
LowerValue = lower_binary(Value),
{NumberBin, Unit} = split_numeric_suffix(LowerValue),
case maps:get(Unit, Multipliers, undefined) of
undefined ->
throw({error, <<"invalid value for ", Field/binary, ": ", Value0/binary>>});
Multiplier ->
trunc(parse_decimal(NumberBin, Field) * Multiplier)
end.
-spec parse_uint32(binary(), binary()) -> non_neg_integer().
parse_uint32(Value, Field) when is_binary(Value), is_binary(Field) ->
Parsed = parse_uint64(Value, Field),
case Parsed =< 16#FFFFFFFF of
true ->
Parsed;
false ->
throw({error, <<"value overflow for ", Field/binary>>})
end.
-spec parse_uint64(binary(), binary()) -> non_neg_integer().
parse_uint64(Value0, Field) when is_binary(Value0), is_binary(Field) ->
Value = trim_binary(Value0),
case catch binary_to_integer(Value) of
Parsed when is_integer(Parsed), Parsed >= 0 ->
Parsed;
_ ->
throw({error, <<"invalid unsigned integer for ", Field/binary, ": ", Value0/binary>>})
end.
-spec parse_decimal(binary(), binary()) -> float().
parse_decimal(Value, Field) when is_binary(Value), is_binary(Field) ->
case catch binary_to_integer(Value) of
ParsedInt when is_integer(ParsedInt), ParsedInt >= 0 ->
float(ParsedInt);
_ ->
case catch binary_to_float(Value) of
ParsedFloat when is_float(ParsedFloat), ParsedFloat >= 0 ->
ParsedFloat;
_ ->
throw({error, <<"invalid number for ", Field/binary, ": ", Value/binary>>})
end
end.
-spec split_numeric_suffix(binary()) -> {binary(), binary()}.
split_numeric_suffix(Value) when is_binary(Value) ->
split_numeric_suffix(Value, <<>>).
-spec split_numeric_suffix(binary(), binary()) -> {binary(), binary()}.
split_numeric_suffix(<<Char, Rest/binary>>, Acc)
when (Char >= $0 andalso Char =< $9) orelse Char =:= $. ->
split_numeric_suffix(Rest, <<Acc/binary, Char>>);
split_numeric_suffix(Rest, <<>>) ->
throw({error, <<"invalid numeric value: ", Rest/binary>>});
split_numeric_suffix(Rest, Acc) ->
{Acc, Rest}.
-spec trim_binary(binary()) -> binary().
trim_binary(Value) when is_binary(Value) ->
Trimmed = string:trim(binary_to_list(Value)),
list_to_binary(Trimmed).
-spec lower_binary(binary()) -> binary().
lower_binary(Value) when is_binary(Value) ->
list_to_binary(string:lowercase(binary_to_list(Value))).