fix

docs/efka_iot_protocol.md

@@ -0,0 +1,255 @@
+# 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 用来表达交互语义：
+
+```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` 表示业务域，目前支持：
+
+- `auth`
+- `container`
+
+## 鉴权请求
+
+初始连接由 `efka` 发起鉴权 request：
+
+```erlang
+{request, Ref, {auth_request, #{
+    uuid => UUID,
+    token => Token,
+    timestamp => Timestamp
+}}}
+```
+
+`iot` 回复：
+
+```erlang
+{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 语义：
+
+```erlang
+{command, Ref, {auth, activate}}
+{command, Ref, {auth, deactivate}}
+```
+
+`efka` 回复：
+
+```erlang
+{command_response, Ref, {auth, ok}}
+{command_response, Ref, {auth, {error, Reason}}}
+```
+
+处理语义：
+
+- `activate`：如果 `efka` 已经是 `activated`，直接回复 `ok`；否则重新发送 `auth_request`，等待鉴权结果后再回复该 command。
+- `deactivate`：`efka` 进入 `restricted` 状态，并回复 `ok`。
+
+## 容器管理命令
+
+`iot` 对 `efka` 的容器管理使用 command 语义：
+
+```erlang
+{command, Ref, {container, CommandMap}}
+```
+
+`efka` 回复：
+
+```erlang
+{command_response, Ref, {container, Reply}}
+```
+
+`Reply` 取值：
+
+```erlang
+ok
+{ok, Result}
+{error, Reason}
+```
+
+### 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
+}}}
+```
+
+### 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` 到达，会被视为未预期响应。
+
+## 兼容性
+
+当前协议不兼容旧 tuple：
+
+- 旧容器管理：`{request, Ref, {container_request, ...}}`
+- 旧容器回复：`{response, Ref, {container_response, ...}}`
+- 旧授权控制：`{message, {auth_control, Command}}`
+
+如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。

src/host/iot_host.erl

@@ -128,14 +128,14 @@ remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName)
     ok | {ok, Result :: term()} | {error, Reason :: term()}.
 await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) ->
     receive
-        {request_reply, Ref, ok} ->
+        {command_reply, Ref, ok} ->
             ok;
-        {request_reply, Ref, {ok, Result}} ->
+        {command_reply, Ref, {ok, Result}} ->
             {ok, Result};
-        {request_reply, Ref, {error, Reason}} ->
+        {command_reply, Ref, {error, Reason}} ->
             {error, Reason}
     after Timeout ->
-        ok = gen_statem:call(Pid, {cancel_request_call, Ref}),
+        ok = gen_statem:call(Pid, {cancel_command_call, Ref}),
         flush_reply(Ref),
         {error, timeout}
     end.
@@ -217,13 +217,13 @@ handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #s
             {keep_state, State, [{reply, From, {ok, Ref}}]};
         false ->
             logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]),
-            {keep_state, State, [{reply, From, {error, <<"主机离线，发送请求失败"/utf8>>}}]}
+            {keep_state, State, [{reply, From, {error, <<"主机离线，发送命令失败"/utf8>>}}]}
     end;
 
