2026-05-08 00:05:54 +08:00

5.3 KiB

Raw Blame History

EFKA 与 IOT 交互协议

本文档描述 efka 与 iot 之间的 TLS 长连接协议。当前协议由 Erlang term 直接序列化，发送端使用 term_to_binary/1，接收端使用 binary_to_term(PacketBin, [safe])。

传输层

efka 作为 TLS client 连接 iot。
iot 作为 TLS server 接收多个 efka 连接，一个连接对应一个 ssl_channel 进程。
socket 使用 {packet, 4}，每个 Erlang term binary 作为一个完整包发送。
Ref 使用 make_ref() 生成，只在当前连接的 inflight 表内匹配。

顶层帧

协议顶层 tuple 用来表达交互语义：

{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 表示业务域，目前支持：

auth
container

鉴权请求

初始连接由 efka 发起鉴权 request：

{request, Ref, {auth_request, #{
    uuid => UUID,
    token => Token,
    timestamp => Timestamp
}}}

iot 回复：

{response, Ref, {auth_response, ok}}
{response, Ref, {auth_response, {error, {denied, Reason}}}}
{response, Ref, {auth_response, {error, {failed, Reason}}}}

处理语义：

ok：efka 进入 activated 状态。
{error, {denied, Reason}}：efka 进入 restricted 状态，不能正常上报数据，但仍可接收部分命令。
{error, {failed, Reason}}：鉴权失败，连接关闭后重连。

授权控制命令

iot 对 efka 的授权控制使用 command 语义：

{command, Ref, {auth, activate}}
{command, Ref, {auth, deactivate}}

efka 回复：

{command_response, Ref, {auth, ok}}
{command_response, Ref, {auth, {error, Reason}}}

处理语义：

activate：如果 efka 已经是 activated，直接回复 ok；否则重新发送 auth_request，等待鉴权结果后再回复该 command。
deactivate：efka 进入 restricted 状态，并回复 ok。
iot 侧会异步提交 auth command 并等待对应 Ref 的 command_response，再向 /host/activate HTTP 调用方返回结果；等待超时为 10 秒，超时返回 timeout，无效响应返回 invalid response。

容器管理命令

iot 对 efka 的容器管理使用 command 语义：

{command, Ref, {container, CommandMap}}

efka 回复：

{command_response, Ref, {container, Reply}}

Reply 取值：

ok
{ok, Result}
{error, Reason}

list

#{action => list}

返回当前 efka 主机上的容器列表。

deploy

#{
    action => deploy,
    task_id => TaskId,
    params => Params
}

触发容器部署。部署过程中的流式日志不通过该 command response 返回，而是通过 message 的 task_event 上报。

start

#{
    action => start,
    target => Target
}

stop

#{
    action => stop,
    target => Target,
    timeout_seconds => TimeoutSeconds
}

kill

#{
    action => kill,
    target => Target,
    signal => Signal
}

remove

#{
    action => remove,
    target => Target,
    force => Force,
    remove_volumes => RemoveVolumes
}

config

#{
    action => config,
    target => Target,
    config => Config
}

更新容器配置文件。

Target

容器目标使用 map 表示：

#{
    name => ContainerName,
    id => ContainerId
}

name 和 id 至少一个非空；优先使用 name，name 为空时使用 id。

异步消息

message 不带 Ref，不要求对端回复。

efka -> iot: data

{message, {data, #{
    route_key => RouteKey,
    metric => Metric
}}}

用于 efka 上报业务指标数据。

efka -> iot: task_event

{message, {task_event, #{
    task_id => TaskId,
    type => Type,
    stream => Stream
}}}

任务事件流关闭时：

{message, {task_event, #{
    task_id => TaskId,
    type => <<"close">>,
    stream => Reason
}}}

iot -> efka: pub

{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 到达，会被视为未预期响应。

兼容性

当前协议不兼容旧 tuple：

旧容器管理：{request, Ref, {container_request, ...}}
旧容器回复：{response, Ref, {container_response, ...}}
旧授权控制：{message, {auth_control, Command}}

如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。

5.3 KiB Raw Blame History Unescape Escape