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

Compare commits

base: anlicheng:d506ec98d5cde5a042696e9e222c28f98c81e930
Branches Tags
anlicheng:main
anlicheng:protobuf_mst
anlicheng:docker_mst
anlicheng:dets_mst
anlicheng:v1.0
..
compare: anlicheng:ba7c2b8ecd31b0aa4a1dd4c27530fd1f0ac2c750
Branches Tags
anlicheng:main
anlicheng:protobuf_mst
anlicheng:docker_mst
anlicheng:dets_mst
anlicheng:v1.0

2 Commits
d506ec98d5 ... ba7c2b8ecd

Author SHA1 Message Date
anlicheng
ba7c2b8ecd fix binary_to_term 2026-05-09 22:55:23 +08:00
anlicheng
c4d17b9024 fix docs 2026-05-09 22:49:58 +08:00
7 changed files with 311 additions and 237 deletions
Show Stats Expand all files Collapse all files

148
docs/container_command_docker_json.md
View File

@ -15,31 +15,41 @@
`iot` 通过 TLS 长连接向 `efka` 下发容器命令： `iot` 通过 TLS 长连接向 `efka` 下发容器命令：
```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}
``` ```
只有 `efka_client` 处于 `activated` 状态时，容器命令才会正常执行；处于 `restricted` 或其他状态时会返回错误。 `Ref` 是 `crypto:strong_rand_bytes(16)` 生成的 16 字节 binary。网络帧只使用 `binary_to_term(PacketBin, [safe])` 可解码的 safe term；协议 label、业务 label、map key 和 action 都使用 binary。
只有 `efka_client` 处于 `activated` 状态时，容器命令才会正常执行；处于非 activated 状态时会返回错误。
## 2. 容器命令 map ## 2. 容器命令 map
`efka_client` 收到网络协议里的 binary-key `CommandMap` 后，直接按 binary key 和 binary action 做函数参数匹配，不再做整包 atom-key 转换。
- command 顶层、`target`、deploy `params`、`create` 中间结构都使用 binary key。
- `<<"action">>` 的取值使用 binary：`<<"list">>`、`<<"deploy">>`、`<<"start">>`、`<<"stop">>`、`<<"kill">>`、`<<"remove">>`、`<<"config">>`。
- `<<"__undefined__">>` 只在 Docker 参数构造读取可选字段时按需当成 `undefined` 处理。
下面各小节描述的是 `efka` 直接接收和处理的 map 格式。
### list ### list
```erlang ```erlang
#{action => list} #{<<"action">> => <<"list">>}
``` ```
执行： 执行：
@ -58,9 +68,9 @@ GET /containers/json?all=true
```erlang ```erlang
#{ #{
action => deploy, <<"action">> => <<"deploy">>,
task_id => TaskId, <<"task_id">> => TaskId,
params => Params <<"params">> => Params
} }
``` ```
@ -76,8 +86,8 @@ docker_deploy_manager:deploy(TaskId, Params)
```erlang ```erlang
#{ #{
action => start, <<"action">> => <<"start">>,
target => Target <<"target">> => Target
} }
``` ```
@ -97,9 +107,9 @@ POST /containers/{name_or_id}/start
```erlang ```erlang
#{ #{
action => stop, <<"action">> => <<"stop">>,
target => Target, <<"target">> => Target,
timeout_seconds => TimeoutSeconds <<"timeout_seconds">> => TimeoutSeconds
} }
``` ```
@ -119,9 +129,9 @@ POST /containers/{name_or_id}/stop?t={TimeoutSeconds}
```erlang ```erlang
#{ #{
action => kill, <<"action">> => <<"kill">>,
target => Target, <<"target">> => Target,
signal => Signal <<"signal">> => Signal
} }
``` ```
@ -135,10 +145,10 @@ docker_commands:kill_container(ContainerNameOrId, Signal)
```erlang ```erlang
#{ #{
action => remove, <<"action">> => <<"remove">>,
target => Target, <<"target">> => Target,
force => Force, <<"force">> => Force,
remove_volumes => RemoveVolumes <<"remove_volumes">> => RemoveVolumes
} }
``` ```
@ -152,9 +162,9 @@ docker_commands:remove_container(ContainerNameOrId, Force, RemoveVolumes)
```erlang ```erlang
#{ #{
action => config, <<"action">> => <<"config">>,
target => Target, <<"target">> => Target,
config => Config <<"config">> => Config
} }
``` ```
@ -172,8 +182,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
```erlang ```erlang
#{ #{
name => ContainerName, <<"name">> => ContainerName,
id => ContainerId <<"id">> => ContainerId
} }
``` ```
@ -189,8 +199,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
```erlang ```erlang
#{ #{
container_name => ContainerName, <<"container_name">> => ContainerName,
create => Create <<"create">> => Create
} }
``` ```
@ -215,9 +225,9 @@ efka.root_dir/container_name/
```erlang ```erlang
#{ #{
config => ContainerConfig, <<"config">> => ContainerConfig,
host_config => HostConfig, <<"host_config">> => HostConfig,
networking_config => NetworkingConfig <<"networking_config">> => NetworkingConfig
} }
``` ```
@ -340,8 +350,8 @@ Body = iolist_to_binary(json:encode(Options))
```erlang ```erlang
[ [
#{container_port => 80, protocol => <<"tcp">>}, #{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>},
#{container_port => 53, protocol => <<"udp">>} #{<<"container_port">> => 53, <<"protocol">> => <<"udp">>}
] ]
``` ```
@ -369,10 +379,10 @@ Body = iolist_to_binary(json:encode(Options))
```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
} }
``` ```
@ -440,7 +450,7 @@ efka 自动补丁后输出：
输入： 输入：
```erlang ```erlang
#{name => <<"always">>, maximum_retry_count => 0} #{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0}
``` ```
输出： 输出：
@ -458,7 +468,7 @@ efka 自动补丁后输出：
输入： 输入：
```erlang ```erlang
#{name => <<"on-failure">>, maximum_retry_count => 3} #{<<"name">> => <<"on-failure">>, <<"maximum_retry_count">> => 3}
``` ```
输出： 输出：
@ -481,9 +491,9 @@ efka 自动补丁后输出：
```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">>
} }
] ]
``` ```
@ -512,10 +522,10 @@ efka 自动补丁后输出：
```erlang ```erlang
#{ #{
memory => 536870912, <<"memory">> => 536870912,
memory_reservation => 268435456, <<"memory_reservation">> => 268435456,
nano_cpus => 1500000000, <<"nano_cpus">> => 1500000000,
cpu_shares => 512 <<"cpu_shares">> => 512
} }
``` ```
@ -540,7 +550,7 @@ efka 自动补丁后输出：
```erlang ```erlang
[ [
#{name => <<"nofile">>, soft => 1024, hard => 2048} #{<<"name">> => <<"nofile">>, <<"soft">> => 1024, <<"hard">> => 2048}
] ]
``` ```
@ -628,9 +638,9 @@ efka 自动补丁后输出：
```erlang ```erlang
#{ #{
endpoints => [ <<"endpoints">> => [
#{name => <<"bridge">>}, #{<<"name">> => <<"bridge">>},
#{name => <<"mynet">>} #{<<"name">> => <<"mynet">>}
] ]
} }
``` ```
@ -661,26 +671,26 @@ efka 自动补丁后输出：
收到的 deploy command： 收到的 deploy command：
```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;">>],
env => [<<"ENV=prod">>], <<"env">> => [<<"ENV=prod">>],
volumes => [<<"/data">>], <<"volumes">> => [<<"/data">>],
exposed_ports => [#{container_port => 80, protocol => <<"tcp">>}] <<"exposed_ports">> => [#{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>}]
}, },
host_config => #{ <<"host_config">> => #{
binds => [<<"/host/data:/data">>], <<"binds">> => [<<"/host/data:/data">>],
restart_policy => #{name => <<"always">>, maximum_retry_count => 0}, <<"restart_policy">> => #{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0},
memory => 536870912 <<"memory">> => 536870912
}, },
networking_config => #{ <<"networking_config">> => #{
endpoints => [#{name => <<"bridge">>}] <<"endpoints">> => [#{<<"name">> => <<"bridge">>}]
} }
} }
} }

