This commit is contained in:
anlicheng 2026-05-09 22:49:58 +08:00
parent d506ec98d5
commit c4d17b9024
2 changed files with 146 additions and 131 deletions

View File

@ -15,31 +15,41 @@
`iot` 通过 TLS 长连接向 `efka` 下发容器命令:
```erlang
{command, Ref, {container, CommandMap}}
{<<"command">>, Ref, {<<"container">>, CommandMap}}
```
`efka` 执行后回复:
```erlang
{command_response, Ref, {container, Reply}}
{<<"command_response">>, Ref, {<<"container">>, Reply}}
```
`Reply` 取值:
```erlang
ok
{ok, Result}
{error, Reason}
<<"ok">>
{<<"ok">>, Result}
{<<"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
`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
```erlang
#{action => list}
#{<<"action">> => <<"list">>}
```
执行:
@ -58,9 +68,9 @@ GET /containers/json?all=true
```erlang
#{
action => deploy,
task_id => TaskId,
params => Params
<<"action">> => <<"deploy">>,
<<"task_id">> => TaskId,
<<"params">> => Params
}
```
@ -76,8 +86,8 @@ docker_deploy_manager:deploy(TaskId, Params)
```erlang
#{
action => start,
target => Target
<<"action">> => <<"start">>,
<<"target">> => Target
}
```
@ -97,9 +107,9 @@ POST /containers/{name_or_id}/start
```erlang
#{
action => stop,
target => Target,
timeout_seconds => TimeoutSeconds
<<"action">> => <<"stop">>,
<<"target">> => Target,
<<"timeout_seconds">> => TimeoutSeconds
}
```
@ -119,9 +129,9 @@ POST /containers/{name_or_id}/stop?t={TimeoutSeconds}
```erlang
#{
action => kill,
target => Target,
signal => Signal
<<"action">> => <<"kill">>,
<<"target">> => Target,
<<"signal">> => Signal
}
```
@ -135,10 +145,10 @@ docker_commands:kill_container(ContainerNameOrId, Signal)
```erlang
#{
action => remove,
target => Target,
force => Force,
remove_volumes => RemoveVolumes
<<"action">> => <<"remove">>,
<<"target">> => Target,
<<"force">> => Force,
<<"remove_volumes">> => RemoveVolumes
}
```
@ -152,9 +162,9 @@ docker_commands:remove_container(ContainerNameOrId, Force, RemoveVolumes)
```erlang
#{
action => config,
target => Target,
config => Config
<<"action">> => <<"config">>,
<<"target">> => Target,
<<"config">> => Config
}
```
@ -172,8 +182,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
```erlang
#{
name => ContainerName,
id => ContainerId
<<"name">> => ContainerName,
<<"id">> => ContainerId
}
```
@ -189,8 +199,8 @@ docker_helper:update_container_config(ContainerNameOrId, iolist_to_binary(Config
```erlang
#{
container_name => ContainerName,
create => Create
<<"container_name">> => ContainerName,
<<"create">> => Create
}
```
@ -215,9 +225,9 @@ efka.root_dir/container_name/
```erlang
#{
config => ContainerConfig,
host_config => HostConfig,
networking_config => NetworkingConfig
<<"config">> => ContainerConfig,
<<"host_config">> => HostConfig,
<<"networking_config">> => NetworkingConfig
}
```
@ -340,8 +350,8 @@ Body = iolist_to_binary(json:encode(Options))
```erlang
[
#{container_port => 80, protocol => <<"tcp">>},
#{container_port => 53, protocol => <<"udp">>}
#{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>},
#{<<"container_port">> => 53, <<"protocol">> => <<"udp">>}
]
```
@ -369,10 +379,10 @@ Body = iolist_to_binary(json:encode(Options))
```erlang
#{
test => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
interval_ns => 30000000000,
timeout_ns => 10000000000,
retries => 3
<<"test">> => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
<<"interval_ns">> => 30000000000,
<<"timeout_ns">> => 10000000000,
<<"retries">> => 3
}
```
@ -440,7 +450,7 @@ efka 自动补丁后输出:
输入:
```erlang
#{name => <<"always">>, maximum_retry_count => 0}
#{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0}
```
输出:
@ -458,7 +468,7 @@ efka 自动补丁后输出:
输入:
```erlang
#{name => <<"on-failure">>, maximum_retry_count => 3}
#{<<"name">> => <<"on-failure">>, <<"maximum_retry_count">> => 3}
```
输出:
@ -481,9 +491,9 @@ efka 自动补丁后输出:
```erlang
[
#{
path_on_host => <<"/dev/ttyUSB0">>,
path_in_container => <<"/dev/ttyUSB0">>,
cgroup_permissions => <<"rwm">>
<<"path_on_host">> => <<"/dev/ttyUSB0">>,
<<"path_in_container">> => <<"/dev/ttyUSB0">>,
<<"cgroup_permissions">> => <<"rwm">>
}
]
```
@ -512,10 +522,10 @@ efka 自动补丁后输出:
```erlang
#{
memory => 536870912,
memory_reservation => 268435456,
nano_cpus => 1500000000,
cpu_shares => 512
<<"memory">> => 536870912,
<<"memory_reservation">> => 268435456,
<<"nano_cpus">> => 1500000000,
<<"cpu_shares">> => 512
}
```
@ -540,7 +550,7 @@ efka 自动补丁后输出:
```erlang
[
#{name => <<"nofile">>, soft => 1024, hard => 2048}
#{<<"name">> => <<"nofile">>, <<"soft">> => 1024, <<"hard">> => 2048}
]
```
@ -628,9 +638,9 @@ efka 自动补丁后输出:
```erlang
#{
endpoints => [
#{name => <<"bridge">>},
#{name => <<"mynet">>}
<<"endpoints">> => [
#{<<"name">> => <<"bridge">>},
#{<<"name">> => <<"mynet">>}
]
}
```
@ -661,26 +671,26 @@ efka 自动补丁后输出:
收到的 deploy command
```erlang
{command, Ref, {container, #{
action => deploy,
task_id => 1001,
params => #{
container_name => <<"my_nginx">>,
create => #{
config => #{
image => <<"docker.io/library/nginx:latest">>,
cmd => [<<"nginx">>, <<"-g">>, <<"daemon off;">>],
env => [<<"ENV=prod">>],
volumes => [<<"/data">>],
exposed_ports => [#{container_port => 80, protocol => <<"tcp">>}]
{<<"command">>, Ref, {<<"container">>, #{
<<"action">> => <<"deploy">>,
<<"task_id">> => 1001,
<<"params">> => #{
<<"container_name">> => <<"my_nginx">>,
<<"create">> => #{
<<"config">> => #{
<<"image">> => <<"docker.io/library/nginx:latest">>,
<<"cmd">> => [<<"nginx">>, <<"-g">>, <<"daemon off;">>],
<<"env">> => [<<"ENV=prod">>],
<<"volumes">> => [<<"/data">>],
<<"exposed_ports">> => [#{<<"container_port">> => 80, <<"protocol">> => <<"tcp">>}]
},
host_config => #{
binds => [<<"/host/data:/data">>],
restart_policy => #{name => <<"always">>, maximum_retry_count => 0},
memory => 536870912
<<"host_config">> => #{
<<"binds">> => [<<"/host/data:/data">>],
<<"restart_policy">> => #{<<"name">> => <<"always">>, <<"maximum_retry_count">> => 0},
<<"memory">> => 536870912
},
networking_config => #{
endpoints => [#{name => <<"bridge">>}]
<<"networking_config">> => #{
<<"endpoints">> => [#{<<"name">> => <<"bridge">>}]
}
}
}

View File

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