258 lines
7.0 KiB
Markdown
258 lines
7.0 KiB
Markdown
# EFKA 与 IOT 交互协议
|
||
|
||
本文档描述 `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` 使用 `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}` | 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">>`
|
||
|
||
## 鉴权请求
|
||
|
||
初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。
|
||
|
||
```erlang
|
||
{<<"request">>, Ref, {<<"auth_request">>, #{
|
||
<<"uuid">> => UUID,
|
||
<<"token">> => Token,
|
||
<<"timestamp">> => Timestamp
|
||
}}}
|
||
```
|
||
|
||
`iot` 回复：
|
||
|
||
```erlang
|
||
{<<"response">>, Ref, {<<"auth_response">>, <<"ok">>}}
|
||
{<<"response">>, Ref, {<<"auth_response">>, {<<"error">>, {<<"failed">>, Reason}}}}
|
||
```
|
||
|
||
处理语义：
|
||
|
||
- `<<"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 可以继续使用。
|
||
|
||
## 容器管理命令
|
||
|
||
`iot` 对 `efka` 的容器管理使用 command 语义：
|
||
|
||
```erlang
|
||
{<<"command">>, Ref, {<<"container">>, CommandMap}}
|
||
```
|
||
|
||
`efka` 回复：
|
||
|
||
```erlang
|
||
{<<"command_response">>, Ref, {<<"container">>, Reply}}
|
||
```
|
||
|
||
`Reply` 取值：
|
||
|
||
```erlang
|
||
<<"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">>}
|
||
```
|
||
|
||
返回当前 `efka` 主机上的容器列表。
|
||
|
||
### deploy
|
||
|
||
```erlang
|
||
#{
|
||
<<"action">> => <<"deploy">>,
|
||
<<"task_id">> => TaskId,
|
||
<<"params">> => Params
|
||
}
|
||
```
|
||
|
||
触发容器部署。部署过程中的流式日志不通过该 command response 返回，而是通过 `message` 的 `task_event` 上报。
|
||
|
||
### start
|
||
|
||
```erlang
|
||
#{
|
||
<<"action">> => <<"start">>,
|
||
<<"target">> => Target
|
||
}
|
||
```
|
||
|
||
### stop
|
||
|
||
```erlang
|
||
#{
|
||
<<"action">> => <<"stop">>,
|
||
<<"target">> => Target,
|
||
<<"timeout_seconds">> => TimeoutSeconds
|
||
}
|
||
```
|
||
|
||
### kill
|
||
|
||
```erlang
|
||
#{
|
||
<<"action">> => <<"kill">>,
|
||
<<"target">> => Target,
|
||
<<"signal">> => Signal
|
||
}
|
||
```
|
||
|
||
### remove
|
||
|
||
```erlang
|
||
#{
|
||
<<"action">> => <<"remove">>,
|
||
<<"target">> => Target,
|
||
<<"force">> => Force,
|
||
<<"remove_volumes">> => RemoveVolumes
|
||
}
|
||
```
|
||
|
||
### config
|
||
|
||
```erlang
|
||
#{
|
||
<<"action">> => <<"config">>,
|
||
<<"target">> => Target,
|
||
<<"config">> => Config
|
||
}
|
||
```
|
||
|
||
更新容器配置文件。
|
||
|
||
### Target
|
||
|
||
容器目标使用 map 表示：
|
||
|
||
```erlang
|
||
#{
|
||
<<"name">> => ContainerName,
|
||
<<"id">> => ContainerId
|
||
}
|
||
```
|
||
|
||
`name` 和 `id` 至少一个非空；优先使用 `name`，`name` 为空时使用 `id`。
|
||
|
||
## 异步消息
|
||
|
||
`message` 不带 `Ref`，不要求对端回复。
|
||
|
||
### efka -> iot: data
|
||
|
||
```erlang
|
||
{<<"message">>, {<<"data">>, #{
|
||
<<"route_key">> => RouteKey,
|
||
<<"metric">> => Metric
|
||
}}}
|
||
```
|
||
|
||
用于 `efka` 上报业务指标数据。
|
||
|
||
### efka -> iot: task_event
|
||
|
||
```erlang
|
||
{<<"message">>, {<<"task_event">>, #{
|
||
<<"task_id">> => TaskId,
|
||
<<"type">> => Type,
|
||
<<"stream">> => Stream
|
||
}}}
|
||
```
|
||
|
||
任务事件流关闭时：
|
||
|
||
```erlang
|
||
{<<"message">>, {<<"task_event">>, #{
|
||
<<"task_id">> => TaskId,
|
||
<<"type">> => <<"close">>,
|
||
<<"stream">> => Reason
|
||
}}}
|
||
```
|
||
|
||
`task_event` 本身不携带 `uuid`。`iot` 在接收该消息时使用当前已鉴权 `ssl_channel` 绑定的 host UUID，把事件路由到内部任务进程 `{UUID, TaskId}`。HTTP 页面通过 SSE 订阅时也必须使用同一组参数：
|
||
|
||
```http
|
||
GET /event_stream?uuid=<host_uuid>&task_id=<task_id>
|
||
```
|
||
|
||
`iot` 会为每个 `{UUID, TaskId}` 维护一个独立的任务进程，用于缓存最近的部署日志、支持多个 SSE listener，并在收到 close 事件后结束事件流。
|
||
|
||
### iot -> efka: pub
|
||
|
||
```erlang
|
||
{<<"message">>, {<<"pub">>, #{
|
||
<<"topic">> => Topic,
|
||
<<"qos">> => Qos,
|
||
<<"content">> => Content
|
||
}}}
|
||
```
|
||
|
||
用于 `iot` 向 `efka` 本地订阅系统发布 topic 消息。
|
||
|
||
## 状态与超时
|
||
|
||
- `efka` 鉴权超时时间：5 秒。
|
||
- `iot` command inflight 超时时间：60 秒。
|
||
- `iot` 管理多个 `efka` 时，每个连接有独立 `ssl_channel` 和独立 inflight 表。
|
||
- command 超时后，`iot` 删除 inflight 记录；之后如果迟到的 `command_response` 到达，会被视为未预期响应。
|
||
|
||
## UDP 心跳
|
||
|
||
`efka` 通过独立的 `efka_heartbeat` 进程向 `iot` 发送 UDP 心跳。TLS control channel 和 UDP 心跳共用 `iot_server.host`，分别使用 `tls_port` 和 `udp_port`。UDP 心跳包使用 HMAC-SHA256 校验，HMAC key 为 `SHA256(auth.token)`。
|
||
|
||
详细格式见 [heartbeat.md](heartbeat.md)。
|
||
|
||
## 兼容性
|
||
|
||
当前协议不兼容旧 tuple：
|
||
|
||
- 旧容器管理：`{request, Ref, {container_request, ...}}`
|
||
- 旧容器回复：`{response, Ref, {container_response, ...}}`
|
||
- 旧授权控制：`{message, {auth_control, Command}}`
|
||
- 已移除的 auth command：`{command, Ref, {auth, activate | deactivate}}`
|
||
- 旧 Ref：Erlang `reference()`，例如 `make_ref()` 生成的值。
|
||
|
||
如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。
|