fix heartbeat

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, [

docs/efka_iot_protocol.md

@@ -234,6 +234,12 @@ GET /event_stream?uuid=<host_uuid>&task_id=<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：

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 = <<Version:8, UuidLen:16, UUID:UuidLen/binary, Timestamp:64/unsigned-big, Nonce:16/binary>>,
+Mac = crypto:mac(hmac, sha256, HeartbeatSecret, Payload)
+```
+
+其中 `Token` 是 `auth.token`。`iot` 侧注册 efka client 时保存同样的 `SHA256(Token)` 派生值，并用它校验心跳。

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, []},

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},

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 = <<?HEARTBEAT_VERSION:8, UUIDLen:16, UUID:UUIDLen/binary, Timestamp:64/unsigned-big, Nonce/binary>>,
+    Mac = crypto:mac(hmac, sha256, HeartbeatSecret, Payload),
+    <<Payload/binary, Mac/binary>>.