-handle_event({call, From}, {cancel_request_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
+handle_event({call, From}, {cancel_command_call, Ref}, _, State = #state{channel_pid = ChannelPid}) ->
     case is_pid(ChannelPid) of
         true ->
-            ok = ssl_channel:cancel_request_call(ChannelPid, Ref),
+            ok = ssl_channel:cancel_command_call(ChannelPid, Ref),
             {keep_state, State, [{reply, From, ok}]};
         false ->
             {keep_state, State, [{reply, From, ok}]}
@@ -234,7 +234,7 @@ handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State =
     case HasSession andalso is_pid(ChannelPid) of
         true ->
             logger:debug("[iot_host] host: ~p, publish to topic: ~p, content: ~p", [UUID, Topic, Content]),
-            %% 通过websocket发送请求
+            %% 通过websocket发送消息
             ssl_channel:pub(ChannelPid, Topic, Qos, Content),
 
             {keep_state, State, [{reply, From, ok}]};
@@ -375,7 +375,7 @@ state_map(#state{host_id = HostId, uuid = UUID, has_session = HasSession, heartb
 
 flush_reply(Ref) ->
     receive
-        {request_reply, Ref, _Reply} ->
+        {command_reply, Ref, _Reply} ->
             ok
     after 0 ->
         ok

src/transport/tcp/ssl_channel.erl

@@ -13,7 +13,7 @@
 -define(INFLIGHT_TIMEOUT, 60000).
 
 %% API
--export([pub/4, container_call/3, cancel_request_call/2, command/2, activate/2]).
+-export([pub/4, container_call/3, cancel_command_call/2, command/2, activate/2]).
 
 -export([start_link/3, stop/2]).
 %% gen_server callbacks
@@ -27,12 +27,12 @@
     %% 用户进程id
     host_pid = undefined,
 
-    %% 请求响应的对应关系
+    %% iot 发起的 command 与 command_response 的对应关系
     inflight = #{}
 }).
 
--record(inflight_request, {
+-record(inflight_command, {
-    receiver_pid :: pid(),
+    receiver_pid :: undefined | pid(),
     timer_ref :: reference()
 }).
 
@@ -54,12 +54,12 @@ activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) ->
 -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: map()) -> Ref :: reference().
 container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_map(Request) ->
     Ref = make_ref(),
-    gen_server:cast(Pid, {request_call, ReceiverPid, Ref, {container_request, Request}}),
+    gen_server:cast(Pid, {command_call, ReceiverPid, Ref, {container, Request}}),
     Ref.
 
--spec cancel_request_call(Pid :: pid(), Ref :: reference()) -> ok.
+-spec cancel_command_call(Pid :: pid(), Ref :: reference()) -> ok.
-cancel_request_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
+cancel_command_call(Pid, Ref) when is_pid(Pid), is_reference(Ref) ->
-    gen_server:call(Pid, {cancel_request_call, Ref}).
+    gen_server:call(Pid, {cancel_command_call, Ref}).
 
 %% 关闭方法
 -spec stop(Pid :: pid(), Reason :: any()) -> no_return().
@@ -83,9 +83,9 @@ init(Ref, Transport, _Opts = []) ->
     % erlang:start_timer(?PING_TICKER, self(), ping_ticker),
     gen_server:enter_loop(?MODULE, [], #state{transport = Transport, socket = Socket}).
 
-handle_call({cancel_request_call, Ref}, _From, State = #state{inflight = Inflight}) ->
+handle_call({cancel_command_call, Ref}, _From, State = #state{inflight = Inflight}) ->
     case maps:take(Ref, Inflight) of
-        {#inflight_request{timer_ref = TimerRef}, NInflight} ->
+        {#inflight_command{timer_ref = TimerRef}, NInflight} ->
             erlang:cancel_timer(TimerRef),
             {reply, ok, State#state{inflight = NInflight}};
         error ->
@@ -101,25 +101,30 @@ handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, so
     {noreply, State};
 
 %% 发送Command消息
-handle_cast({command, Command}, State = #state{transport = Transport, socket = Socket}) ->
+handle_cast({command, Command}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
-    Packet = term_to_binary({message, {auth_control, Command}}),
+    Ref = make_ref(),
-    Transport:send(Socket, Packet),
+    Packet = term_to_binary({command, Ref, {auth, Command}}),
-    {noreply, State};
-
-%% 推送需要响应的请求
-handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
-    Packet = term_to_binary({request, Ref, Body}),
     Transport:send(Socket, Packet),
 
-    TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {request_timeout, Ref}),
+    TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}),
-    RequestInfo = #inflight_request{receiver_pid = ReceiverPid, timer_ref = TimerRef},
+    CommandInfo = #inflight_command{receiver_pid = undefined, timer_ref = TimerRef},
 
-    {noreply, State#state{inflight = maps:put(Ref, RequestInfo, Inflight)}}.
+    {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}};
 
-handle_info({timeout, TimerRef, {request_timeout, Ref}}, State = #state{inflight = Inflight}) ->
+%% iot 请求 efka 时使用 command/command_response 语义。
+handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) ->
+    Packet = term_to_binary({command, Ref, Body}),
+    Transport:send(Socket, Packet),
+
+    TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}),
+    CommandInfo = #inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef},
+
+    {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}}.
+
+handle_info({timeout, TimerRef, {command_timeout, Ref}}, State = #state{inflight = Inflight}) ->
     case maps:get(Ref, Inflight, undefined) of
-        #inflight_request{timer_ref = TimerRef} ->
+        #inflight_command{timer_ref = TimerRef} ->
-            logger:warning("[ws_channel] request timeout, ref: ~p", [Ref]),
+            logger:warning("[ws_channel] command timeout, ref: ~p", [Ref]),
             {noreply, State#state{inflight = maps:remove(Ref, Inflight)}};
         _ ->
             {noreply, State}
@@ -140,8 +145,8 @@ handle_info({ssl, Socket, PacketBin}, State = #state{transport = Transport, sock
             handle_request_frame(Ref, Body, Transport, Socket, State);
         {message, Body} ->
             handle_message_frame(Body, HostPid, State);
-        {response, Ref, Response} ->
+        {command_response, Ref, Response} ->
-            handle_response_frame(Ref, Response, Inflight, State);
+            handle_command_response_frame(Ref, Response, Inflight, State);
         Other ->
             logger:warning("[ssl_channel] unsupported packet: ~p", [Other]),
             {stop, bad_packet, State}
@@ -213,8 +218,8 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timesta
             logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]),
             {stop, Reason, State}
     end;
-handle_request_frame(Ref, {container_request, ContainerRequest}, _Transport, _Socket, State) ->
+handle_request_frame(Ref, {container, ContainerCommand}, _Transport, _Socket, State) ->
-    logger:warning("[ws_channel] unsupported request message type: container_request, ref: ~p, request: ~p", [Ref, ContainerRequest]),
+    logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]),
     {stop, normal, State};
 handle_request_frame(Ref, Body, _Transport, _Socket, State) ->
     logger:warning("[ws_channel] unsupported request body, ref: ~p, body: ~p", [Ref, Body]),
@@ -239,24 +244,29 @@ handle_event_stream_frame(#{task_id := TaskId, type := Type, stream := Stream})
     logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]),
     iot_event_stream_observer:stream_data(TaskId, Type, Stream).
 
--spec handle_response_frame(reference(), tuple(), map(), #state{}) ->
+-spec handle_command_response_frame(reference(), tuple(), map(), #state{}) ->
     {noreply, #state{}}.
-handle_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref) ->
+handle_command_response_frame(Ref, Reply, Inflight, State) when is_reference(Ref) ->
     case maps:take(Ref, Inflight) of
         error ->
             {noreply, State};
-        {#inflight_request{receiver_pid = ReceiverPid, timer_ref = TimerRef}, NInflight} ->
+        {#inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef}, NInflight} ->
             erlang:cancel_timer(TimerRef),
-            case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
+            case ReceiverPid of
+                undefined ->
+                    ok;
+                _ when is_pid(ReceiverPid) ->
+                    case is_process_alive(ReceiverPid) of
                         true ->
-                    ReceiverPid ! {request_reply, Ref, decode_reply(Reply)};
+                            ReceiverPid ! {command_reply, Ref, decode_command_response(Reply)};
                         false ->
-                    logger:warning("[ws_channel] get reply message: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref])
+                            logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref])
+                    end
             end,
             {noreply, State#state{inflight = NInflight}}
     end;
-handle_response_frame(Ref, Reply, _Inflight, State) ->
+handle_command_response_frame(Ref, Reply, _Inflight, State) ->
-    logger:warning("[ws_channel] unexpected response frame, ref: ~p, reply: ~p", [Ref, Reply]),
+    logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]),
     {noreply, State}.
 
 -spec send_reply_frame(module(), any(), reference(), tuple()) -> any().
@@ -264,15 +274,15 @@ send_reply_frame(Transport, Socket, Ref, Reply) ->
     Packet = term_to_binary({response, Ref, Reply}),
     Transport:send(Socket, Packet).
 
--spec decode_reply({container_response, ok | {ok, term()} | {error, term()}} | tuple()) ->
+-spec decode_command_response({container, ok | {ok, term()} | {error, term()}} | tuple()) ->
     ok | {ok, term()} | {error, term()}.
-decode_reply({container_response, ok}) ->
+decode_command_response({container, ok}) ->
     ok;
-decode_reply({container_response, {ok, Result}}) ->
+decode_command_response({container, {ok, Result}}) ->
     {ok, Result};
-decode_reply({container_response, {error, Reason}}) ->
+decode_command_response({container, {error, Reason}}) ->
     {error, Reason};
-decode_reply(_Reply) ->
+decode_command_response(_Reply) ->
     {error, invalid_response}.
 
 %% 检测token是否是合法值