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

Compare commits

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

No commits in common. "ba7c2b8ecd31b0aa4a1dd4c27530fd1f0ac2c750" and "d506ec98d5cde5a042696e9e222c28f98c81e930" have entirely different histories.
ba7c2b8ecd ... d506ec98d5

7 changed files with 237 additions and 311 deletions
Show Stats Expand all files Collapse all files

148
docs/container_command_docker_json.md
View File

@ -15,41 +15,31 @@
`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}
``` ```
`Ref` 是 `crypto:strong_rand_bytes(16)` 生成的 16 字节 binary。网络帧只使用 `binary_to_term(PacketBin, [safe])` 可解码的 safe term；协议 label、业务 label、map key 和 action 都使用 binary。 只有 `efka_client` 处于 `activated` 状态时，容器命令才会正常执行；处于 `restricted` 或其他状态时会返回错误。
只有 `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}
``` ```
执行： 执行：
@ -68,9 +58,9 @@ GET /containers/json?all=true
```erlang ```erlang
#{ #{
<<"action">> => <<"deploy">>, action => deploy,
<<"task_id">> => TaskId, task_id => TaskId,
<<"params">> => Params params => Params
} }
``` ```
@ -86,8 +76,8 @@ docker_deploy_manager:deploy(TaskId, Params)
```erlang ```erlang
#{ #{
<<"action">> => <<"start">>, action => start,
<<"target">> => Target target => Target
} }
``` ```
@ -107,9 +97,9 @@ POST /containers/{name_or_id}/start
```erlang ```erlang
#{ #{
<<"action">> => <<"stop">>, action => stop,
<<"target">> => Target, target => Target,
<<"timeout_seconds">> => TimeoutSeconds timeout_seconds => TimeoutSeconds
} }
``` ```
@ -129,9 +119,9 @@ POST /containers/{name_or_id}/stop?t={TimeoutSeconds}
```erlang ```erlang
#{ #{
<<"action">> => <<"kill">>, action => kill,
<<"target">> => Target, target => Target,
<<"signal">> => Signal signal => Signal
} }
``` ```
@ -145,10 +135,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
} }
``` ```
@ -162,9 +152,9 @@ docker_commands:remove_container(ContainerNameOrId, Force, RemoveVolumes)
```erlang ```erlang
#{ #{
<<"action">> => <<"config">>, action => config,
<<"target">> => Target, target => Target,
<<"config">> => Config config => Config
} }
``` ```
@ -182,8 +172,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
```erlang ```erlang
#{ #{
<<"name">> => ContainerName, name => ContainerName,
<<"id">> => ContainerId id => ContainerId
} }
``` ```
@ -199,8 +189,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
```erlang ```erlang
#{ #{
<<"container_name">> => ContainerName, container_name => ContainerName,
<<"create">> => Create create => Create
} }
``` ```
@ -225,9 +215,9 @@ efka.root_dir/container_name/
```erlang ```erlang
#{ #{
<<"config">> => ContainerConfig, config => ContainerConfig,
<<"host_config">> => HostConfig, host_config => HostConfig,
<<"networking_config">> => NetworkingConfig networking_config => NetworkingConfig
} }
``` ```
@ -350,8 +340,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">>}
] ]
``` ```
@ -379,10 +369,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
} }
``` ```
@ -450,7 +440,7 @@ efka 自动补丁后输出：
输入： 输入：
```erlang ```erlang
#{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0} #{name => <<"always">>, maximum_retry_count => 0}
``` ```
输出： 输出：
@ -468,7 +458,7 @@ efka 自动补丁后输出：
输入： 输入：
```erlang ```erlang
#{<<"name">> => <<"on-failure">>, <<"maximum_retry_count">> => 3} #{name => <<"on-failure">>, maximum_retry_count => 3}
``` ```
输出： 输出：
@ -491,9 +481,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">>
} }
] ]
``` ```
@ -522,10 +512,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
} }
``` ```
@ -550,7 +540,7 @@ efka 自动补丁后输出：
```erlang ```erlang
[ [
#{<<"name">> => <<"nofile">>, <<"soft">> => 1024, <<"hard">> => 2048} #{name => <<"nofile">>, soft => 1024, hard => 2048}
] ]
``` ```
@ -638,9 +628,9 @@ efka 自动补丁后输出：
```erlang ```erlang
#{ #{
<<"endpoints">> => [ endpoints => [
#{<<"name">> => <<"bridge">>}, #{name => <<"bridge">>},
#{<<"name">> => <<"mynet">>} #{name => <<"mynet">>}
] ]
} }
``` ```
@ -671,26 +661,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,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
}}} }}}
``` ```
@ -252,6 +248,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()` 生成的值。
如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。 如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。

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(field(Create0, <<"config">>, undefined))), Config = patch_container_config(ContainerName, ensure_container_config(maps:get(config, Create0, undefined))),
HostConfig = patch_host_config(ConfigFile, ensure_host_config(field(Create0, <<"host_config">>, undefined))), HostConfig = patch_host_config(ConfigFile, ensure_host_config(maps:get(host_config, Create0, 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 <- field(Config0, <<"env">>, [])], Env0 = [to_binary(EnvItem) || EnvItem <- maps:get(env, Config0, [])],
Volumes0 = [to_binary(Volume) || Volume <- field(Config0, <<"volumes">>, [])], Volumes0 = [to_binary(Volume) || Volume <- maps:get(volumes, Config0, [])],
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 <- field(HostConfig0, <<"binds">>, [])], Binds0 = [to_binary(Bind) || Bind <- maps:get(binds, HostConfig0, [])],
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 = field(Create0, <<"config">>, #{}), Config = maps:get(config, Create0, #{}),
HostConfig = field(Create0, <<"host_config">>, #{}), HostConfig = maps:get(host_config, Create0, #{}),
Endpoints = networking_config_endpoints(field(Create0, <<"networking_config">>, undefined)), Endpoints = networking_config_endpoints(maps:get(networking_config, Create0, undefined)),
#{ #{
<<"Image">> => to_binary(field(Config, <<"image">>, <<>>)), <<"Image">> => to_binary(maps:get(image, Config, <<>>)),
<<"Cmd">> => [to_binary(CommandItem) || CommandItem <- field(Config, <<"cmd">>, [])], <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- maps:get(cmd, Config, [])],
<<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- field(Config, <<"entrypoint">>, [])], <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- maps:get(entrypoint, Config, [])],
<<"Env">> => [to_binary(EnvItem) || EnvItem <- field(Config, <<"env">>, [])], <<"Env">> => [to_binary(EnvItem) || EnvItem <- maps:get(env, Config, [])],
<<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(field(Config, <<"labels">>, #{}))]), <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(maps:get(labels, Config, #{}))]),
<<"Volumes">> => build_volumes(field(Config, <<"volumes">>, [])), <<"Volumes">> => build_volumes(maps:get(volumes, Config, [])),
<<"User">> => to_binary(field(Config, <<"user">>, <<>>)), <<"User">> => to_binary(maps:get(user, Config, <<>>)),
<<"WorkingDir">> => to_binary(field(Config, <<"working_dir">>, <<>>)), <<"WorkingDir">> => to_binary(maps:get(working_dir, Config, <<>>)),
<<"Hostname">> => to_binary(field(Config, <<"hostname">>, <<>>)), <<"Hostname">> => to_binary(maps:get(hostname, Config, <<>>)),
<<"ExposedPorts">> => build_expose(field(Config, <<"exposed_ports">>, [])), <<"ExposedPorts">> => build_expose(maps:get(exposed_ports, Config, [])),
<<"NetworkingConfig">> => build_networking_config(Endpoints), <<"NetworkingConfig">> => build_networking_config(Endpoints),
<<"Healthcheck">> => build_healthcheck(field(Config, <<"healthcheck">>, undefined)), <<"Healthcheck">> => build_healthcheck(maps:get(healthcheck, Config, undefined)),
<<"HostConfig">> => fold_merge([ <<"HostConfig">> => fold_merge([
build_binds(field(HostConfig, <<"binds">>, [])), build_binds(maps:get(binds, HostConfig, [])),
build_network_mode(field(HostConfig, <<"network_mode">>, <<>>)), build_network_mode(maps:get(network_mode, HostConfig, <<>>)),
build_restart(field(HostConfig, <<"restart_policy">>, undefined)), build_restart(maps:get(restart_policy, HostConfig, undefined)),
build_privileged(field(HostConfig, <<"privileged">>, false)), build_privileged(maps:get(privileged, HostConfig, false)),
build_cap_add_drop(field(HostConfig, <<"cap_add">>, []), field(HostConfig, <<"cap_drop">>, [])), build_cap_add_drop(maps:get(cap_add, HostConfig, []), maps:get(cap_drop, HostConfig, [])),
build_devices(field(HostConfig, <<"devices">>, [])), build_devices(maps:get(devices, HostConfig, [])),
build_resources( build_resources(
field(HostConfig, <<"memory">>, 0), maps:get(memory, HostConfig, 0),
field(HostConfig, <<"memory_reservation">>, 0), maps:get(memory_reservation, HostConfig, 0),
field(HostConfig, <<"nano_cpus">>, 0), maps:get(nano_cpus, HostConfig, 0),
field(HostConfig, <<"cpu_shares">>, 0) maps:get(cpu_shares, HostConfig, 0)
), ),
build_ulimits(field(HostConfig, <<"ulimits">>, [])), build_ulimits(maps:get(ulimits, HostConfig, [])),
build_tmpfs(field(HostConfig, <<"tmpfs">>, #{})), build_tmpfs(maps:get(tmpfs, HostConfig, #{})),
build_sysctls(field(HostConfig, <<"sysctls">>, #{})), build_sysctls(maps:get(sysctls, HostConfig, #{})),
build_extra_hosts(field(HostConfig, <<"extra_hosts">>, [])) build_extra_hosts(maps:get(extra_hosts, HostConfig, []))
]) ])
}. }.
@ -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) ->
field(NetworkingConfig, <<"endpoints">>, []). maps:get(endpoints, NetworkingConfig, []).
-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 <- field(Healthcheck, <<"test">>, [])], <<"Test">> => [to_binary(Item) || Item <- maps:get(test, Healthcheck, [])],
<<"Interval">> => field(Healthcheck, <<"interval_ns">>, 0), <<"Interval">> => maps:get(interval_ns, Healthcheck, 0),
<<"Timeout">> => field(Healthcheck, <<"timeout_ns">>, 0), <<"Timeout">> => maps:get(timeout_ns, Healthcheck, 0),
<<"Retries">> => field(Healthcheck, <<"retries">>, 0) <<"Retries">> => maps:get(retries, Healthcheck, 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(field(RestartPolicy0, <<"name">>, <<>>)) <<"Name">> => to_binary(maps:get(name, RestartPolicy0, <<>>))
}, },
case field(RestartPolicy0, <<"maximum_retry_count">>, 0) of case maps:get(maximum_retry_count, RestartPolicy0, 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,16 +316,6 @@ 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 | binary(), auth_ref = undefined :: undefined | reference(),
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,10 +127,9 @@ 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 = request_ref(), Ref = make_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(),
@ -158,16 +157,8 @@ 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) ->
try binary_to_term(PacketBin, [safe]) of Packet = binary_to_term(PacketBin, [safe]),
Packet -> {keep_state, State, [{next_event, internal, 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),
@ -180,32 +171,57 @@ 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">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(internal, {command, Ref, {container, #{action := list}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
handle_container_command(Ref, Request, Socket), Reply = docker_commands:get_containers(),
send_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) -> handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}}, ?STATE_ACTIVATED, 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};
@ -217,45 +233,10 @@ 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{}}.
@ -266,17 +247,17 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
%%% Internal functions %%% Internal functions
%%%=================================================================== %%%===================================================================
-spec auth_packet(binary()) -> binary(). -spec auth_packet(reference()) -> binary().
auth_packet(Ref) when is_binary(Ref) -> auth_packet(Ref) when is_reference(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()}.
@ -303,45 +284,15 @@ disconnect(Socket) ->
schedule_reconnect() -> schedule_reconnect() ->
erlang:start_timer(5000, self(), create_transport). erlang:start_timer(5000, self(), create_transport).
-spec request_ref() -> binary(). -spec send_container_response(ssl:sslsocket(), reference(), term()) -> ok.
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">>, safe_reply(Reply)}}), Packet = term_to_binary({command_response, Ref, {container, 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 =/= <<>>,