diff --git a/src/endpoint/endpoint_log.erl b/src/endpoint/endpoint_log.erl new file mode 100644 index 0000000..7151701 --- /dev/null +++ b/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, + <> = Content, + {LoggedContent, true}. diff --git a/src/endpoint/endpoint_subscription.erl b/src/endpoint/endpoint_subscription.erl index 75085ec..18b9754 100644 --- a/src/endpoint/endpoint_subscription.erl +++ b/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(). diff --git a/src/endpoint/endpoint_sup_sup.erl b/src/endpoint/endpoint_sup_sup.erl index 8ff2b9b..2d4ee48 100644 --- a/src/endpoint/endpoint_sup_sup.erl +++ b/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}}. \ No newline at end of file + {ok, {SupFlags, ChildSpecs}}.