add endpoint_log

src/endpoint/endpoint_log.erl

@@ -0,0 +1,171 @@
+%%%-------------------------------------------------------------------
+%%% @doc Endpoint diagnostic log.
+%%%
+%%% 当前用于记录 publish 时没有匹配到任何订阅者的数据，避免这类信息进入
+%%% endpoint_subscription 的热路径同步磁盘写入。
+%%%-------------------------------------------------------------------
+-module(endpoint_log).
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0, unmatched_publish/2]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+-define(LOG_NAME, endpoint_unmatched_publish_log).
+-define(LOG_DIR, "endpoint_log").
+-define(LOG_FILE, "unmatched_publish.log").
+-define(DEFAULT_ROOT_DIR, "log").
+-define(DEFAULT_MAX_BYTES, 10485760).
+-define(DEFAULT_MAX_FILES, 10).
+-define(MAX_LOG_CONTENT_BYTES, 32 * 1024).
+
+-record(state, {
+    enabled = false :: boolean(),
+    log_name = ?LOG_NAME :: atom()
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+-spec unmatched_publish(RouteKey :: binary(), Content :: binary()) -> ok.
+unmatched_publish(RouteKey, Content) when is_binary(RouteKey), is_binary(Content) ->
+    case erlang:whereis(?SERVER) of
+        undefined ->
+            ok;
+        Pid when is_pid(Pid) ->
+            gen_server:cast(Pid, {unmatched_publish, RouteKey, Content}),
+            ok
+    end.
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+-spec init(term()) -> {ok, #state{}}.
+init([]) ->
+    ok = iot_log:set_metadata(),
+    case open_log() of
+        ok ->
+            {ok, #state{enabled = true}};
+        {error, Reason} ->
+            logger:warning("[endpoint_log] open disk_log failed, reason: ~p", [Reason]),
+            {ok, #state{enabled = false}}
+    end.
+
+-spec handle_call(term(), gen_server:from(), #state{}) ->
+    {reply, ok, #state{}}.
+handle_call(_Request, _From, State) ->
+    {reply, ok, State}.
+
+-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
+handle_cast({unmatched_publish, RouteKey, Content}, State = #state{enabled = true, log_name = LogName}) ->
+    Entry = encode_unmatched_publish(RouteKey, Content),
+    case disk_log:blog(LogName, Entry) of
+        ok ->
+            {noreply, State};
+        {error, Reason} ->
+            logger:warning("[endpoint_log] write unmatched publish log failed, reason: ~p", [Reason]),
+            {noreply, State#state{enabled = false}}
+    end;
+handle_cast({unmatched_publish, _RouteKey, _Content}, State) ->
+    {noreply, State};
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
+handle_info(Info, State) ->
+    logger:debug("[endpoint_log] unknown info: ~p", [Info]),
+    {noreply, State}.
+
+-spec terminate(term(), #state{}) -> ok.
+terminate(_Reason, #state{enabled = true, log_name = LogName}) ->
+    _ = disk_log:close(LogName),
+    ok;
+terminate(_Reason, _State) ->
+    ok.
+
+-spec code_change(term() | {down, term()}, #state{}, term()) ->
+    {ok, #state{}} | {error, term()}.
+code_change(_OldVsn, State = #state{}, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+-spec open_log() -> ok | {error, term()}.
+open_log() ->
+    Path = log_path(),
+    ok = filelib:ensure_dir(Path),
+    case disk_log:open([
+        {name, ?LOG_NAME},
+        {file, Path},
+        {type, wrap},
+        {format, external},
+        {size, {?DEFAULT_MAX_BYTES, ?DEFAULT_MAX_FILES}},
+        {linkto, self()},
+        {repair, true}
+    ]) of
+        {ok, ?LOG_NAME} ->
+            ok;
+        {repaired, ?LOG_NAME, _Recovered, _BadBytes} ->
+            ok;
+        {error, Reason} ->
+            {error, Reason}
+    end.
+
+-spec log_path() -> file:filename_all().
+log_path() ->
+    filename:join([root_dir(), ?LOG_DIR, ?LOG_FILE]).
+
+-spec root_dir() -> file:filename_all().
+root_dir() ->
+    case application:get_env(iot, endpoints) of
+        {ok, Endpoints} ->
+            proplists:get_value(root_dir, Endpoints, ?DEFAULT_ROOT_DIR);
+        undefined ->
+            ?DEFAULT_ROOT_DIR
+    end.
+
+-spec encode_unmatched_publish(binary(), binary()) -> binary().
+encode_unmatched_publish(RouteKey, Content) ->
+    {LoggedContent, Truncated} = maybe_truncate(Content),
+    iolist_to_binary([
+        format_local_time(), <<"\t">>,
+        <<"unmatched_publish">>, <<"\t">>,
+        RouteKey, <<"\t">>,
+        integer_to_binary(byte_size(Content)), <<"\t">>,
+        boolean_to_binary(Truncated), <<"\t">>,
+        base64:encode(LoggedContent), <<"\n">>
+    ]).
+
+-spec boolean_to_binary(boolean()) -> binary().
+boolean_to_binary(true) ->
+    <<"true">>;
+boolean_to_binary(false) ->
+    <<"false">>.
+
+-spec format_local_time() -> binary().
+format_local_time() ->
+    {{Year, Month, Day}, {Hour, Minute, Second}} = calendar:local_time(),
+    iolist_to_binary(io_lib:format(
+        "~4..0B-~2..0B-~2..0B ~2..0B:~2..0B:~2..0B",
+        [Year, Month, Day, Hour, Minute, Second]
+    )).
+
+-spec maybe_truncate(binary()) -> {binary(), boolean()}.
+maybe_truncate(Content) when byte_size(Content) =< ?MAX_LOG_CONTENT_BYTES ->
+    {Content, false};
+maybe_truncate(Content) ->
+    MaxBytes = ?MAX_LOG_CONTENT_BYTES,
+    <<LoggedContent:MaxBytes/binary, _/binary>> = Content,
+    {LoggedContent, true}.

src/endpoint/endpoint_subscription.erl

@@ -55,7 +55,7 @@ unsubscribe(Topic, SubscriberPid) when is_binary(Topic), is_pid(SubscriberPid) -
 get_subscribers() ->
     gen_server:call(?SERVER, get_subscribers).
 
--spec publish(RouteKey :: binary(), Content :: binary()) -> no_return().
+-spec publish(RouteKey :: binary(), Content :: binary()) -> ok.
 publish(RouteKey, Content) when is_binary(RouteKey), is_binary(Content) ->
     case ets:info(?SUBSCRIBER_TAB) of
         undefined ->
@@ -66,6 +66,7 @@ publish(RouteKey, Content) when is_binary(RouteKey), is_binary(Content) ->
             lists:foreach(fun(#subscriber{subscriber_pid = SubscriberPid}) ->
                 endpoint:forward(SubscriberPid, Content)
             end, MatchedSubscribers),
+            maybe_log_unmatched_publish(RouteKey, Content, MatchedSubscribers),
             logger:debug("[efka_subscription] route_key: ~p, metric: ~p, match subscribers: ~p", [RouteKey, Content, MatchedSubscribers]),
             ok
     end.
@@ -201,6 +202,12 @@ match_subscribers(Subscribers, Topic) when is_list(Subscribers), is_binary(Topic
     Sorted = lists:sort(fun compare_subscriber/2, Matched),
     dedupe_subscribers(Sorted).
 
+-spec maybe_log_unmatched_publish(binary(), binary(), [#subscriber{}]) -> ok.
+maybe_log_unmatched_publish(RouteKey, Content, []) ->
+    endpoint_log:unmatched_publish(RouteKey, Content);
+maybe_log_unmatched_publish(_RouteKey, _Content, _MatchedSubscribers) ->
+    ok.
+
 %% 开始对比订阅的topic和发布的topic的Components信息
 %% *表示单级匹配，+表示多级匹配；+只能出现一次，并且只能在末尾
 -spec match_components(list(), list()) -> boolean().

src/endpoint/endpoint_sup_sup.erl

@@ -28,6 +28,15 @@ start_link() ->
 init([]) ->
     SupFlags = #{strategy => one_for_all, intensity => 1000, period => 3600},
     ChildSpecs = [
+        #{
+            id => endpoint_log,
+            start => {'endpoint_log', start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => worker,
+            modules => ['endpoint_log']
+        },
+
         #{
             id => endpoint_subscription,
             start => {'endpoint_subscription', start_link, []},
@@ -46,4 +55,4 @@ init([]) ->
             modules => ['endpoint_sup']
         }
     ],
-    {ok, {SupFlags, ChildSpecs}}.
+    {ok, {SupFlags, ChildSpecs}}.