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

Compare commits

base: anlicheng:0dd6f9715a3b5f46ef4acb65c246076b63c4d3ce
Branches Tags
anlicheng:docker_mst
anlicheng:main
anlicheng:v1.0
..
compare: anlicheng:7e49d2df0cdfa854cb67cf53887072fa7f2854bd
Branches Tags
anlicheng:main
anlicheng:docker_mst
anlicheng:v1.0

No commits in common. "0dd6f9715a3b5f46ef4acb65c246076b63c4d3ce" and "7e49d2df0cdfa854cb67cf53887072fa7f2854bd" have entirely different histories.
0dd6f9715a ... 7e49d2df0c

4 changed files with 178 additions and 215 deletions
Show Stats Expand all files Collapse all files

148
docs/container_deploy_request.md
View File

@ -195,14 +195,14 @@ miss requied parameter: <<"image">>
```erlang ```erlang
#{ #{
<<"action">> => <<"deploy">>, action => deploy,
<<"task_id">> => TaskId, task_id => TaskId,
<<"params">> => #{ params => #{
<<"container_name">> => ContainerName, container_name => ContainerName,
<<"create">> => #{ create => #{
<<"config">> => ContainerConfig, config => ContainerConfig,
<<"host_config">> => HostConfig, host_config => HostConfig,
<<"networking_config">> => NetworkingConfig networking_config => NetworkingConfig
} }
} }
} }
@ -211,21 +211,19 @@ miss requied parameter: <<"image">>
该 map 会通过 efka/iot 长连接协议下发： 该 map 会通过 efka/iot 长连接协议下发：
```erlang ```erlang
{<<"command">>, Ref, {<<"container">>, #{ {command, Ref, {container, #{
<<"action">> => <<"deploy">>, action => deploy,
<<"task_id">> => TaskId, task_id => TaskId,
<<"params">> => Params params => Params
}}} }}}
``` ```
efka 返回： efka 返回：
```erlang ```erlang
{<<"command_response">>, Ref, {<<"container">>, Reply}} {command_response, Ref, {container, Reply}}
``` ```
其中 `Ref` 是 `crypto:strong_rand_bytes(16)` 生成的 16 字节 binary。网络帧只使用 safe term，协议 label、command map key 和 action 使用 binary；`efka` 收到后直接按 binary key/action 处理。
HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行错误通常返回 HTTP 400；找不到 host 返回业务错误 code 404。 HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行错误通常返回 HTTP 400；找不到 host 返回业务错误 code 404。
## 7. Docker create config 映射 ## 7. Docker create config 映射
@ -273,7 +271,7 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
`create.networking_config` 由 `build_docker_networking_config/1` 生成： `create.networking_config` 由 `build_docker_networking_config/1` 生成：
```erlang ```erlang
#{<<"endpoints">> => [#{<<"name">> => Network} || Network <- Networks]} #{endpoints => [#{name => Network} || Network <- Networks]}
``` ```
来源字段： 来源字段：
@ -285,9 +283,9 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
转换结果： 转换结果：
```erlang ```erlang
#{<<"endpoints">> => [ #{endpoints => [
#{<<"name">> => <<"bridge">>}, #{name => <<"bridge">>},
#{<<"name">> => <<"mynet">>} #{name => <<"mynet">>}
]} ]}
``` ```
@ -304,7 +302,7 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
转换： 转换：
```erlang ```erlang
#{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0} #{name => <<"always">>, maximum_retry_count => 0}
``` ```
输入： 输入：
@ -316,7 +314,7 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
转换： 转换：
```erlang ```erlang
#{<<"name">> => <<"on-failure">>, <<"maximum_retry_count">> => 3} #{name => <<"on-failure">>, maximum_retry_count => 3}
``` ```
### expose ### expose
@ -331,9 +329,9 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
```erlang ```erlang
[ [
#{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>}, #{container_port => 80, protocol => <<"tcp">>},
#{<<"container_port">> => 443, <<"protocol">> => <<"tcp">>}, #{container_port => 443, protocol => <<"tcp">>},
#{<<"container_port">> => 53, <<"protocol">> => <<"udp">>} #{container_port => 53, protocol => <<"udp">>}
] ]
``` ```
@ -383,14 +381,14 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
```erlang ```erlang
[ [
#{ #{
<<"path_on_host">> => <<"/dev/ttyUSB0">>, path_on_host => <<"/dev/ttyUSB0">>,
<<"path_in_container">> => <<"/dev/ttyUSB0">>, path_in_container => <<"/dev/ttyUSB0">>,
<<"cgroup_permissions">> => <<"rwm">> cgroup_permissions => <<"rwm">>
}, },
#{ #{
<<"path_on_host">> => <<"/dev/snd">>, path_on_host => <<"/dev/snd">>,
<<"path_in_container">> => <<"/dev/snd">>, path_in_container => <<"/dev/snd">>,
<<"cgroup_permissions">> => <<"rwm">> cgroup_permissions => <<"rwm">>
} }
] ]
``` ```
@ -416,8 +414,8 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
```erlang ```erlang
[ [
#{<<"name">> => <<"nofile">>, <<"soft">> => 1024, <<"hard">> => 2048}, #{name => <<"nofile">>, soft => 1024, hard => 2048},
#{<<"name">> => <<"nproc">>, <<"soft">> => 4096, <<"hard">> => 4096} #{name => <<"nproc">>, soft => 4096, hard => 4096}
] ]
``` ```
@ -467,10 +465,10 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
```erlang ```erlang
#{ #{
<<"test">> => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>], test => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
<<"interval_ns">> => 30000000000, interval_ns => 30000000000,
<<"timeout_ns">> => 10000000000, timeout_ns => 10000000000,
<<"retries">> => 3 retries => 3
} }
``` ```
@ -514,8 +512,8 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
转换： 转换：
```erlang ```erlang
<<"memory">> => 536870912, memory => 536870912,
<<"memory_reservation">> => 1073741824 memory_reservation => 1073741824
``` ```
支持单位： 支持单位：
@ -541,7 +539,7 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
转换： 转换：
```erlang ```erlang
<<"nano_cpus">> => 1500000000 nano_cpus => 1500000000
``` ```
规则： 规则：
@ -565,46 +563,46 @@ HTTP handler 最多等待 10 秒。超时返回 HTTP 504，其他参数或执行
} }
``` ```
对应下发给 `efka` 的协议命令示意： 对应内部命令示意：
```erlang ```erlang
{<<"command">>, Ref, {<<"container">>, #{ {command, Ref, {container, #{
<<"action">> => <<"deploy">>, action => deploy,
<<"task_id">> => 1001, task_id => 1001,
<<"params">> => #{ params => #{
<<"container_name">> => <<"my_nginx">>, container_name => <<"my_nginx">>,
<<"create">> => #{ create => #{
<<"config">> => #{ config => #{
<<"image">> => <<"docker.io/library/nginx:latest">>, image => <<"docker.io/library/nginx:latest">>,
<<"cmd">> => [<<"nginx">>, <<"-g">>, <<"daemon off;">>], cmd => [<<"nginx">>, <<"-g">>, <<"daemon off;">>],
<<"entrypoint">> => [], entrypoint => [],
<<"env">> => [], env => [],
<<"labels">> => #{}, labels => #{},
<<"volumes">> => [], volumes => [],
<<"user">> => <<>>, user => <<>>,
<<"working_dir">> => <<>>, working_dir => <<>>,
<<"hostname">> => <<>>, hostname => <<>>,
<<"exposed_ports">> => [], exposed_ports => [],
<<"healthcheck">> => undefined healthcheck => undefined
}, },
<<"host_config">> => #{ host_config => #{
<<"binds">> => [], binds => [],
<<"network_mode">> => <<>>, network_mode => <<>>,
<<"restart_policy">> => #{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0}, restart_policy => #{name => <<"always">>, maximum_retry_count => 0},
<<"privileged">> => false, privileged => false,
<<"cap_add">> => [], cap_add => [],
<<"cap_drop">> => [], cap_drop => [],
<<"devices">> => [], devices => [],
<<"memory">> => 0, memory => 0,
<<"memory_reservation">> => 0, memory_reservation => 0,
<<"nano_cpus">> => 0, nano_cpus => 0,
<<"cpu_shares">> => 0, cpu_shares => 0,
<<"ulimits">> => [], ulimits => [],
<<"tmpfs">> => #{}, tmpfs => #{},
<<"sysctls">> => #{}, sysctls => #{},
<<"extra_hosts">> => [] extra_hosts => []
}, },
<<"networking_config">> => #{<<"endpoints">> => []} networking_config => #{endpoints => []}
} }
} }
}}} }}}