129
docs/efka_iot_protocol.md
View File

@ -2,95 +2,99 @@
本文档描述 `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` 使用 `make_ref()` 生成，只在当前连接的 inflight 表内匹配。 - `Ref` 使用 `crypto:strong_rand_bytes(16)` 生成，只在当前连接的 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` 主机上的容器列表。
@ -99,9 +103,9 @@ ok
```erlang ```erlang
#{ #{
action => deploy, <<"action">> => <<"deploy">>,
task_id => TaskId, <<"task_id">> => TaskId,
params => Params <<"params">> => Params
} }
``` ```
@ -111,8 +115,8 @@ ok
```erlang ```erlang
#{ #{
action => start, <<"action">> => <<"start">>,
target => Target <<"target">> => Target
} }
``` ```
@ -120,9 +124,9 @@ ok
```erlang ```erlang
#{ #{
action => stop, <<"action">> => <<"stop">>,
target => Target, <<"target">> => Target,
timeout_seconds => TimeoutSeconds <<"timeout_seconds">> => TimeoutSeconds
} }
``` ```
@ -130,9 +134,9 @@ ok
```erlang ```erlang
#{ #{
action => kill, <<"action">> => <<"kill">>,
target => Target, <<"target">> => Target,
signal => Signal <<"signal">> => Signal
} }
``` ```
@ -140,10 +144,10 @@ ok
```erlang ```erlang
#{ #{
action => remove, <<"action">> => <<"remove">>,
target => Target, <<"target">> => Target,
force => Force, <<"force">> => Force,
remove_volumes => RemoveVolumes <<"remove_volumes">> => RemoveVolumes
} }
``` ```
@ -151,9 +155,9 @@ ok
```erlang ```erlang
#{ #{
action => config, <<"action">> => <<"config">>,
target => Target, <<"target">> => Target,
config => Config <<"config">> => Config
} }
``` ```
@ -165,8 +169,8 @@ ok
```erlang ```erlang
#{ #{
name => ContainerName, <<"name">> => ContainerName,
id => ContainerId <<"id">> => ContainerId
} }
``` ```
@ -179,9 +183,9 @@ ok
### efka -> iot: data ### efka -> iot: data
```erlang ```erlang
{message, {data, #{ {<<"message">>, {<<"data">>, #{
route_key => RouteKey, <<"route_key">> => RouteKey,
metric => Metric <<"metric">> => Metric
}}} }}}
``` ```
@ -190,20 +194,20 @@ ok
### 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
}}} }}}
``` ```
@ -218,10 +222,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
}}} }}}
``` ```
@ -248,5 +252,6 @@ 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()` 生成的值。
如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。 如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。

