anlicheng/iot_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
anlicheng 2026-04-19 14:47:50 +08:00
parent 099af2dfb6
commit 881284c79c
6 changed files with 4360 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
include/message_pb.hrl
View File

@ -12,7 +12,7 @@
-define('REQUESTFRAME_PB_H', true).
-record('RequestFrame',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
body :: {auth_request, message_pb:'AuthRequest'()} | {rpc_request, message_pb:'RpcRequest'()} | undefined % oneof
body :: {auth_request, message_pb:'AuthRequest'()} | {rpc_request, message_pb:'RpcRequest'()} | {container_request, message_pb:'ContainerRequest'()} | undefined % oneof
}).
-endif.
@ -31,6 +31,185 @@
}).
-endif.
-ifndef('CONTAINERREF_PB_H').
-define('CONTAINERREF_PB_H', true).
-record('ContainerRef',
{id = <<>> :: unicode:chardata() | undefined, % = 1, optional
name = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.LIST_PB_H').
-define('CONTAINERREQUEST.LIST_PB_H', true).
-record('ContainerRequest.List',
{all = false :: boolean() | 0 | 1 | undefined % = 1, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.DEPLOY_PB_H').
-define('CONTAINERREQUEST.DEPLOY_PB_H', true).
-record('ContainerRequest.Deploy',
{task_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
params = undefined :: message_pb:'ContainerDeployParams'() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.START_PB_H').
-define('CONTAINERREQUEST.START_PB_H', true).
-record('ContainerRequest.Start',
{target = undefined :: message_pb:'ContainerRef'() | undefined % = 1, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.STOP_PB_H').
-define('CONTAINERREQUEST.STOP_PB_H', true).
-record('ContainerRequest.Stop',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
timeout_seconds = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
}).
-endif.
-ifndef('CONTAINERREQUEST.KILL_PB_H').
-define('CONTAINERREQUEST.KILL_PB_H', true).
-record('ContainerRequest.Kill',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
signal = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.REMOVE_PB_H').
-define('CONTAINERREQUEST.REMOVE_PB_H', true).
-record('ContainerRequest.Remove',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
force = false :: boolean() | 0 | 1 | undefined, % = 2, optional
remove_volumes = false :: boolean() | 0 | 1 | undefined % = 3, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.CONFIG_PB_H').
-define('CONTAINERREQUEST.CONFIG_PB_H', true).
-record('ContainerRequest.Config',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
config = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST_PB_H').
-define('CONTAINERREQUEST_PB_H', true).
-record('ContainerRequest',
{action :: {list, message_pb:'ContainerRequest.List'()} | {deploy, message_pb:'ContainerRequest.Deploy'()} | {start, message_pb:'ContainerRequest.Start'()} | {stop, message_pb:'ContainerRequest.Stop'()} | {kill, message_pb:'ContainerRequest.Kill'()} | {remove, message_pb:'ContainerRequest.Remove'()} | {config, message_pb:'ContainerRequest.Config'()} | undefined % oneof
}).
-endif.
-ifndef('CONTAINERDEPLOYPARAMS_PB_H').
-define('CONTAINERDEPLOYPARAMS_PB_H', true).
-record('ContainerDeployParams',
{container_name = <<>> :: unicode:chardata() | undefined, % = 1, optional
container_dir = <<>> :: unicode:chardata() | undefined, % = 2, optional
spec = undefined :: message_pb:'ContainerSpec'() | undefined % = 3, optional
}).
-endif.
-ifndef('CONTAINERSPEC_PB_H').
-define('CONTAINERSPEC_PB_H', true).
-record('ContainerSpec',
{image = <<>> :: unicode:chardata() | undefined, % = 1, optional
command = [] :: [unicode:chardata()] | undefined, % = 2, repeated
entrypoint = [] :: [unicode:chardata()] | undefined, % = 3, repeated
env = [] :: [unicode:chardata()] | undefined, % = 4, repeated
labels = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 5
volumes = [] :: [message_pb:'VolumeBind'()] | undefined, % = 6, repeated
user = <<>> :: unicode:chardata() | undefined, % = 7, optional
working_dir = <<>> :: unicode:chardata() | undefined, % = 8, optional
hostname = <<>> :: unicode:chardata() | undefined, % = 9, optional
expose = [] :: [message_pb:'PortExpose'()] | undefined, % = 10, repeated
networks = [] :: [unicode:chardata()] | undefined, % = 11, repeated
network_mode = <<>> :: unicode:chardata() | undefined, % = 12, optional
healthcheck = undefined :: message_pb:'Healthcheck'() | undefined, % = 13, optional
restart = undefined :: message_pb:'RestartPolicy'() | undefined, % = 14, optional
privileged = false :: boolean() | 0 | 1 | undefined, % = 15, optional
cap_add = [] :: [unicode:chardata()] | undefined, % = 16, repeated
cap_drop = [] :: [unicode:chardata()] | undefined, % = 17, repeated
devices = [] :: [message_pb:'DeviceMapping'()] | undefined, % = 18, repeated
resources = undefined :: message_pb:'ResourceLimits'() | undefined, % = 19, optional
ulimits = [] :: [message_pb:'Ulimit'()] | undefined, % = 20, repeated
tmpfs = [] :: [message_pb:'TmpfsMount'()] | undefined, % = 21, repeated
sysctls = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 22
extra_hosts = [] :: [unicode:chardata()] | undefined % = 23, repeated
}).
-endif.
-ifndef('VOLUMEBIND_PB_H').
-define('VOLUMEBIND_PB_H', true).
-record('VolumeBind',
{host_path = <<>> :: unicode:chardata() | undefined, % = 1, optional
container_path = <<>> :: unicode:chardata() | undefined, % = 2, optional
read_only = false :: boolean() | 0 | 1 | undefined % = 3, optional
}).
-endif.
-ifndef('PORTEXPOSE_PB_H').
-define('PORTEXPOSE_PB_H', true).
-record('PortExpose',
{container_port = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
protocol = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('HEALTHCHECK_PB_H').
-define('HEALTHCHECK_PB_H', true).
-record('Healthcheck',
{test = [] :: [unicode:chardata()] | undefined, % = 1, repeated
interval_ns = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
timeout_ns = 0 :: non_neg_integer() | undefined, % = 3, optional, 64 bits
retries = 0 :: non_neg_integer() | undefined % = 4, optional, 32 bits
}).
-endif.
-ifndef('RESTARTPOLICY_PB_H').
-define('RESTARTPOLICY_PB_H', true).
-record('RestartPolicy',
{name = <<>> :: unicode:chardata() | undefined, % = 1, optional
maximum_retry_count = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
}).
-endif.
-ifndef('DEVICEMAPPING_PB_H').
-define('DEVICEMAPPING_PB_H', true).
-record('DeviceMapping',
{host_path = <<>> :: unicode:chardata() | undefined, % = 1, optional
container_path = <<>> :: unicode:chardata() | undefined, % = 2, optional
cgroup_permissions = <<>> :: unicode:chardata() | undefined % = 3, optional
}).
-endif.
-ifndef('RESOURCELIMITS_PB_H').
-define('RESOURCELIMITS_PB_H', true).
-record('ResourceLimits',
{memory_bytes = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
memory_reservation_bytes = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
nano_cpus = 0 :: non_neg_integer() | undefined, % = 3, optional, 64 bits
cpu_shares = 0 :: non_neg_integer() | undefined % = 4, optional, 64 bits
}).
-endif.
-ifndef('ULIMIT_PB_H').
-define('ULIMIT_PB_H', true).
-record('Ulimit',
{name = <<>> :: unicode:chardata() | undefined, % = 1, optional
soft = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
hard = 0 :: non_neg_integer() | undefined % = 3, optional, 64 bits
}).
-endif.
-ifndef('TMPFSMOUNT_PB_H').
-define('TMPFSMOUNT_PB_H', true).
-record('TmpfsMount',
{path = <<>> :: unicode:chardata() | undefined, % = 1, optional
options = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('AUTHREQUEST_PB_H').
-define('AUTHREQUEST_PB_H', true).
-record('AuthRequest',

10
proto/message.proto
View File

@ -6,6 +6,7 @@ message RequestFrame {
oneof body {
AuthRequest auth_request = 2;
RpcRequest rpc_request = 3;
ContainerRequest container_request = 4;
}
}
@ -39,7 +40,8 @@ message ContainerRequest {
}
message Deploy {
ContainerDeployParams params = 1;
uint32 task_id = 1;
ContainerDeployParams params = 2;
}
message Start {
@ -62,6 +64,11 @@ message ContainerRequest {
bool remove_volumes = 3;
}
message Config {
ContainerRef target = 1;
bytes config = 2;
}
oneof action {
List list = 10;
Deploy deploy = 11;
@ -69,6 +76,7 @@ message ContainerRequest {
Stop stop = 13;
Kill kill = 14;
Remove remove = 15;
Config config = 16;
}
}

368
src/host/iot_host.erl
View File

@ -95,39 +95,49 @@ attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
-spec get_containers(Pid :: pid()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
get_containers(Pid) when is_pid(Pid) ->
gen_statem:call(Pid, {jsonrpc_call, self(), {<<"get_containers">>, <<"{}">>}}).
container_call(Pid, #'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = true}}}).
-spec config_container(Pid :: pid(), ContainerName :: binary(), ConfigJson :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
config_container(Pid, ContainerName, ConfigJson) when is_pid(Pid), is_binary(ContainerName), is_binary(ConfigJson) ->
Params = jiffy:encode(#{
<<"container_name">> => ContainerName,
<<"config">> => ConfigJson
}, [force_utf8]),
gen_statem:call(Pid, {jsonrpc_call, self(), {<<"config_container">>, Params}}).
container_call(Pid, #'ContainerRequest'{action = {config, #'ContainerRequest.Config'{
target = container_ref(ContainerName),
config = ConfigJson
}}}).
-spec deploy_container(Pid :: pid(), TaskId :: integer(), Params :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
deploy_container(Pid, TaskId, Params) when is_pid(Pid), is_integer(TaskId), is_binary(Params) ->
gen_statem:call(Pid, {jsonrpc_call, self(), {<<"deploy">>, Params}}).
-spec deploy_container(Pid :: pid(), TaskId :: integer(), Config :: map()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
deploy_container(Pid, TaskId, Config) when is_pid(Pid), is_integer(TaskId), is_map(Config) ->
case build_deploy_request(TaskId, Config) of
{ok, Request} ->
container_call(Pid, Request);
{error, Reason} ->
{error, Reason}
end.
-spec start_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
start_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
Params = jiffy:encode(#{<<"container_name">> => ContainerName}, [force_utf8]),
gen_statem:call(Pid, {jsonrpc_call, self(), {<<"start_container">>, Params}}).
container_call(Pid, #'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = container_ref(ContainerName)}}}).
-spec stop_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
stop_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
Params = jiffy:encode(#{<<"container_name">> => ContainerName}, [force_utf8]),
gen_statem:call(Pid, {jsonrpc_call, self(), {<<"stop_container">>, Params}}).
container_call(Pid, #'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{
target = container_ref(ContainerName),
timeout_seconds = 0
}}}).
-spec kill_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
kill_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
Params = jiffy:encode(#{<<"container_name">> => ContainerName}, [force_utf8]),
gen_statem:call(Pid, {jsonrpc_call, self(), {<<"kill_container">>, Params}}).
container_call(Pid, #'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{
target = container_ref(ContainerName),
signal = <<>>
}}}).
-spec remove_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
Params = jiffy:encode(#{<<"container_name">> => ContainerName}, [force_utf8]),
gen_statem:call(Pid, {jsonrpc_call, self(), {<<"remove_container">>, Params}}).
container_call(Pid, #'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{
target = container_ref(ContainerName),
force = false,
remove_volumes = false
}}}).
-spec await_reply(Pid :: pid(), Ref :: reference(), Timeout :: integer()) ->
{ok, Result :: binary()} | {error, Code :: integer(), Reason :: binary()}.
@ -229,6 +239,16 @@ handle_event({call, From}, {jsonrpc_call, ReceiverPid, RpcCall}, _, State = #sta
{keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
end;
handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) ->
case HasSession andalso is_pid(ChannelPid) of
true ->
Ref = tcp_channel:container_call(ChannelPid, ReceiverPid, Request),
{keep_state, State, [{reply, From, {ok, Ref}}]};
false ->
logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]),
{keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
end;
handle_event({call, From}, {cancel_jsonrpc_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
case is_pid(ChannelPid) of
true ->
@ -381,6 +401,320 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
%%% Internal functions
%%%===================================================================
-spec container_call(Pid :: pid(), Request :: message_pb:'ContainerRequest'()) ->
{ok, Ref :: reference()} | {error, Reason :: any()}.
container_call(Pid, Request) when is_pid(Pid), is_record(Request, 'ContainerRequest') ->
gen_statem:call(Pid, {container_call, self(), Request}).
-spec container_ref(ContainerName :: binary()) -> message_pb:'ContainerRef'().
container_ref(ContainerName) when is_binary(ContainerName) ->
#'ContainerRef'{name = ContainerName}.
-spec build_deploy_request(TaskId :: integer(), Config :: map()) ->
{ok, message_pb:'ContainerRequest'()} | {error, binary()}.
build_deploy_request(TaskId, Config) when is_integer(TaskId), is_map(Config), TaskId >= 0 ->
try
ensure_supported_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 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 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, <<>>),
Spec = build_container_spec(Config),
#'ContainerDeployParams'{
container_name = ContainerName,
container_dir = ContainerDir,
spec = Spec
}.
-spec build_container_spec(Config :: map()) -> message_pb:'ContainerSpec'().
build_container_spec(Config) when is_map(Config) ->
Volumes = build_volume_binds(maps:get(<<"volumes">>, Config, [])),
Expose = build_port_exposes(maps:get(<<"expose">>, Config, [])),
Restart = build_restart_policy(maps:get(<<"restart">>, Config)),
Devices = build_device_mappings(maps:get(<<"devices">>, Config, [])),
Resources = build_resource_limits(Config),
Ulimits = build_ulimits(maps:get(<<"ulimits">>, Config, #{})),
Tmpfs = build_tmpfs_mounts(maps:get(<<"tmpfs">>, Config, [])),
Healthcheck = build_healthcheck(maps:get(<<"healthcheck">>, Config, undefined)),
#'ContainerSpec'{
image = maps:get(<<"image">>, Config),
command = 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, <<>>),
expose = Expose,
networks = maps:get(<<"networks">>, Config, []),
network_mode = maps:get(<<"network_mode">>, Config, <<>>),
healthcheck = Healthcheck,
restart = Restart,
privileged = maps:get(<<"privileged">>, Config, false),
cap_add = maps:get(<<"cap_add">>, Config, []),
cap_drop = maps:get(<<"cap_drop">>, Config, []),
devices = Devices,
resources = Resources,
ulimits = Ulimits,
tmpfs = Tmpfs,
sysctls = maps:to_list(maps:get(<<"sysctls">>, Config, #{})),
extra_hosts = maps:get(<<"extra_hosts">>, Config, [])
}.
-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 build_resource_limits(map()) -> undefined | message_pb:'ResourceLimits'().
build_resource_limits(Config) when is_map(Config) ->
MemoryBytes = parse_optional_size_bytes(maps:get(<<"mem_limit">>, Config, undefined), <<"mem_limit">>),
ReservationBytes = parse_optional_size_bytes(maps:get(<<"mem_reservation">>, Config, undefined), <<"mem_reservation">>),
NanoCpus = parse_optional_nano_cpus(maps:get(<<"cpus">>, Config, undefined)),
CpuShares = maps:get(<<"cpu_shares">>, Config, undefined),
case {MemoryBytes, ReservationBytes, NanoCpus, CpuShares} of
{undefined, undefined, undefined, undefined} ->
undefined;
_ ->
#'ResourceLimits'{
memory_bytes = default_uint64(MemoryBytes),
memory_reservation_bytes = default_uint64(ReservationBytes),
nano_cpus = default_uint64(NanoCpus),
cpu_shares = default_uint64(CpuShares)
}
end.
-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_volume_binds([binary()]) -> [message_pb:'VolumeBind'()].
build_volume_binds(VolumeSpecs) when is_list(VolumeSpecs) ->
[build_volume_bind(VolumeSpec) || VolumeSpec <- VolumeSpecs].
-spec build_volume_bind(binary()) -> message_pb:'VolumeBind'().
build_volume_bind(VolumeSpec) when is_binary(VolumeSpec) ->
case binary:split(VolumeSpec, <<":">>, [global]) of
[HostPath, ContainerPath] when HostPath =/= <<>>, ContainerPath =/= <<>> ->
#'VolumeBind'{host_path = HostPath, container_path = ContainerPath, read_only = false};
[HostPath, ContainerPath | Modes] when HostPath =/= <<>>, ContainerPath =/= <<>> ->
#'VolumeBind'{host_path = HostPath, container_path = ContainerPath, read_only = lists:member(<<"ro">>, Modes)};
_ ->
throw({error, <<"invalid volume binding">>})
end.
-spec build_port_exposes([binary()]) -> [message_pb:'PortExpose'()].
build_port_exposes(ExposeSpecs) when is_list(ExposeSpecs) ->
[build_port_expose(ExposeSpec) || ExposeSpec <- ExposeSpecs].
-spec build_port_expose(binary()) -> message_pb:'PortExpose'().
build_port_expose(ExposeSpec) when is_binary(ExposeSpec) ->
case binary:split(ExposeSpec, <<"/">>) of
[PortBin] ->
#'PortExpose'{container_port = parse_uint32(PortBin, <<"expose">>), protocol = <<"tcp">>};
[PortBin, Protocol] ->
#'PortExpose'{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'{host_path = HostPath, container_path = ContainerPath, cgroup_permissions = <<"rwm">>};
[HostPath, ContainerPath, Permissions] when HostPath =/= <<>>, ContainerPath =/= <<>>, Permissions =/= <<>> ->
#'DeviceMapping'{host_path = HostPath, container_path = 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_mounts([binary()]) -> [message_pb:'TmpfsMount'()].
build_tmpfs_mounts(TmpfsSpecs) when is_list(TmpfsSpecs) ->
[build_tmpfs_mount(TmpfsSpec) || TmpfsSpec <- TmpfsSpecs].
-spec build_tmpfs_mount(binary()) -> message_pb:'TmpfsMount'().
build_tmpfs_mount(TmpfsSpec) when is_binary(TmpfsSpec) ->
case binary:split(TmpfsSpec, <<":">>) of
[Path] when Path =/= <<>> ->
#'TmpfsMount'{path = Path, options = <<>>};
[Path, Options] when Path =/= <<>> ->
#'TmpfsMount'{path = Path, options = 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))).
-spec get_route_key(binary()) -> binary().
get_route_key(<<"">>) ->
<<"/">>;

3806
src/protobuf/message_pb.erl
View File

File diff suppressed because it is too large Load Diff

9
src/transport/http/container_handler.erl
View File

@ -69,11 +69,7 @@ handle_request("POST", "/container/deploy", _, #{<<"uuid">> := UUID, <<"task_id"
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
Pid when is_pid(Pid) ->
Params = jiffy:encode(#{
<<"task_id">> => TaskId,
<<"config">> => Config
}, [force_utf8]),
case iot_host:deploy_container(Pid, TaskId, Params) of
case iot_host:deploy_container(Pid, TaskId, Config) of
{ok, Ref} ->
case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of
{ok, Result} ->
@ -185,6 +181,7 @@ validate_config(Config) when is_map(Config) ->
%% 可选参数（附带默认值）
Optional = [
{<<"privileged">>, boolean},
{<<"entrypoint">>, {list, binary}},
{<<"envs">>, {list, binary}},
{<<"ports">>, {list, binary}},
{<<"expose">>, {list, binary}},
@ -194,6 +191,8 @@ validate_config(Config) when is_map(Config) ->
{<<"user">>, binary},
{<<"working_dir">>, binary},
{<<"hostname">>, binary},
{<<"container_dir">>, binary},
{<<"network_mode">>, binary},
{<<"cap_add">>, {list, binary}},
{<<"cap_drop">>, {list, binary}},
{<<"devices">>, {list, binary}},

28
src/transport/tcp/tcp_channel.erl
View File

@ -16,7 +16,7 @@
-define(INFLIGHT_TIMEOUT, 60000).
%% API
-export([pub/4, jsonrpc_call/3, cancel_jsonrpc_call/2, command/3]).
-export([pub/4, jsonrpc_call/3, container_call/3, cancel_jsonrpc_call/2, command/3]).
-export([start_link/3, stop/2]).
%% gen_server callbacks
@ -54,9 +54,16 @@ command(Pid, CommandType, Command) when is_pid(Pid), is_integer(CommandType), is
%% 向通道中写入消息
-spec jsonrpc_call(Pid :: pid(), ReceiverPid :: pid(), Request :: {Method :: binary(), Params :: binary()}) -> Ref :: reference().
jsonrpc_call(Pid, ReceiverPid, Request = {Method, Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method), is_binary(Params) ->
jsonrpc_call(Pid, ReceiverPid, {Method, Params}) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Method), is_binary(Params) ->
Ref = make_ref(),
gen_server:cast(Pid, {jsonrpc_call, ReceiverPid, Ref, Request}),
RpcRequest = #'RpcRequest'{method = Method, params = Params},
gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {rpc_request, RpcRequest}}),
Ref.
-spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: message_pb:'ContainerRequest'()) -> Ref :: reference().
container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_record(Request, 'ContainerRequest') ->
Ref = make_ref(),
gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {container_request, Request}}),
Ref.
-spec cancel_jsonrpc_call(Pid :: pid(), Ref :: reference()) -> ok.
@ -112,15 +119,15 @@ handle_cast({command, CommandType, Command}, State = #state{transport = Transpor
Transport:send(Socket, <<?FRAME_CAST, Encoded/binary>>),
{noreply, State};
%% 推送消息
handle_cast({jsonrpc_call, ReceiverPid, Ref, {Method, Params}}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
%% 推送需要响应的请求
handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
case next_packet_id(PacketId, Inflight) of
{ok, NPacketId, NextPacketId} ->
Encoded = message_pb:encode_msg(#'RequestFrame'{
packet_id = NPacketId,
body = {rpc_request, #'RpcRequest'{method = Method, params = Params}}
body = Body
}),
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {jsonrpc_timeout, NPacketId}),
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, NPacketId}),
Transport:send(Socket, <<?FRAME_REQUEST, Encoded/binary>>),
RequestInfo = #inflight_request{receiver_pid = ReceiverPid, ref = Ref, timer_ref = TimerRef},
@ -144,10 +151,10 @@ handle_info({tcp, Socket, <<?FRAME_RESPONSE, FrameBin/binary>>}, State = #state{
ResponseFrame = message_pb:decode_msg(FrameBin, 'ResponseFrame'),
handle_response_frame(ResponseFrame, Inflight, State);
handle_info({timeout, TimerRef, {jsonrpc_timeout, PacketId}}, State = #state{inflight = Inflight}) ->
handle_info({timeout, TimerRef, {request_timeout, PacketId}}, State = #state{inflight = Inflight}) ->
case maps:get(PacketId, Inflight, undefined) of
#inflight_request{ref = Ref, timer_ref = TimerRef} ->
logger:warning("[ws_channel] jsonrpc request timeout, packet_id: ~p, ref: ~p", [PacketId, Ref]),
logger:warning("[ws_channel] request timeout, packet_id: ~p, ref: ~p", [PacketId, Ref]),
{noreply, State#state{inflight = maps:remove(PacketId, Inflight)}};
_ ->
{noreply, State}
@ -270,6 +277,9 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId,
handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {rpc_request, RpcRequest}}, _Transport, _Socket, State) ->
logger:warning("[ws_channel] unsupported request message type: rpc_request, packet_id: ~p, request: ~p", [PacketId, RpcRequest]),
{stop, State};
handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, ContainerRequest}}, _Transport, _Socket, State) ->
logger:warning("[ws_channel] unsupported request message type: container_request, packet_id: ~p, request: ~p", [PacketId, ContainerRequest]),
{stop, State};
handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _Transport, _Socket, State) ->
logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]),
{stop, State}.