129
docs/efka_iot_protocol.md
View File

@ -2,99 +2,95 @@
本文档描述 `efka` 与 `iot` 之间的 TLS 长连接协议。当前协议由 Erlang term 直接序列化，发送端使用 `term_to_binary/1`，接收端使用 `binary_to_term(PacketBin, [safe])`。 本文档描述 `efka` 与 `iot` 之间的 TLS 长连接协议。当前协议由 Erlang term 直接序列化，发送端使用 `term_to_binary/1`，接收端使用 `binary_to_term(PacketBin, [safe])`。
协议帧只使用 safe external term：顶层 label、业务 label、map key 使用 binary；`Ref` 使用 `crypto:strong_rand_bytes(16)` 生成，是 16 字节 binary。网络协议里不发送 Erlang `reference()`，也不依赖动态创建 atom。
## 传输层 ## 传输层
- `efka` 作为 TLS client 连接 `iot`。 - `efka` 作为 TLS client 连接 `iot`。
- `iot` 作为 TLS server 接收多个 `efka` 连接，一个连接对应一个 `ssl_channel` 进程。 - `iot` 作为 TLS server 接收多个 `efka` 连接，一个连接对应一个 `ssl_channel` 进程。
- socket 使用 `{packet, 4}`，每个 Erlang term binary 作为一个完整包发送。 - socket 使用 `{packet, 4}`，每个 Erlang term binary 作为一个完整包发送。
- `Ref` 使用 `crypto:strong_rand_bytes(16)` 生成，只在当前连接的 inflight 表内匹配。 - `Ref` 使用 `make_ref()` 生成，只在当前连接的 inflight 表内匹配。
## 顶层帧 ## 顶层帧
协议顶层 tuple 用来表达交互语义： 协议顶层 tuple 用来表达交互语义：
```erlang ```erlang
{<<"request">>, Ref, Body} {request, Ref, Body}
{<<"response">>, Ref, Reply} {response, Ref, Reply}
{<<"command">>, Ref, {Domain, Payload}} {command, Ref, {Domain, Payload}}
{<<"command_response">>, Ref, {Domain, Reply}} {command_response, Ref, {Domain, Reply}}
{<<"message">>, Body} {message, Body}
``` ```
语义说明： 语义说明：
| 帧 | 方向 | 语义 | | 帧 | 方向 | 语义 |
| --- | --- | --- | | --- | --- | --- |
| `{<<"request">>, Ref, Body}` | efka -> iot | efka 发起请求，需要 iot 回复 | | `{request, Ref, Body}` | efka -> iot | efka 发起请求，需要 iot 回复 |
| `{<<"response">>, Ref, Reply}` | iot -> efka | iot 对 efka request 的回复 | | `{response, Ref, Reply}` | iot -> efka | iot 对 efka request 的回复 |
| `{<<"command">>, Ref, {Domain, Payload}}` | iot -> efka | iot 下发命令，需要 efka 回复 | | `{command, Ref, {Domain, Payload}}` | iot -> efka | iot 下发命令，需要 efka 回复 |
| `{<<"command_response">>, Ref, {Domain, Reply}}` | efka -> iot | efka 对 iot command 的回复 | | `{command_response, Ref, {Domain, Reply}}` | efka -> iot | efka 对 iot command 的回复 |
| `{<<"message">>, Body}` | 双向 | 异步消息，不要求回复 | | `{message, Body}` | 双向 | 异步消息，不要求回复 |
`command` 和 `command_response` 的 `Domain` 表示业务域，目前支持： `command` 和 `command_response` 的 `Domain` 表示业务域，目前支持：
- `<<"container">>` - `container`
## 鉴权请求 ## 鉴权请求
初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。 初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。
```erlang ```erlang
{<<"request">>, Ref, {<<"auth_request">>, #{ {request, Ref, {auth_request, #{
<<"uuid">> => UUID, uuid => UUID,
<<"token">> => Token, token => Token,
<<"timestamp">> => Timestamp timestamp => Timestamp
}}} }}}
``` ```
`iot` 回复： `iot` 回复：
```erlang ```erlang
{<<"response">>, Ref, {<<"auth_response">>, <<"ok">>}} {response, Ref, {auth_response, ok}}
{<<"response">>, Ref, {<<"auth_response">>, {<<"error">>, {<<"failed">>, Reason}}}} {response, Ref, {auth_response, {error, {failed, Reason}}}}
``` ```
处理语义： 处理语义：
- `<<"ok">>`：`efka` 进入 `activated` 状态。 - `ok`：`efka` 进入 `activated` 状态。
- `{<<"error">>, {<<"failed">>, Reason}}`：鉴权失败，`iot` 返回失败响应后关闭连接；`efka` 进入重连流程。 - `{error, {failed, Reason}}`：鉴权失败，`iot` 返回失败响应后关闭连接；`efka` 进入重连流程。
## 授权控制 ## 授权控制
`/host/activate` 只修改 `iot` 本地和持久化的 host 授权状态，不再向 `efka` 下发 auth command。`efka` 可以继续保持连接并发送数据，是否处理这些数据由 `iot_host` 当前状态决定。 `/host/activate` 只修改 `iot` 本地和持久化的 host 授权状态，不再向 `efka` 下发 auth command。`efka` 可以继续保持连接并发送数据，是否处理这些数据由 `iot_host` 当前状态决定。
因此当前协议没有 `{<<"command">>, Ref, {<<"auth">>, ...}}` 和 `{<<"command_response">>, Ref, {<<"auth">>, ...}}`。授权关闭时，`iot_host` 保持 channel 在线，但不处理上报数据；授权重新打开后，已在线的 channel 可以继续使用。 因此当前协议没有 `{command, Ref, {auth, ...}}` 和 `{command_response, Ref, {auth, ...}}`。授权关闭时，`iot_host` 保持 channel 在线，但不处理上报数据；授权重新打开后，已在线的 channel 可以继续使用。
## 容器管理命令 ## 容器管理命令
`iot` 对 `efka` 的容器管理使用 command 语义： `iot` 对 `efka` 的容器管理使用 command 语义：
```erlang ```erlang
{<<"command">>, Ref, {<<"container">>, CommandMap}} {command, Ref, {container, CommandMap}}
``` ```
`efka` 回复： `efka` 回复：
```erlang ```erlang
{<<"command_response">>, Ref, {<<"container">>, Reply}} {command_response, Ref, {container, Reply}}
``` ```
`Reply` 取值： `Reply` 取值：
```erlang ```erlang
<<"ok">> ok
{<<"ok">>, Result} {ok, Result}
{<<"error">>, Reason} {error, Reason}
``` ```
`CommandMap` 使用 binary key 和 binary action；`efka` 接收后直接按 binary key/action 匹配，Docker 参数链路继续使用 binary-key map，不再转换成 atom-key map。
### list ### list
```erlang ```erlang
#{<<"action">> => <<"list">>} #{action => list}
``` ```
返回当前 `efka` 主机上的容器列表。 返回当前 `efka` 主机上的容器列表。
@ -103,9 +99,9 @@
```erlang ```erlang
#{ #{
<<"action">> => <<"deploy">>, action => deploy,
<<"task_id">> => TaskId, task_id => TaskId,
<<"params">> => Params params => Params
} }
``` ```
@ -115,8 +111,8 @@
```erlang ```erlang
#{ #{
<<"action">> => <<"start">>, action => start,
<<"target">> => Target target => Target
} }
``` ```
@ -124,9 +120,9 @@
```erlang ```erlang
#{ #{
<<"action">> => <<"stop">>, action => stop,
<<"target">> => Target, target => Target,
<<"timeout_seconds">> => TimeoutSeconds timeout_seconds => TimeoutSeconds
} }
``` ```
@ -134,9 +130,9 @@
```erlang ```erlang
#{ #{
<<"action">> => <<"kill">>, action => kill,
<<"target">> => Target, target => Target,
<<"signal">> => Signal signal => Signal
} }
``` ```
@ -144,10 +140,10 @@
```erlang ```erlang
#{ #{
<<"action">> => <<"remove">>, action => remove,
<<"target">> => Target, target => Target,
<<"force">> => Force, force => Force,
<<"remove_volumes">> => RemoveVolumes remove_volumes => RemoveVolumes
} }
``` ```
@ -155,9 +151,9 @@
```erlang ```erlang
#{ #{
<<"action">> => <<"config">>, action => config,
<<"target">> => Target, target => Target,
<<"config">> => Config config => Config
} }
``` ```
@ -169,8 +165,8 @@
```erlang ```erlang
#{ #{
<<"name">> => ContainerName, name => ContainerName,
<<"id">> => ContainerId id => ContainerId
} }
``` ```
@ -183,9 +179,9 @@
### efka -> iot: data ### efka -> iot: data
```erlang ```erlang
{<<"message">>, {<<"data">>, #{ {message, {data, #{
<<"route_key">> => RouteKey, route_key => RouteKey,
<<"metric">> => Metric metric => Metric
}}} }}}
``` ```
@ -194,20 +190,20 @@
### efka -> iot: task_event ### efka -> iot: task_event
```erlang ```erlang
{<<"message">>, {<<"task_event">>, #{ {message, {task_event, #{
<<"task_id">> => TaskId, task_id => TaskId,
<<"type">> => Type, type => Type,
<<"stream">> => Stream stream => Stream
}}} }}}
``` ```
任务事件流关闭时： 任务事件流关闭时：
```erlang ```erlang
{<<"message">>, {<<"task_event">>, #{ {message, {task_event, #{
<<"task_id">> => TaskId, task_id => TaskId,
<<"type">> => <<"close">>, type => <<"close">>,
<<"stream">> => Reason stream => Reason
}}} }}}
``` ```
@ -222,10 +218,10 @@ GET /event_stream?uuid=<host_uuid>&task_id=<task_id>
### iot -> efka: pub ### iot -> efka: pub
```erlang ```erlang
{<<"message">>, {<<"pub">>, #{ {message, {pub, #{
<<"topic">> => Topic, topic => Topic,
<<"qos">> => Qos, qos => Qos,
<<"content">> => Content content => Content
}}} }}}
``` ```
@ -246,6 +242,5 @@ GET /event_stream?uuid=<host_uuid>&task_id=<task_id>
- 旧容器回复：`{response, Ref, {container_response, ...}}` - 旧容器回复：`{response, Ref, {container_response, ...}}`
- 旧授权控制：`{message, {auth_control, Command}}` - 旧授权控制：`{message, {auth_control, Command}}`
- 已移除的 auth command：`{command, Ref, {auth, activate | deactivate}}` - 已移除的 auth command：`{command, Ref, {auth, activate | deactivate}}`
- 旧 Ref：Erlang `reference()`，例如 `make_ref()` 生成的值。
如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。 如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。