2
src/docker/docker_commands.erl
View File

@ -35,7 +35,7 @@ check_image_exist(Image) when is_binary(Image) ->
-spec create_container(ContainerDir :: string(), Params :: map()) -> -spec create_container(ContainerDir :: string(), Params :: map()) ->
{ok, ContainerId :: binary()} | {error, Reason :: any()}. {ok, ContainerId :: binary()} | {error, Reason :: any()}.
create_container(ContainerDir, #{container_name := ContainerName0, create := CreateOpts}) create_container(ContainerDir, #{<<"container_name">> := ContainerName0, <<"create">> := CreateOpts})
when is_list(ContainerDir), is_binary(ContainerName0), is_map(CreateOpts) -> when is_list(ContainerDir), is_binary(ContainerName0), is_map(CreateOpts) ->
Url = lists:flatten(io_lib:format("/containers/create?name=~s", [binary_to_list(ContainerName0)])), Url = lists:flatten(io_lib:format("/containers/create?name=~s", [binary_to_list(ContainerName0)])),
Options = docker_container_builder:build_options(ContainerName0, ContainerDir, CreateOpts), Options = docker_container_builder:build_options(ContainerName0, ContainerDir, CreateOpts),

112
src/docker/docker_container_builder.erl
View File

@ -23,30 +23,30 @@ patch_create_options(ContainerName, ConfigFile, undefined) ->
patch_create_options(ContainerName, ConfigFile, #{}); patch_create_options(ContainerName, ConfigFile, #{});
patch_create_options(ContainerName, ConfigFile, Create0) patch_create_options(ContainerName, ConfigFile, Create0)
when is_binary(ContainerName), is_binary(ConfigFile), is_map(Create0) -> when is_binary(ContainerName), is_binary(ConfigFile), is_map(Create0) ->
Config = patch_container_config(ContainerName, ensure_container_config(maps:get(config, Create0, undefined))), Config = patch_container_config(ContainerName, ensure_container_config(field(Create0, <<"config">>, undefined))),
HostConfig = patch_host_config(ConfigFile, ensure_host_config(maps:get(host_config, Create0, undefined))), HostConfig = patch_host_config(ConfigFile, ensure_host_config(field(Create0, <<"host_config">>, undefined))),
Create0#{ Create0#{
config => Config, <<"config">> => Config,
host_config => HostConfig <<"host_config">> => HostConfig
}. }.
-spec patch_container_config(binary(), map()) -> map(). -spec patch_container_config(binary(), map()) -> map().
patch_container_config(ContainerName, Config0) when is_binary(ContainerName), is_map(Config0) -> patch_container_config(ContainerName, Config0) when is_binary(ContainerName), is_map(Config0) ->
Env0 = [to_binary(EnvItem) || EnvItem <- maps:get(env, Config0, [])], Env0 = [to_binary(EnvItem) || EnvItem <- field(Config0, <<"env">>, [])],
Volumes0 = [to_binary(Volume) || Volume <- maps:get(volumes, Config0, [])], Volumes0 = [to_binary(Volume) || Volume <- field(Config0, <<"volumes">>, [])],
ConfigVolume = <<"/usr/local/etc/service.conf">>, ConfigVolume = <<"/usr/local/etc/service.conf">>,
Envs = add_unique_front([<<"CONTAINER_NAME=", ContainerName/binary>>], Env0), Envs = add_unique_front([<<"CONTAINER_NAME=", ContainerName/binary>>], Env0),
Volumes = add_unique_front([ConfigVolume], Volumes0), Volumes = add_unique_front([ConfigVolume], Volumes0),
Config0#{ Config0#{
env => Envs, <<"env">> => Envs,
volumes => Volumes <<"volumes">> => Volumes
}. }.
-spec patch_host_config(binary(), map()) -> map(). -spec patch_host_config(binary(), map()) -> map().
patch_host_config(ConfigFile, HostConfig0) when is_binary(ConfigFile), is_map(HostConfig0) -> patch_host_config(ConfigFile, HostConfig0) when is_binary(ConfigFile), is_map(HostConfig0) ->
Binds0 = [to_binary(Bind) || Bind <- maps:get(binds, HostConfig0, [])], Binds0 = [to_binary(Bind) || Bind <- field(HostConfig0, <<"binds">>, [])],
ConfigBind = <<ConfigFile/binary, ":/usr/local/etc/service.conf">>, ConfigBind = <<ConfigFile/binary, ":/usr/local/etc/service.conf">>,
HostConfig0#{binds => add_unique_front([ConfigBind], Binds0)}. HostConfig0#{<<"binds">> => add_unique_front([ConfigBind], Binds0)}.
-spec ensure_container_config(map() | undefined) -> map(). -spec ensure_container_config(map() | undefined) -> map().
ensure_container_config(undefined) -> ensure_container_config(undefined) ->
@ -62,39 +62,39 @@ ensure_host_config(HostConfig) when is_map(HostConfig) ->
-spec build_create_options(map()) -> map(). -spec build_create_options(map()) -> map().
build_create_options(Create0) when is_map(Create0) -> build_create_options(Create0) when is_map(Create0) ->
Config = maps:get(config, Create0, #{}), Config = field(Create0, <<"config">>, #{}),
HostConfig = maps:get(host_config, Create0, #{}), HostConfig = field(Create0, <<"host_config">>, #{}),
Endpoints = networking_config_endpoints(maps:get(networking_config, Create0, undefined)), Endpoints = networking_config_endpoints(field(Create0, <<"networking_config">>, undefined)),
#{ #{
<<"Image">> => to_binary(maps:get(image, Config, <<>>)), <<"Image">> => to_binary(field(Config, <<"image">>, <<>>)),
<<"Cmd">> => [to_binary(CommandItem) || CommandItem <- maps:get(cmd, Config, [])], <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- field(Config, <<"cmd">>, [])],
<<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- maps:get(entrypoint, Config, [])], <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- field(Config, <<"entrypoint">>, [])],
<<"Env">> => [to_binary(EnvItem) || EnvItem <- maps:get(env, Config, [])], <<"Env">> => [to_binary(EnvItem) || EnvItem <- field(Config, <<"env">>, [])],
<<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(maps:get(labels, Config, #{}))]), <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(field(Config, <<"labels">>, #{}))]),
<<"Volumes">> => build_volumes(maps:get(volumes, Config, [])), <<"Volumes">> => build_volumes(field(Config, <<"volumes">>, [])),
<<"User">> => to_binary(maps:get(user, Config, <<>>)), <<"User">> => to_binary(field(Config, <<"user">>, <<>>)),
<<"WorkingDir">> => to_binary(maps:get(working_dir, Config, <<>>)), <<"WorkingDir">> => to_binary(field(Config, <<"working_dir">>, <<>>)),
<<"Hostname">> => to_binary(maps:get(hostname, Config, <<>>)), <<"Hostname">> => to_binary(field(Config, <<"hostname">>, <<>>)),
<<"ExposedPorts">> => build_expose(maps:get(exposed_ports, Config, [])), <<"ExposedPorts">> => build_expose(field(Config, <<"exposed_ports">>, [])),
<<"NetworkingConfig">> => build_networking_config(Endpoints), <<"NetworkingConfig">> => build_networking_config(Endpoints),
<<"Healthcheck">> => build_healthcheck(maps:get(healthcheck, Config, undefined)), <<"Healthcheck">> => build_healthcheck(field(Config, <<"healthcheck">>, undefined)),
<<"HostConfig">> => fold_merge([ <<"HostConfig">> => fold_merge([
build_binds(maps:get(binds, HostConfig, [])), build_binds(field(HostConfig, <<"binds">>, [])),
build_network_mode(maps:get(network_mode, HostConfig, <<>>)), build_network_mode(field(HostConfig, <<"network_mode">>, <<>>)),
build_restart(maps:get(restart_policy, HostConfig, undefined)), build_restart(field(HostConfig, <<"restart_policy">>, undefined)),
build_privileged(maps:get(privileged, HostConfig, false)), build_privileged(field(HostConfig, <<"privileged">>, false)),
build_cap_add_drop(maps:get(cap_add, HostConfig, []), maps:get(cap_drop, HostConfig, [])), build_cap_add_drop(field(HostConfig, <<"cap_add">>, []), field(HostConfig, <<"cap_drop">>, [])),
build_devices(maps:get(devices, HostConfig, [])), build_devices(field(HostConfig, <<"devices">>, [])),
build_resources( build_resources(
maps:get(memory, HostConfig, 0), field(HostConfig, <<"memory">>, 0),
maps:get(memory_reservation, HostConfig, 0), field(HostConfig, <<"memory_reservation">>, 0),
maps:get(nano_cpus, HostConfig, 0), field(HostConfig, <<"nano_cpus">>, 0),
maps:get(cpu_shares, HostConfig, 0) field(HostConfig, <<"cpu_shares">>, 0)
), ),
build_ulimits(maps:get(ulimits, HostConfig, [])), build_ulimits(field(HostConfig, <<"ulimits">>, [])),
build_tmpfs(maps:get(tmpfs, HostConfig, #{})), build_tmpfs(field(HostConfig, <<"tmpfs">>, #{})),
build_sysctls(maps:get(sysctls, HostConfig, #{})), build_sysctls(field(HostConfig, <<"sysctls">>, #{})),
build_extra_hosts(maps:get(extra_hosts, HostConfig, [])) build_extra_hosts(field(HostConfig, <<"extra_hosts">>, []))
]) ])
}. }.
@ -102,7 +102,7 @@ build_create_options(Create0) when is_map(Create0) ->
networking_config_endpoints(undefined) -> networking_config_endpoints(undefined) ->
[]; [];
networking_config_endpoints(NetworkingConfig) when is_map(NetworkingConfig) -> networking_config_endpoints(NetworkingConfig) when is_map(NetworkingConfig) ->
maps:get(endpoints, NetworkingConfig, []). field(NetworkingConfig, <<"endpoints">>, []).
-spec fold_merge([map()]) -> map(). -spec fold_merge([map()]) -> map().
fold_merge(List) -> fold_merge(List) ->
@ -141,7 +141,7 @@ build_networking_config(Endpoints) when is_list(Endpoints) ->
[] -> [] ->
#{}; #{};
_ -> _ ->
NetCfg = maps:from_list([{to_binary(Name), #{}} || #{name := Name} <- Endpoints]), NetCfg = maps:from_list([{to_binary(Name), #{}} || #{<<"name">> := Name} <- Endpoints]),
#{<<"EndpointsConfig">> => NetCfg} #{<<"EndpointsConfig">> => NetCfg}
end. end.
@ -156,10 +156,10 @@ build_healthcheck(undefined) ->
#{}; #{};
build_healthcheck(Healthcheck) when is_map(Healthcheck) -> build_healthcheck(Healthcheck) when is_map(Healthcheck) ->
#{ #{
<<"Test">> => [to_binary(Item) || Item <- maps:get(test, Healthcheck, [])], <<"Test">> => [to_binary(Item) || Item <- field(Healthcheck, <<"test">>, [])],
<<"Interval">> => maps:get(interval_ns, Healthcheck, 0), <<"Interval">> => field(Healthcheck, <<"interval_ns">>, 0),
<<"Timeout">> => maps:get(timeout_ns, Healthcheck, 0), <<"Timeout">> => field(Healthcheck, <<"timeout_ns">>, 0),
<<"Retries">> => maps:get(retries, Healthcheck, 0) <<"Retries">> => field(Healthcheck, <<"retries">>, 0)
}. }.
-spec build_restart(map() | undefined) -> map(). -spec build_restart(map() | undefined) -> map().
@ -167,9 +167,9 @@ build_restart(undefined) ->
#{}; #{};
build_restart(RestartPolicy0) when is_map(RestartPolicy0) -> build_restart(RestartPolicy0) when is_map(RestartPolicy0) ->
RestartPolicy = #{ RestartPolicy = #{
<<"Name">> => to_binary(maps:get(name, RestartPolicy0, <<>>)) <<"Name">> => to_binary(field(RestartPolicy0, <<"name">>, <<>>))
}, },
case maps:get(maximum_retry_count, RestartPolicy0, 0) of case field(RestartPolicy0, <<"maximum_retry_count">>, 0) of
0 -> 0 ->
#{<<"RestartPolicy">> => RestartPolicy}; #{<<"RestartPolicy">> => RestartPolicy};
RetryCount -> RetryCount ->
@ -210,9 +210,9 @@ build_devices(Devices) when is_list(Devices) ->
<<"PathInContainer">> => to_binary(ContainerPath), <<"PathInContainer">> => to_binary(ContainerPath),
<<"CgroupPermissions">> => device_permissions(Permissions) <<"CgroupPermissions">> => device_permissions(Permissions)
} || #{ } || #{
path_on_host := HostPath, <<"path_on_host">> := HostPath,
path_in_container := ContainerPath, <<"path_in_container">> := ContainerPath,
cgroup_permissions := Permissions <<"cgroup_permissions">> := Permissions
} <- Devices], } <- Devices],
#{<<"Devices">> => DevObjs} #{<<"Devices">> => DevObjs}
end. end.
@ -255,7 +255,7 @@ build_ulimits(Ulimits) when is_list(Ulimits) ->
<<"Name">> => to_binary(Name), <<"Name">> => to_binary(Name),
<<"Soft">> => Soft, <<"Soft">> => Soft,
<<"Hard">> => Hard <<"Hard">> => Hard
} || #{name := Name, soft := Soft, hard := Hard} <- Ulimits]} } || #{<<"name">> := Name, <<"soft">> := Soft, <<"hard">> := Hard} <- Ulimits]}
end. end.
-spec build_sysctls(map()) -> map(). -spec build_sysctls(map()) -> map().
@ -286,7 +286,7 @@ build_extra_hosts(Hosts) when is_list(Hosts) ->
end. end.
-spec normalize_expose_port(map()) -> binary(). -spec normalize_expose_port(map()) -> binary().
normalize_expose_port(#{container_port := Port, protocol := Protocol}) -> normalize_expose_port(#{<<"container_port">> := Port, <<"protocol">> := Protocol}) ->
PortBin = integer_to_binary(Port), PortBin = integer_to_binary(Port),
ProtocolBin = to_binary(Protocol), ProtocolBin = to_binary(Protocol),
case ProtocolBin of case ProtocolBin of
@ -316,6 +316,16 @@ add_unique_front([Item | Rest], List) ->
end, end,
add_unique_front(Rest, NList). add_unique_front(Rest, NList).
-spec field(map(), binary(), term()) -> term().
field(Map, Key, Default) when is_map(Map), is_binary(Key) ->
safe_value(maps:get(Key, Map, Default)).
-spec safe_value(term()) -> term().
safe_value(<<"__undefined__">>) ->
undefined;
safe_value(Value) ->
Value.
-spec to_binary(binary() | list() | atom() | any()) -> binary(). -spec to_binary(binary() | list() | atom() | any()) -> binary().
to_binary(Value) when is_binary(Value) -> to_binary(Value) when is_binary(Value) ->
Value; Value;

2
src/docker/docker_deploy_manager.erl
View File

@ -49,7 +49,7 @@ init([]) ->
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}. -spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.
handle_call({deploy, TaskId, Params}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) handle_call({deploy, TaskId, Params}, _From, State = #state{root_dir = RootDir, task_map = TaskMap})
when is_map(Params) -> when is_map(Params) ->
ContainerName = maps:get(container_name, Params), ContainerName = maps:get(<<"container_name">>, Params),
{ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName), {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName),
{ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params), {ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params),
logger:debug("[docker_deploy_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]), logger:debug("[docker_deploy_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]),

4
src/docker/docker_deployer.erl
View File

@ -199,13 +199,13 @@ short_container_id(ContainerId) when is_binary(ContainerId) ->
ContainerId. ContainerId.
-spec deploy_container_name(map()) -> binary(). -spec deploy_container_name(map()) -> binary().
deploy_container_name(#{container_name := ContainerName}) when is_binary(ContainerName) -> deploy_container_name(#{<<"container_name">> := ContainerName}) when is_binary(ContainerName) ->
ContainerName; ContainerName;
deploy_container_name(_) -> deploy_container_name(_) ->
throw({deploy_error, <<"invalid deploy params: container_name missing">>}). throw({deploy_error, <<"invalid deploy params: container_name missing">>}).
-spec deploy_image(map()) -> binary(). -spec deploy_image(map()) -> binary().
deploy_image(#{create := #{config := #{image := Image}}}) when is_binary(Image) -> deploy_image(#{<<"create">> := #{<<"config">> := #{<<"image">> := Image}}}) when is_binary(Image) ->
Image; Image;
deploy_image(_) -> deploy_image(_) ->
throw({deploy_error, <<"invalid deploy params: image missing">>}). throw({deploy_error, <<"invalid deploy params: image missing">>}).

151
src/transport/efka_client.erl
View File

@ -32,7 +32,7 @@
-record(state, { -record(state, {
socket :: undefined | ssl:sslsocket(), socket :: undefined | ssl:sslsocket(),
%% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系 %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
auth_ref = undefined :: undefined | reference(), auth_ref = undefined :: undefined | binary(),
dropped_message_count = 0 :: non_neg_integer() dropped_message_count = 0 :: non_neg_integer()
}). }).
@ -90,7 +90,7 @@ callback_mode() ->
%% 异步发送数据, 连接存在时候直接发送；否则缓存到DETS %% 异步发送数据, 连接存在时候直接发送；否则缓存到DETS
-spec handle_event(term(), term(), atom(), #state{}) -> term(). -spec handle_event(term(), term(), atom(), #state{}) -> term().
handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) -> handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
Packet = term_to_binary({message, {data, #{route_key => RouteKey, metric => Metric}}}), Packet = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}),
case StateName of case StateName of
?STATE_ACTIVATED -> ?STATE_ACTIVATED ->
ok = ssl:send(Socket, Packet), ok = ssl:send(Socket, Packet),
@ -103,12 +103,12 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so
%% Task的stream流，只做实时的 %% Task的stream流，只做实时的
handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]), logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => Type, stream => Stream}}}), Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}),
ok = ssl:send(Socket, Packet), ok = ssl:send(Socket, Packet),
{keep_state, State}; {keep_state, State};
handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => <<"close">>, stream => Reason}}}), Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
ok = ssl:send(Socket, Packet), ok = ssl:send(Socket, Packet),
{keep_state, State}; {keep_state, State};
@ -127,9 +127,10 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro
handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) -> handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
case connect_socket() of case connect_socket() of
{ok, Socket} -> {ok, Socket} ->
Ref = make_ref(), Ref = request_ref(),
AuthPacket = auth_packet(Ref), AuthPacket = auth_packet(Ref),
ok = ssl:send(Socket, AuthPacket), ok = ssl:send(Socket, AuthPacket),
logger:debug("[efka_client] send auth request, ref: ~p", [Ref]),
{next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]}; {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
{error, _Reason} -> {error, _Reason} ->
schedule_reconnect(), schedule_reconnect(),
@ -157,8 +158,16 @@ handle_event(info, flush_cache, _, State) ->
%% 处理收到的ssl消息 %% 处理收到的ssl消息
handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) -> handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
Packet = binary_to_term(PacketBin, [safe]), try binary_to_term(PacketBin, [safe]) of
{keep_state, State, [{next_event, internal, Packet}]}; Packet ->
{keep_state, State, [{next_event, internal, Packet}]}
catch
error:Error ->
logger:warning("[efka_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
disconnect(Socket),
schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}
end;
handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) -> handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) ->
logger:debug("[efka_client] ssl error: ~p", [Reason]), logger:debug("[efka_client] ssl error: ~p", [Reason]),
disconnect(Socket), disconnect(Socket),
@ -171,57 +180,32 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配 %%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
%% 容器管理命令由 iot 发起，使用 command/command_response 语义。 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
handle_event(internal, {command, Ref, {container, #{action := list}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:get_containers(), handle_container_command(Ref, Request, Socket),
send_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
Reply = docker_deploy_manager:deploy(TaskId, Params),
send_container_response(Socket, Ref, Reply),
{keep_state, State};
handle_event(internal, {command, Ref, {container, #{action := start, target := Target}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:start_container(container_target(Target)),
send_container_response(Socket, Ref, Reply),
{keep_state, State};
handle_event(internal, {command, Ref, {container, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
send_container_response(Socket, Ref, Reply),
{keep_state, State};
handle_event(internal, {command, Ref, {container, #{action := kill, target := Target, signal := Signal}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
send_container_response(Socket, Ref, Reply),
{keep_state, State};
handle_event(internal, {command, Ref, {container, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
send_container_response(Socket, Ref, Reply),
{keep_state, State};
handle_event(internal, {command, Ref, {container, #{action := config, target := Target, config := Config}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
send_container_response(Socket, Ref, Reply),
{keep_state, State};
handle_event(internal, {command, Ref, {container, Request}}, _StateName, State = #state{socket = Socket}) ->
logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]), logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]),
send_container_response(Socket, Ref, {error, <<"agent invalid">>}), send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
{keep_state, State}; {keep_state, State};
%% 处理response %% 处理response
handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
logger:debug("[efka_client] auth success"), logger:debug("[efka_client] auth success"),
{next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]}; {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]};
handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) -> handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
logger:debug("[efka_client] auth failed, reason: ~p", [Reason]), logger:debug("[efka_client] auth failed, reason: ~p", [Reason]),
disconnect(Socket), disconnect(Socket),
schedule_reconnect(), schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
handle_event(internal, {response, _Ref, Reply}, StateName, State) -> handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) ->
logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]), logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
{keep_state, State}; {keep_state, State};
handle_event(internal, {command_response, _Ref, Reply}, StateName, State) -> handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) ->
logger:warning("[efka_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]), logger:warning("[efka_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
{keep_state, State}; {keep_state, State};
%% 处理Pub/Sub机制 %% 处理Pub/Sub机制
handle_event(internal, {message, {pub, #{topic := Topic, qos := Qos, content := Content}}}, ?STATE_ACTIVATED, State) -> handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]), logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
efka_subscription:publish(Topic, Qos, Content), efka_subscription:publish(Topic, Qos, Content),
{keep_state, State}; {keep_state, State};
@ -233,10 +217,45 @@ handle_event(info, Info, _, State = #state{}) ->
logger:notice("[efka_client] get unknown info: ~p", [Info]), logger:notice("[efka_client] get unknown info: ~p", [Info]),
{keep_state, State}. {keep_state, State}.
-spec handle_container_command(binary(), term(), ssl:sslsocket()) -> ok.
handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) ->
Reply = docker_commands:get_containers(),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
Reply = docker_deploy_manager:deploy(TaskId, Params),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
Reply = docker_commands:start_container(container_target(Target)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, Request, Socket) ->
logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]),
send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
ok.
-spec terminate(term(), atom(), #state{}) -> ok. -spec terminate(term(), atom(), #state{}) -> ok.
terminate(_Reason, _StateName, _State = #state{socket = Socket}) -> terminate(Reason, _StateName, _State = #state{socket = Socket}) ->
disconnect(Socket), disconnect(Socket),
efka_client_cache:close(), efka_client_cache:close(),
logger:notice("[efka_client] terminate with reason: ~p", [Reason]),
ok. ok.
-spec code_change(term(), atom(), #state{}, term()) -> {ok, atom(), #state{}}. -spec code_change(term(), atom(), #state{}, term()) -> {ok, atom(), #state{}}.
@ -247,17 +266,17 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
%%% Internal functions %%% Internal functions
%%%=================================================================== %%%===================================================================
-spec auth_packet(reference()) -> binary(). -spec auth_packet(binary()) -> binary().
auth_packet(Ref) when is_reference(Ref) -> auth_packet(Ref) when is_binary(Ref) ->
{ok, AuthInfo} = application:get_env(efka, auth), {ok, AuthInfo} = application:get_env(efka, auth),
UUID = proplists:get_value(uuid, AuthInfo), UUID = proplists:get_value(uuid, AuthInfo),
Token = proplists:get_value(token, AuthInfo), Token = proplists:get_value(token, AuthInfo),
Timestamp = efka_util:timestamp(), Timestamp = efka_util:timestamp(),
term_to_binary({request, Ref, {auth_request, #{ term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
uuid => list_to_binary(UUID), <<"uuid">> => list_to_binary(UUID),
token => list_to_binary(Token), <<"token">> => list_to_binary(Token),
timestamp => Timestamp <<"timestamp">> => Timestamp
}}}). }}}).
-spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}. -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
@ -284,15 +303,45 @@ disconnect(Socket) ->
schedule_reconnect() -> schedule_reconnect() ->
erlang:start_timer(5000, self(), create_transport). erlang:start_timer(5000, self(), create_transport).
-spec send_container_response(ssl:sslsocket(), reference(), term()) -> ok. -spec request_ref() -> binary().
request_ref() ->
crypto:strong_rand_bytes(16).
-spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
send_container_response(Socket, Ref, Reply) -> send_container_response(Socket, Ref, Reply) ->
Packet = term_to_binary({command_response, Ref, {container, Reply}}), Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
ok = ssl:send(Socket, Packet). ok = ssl:send(Socket, Packet).
-spec safe_reply(term()) -> term().
safe_reply(ok) ->
<<"ok">>;
safe_reply({ok, Result}) ->
{<<"ok">>, safe_term(Result)};
safe_reply({error, Reason}) ->
{<<"error">>, safe_term(Reason)}.
-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.
-spec container_target(map()) -> binary(). -spec container_target(map()) -> binary().
container_target(Target) when is_map(Target) -> container_target(Target) when is_map(Target) ->
NameBin = to_binary(maps:get(name, Target, <<>>)), NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)),
IdBin = to_binary(maps:get(id, Target, <<>>)), IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)),
case NameBin of case NameBin of
<<>> -> <<>> ->
true = IdBin =/= <<>>, true = IdBin =/= <<>>,