%%%------------------------------------------------------------------- %%% @author %%% @copyright (C) 2023, %%% @doc %%% %%% @end %%% Created : 22. 9月 2023 16:38 %%%------------------------------------------------------------------- -module(iot_host). -author("aresei"). -include("iot.hrl"). -include("domain_model.hrl"). -behaviour(gen_statem). %% 心跳包检测时间间隔, 15分钟检测一次 -define(HEARTBEAT_INTERVAL, 900 * 1000). %% 状态 -define(STATE_DENIED, denied). -define(STATE_ACTIVATED, activated). %% API -export([start_link/2, get_name/1, get_alias_name/1, get_pid/1, handle/2, activate/2]). -export([lookup_pid/1]). -export([get_metric/1, get_status/1, kill/1]). %% 通讯相关 -export([pub/4, attach_channel/2]). -export([deploy_container/3, start_container/2, stop_container/2, remove_container/2, kill_container/2, config_container/3, get_containers/1, await_reply/3]). -export([heartbeat/1]). %% gen_statem callbacks -export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). -record(state, { host_id :: integer(), %% 从数据库里面读取到的数据 uuid :: binary(), has_session = false :: boolean(), %% 心跳计数器 heartbeat_counter = 0 :: integer(), %% websocket相关 channel_pid :: undefined | pid(), %% 主机的相关信息 metrics = #{} :: map() }). %%%=================================================================== %%% API %%%=================================================================== -spec get_pid(UUID :: binary()) -> undefined | pid(). get_pid(UUID) when is_binary(UUID) -> Name = get_name(UUID), gproc:whereis_name({n, l, Name}). -spec lookup_pid(UUID :: binary()) -> {ok, pid()} | {error, Reason :: any()}. lookup_pid(UUID) when is_binary(UUID) -> case get_pid(UUID) of undefined -> {error, <<"host not found">>}; HostPid -> {ok, HostPid} end. -spec get_name(UUID :: binary()) -> term(). get_name(UUID) when is_binary(UUID) -> {iot_host, UUID}. -spec get_alias_name(HostId :: integer()) -> term(). get_alias_name(HostId0) when is_integer(HostId0) -> {iot_host_id, HostId0}. -spec kill(UUID :: binary()) -> ok | true. kill(UUID) when is_binary(UUID) -> case get_pid(UUID) of undefined -> ok; Pid -> exit(Pid, kill) end. %% 处理消息 -spec handle(Pid :: pid(), Packet :: tuple()) -> ok. handle(Pid, Packet) when is_pid(Pid) -> gen_statem:cast(Pid, {handle, Packet}). -spec get_status(Pid :: pid()) -> {ok, Status :: map()}. get_status(Pid) when is_pid(Pid) -> gen_statem:call(Pid, get_status). %% 激活主机, true 表示激活； false表示关闭激活 -spec activate(Pid :: pid(), Auth :: boolean()) -> ok | {error, term()}. activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) -> gen_statem:call(Pid, {activate, Auth}). -spec get_metric(Pid :: pid()) -> {ok, MetricInfo :: map()}. get_metric(Pid) when is_pid(Pid) -> gen_statem:call(Pid, get_metric). -spec attach_channel(pid(), pid()) -> ok | {error, Reason :: binary()}. attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) -> gen_statem:call(Pid, {attach_channel, ChannelPid}). -type request_ref() :: binary(). -spec get_containers(Pid :: pid()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. get_containers(Pid) when is_pid(Pid) -> container_call(Pid, docker_container_builder:list_request()). -spec config_container(Pid :: pid(), ContainerName :: binary(), ConfigJson :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. config_container(Pid, ContainerName, ConfigJson) when is_pid(Pid), is_binary(ContainerName), is_binary(ConfigJson) -> container_call(Pid, docker_container_builder:config_request(ContainerName, ConfigJson)). -spec deploy_container(Pid :: pid(), TaskId :: integer(), Config :: map()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. deploy_container(Pid, TaskId, Config) when is_pid(Pid), is_integer(TaskId), is_map(Config) -> case docker_container_builder:deploy_request(TaskId, Config) of {ok, Request} -> container_call(Pid, Request); {error, Reason} -> {error, Reason} end. -spec start_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. start_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:start_request(ContainerName)). -spec stop_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. stop_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:stop_request(ContainerName)). -spec kill_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. kill_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:kill_request(ContainerName)). -spec remove_container(Pid :: pid(), ContainerName :: binary()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName) -> container_call(Pid, docker_container_builder:remove_request(ContainerName)). -spec await_reply(Pid :: pid(), Ref :: request_ref(), Timeout :: integer()) -> ok | {ok, Result :: term()} | {error, Reason :: term()}. await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_binary(Ref), is_integer(Timeout) -> receive {command_reply, Ref, ok} -> ok; {command_reply, Ref, {ok, Result}} -> {ok, Result}; {command_reply, Ref, {error, Reason}} -> {error, Reason} after Timeout -> ok = gen_statem:call(Pid, {cancel_command_call, Ref}), flush_reply(Ref), {error, timeout} end. -spec pub(Pid :: pid(), Topic :: binary(), Qos :: integer(), Content :: binary()) -> ok | {error, Reason :: any()}. pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos), is_binary(Content) -> gen_statem:call(Pid, {pub, Topic, Qos, Content}). -spec heartbeat(Pid :: undefined | pid()) -> ok. heartbeat(undefined) -> ok; heartbeat(Pid) when is_pid(Pid) -> gen_statem:cast(Pid, heartbeat). %% @doc Creates a gen_statem process which calls Module:init/1 to %% initialize. To ensure a synchronized start-up procedure, this %% function does not return until Module:init/1 has returned. start_link(Name, UUID) when is_binary(UUID) -> gen_statem:start_link({via, gproc, {n, l, Name}}, ?MODULE, [UUID], []). %%%=================================================================== %%% gen_statem callbacks %%%=================================================================== %% @private %% @doc Whenever a gen_statem is started using gen_statem:start/[3,4] or %% gen_statem:start_link/[3,4], this function is called by the new %% process to initialize. init([UUID]) -> ok = iot_log:set_metadata(), case iot_api_client:get_host_by_uuid(UUID) of {ok, #host_info{id = HostId, authorize_status = AuthorizeStatus}} -> %% 通过host_id注册别名, 可以避免通过查询数据库获取HostPid AliasName = get_alias_name(HostId), global:register_name(AliasName, self()), %% 心跳检测机制 erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker), StateName = case AuthorizeStatus =:= ?HOST_AUTHORIZED of true -> ?STATE_ACTIVATED; false -> ?STATE_DENIED end, {ok, StateName, #state{host_id = HostId, uuid = UUID, has_session = false}}; undefined -> logger:warning("[iot_host] host uuid: ~p, load host info failed", [UUID]), ignore end. %% @private %% @doc This function is called by a gen_statem when it needs to find out %% the callback mode of the callback module. callback_mode() -> handle_event_function. %% @private %% @doc If callback_mode is handle_event_function, then whenever a %% gen_statem receives an event from call/2, cast/2, or as a normal %% process message, this function is called. handle_event({call, From}, get_metric, _, State = #state{metrics = Metrics}) -> {keep_state, State, [{reply, From, {ok, Metrics}}]}; %% 获取主机的状态 handle_event({call, From}, get_status, _, State = #state{channel_pid = ChannelPid, heartbeat_counter = HeartbeatCounter, metrics = Metrics, has_session = HasSession}) -> HasChannel = (ChannelPid /= undefined), Reply = #{ <<"has_channel">> => HasChannel, <<"has_session">> => HasSession, <<"heartbeat_counter">> => HeartbeatCounter, <<"metrics">> => Metrics }, {keep_state, State, [{reply, From, {ok, Reply}}]}; handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> case HasSession andalso is_pid(ChannelPid) of true -> Ref = request_ref(), ok = ssl_channel:container_call(ChannelPid, ReceiverPid, Ref, Request), {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>>}}]} end; handle_event({call, From}, {cancel_command_call, Ref}, _, State = #state{channel_pid = ChannelPid}) -> case is_pid(ChannelPid) of true -> ok = ssl_channel:cancel_command_call(ChannelPid, Ref), {keep_state, State, [{reply, From, ok}]}; false -> {keep_state, State, [{reply, From, ok}]} end; %% 发送指令时, pub/sub handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> case HasSession andalso is_pid(ChannelPid) of true -> logger:debug("[iot_host] host: ~p, publish to topic: ~p, content: ~p", [UUID, Topic, Content]), %% 通过websocket发送消息 ssl_channel:pub(ChannelPid, Topic, Qos, Content), {keep_state, State, [{reply, From, ok}]}; false -> logger:debug("[iot_host] uuid: ~p, publish to topic: ~p, content: ~p, invalid state: ~p", [UUID, Topic, Content, state_map(State)]), {keep_state, State, [{reply, From, {error, <<"主机离线，发送失败"/utf8>>}}]} end; %% 激活/关闭授权只修改 iot 本地授权状态；efka 连接保持在线，数据是否处理由 host 状态决定。 handle_event({call, From}, {activate, Auth}, _, State = #state{uuid = UUID}) -> NStateName = case Auth of true -> ?STATE_ACTIVATED; false -> ?STATE_DENIED end, logger:debug("[iot_host] uuid: ~p, state_name change to ~p", [UUID, NStateName]), {next_state, NStateName, State, [{reply, From, ok}]}; %% 绑定channel handle_event({call, From}, {attach_channel, ChannelPid}, StateName, State = #state{uuid = UUID, channel_pid = OldChannelPid}) -> case OldChannelPid == undefined orelse OldChannelPid =:= ChannelPid of true -> case StateName of ?STATE_ACTIVATED -> erlang:monitor(process, ChannelPid), %% 更新主机为在线状态 ChangeResult = iot_api_client:change_host_status(UUID, ?HOST_ONLINE), logger:debug("[iot_host] host_id(attach_channel) uuid: ~p, will change status, result: ~p", [UUID, ChangeResult]), {keep_state, State#state{channel_pid = ChannelPid, has_session = true}, [{reply, From, ok}]}; %% 主机未激活 ?STATE_DENIED -> logger:notice("[iot_host] attach_channel host_id uuid: ~p, channel: ~p, host denied locally", [UUID, ChannelPid]), erlang:monitor(process, ChannelPid), ChangeResult = iot_api_client:change_host_status(UUID, ?HOST_ONLINE), logger:debug("[iot_host] host_id(attach_channel) uuid: ~p, denied but online, change status result: ~p", [UUID, ChangeResult]), {keep_state, State#state{channel_pid = ChannelPid, has_session = true}, [{reply, From, ok}]} end; false -> logger:notice("[iot_host] attach_channel host_id uuid: ~p, old channel exists: ~p", [UUID, OldChannelPid]), {keep_state, State, [{reply, From, {error, <<"channel existed">>}}]} end; %% 数据分发 handle_event(cast, {handle, {data, RouteKey, MetricBin}}, ?STATE_ACTIVATED, State = #state{uuid = UUID, has_session = true}) -> logger:debug("[iot_host] metric_data host: ~p, route_key: ~p, metric: ~p", [UUID, RouteKey, MetricBin]), endpoint_subscription:publish(get_route_key(RouteKey), MetricBin), {keep_state, State}; handle_event(cast, {handle, {data, RouteKey, MetricBin}}, StateName, State = #state{uuid = UUID, has_session = true}) -> logger:notice("[iot_host] host_uuid: ~p, state_name: ~p, metric_data route_key: ~p, metric: ~p, discard", [UUID, StateName, RouteKey, MetricBin]), {keep_state, State}; %% ping的数据是通过aes加密后的，因此需要在有会话的情况下才行 handle_event(cast, {handle, {ping, Metrics}}, ?STATE_ACTIVATED, State = #state{uuid = UUID, has_session = true}) -> logger:debug("[iot_host] ping host_id uuid: ~p, get ping: ~p", [UUID, Metrics]), {keep_state, State#state{metrics = Metrics}}; %% 心跳机制 handle_event(cast, heartbeat, _, State = #state{uuid = UUID, heartbeat_counter = HeartbeatCounter}) -> maybe_mark_host_online(UUID), {keep_state, State#state{heartbeat_counter = HeartbeatCounter + 1}}; %% UDP 心跳丢失但 SSL channel 仍在时，不能把 host 标记为离线。 handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{uuid = UUID, heartbeat_counter = 0, channel_pid = ChannelPid}) when is_pid(ChannelPid) -> logger:warning("[iot_host] uuid: ~p, udp heartbeat lost but ssl channel is alive: ~p", [UUID, ChannelPid]), maybe_mark_host_online(UUID), erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker), {keep_state, State#state{heartbeat_counter = 0}}; %% 没有收到 UDP 心跳且没有 SSL channel，主机下线, 设备状态不变 handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{uuid = UUID, heartbeat_counter = 0}) -> logger:warning("[iot_host] uuid: ~p, heartbeat lost, devices will unknown", [UUID]), maybe_mark_host_offline(UUID), erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker), {keep_state, State#state{channel_pid = undefined, has_session = false, heartbeat_counter = 0}}; %% 其他情况下需要重置系统计数器 handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{}) -> erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker), {keep_state, State#state{heartbeat_counter = 0}}; %% 当websocket断开的时候，主机的状态不一定改变；主机的状态改变通过心跳机制，会话状态需要改变 handle_event(info, {'DOWN', _Ref, process, ChannelPid, Reason}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> logger:warning("[iot_host] uuid: ~p, channel: ~p, down with reason: ~p, has_session: ~p, state: ~p", [UUID, ChannelPid, Reason, HasSession, State]), {keep_state, State#state{channel_pid = undefined, has_session = false}}; handle_event(info, {'DOWN', _Ref, process, Pid, Reason}, _, State = #state{uuid = UUID}) -> logger:debug("[iot_host] uuid: ~p, process_pid: ~p, down with reason: ~p, state: ~p", [UUID, Pid, Reason, State]), {keep_state, State}; handle_event(Event, Info, StateName, State = #state{uuid = UUID}) -> logger:warning("[iot_host] host: ~p, event: ~p, unknown message: ~p, state_name: ~p, state: ~p", [UUID, Event, Info, StateName, state_map(State)]), {keep_state, State}. %% @private %% @doc This function is called by a gen_statem when it is about to %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_statem terminates with %% Reason. The return value is ignored. terminate(Reason, _StateName, _State = #state{uuid = UUID, has_session = HasSession}) -> logger:debug("[iot_host] host: ~p, terminate with reason: ~p, has_session: ~p", [UUID, Reason, HasSession]), ok. %% @private %% @doc Convert process state when code is changed code_change(_OldVsn, StateName, State = #state{}, _Extra) -> {ok, StateName, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== -spec container_call(Pid :: pid(), Request :: term()) -> {ok, Ref :: request_ref()} | {error, Reason :: any()}. container_call(Pid, Request) when is_pid(Pid) -> gen_statem:call(Pid, {container_call, self(), Request}). -spec get_route_key(binary()) -> binary(). get_route_key(<<"">>) -> <<"/">>; get_route_key(RouteKey) when is_binary(RouteKey) -> RouteKey. %% 将当前的state转换成map state_map(#state{host_id = HostId, uuid = UUID, has_session = HasSession, heartbeat_counter = HeartbeatCounter, channel_pid = ChannelPid, metrics = Metrics}) -> #{ host_id => HostId, uuid => UUID, has_session => HasSession, heartbeat_counter => HeartbeatCounter, channel_pid => ChannelPid, metrics => Metrics }. flush_reply(Ref) -> receive {command_reply, Ref, _Reply} -> ok after 0 -> ok end. -spec request_ref() -> request_ref(). request_ref() -> crypto:strong_rand_bytes(16). -spec maybe_mark_host_offline(binary()) -> ok. maybe_mark_host_offline(UUID) -> case iot_api_client:get_host_by_uuid(UUID) of {ok, #host_info{status = ?HOST_NOT_JOINED}} -> logger:debug("[iot_host] host: ~p, host_maybe_offline, host not joined, can not change to offline", [UUID]), ok; {ok, #host_info{status = ?HOST_OFFLINE}} -> logger:debug("[iot_host] host: ~p, host_maybe_offline, host now is offline, do nothing", [UUID]), ok; {ok, #host_info{status = ?HOST_ONLINE}} -> _ = iot_api_client:change_host_status(UUID, ?HOST_OFFLINE), ok; Other -> logger:warning("[iot_host] host: ~p, load status failed while marking offline: ~p", [UUID, Other]), ok end. -spec maybe_mark_host_online(binary()) -> ok. maybe_mark_host_online(UUID) -> case iot_api_client:get_host_by_uuid(UUID) of {ok, #host_info{status = ?HOST_OFFLINE}} -> _ = iot_api_client:change_host_status(UUID, ?HOST_ONLINE), ok; {ok, #host_info{status = ?HOST_ONLINE}} -> ok; {ok, #host_info{status = ?HOST_NOT_JOINED}} -> ok; Other -> logger:warning("[iot_host] host: ~p, load status failed while marking online: ~p", [UUID, Other]), ok end.