28
src/host/iot_host.erl
View File

@ -101,17 +101,15 @@ get_metric(Pid) when is_pid(Pid) ->
attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) -> attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
gen_statem:call(Pid, {attach_channel, ChannelPid}). gen_statem:call(Pid, {attach_channel, ChannelPid}).
-type request_ref() :: binary(). -spec get_containers(Pid :: pid()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
-spec get_containers(Pid :: pid()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}.
get_containers(Pid) when is_pid(Pid) -> get_containers(Pid) when is_pid(Pid) ->
container_call(Pid, docker_container_builder:list_request()). container_call(Pid, docker_container_builder:list_request()).
-spec config_container(Pid :: pid(), ContainerName :: binary(), ConfigJson :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. -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) -> config_container(Pid, ContainerName, ConfigJson) when is_pid(Pid), is_binary(ContainerName), is_binary(ConfigJson) ->
container_call(Pid, docker_container_builder:config_request(ContainerName, ConfigJson)). container_call(Pid, docker_container_builder:config_request(ContainerName, ConfigJson)).
-spec deploy_container(Pid :: pid(), TaskId :: integer(), Config :: map()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. -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) -> deploy_container(Pid, TaskId, Config) when is_pid(Pid), is_integer(TaskId), is_map(Config) ->
case docker_container_builder:deploy_request(TaskId, Config) of case docker_container_builder:deploy_request(TaskId, Config) of
{ok, Request} -> {ok, Request} ->
@ -120,25 +118,25 @@ deploy_container(Pid, TaskId, Config) when is_pid(Pid), is_integer(TaskId), is_m
{error, Reason} {error, Reason}
end. end.
-spec start_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. -spec start_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
start_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> start_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
container_call(Pid, docker_container_builder:start_request(ContainerName)). container_call(Pid, docker_container_builder:start_request(ContainerName)).
-spec stop_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. -spec stop_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
stop_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> stop_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
container_call(Pid, docker_container_builder:stop_request(ContainerName)). container_call(Pid, docker_container_builder:stop_request(ContainerName)).
-spec kill_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. -spec kill_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
kill_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> kill_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
container_call(Pid, docker_container_builder:kill_request(ContainerName)). container_call(Pid, docker_container_builder:kill_request(ContainerName)).
-spec remove_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. -spec remove_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: reference()} | {error, Reason :: any()}.
remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) ->
container_call(Pid, docker_container_builder:remove_request(ContainerName)). container_call(Pid, docker_container_builder:remove_request(ContainerName)).
-spec await_reply(Pid :: pid(), Ref :: request_ref(), Timeout :: integer()) -> -spec await_reply(Pid :: pid(), Ref :: reference(), Timeout :: integer()) ->
ok | {ok, Result :: term()} | {error, Reason :: term()}. ok | {ok, Result :: term()} | {error, Reason :: term()}.
await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_binary(Ref), is_integer(Timeout) -> await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) ->
receive receive
{command_reply, Ref, ok} -> {command_reply, Ref, ok} ->
ok; ok;
@ -225,7 +223,7 @@ handle_event({call, From}, get_status, _, State = #state{channel_pid = ChannelPi
handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> 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 case HasSession andalso is_pid(ChannelPid) of
true -> true ->
Ref = request_ref(), Ref = make_ref(),
ok = ssl_channel:container_call(ChannelPid, ReceiverPid, Ref, Request), ok = ssl_channel:container_call(ChannelPid, ReceiverPid, Ref, Request),
{keep_state, State, [{reply, From, {ok, Ref}}]}; {keep_state, State, [{reply, From, {ok, Ref}}]};
false -> false ->
@ -365,7 +363,7 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
%%%=================================================================== %%%===================================================================
-spec container_call(Pid :: pid(), Request :: term()) -> -spec container_call(Pid :: pid(), Request :: term()) ->
{ok, Ref :: request_ref()} | {error, Reason :: any()}. {ok, Ref :: reference()} | {error, Reason :: any()}.
container_call(Pid, Request) when is_pid(Pid) -> container_call(Pid, Request) when is_pid(Pid) ->
gen_statem:call(Pid, {container_call, self(), Request}). gen_statem:call(Pid, {container_call, self(), Request}).
@ -394,10 +392,6 @@ flush_reply(Ref) ->
ok ok
end. end.
-spec request_ref() -> request_ref().
request_ref() ->
crypto:strong_rand_bytes(16).
-spec maybe_mark_host_offline(binary()) -> ok. -spec maybe_mark_host_offline(binary()) -> ok.
maybe_mark_host_offline(UUID) -> maybe_mark_host_offline(UUID) ->
case iot_api_client:get_host_by_uuid(UUID) of case iot_api_client:get_host_by_uuid(UUID) of

88
src/transport/tcp/ssl_channel.erl
View File

@ -37,20 +37,18 @@
timer_ref :: reference() timer_ref :: reference()
}). }).
-type request_ref() :: binary().
%% 向通道中写入消息 %% 向通道中写入消息
-spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> ok. -spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> ok.
pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos), is_binary(Content) -> pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos), is_binary(Content) ->
gen_server:cast(Pid, {pub, Topic, Qos, Content}). gen_server:cast(Pid, {pub, Topic, Qos, Content}).
-spec container_call(Pid :: pid(), ReceiverPid :: pid(), Ref :: request_ref(), Request :: map()) -> ok. -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Ref :: reference(), Request :: map()) -> ok.
container_call(Pid, ReceiverPid, Ref, Request) when is_pid(Pid), is_pid(ReceiverPid), is_binary(Ref), is_map(Request) -> container_call(Pid, ReceiverPid, Ref, Request) when is_pid(Pid), is_pid(ReceiverPid), is_reference(Ref), is_map(Request) ->
gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}), gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}),
ok. ok.
-spec cancel_command_call(Pid :: pid(), Ref :: request_ref()) -> ok. -spec cancel_command_call(Pid :: pid(), Ref :: reference()) -> ok.
cancel_command_call(Pid, Ref) when is_pid(Pid), is_binary(Ref) -> cancel_command_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
gen_server:call(Pid, {cancel_command_call, Ref}). gen_server:call(Pid, {cancel_command_call, Ref}).
%% 关闭方法 %% 关闭方法
@ -87,7 +85,7 @@ handle_call(_Request, _From, State) ->
%% 发送消息, 基于pub/sub机制 %% 发送消息, 基于pub/sub机制
handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) ->
Packet = term_to_binary({<<"message">>, {<<"pub">>, #{<<"topic">> => Topic, <<"qos">> => Qos, <<"content">> => Content}}}), Packet = term_to_binary({message, {pub, #{topic => Topic, qos => Qos, content => Content}}}),
case Transport:send(Socket, Packet) of case Transport:send(Socket, Packet) of
ok -> ok ->
{noreply, State}; {noreply, State};
@ -98,7 +96,7 @@ handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, so
%% iot 请求 efka 时使用 command/command_response 语义。 %% iot 请求 efka 时使用 command/command_response 语义。
handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) -> handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
Packet = term_to_binary({<<"command">>, Ref, encode_command_body(Body)}), Packet = term_to_binary({command, Ref, Body}),
case Transport:send(Socket, Packet) of case Transport:send(Socket, Packet) of
ok -> ok ->
TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}), TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}),
@ -131,17 +129,17 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h
handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket}) when is_binary(PacketBin) -> handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket}) when is_binary(PacketBin) ->
try binary_to_term(PacketBin, [safe]) of try binary_to_term(PacketBin, [safe]) of
{<<"request">>, Ref, Body} -> {request, Ref, Body} ->
handle_request_frame(Ref, Body, State); handle_request_frame(Ref, Body, State);
{<<"message">>, Body} -> {message, Body} ->
handle_message_frame(Body, State); handle_message_frame(Body, State);
{<<"command_response">>, Ref, Response} -> {command_response, Ref, Response} ->
handle_command_response_frame(Ref, Response, State); handle_command_response_frame(Ref, Response, State);
Other -> Other ->
logger:warning("[ssl_channel] unsupported packet: ~p", [Other]), logger:warning("[ssl_channel] unsupported packet: ~p", [Other]),
{stop, bad_packet, State} {stop, bad_packet, State}
catch error:Error -> catch error:Error ->
logger:warning("[ssl_channel] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]), logger:warning("[ssl_channel] binary_to_term get error: ~p", [Error]),
{stop, bad_packet, State} {stop, bad_packet, State}
end; end;
@ -173,23 +171,23 @@ code_change(_OldVsn, State, _Extra) ->
%%%% helper methods %%%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec handle_request_frame(request_ref(), tuple(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. -spec handle_request_frame(reference(), tuple(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}.
handle_request_frame(Ref, {<<"auth_request">>, #{<<"uuid">> := UUID}}, State = #state{is_authed = true}) -> handle_request_frame(Ref, {auth_request, #{uuid := UUID}}, State = #state{is_authed = true}) ->
logger:warning("[ws_channel] repeated auth request, ref: ~p, uuid: ~p, close channel", [Ref, UUID]), logger:warning("[ws_channel] repeated auth request, ref: ~p, uuid: ~p, close channel", [Ref, UUID]),
{stop, repeated_auth, State}; {stop, repeated_auth, State};
handle_request_frame(Ref, {<<"auth_request">>, #{<<"uuid">> := UUID, <<"token">> := Token, <<"timestamp">> := Timestamp}}, State = #state{transport = Transport, socket = Socket}) -> handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp}}, State = #state{transport = Transport, socket = Socket}) ->
maybe maybe
ok ?= auth(Token, UUID, Timestamp), ok ?= auth(Token, UUID, Timestamp),
{ok, HostPid} ?= iot_host:lookup_pid(UUID), {ok, HostPid} ?= iot_host:lookup_pid(UUID),
ok ?= iot_host:attach_channel(HostPid, self()), ok ?= iot_host:attach_channel(HostPid, self()),
erlang:monitor(process, HostPid), erlang:monitor(process, HostPid),
ok = send_reply_frame(Transport, Socket, Ref, {<<"auth_response">>, <<"ok">>}), ok = send_reply_frame(Transport, Socket, Ref, {auth_response, ok}),
logger:debug("[ws_channel] auth uuid: ~p", [UUID]), logger:debug("[ws_channel] auth uuid: ~p", [UUID]),
{noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}} {noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}}
else else
{error, Reason} -> {error, Reason} ->
logger:warning("[ws_channel] uuid: ~p, auth failed with reason: ~p", [UUID, Reason]), logger:warning("[ws_channel] uuid: ~p, auth failed with reason: ~p", [UUID, Reason]),
case send_reply_frame(Transport, Socket, Ref, {<<"auth_response">>, {<<"error">>, {<<"failed">>, Reason}}}) of case send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}) of
ok -> ok ->
{stop, Reason, State}; {stop, Reason, State};
{error, SendReason} -> {error, SendReason} ->
@ -197,7 +195,7 @@ handle_request_frame(Ref, {<<"auth_request">>, #{<<"uuid">> := UUID, <<"token">>
{stop, {send_failed, SendReason}, State} {stop, {send_failed, SendReason}, State}
end end
end; end;
handle_request_frame(Ref, {<<"container">>, ContainerCommand}, State) -> handle_request_frame(Ref, {container, ContainerCommand}, State) ->
logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]), logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]),
{stop, normal, State}; {stop, normal, State};
handle_request_frame(Ref, Body, State) -> handle_request_frame(Ref, Body, State) ->
@ -206,10 +204,10 @@ handle_request_frame(Ref, Body, State) ->
-spec handle_message_frame(tuple(), #state{}) -> -spec handle_message_frame(tuple(), #state{}) ->
{noreply, #state{}}. {noreply, #state{}}.
handle_message_frame({<<"data">>, #{<<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> handle_message_frame({data, #{route_key := RouteKey, metric := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) ->
iot_host:handle(HostPid, {data, RouteKey, Metric}), iot_host:handle(HostPid, {data, RouteKey, Metric}),
{noreply, State}; {noreply, State};
handle_message_frame({<<"task_event">>, Event}, State = #state{uuid = UUID, host_pid = HostPid}) when is_binary(UUID), is_pid(HostPid) -> handle_message_frame({task_event, Event}, State = #state{uuid = UUID, host_pid = HostPid}) when is_binary(UUID), is_pid(HostPid) ->
handle_event_stream_frame(UUID, Event), handle_event_stream_frame(UUID, Event),
{noreply, State}; {noreply, State};
handle_message_frame(Body, State) -> handle_message_frame(Body, State) ->
@ -217,18 +215,18 @@ handle_message_frame(Body, State) ->
{noreply, State}. {noreply, State}.
-spec handle_event_stream_frame(binary(), map()) -> ok. -spec handle_event_stream_frame(binary(), map()) -> ok.
handle_event_stream_frame(UUID, #{<<"task_id">> := TaskId, <<"type">> := <<"close">>, <<"stream">> := Reason}) -> handle_event_stream_frame(UUID, #{task_id := TaskId, type := <<"close">>, stream := Reason}) ->
iot_container_task_sup:close(UUID, TaskId, Reason); iot_container_task_sup:close(UUID, TaskId, Reason);
handle_event_stream_frame(UUID, #{<<"task_id">> := TaskId, <<"type">> := Type, <<"stream">> := Stream}) -> handle_event_stream_frame(UUID, #{task_id := TaskId, type := Type, stream := Stream}) ->
logger:debug("[ssl_channel] get uuid: ~p, task_id: ~p, type: ~ts, stream: ~ts", [UUID, TaskId, Type, Stream]), logger:debug("[ssl_channel] get uuid: ~p, task_id: ~p, type: ~ts, stream: ~ts", [UUID, TaskId, Type, Stream]),
iot_container_task_sup:stream(UUID, TaskId, Type, Stream); iot_container_task_sup:stream(UUID, TaskId, Type, Stream);
handle_event_stream_frame(UUID, Event) -> handle_event_stream_frame(UUID, Event) ->
logger:warning("[ssl_channel] invalid task_event, uuid: ~p, event: ~p", [UUID, Event]), logger:warning("[ssl_channel] invalid task_event, uuid: ~p, event: ~p", [UUID, Event]),
ok. ok.
-spec handle_command_response_frame(request_ref(), tuple(), #state{}) -> -spec handle_command_response_frame(reference(), tuple(), #state{}) ->
{noreply, #state{}}. {noreply, #state{}}.
handle_command_response_frame(Ref, Reply, State = #state{inflight = Inflight}) when is_binary(Ref) -> handle_command_response_frame(Ref, Reply, State = #state{inflight = Inflight}) when is_reference(Ref) ->
case maps:take(Ref, Inflight) of case maps:take(Ref, Inflight) of
error -> error ->
{noreply, State}; {noreply, State};
@ -241,23 +239,23 @@ handle_command_response_frame(Ref, Reply, State) ->
logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]), logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]),
{noreply, State}. {noreply, State}.
-spec send_reply_frame(module(), any(), request_ref(), tuple()) -> ok | {error, term()}. -spec send_reply_frame(module(), any(), reference(), tuple()) -> ok | {error, term()}.
send_reply_frame(Transport, Socket, Ref, Reply) -> send_reply_frame(Transport, Socket, Ref, Reply) ->
Packet = term_to_binary({<<"response">>, Ref, Reply}), Packet = term_to_binary({response, Ref, Reply}),
Transport:send(Socket, Packet). Transport:send(Socket, Packet).
-spec decode_command_response({binary(), term()} | tuple()) -> -spec decode_command_response({container, ok | {ok, term()} | {error, term()}} | tuple()) ->
ok | {ok, term()} | {error, term()}. ok | {ok, term()} | {error, term()}.
decode_command_response({<<"container">>, <<"ok">>}) -> decode_command_response({container, ok}) ->
ok; ok;
decode_command_response({<<"container">>, {<<"ok">>, Result}}) -> decode_command_response({container, {ok, Result}}) ->
{ok, Result}; {ok, Result};
decode_command_response({<<"container">>, {<<"error">>, Reason}}) -> decode_command_response({container, {error, Reason}}) ->
{error, Reason}; {error, Reason};
decode_command_response(_Reply) -> decode_command_response(_Reply) ->
{error, invalid_response}. {error, invalid_response}.
-spec deliver_command_response(undefined | pid(), request_ref(), tuple()) -> ok. -spec deliver_command_response(undefined | pid(), reference(), tuple()) -> ok.
deliver_command_response(undefined, _Ref, _Reply) -> deliver_command_response(undefined, _Ref, _Reply) ->
ok; ok;
deliver_command_response(ReceiverPid, Ref, Reply) when is_pid(ReceiverPid) -> deliver_command_response(ReceiverPid, Ref, Reply) when is_pid(ReceiverPid) ->
@ -268,21 +266,21 @@ deliver_command_response(ReceiverPid, Ref, Reply) when is_pid(ReceiverPid) ->
logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref]) logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref])
end. end.
-spec deliver_command_error(undefined | pid(), request_ref(), term()) -> ok. -spec deliver_command_error(undefined | pid(), reference(), term()) -> ok.
deliver_command_error(ReceiverPid, Ref, Reason) when is_pid(ReceiverPid) -> deliver_command_error(ReceiverPid, Ref, Reason) when is_pid(ReceiverPid) ->
ReceiverPid ! {command_reply, Ref, {error, Reason}}, ReceiverPid ! {command_reply, Ref, {error, Reason}},
ok; ok;
deliver_command_error(_ReceiverPid, _Ref, _Reason) -> deliver_command_error(_ReceiverPid, _Ref, _Reason) ->
ok. ok.
-spec reply_command_timeout(undefined | pid(), request_ref()) -> ok. -spec reply_command_timeout(undefined | pid(), reference()) -> ok.
reply_command_timeout(ReceiverPid, Ref) when is_pid(ReceiverPid) -> reply_command_timeout(ReceiverPid, Ref) when is_pid(ReceiverPid) ->
ReceiverPid ! {command_reply, Ref, {error, timeout}}, ReceiverPid ! {command_reply, Ref, {error, timeout}},
ok; ok;
reply_command_timeout(_ReceiverPid, _Ref) -> reply_command_timeout(_ReceiverPid, _Ref) ->
ok. ok.
-spec reply_command_closed(request_ref(), #inflight_command{}, term()) -> ok. -spec reply_command_closed(reference(), #inflight_command{}, term()) -> ok.
reply_command_closed(Ref, #inflight_command{receiver_pid = ReceiverPid}, Reason) when is_pid(ReceiverPid) -> reply_command_closed(Ref, #inflight_command{receiver_pid = ReceiverPid}, Reason) when is_pid(ReceiverPid) ->
ReceiverPid ! {command_reply, Ref, {error, {channel_closed, Reason}}}, ReceiverPid ! {command_reply, Ref, {error, {channel_closed, Reason}}},
ok; ok;
@ -306,25 +304,3 @@ auth(Token, UUID, Timestamp) when is_binary(Token), is_binary(UUID), is_integer(
false -> false ->
{error, <<"invalid timestamp">>} {error, <<"invalid timestamp">>}
end. end.
-spec encode_command_body({container, map()}) -> {binary(), map()}.
encode_command_body({container, Request}) ->
{<<"container">>, safe_term(Request)}.
-spec safe_term(term()) -> term().
safe_term(true) ->
true;
safe_term(false) ->
false;
safe_term(undefined) ->
undefined;
safe_term(Value) when is_atom(Value) ->
atom_to_binary(Value, utf8);
safe_term(Value) when is_map(Value) ->
maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]);
safe_term(Value) when is_list(Value) ->
[safe_term(Item) || Item <- Value];
safe_term(Value) when is_tuple(Value) ->
list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
safe_term(Value) ->
Value.