diff --git a/docs/efka_iot_protocol.md b/docs/efka_iot_protocol.md index 76204bf..8f959ec 100644 --- a/docs/efka_iot_protocol.md +++ b/docs/efka_iot_protocol.md @@ -33,7 +33,6 @@ `command` 和 `command_response` 的 `Domain` 表示业务域,目前支持: -- `auth` - `container` ## 鉴权请求 @@ -52,37 +51,19 @@ ```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 语义: +`/host/activate` 只修改 `iot` 本地和持久化的 host 授权状态,不再向 `efka` 下发 auth command。`efka` 可以继续保持连接并发送数据,是否处理这些数据由 `iot_host` 当前状态决定。 -```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` 侧会异步提交 auth command 并等待对应 `Ref` 的 `command_response`,再向 `/host/activate` HTTP 调用方返回结果;等待超时为 10 秒,超时返回 `timeout`,无效响应返回 `invalid response`。 +因此当前协议没有 `{command, Ref, {auth, ...}}` 和 `{command_response, Ref, {auth, ...}}`。授权关闭时,`iot_host` 保持 channel 在线,但不处理上报数据;授权重新打开后,已在线的 channel 可以继续使用。 ## 容器管理命令 @@ -252,5 +233,6 @@ ok - 旧容器管理:`{request, Ref, {container_request, ...}}` - 旧容器回复:`{response, Ref, {container_response, ...}}` - 旧授权控制:`{message, {auth_control, Command}}` +- 已移除的 auth command:`{command, Ref, {auth, activate | deactivate}}` 如果需要滚动升级,应先增加临时兼容分支或引入协议版本协商。 diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index e878169..5d903b3 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -24,9 +24,8 @@ %% 标记当前agent的状态,只有在 activated 状态下才可以正常的发送数据 -define(STATE_DISCONNECTED, disconnected). +%% 等待校验中 -define(STATE_AUTH, auth). -%% 不能推送消息到服务,但是可以接受服务器的部分指令 --define(STATE_RESTRICTED, restricted). %% 激活状态下 -define(STATE_ACTIVATED, activated). @@ -34,8 +33,6 @@ socket :: undefined | ssl:sslsocket(), %% 保存当前auth请求的ref,用来建立auth请求和响应的对应关系 auth_ref = undefined :: undefined | reference(), - %% iot auth 命令触发重新鉴权时,用它保存 command 的 ref。 - auth_control_ref = undefined :: undefined | reference(), dropped_message_count = 0 :: non_neg_integer() }). @@ -127,7 +124,7 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro {keep_state, State, [{reply, From, DroppedCount}]}; %% 异步建立到服务器的连接 -handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = #state{}) -> +handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) -> case connect_socket() of {ok, Socket} -> Ref = make_ref(), @@ -139,12 +136,11 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = {keep_state, State#state{socket = undefined}} end; -handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket, auth_control_ref = AuthControlRef}) -> +handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket}) -> logger:debug("[efka_client] auth request timeout"), - maybe_send_auth_response(Socket, AuthControlRef, {error, timeout}), disconnect(Socket), schedule_reconnect(), - {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; %% 将缓存中的数据推送到服务器端 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> @@ -167,73 +163,56 @@ handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socke logger:debug("[efka_client] ssl error: ~p", [Reason]), disconnect(Socket), schedule_reconnect(), - {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) -> schedule_reconnect(), - {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; %%% 处理内部消息,ssl收到的消息会先 binary_to_term,再由这里按协议结构模式匹配 %% 容器管理命令由 iot 发起,使用 command/command_response 语义。 -handle_event(internal, {command, Ref, {container, #{action := list}}}, - ?STATE_ACTIVATED, State = #state{socket = Socket}) -> +handle_event(internal, {command, Ref, {container, #{action := list}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> Reply = docker_commands:get_containers(), send_container_response(Socket, Ref, Reply), {keep_state, State}; -handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}}, - ?STATE_ACTIVATED, 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}) -> +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}) -> +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}) -> +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}) -> +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}) -> +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}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> - logger:notice("[efka_client] get an invalid command: ~p, agent restricted", [Request]), - send_container_response(Socket, Ref, {error, <<"agent restricted">>}), - {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]), send_container_response(Socket, Ref, {error, <<"agent invalid">>}), {keep_state, State}; %% 处理response -handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) -> +handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> logger:debug("[efka_client] auth success"), - maybe_send_auth_response(Socket, AuthControlRef, ok), - {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined, auth_control_ref = undefined}, [{next_event, info, flush_cache}]}; -handle_event(internal, {response, AuthRef, {auth_response, {error, {denied, Message}}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) -> - logger:debug("[efka_client] auth denied, message: ~p", [Message]), - maybe_send_auth_response(Socket, AuthControlRef, {error, {denied, Message}}), - {next_state, ?STATE_RESTRICTED, State#state{auth_ref = undefined, auth_control_ref = undefined}}; -handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) -> + {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}) -> logger:debug("[efka_client] auth failed, reason: ~p", [Reason]), - maybe_send_auth_response(Socket, AuthControlRef, {error, Reason}), disconnect(Socket), schedule_reconnect(), - {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; handle_event(internal, {response, _Ref, Reply}, StateName, State) -> logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]), {keep_state, State}; @@ -241,23 +220,6 @@ handle_event(internal, {command_response, _Ref, Reply}, StateName, State) -> logger:warning("[efka_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]), {keep_state, State}; -%% 处理命令 -handle_event(internal, {command, CommandRef, {auth, Cmd}}, StateName, State = #state{socket = Socket}) -> - logger:debug("[efka_client] auth cmd: ~p", [Cmd]), - case {Cmd, StateName} of - {activate, ?STATE_ACTIVATED} -> - send_auth_response(Socket, CommandRef, ok), - {keep_state, State}; - {activate, _} -> - Ref = make_ref(), - AuthPacket = auth_packet(Ref), - ok = ssl:send(Socket, AuthPacket), - {next_state, ?STATE_AUTH, State#state{auth_ref = Ref, auth_control_ref = CommandRef}, [{state_timeout, 5000, auth_timeout}]}; - {deactivate, _} -> - send_auth_response(Socket, CommandRef, ok), - {next_state, ?STATE_RESTRICTED, State#state{auth_ref = undefined, auth_control_ref = undefined}} - end; - %% 处理Pub/Sub机制 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]), @@ -327,17 +289,6 @@ send_container_response(Socket, Ref, Reply) -> Packet = term_to_binary({command_response, Ref, {container, Reply}}), ok = ssl:send(Socket, Packet). --spec send_auth_response(ssl:sslsocket(), reference(), term()) -> ok. -send_auth_response(Socket, Ref, Reply) -> - Packet = term_to_binary({command_response, Ref, {auth, Reply}}), - ok = ssl:send(Socket, Packet). - --spec maybe_send_auth_response(ssl:sslsocket(), undefined | reference(), term()) -> ok. -maybe_send_auth_response(_Socket, undefined, _Reply) -> - ok; -maybe_send_auth_response(Socket, Ref, Reply) when is_reference(Ref) -> - send_auth_response(Socket, Ref, Reply). - -spec container_target(map()) -> binary(). container_target(Target) when is_map(Target) -> NameBin = to_binary(maps:get(name, Target, <<>>)),