From 9b70cadf64d1ece29555d31a2d0affa1fc528035 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 9 May 2026 18:29:46 +0800 Subject: [PATCH] fix heartbeat --- config/sys.config | 9 ++- docs/efka_iot_protocol.md | 6 ++ docs/heartbeat.md | 52 ++++++++++++++ src/efka_sup.erl | 9 +++ src/transport/efka_client.erl | 4 +- src/transport/efka_heartbeat.erl | 119 +++++++++++++++++++++++++++++++ 6 files changed, 195 insertions(+), 4 deletions(-) create mode 100644 docs/heartbeat.md create mode 100644 src/transport/efka_heartbeat.erl diff --git a/config/sys.config b/config/sys.config index 3209152..829b99d 100644 --- a/config/sys.config +++ b/config/sys.config @@ -15,9 +15,14 @@ {backlog, 256} ]}, - {tls_server_address, [ + {iot_server, [ {host, "localhost"}, - {port, 443} + {tls_port, 443}, + {udp_port, 18080} + ]}, + + {heartbeat, [ + {interval, 5000} ]}, {auth, [ diff --git a/docs/efka_iot_protocol.md b/docs/efka_iot_protocol.md index c05b99a..d814fe9 100644 --- a/docs/efka_iot_protocol.md +++ b/docs/efka_iot_protocol.md @@ -234,6 +234,12 @@ GET /event_stream?uuid=&task_id= - `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： diff --git a/docs/heartbeat.md b/docs/heartbeat.md new file mode 100644 index 0000000..43cf7da --- /dev/null +++ b/docs/heartbeat.md @@ -0,0 +1,52 @@ +# UDP 心跳 + +`efka` 通过独立的 `efka_heartbeat` 进程向 `iot` 发送 UDP 心跳。该心跳只表示主机存活，不依赖 TLS control channel 是否在线。 + +## 配置 + +TLS 和 UDP 共用同一个 `iot_server.host`： + +```erlang +{iot_server, [ + {host, "localhost"}, + {tls_port, 443}, + {udp_port, 18080} +]}, + +{heartbeat, [ + {interval, 5000} +]}, + +{auth, [ + {uuid, "qbxmjyzrkpntfgswaevodhluicqzxplkm"}, + {token, "zpxlkvmqwnbghytrujsdieofazxcvbnm"} +]} +``` + +- `tls_port` 用于 `efka_client` 建立 TLS 连接。 +- `udp_port` 用于 `efka_heartbeat` 发送 UDP 心跳。 +- `heartbeat.interval` 是发送间隔，单位毫秒，默认 5000。 +- UDP 心跳和 TLS 鉴权复用 `auth.uuid` 和 `auth.token`。 + +## 包格式 + +```erlang +<< + Version:8, + UuidLen:16, + UUID:UuidLen/binary, + Timestamp:64/unsigned-big, + Nonce:16/binary, + Mac:32/binary +>> +``` + +`Mac` 使用 HMAC-SHA256： + +```erlang +HeartbeatSecret = crypto:hash(sha256, Token), +Payload = <>, +Mac = crypto:mac(hmac, sha256, HeartbeatSecret, Payload) +``` + +其中 `Token` 是 `auth.token`。`iot` 侧注册 efka client 时保存同样的 `SHA256(Token)` 派生值，并用它校验心跳。 diff --git a/src/efka_sup.erl b/src/efka_sup.erl index 3a1a6fc..1eb7417 100644 --- a/src/efka_sup.erl +++ b/src/efka_sup.erl @@ -84,6 +84,15 @@ init([]) -> modules => ['efka_client'] }, + #{ + id => 'efka_heartbeat', + start => {'efka_heartbeat', start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => ['efka_heartbeat'] + }, + #{ id => 'docker_task_reporter', start => {'docker_task_reporter', start_link, []}, diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index 5d903b3..3df9196 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -262,9 +262,9 @@ auth_packet(Ref) when is_reference(Ref) -> -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}. connect_socket() -> - {ok, Props} = application:get_env(efka, tls_server_address), + {ok, Props} = application:get_env(efka, iot_server), Host = proplists:get_value(host, Props), - Port = proplists:get_value(port, Props), + Port = proplists:get_value(tls_port, Props), SslOptions = [ binary, {active, true}, diff --git a/src/transport/efka_heartbeat.erl b/src/transport/efka_heartbeat.erl new file mode 100644 index 0000000..21b103d --- /dev/null +++ b/src/transport/efka_heartbeat.erl @@ -0,0 +1,119 @@ +%%%------------------------------------------------------------------- +%%% @doc UDP heartbeat sender for iot host liveness. +%%% @end +%%%------------------------------------------------------------------- +-module(efka_heartbeat). + +-behaviour(gen_server). + +%% API +-export([start_link/0]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). +-define(HEARTBEAT_VERSION, 1). +-define(HEARTBEAT_NONCE_BYTES, 16). +-define(DEFAULT_INTERVAL, 5000). + +-record(state, { + socket :: gen_udp:socket(), + host :: inet:hostname() | inet:ip_address(), + port :: inet:port_number(), + interval :: pos_integer(), + uuid :: binary(), + heartbeat_secret :: binary() +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +-spec start_link() -> {ok, pid()} | ignore | {error, term()}. +start_link() -> + gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +-spec init(list()) -> {ok, #state{}} | {stop, term()}. +init([]) -> + ok = application:ensure_started(crypto), + {ok, ServerProps} = application:get_env(efka, iot_server), + {ok, AuthProps} = application:get_env(efka, auth), + HeartbeatProps = case application:get_env(efka, heartbeat) of + {ok, Props} -> Props; + undefined -> [] + end, + Host = proplists:get_value(host, ServerProps), + UdpPort = proplists:get_value(udp_port, ServerProps), + Interval = proplists:get_value(interval, HeartbeatProps, ?DEFAULT_INTERVAL), + UUID = list_to_binary(proplists:get_value(uuid, AuthProps)), + Token = list_to_binary(proplists:get_value(token, AuthProps)), + HeartbeatSecret = crypto:hash(sha256, Token), + case gen_udp:open(0, [binary]) of + {ok, Socket} -> + erlang:send_after(0, self(), heartbeat), + {ok, #state{ + socket = Socket, + host = Host, + port = UdpPort, + interval = Interval, + uuid = UUID, + heartbeat_secret = HeartbeatSecret + }}; + {error, Reason} -> + {stop, Reason} + end. + +-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, ok, #state{}}. +handle_call(_Request, _From, State) -> + {reply, ok, State}. + +-spec handle_cast(term(), #state{}) -> {noreply, #state{}}. +handle_cast(_Request, State) -> + {noreply, State}. + +-spec handle_info(term(), #state{}) -> {noreply, #state{}}. +handle_info(heartbeat, State = #state{interval = Interval}) -> + ok = send_heartbeat(State), + erlang:send_after(Interval, self(), heartbeat), + {noreply, State}; +handle_info(Info, State) -> + logger:warning("[efka_heartbeat] ignore unknown info: ~p", [Info]), + {noreply, State}. + +-spec terminate(term(), #state{}) -> ok. +terminate(_Reason, #state{socket = Socket}) -> + gen_udp:close(Socket), + ok. + +-spec code_change(term(), #state{}, term()) -> {ok, #state{}}. +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +-spec send_heartbeat(#state{}) -> ok. +send_heartbeat(#state{socket = Socket, host = Host, port = Port, uuid = UUID, heartbeat_secret = HeartbeatSecret}) -> + Packet = heartbeat_packet(UUID, HeartbeatSecret), + case gen_udp:send(Socket, Host, Port, Packet) of + ok -> + ok; + {error, Reason} -> + logger:warning("[efka_heartbeat] send heartbeat failed, reason: ~p", [Reason]), + ok + end. + +-spec heartbeat_packet(binary(), binary()) -> binary(). +heartbeat_packet(UUID, HeartbeatSecret) when is_binary(UUID), is_binary(HeartbeatSecret) -> + UUIDLen = byte_size(UUID), + Timestamp = efka_util:timestamp(), + Nonce = crypto:strong_rand_bytes(?HEARTBEAT_NONCE_BYTES), + Payload = <>, + Mac = crypto:mac(hmac, sha256, HeartbeatSecret, Payload), + <>.