diff --git a/src/docker/docker_helper.erl b/src/docker/docker_helper.erl index 6158101..ef1fffc 100644 --- a/src/docker/docker_helper.erl +++ b/src/docker/docker_helper.erl @@ -11,6 +11,7 @@ %% API -export([ensure_container_dir/2, ensure_container_dir/3, get_container_dir/2, get_config_file/1]). +-export([update_container_config/2]). -spec ensure_container_dir(RootDir :: string(), ContainerName :: binary()) -> {ok, ServerRootDir :: string()}. ensure_container_dir(RootDir, ContainerName) when is_list(RootDir), is_binary(ContainerName) -> @@ -49,6 +50,24 @@ get_container_dir(RootDir, ContainerName) when is_list(RootDir), is_binary(Conta error end. +-spec update_container_config(binary(), binary()) -> ok | {error, binary()}. +update_container_config(ContainerName, Config) when is_binary(ContainerName), is_binary(Config) -> + {ok, RootDir} = application:get_env(efka, root_dir), + case get_container_dir(RootDir, ContainerName) of + {ok, ContainerDir} -> + ConfigFile = get_config_file(ContainerDir), + case file:write_file(ConfigFile, Config, [write, binary]) of + ok -> + logger:warning("[efka_client] write config file: ~p success", [ConfigFile]), + ok; + {error, Reason} -> + logger:warning("[efka_client] write config file: ~p, get error: ~p", [ConfigFile, Reason]), + {error, <<"write config failed">>} + end; + error -> + {error, <<"error">>} + end. + -spec default_container_dir(RootDir :: string(), ContainerName :: binary()) -> string(). default_container_dir(RootDir, ContainerName) -> normalize_container_dir(RootDir ++ "/" ++ binary_to_list(ContainerName)). diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index 3eba2ed..696b24c 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -21,9 +21,6 @@ -export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). -define(SERVER, ?MODULE). --define(CACHE_TAB, cache). --define(MAX_CACHE_ITEMS, 5000000). --define(MAX_CACHE_FILE_SIZE, 1073741824). %% 标记当前agent的状态，只有在 activated 状态下才可以正常的发送数据 -define(STATE_DISCONNECTED, disconnected). @@ -79,7 +76,7 @@ start_link() -> -spec init(list()) -> {ok, atom(), #state{}}. init([]) -> - case open_cache_table() of + case efka_client_cache:open() of ok -> erlang:start_timer(0, self(), create_transport), {ok, ?STATE_DISCONNECTED, #state{socket = undefined}}; @@ -91,7 +88,7 @@ init([]) -> callback_mode() -> handle_event_function. -%% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia +%% 异步发送数据, 连接存在时候直接发送；否则缓存到DETS -spec handle_event(term(), term(), atom(), #state{}) -> term(). handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) -> Packet = term_to_binary({message, {data, #{route_key => RouteKey, metric => Metric}}}), @@ -100,7 +97,7 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so ok = ssl:send(Socket, Packet), {keep_state, State}; _ -> - {ok, DroppedCount} = cache_insert(Packet), + {ok, DroppedCount} = efka_client_cache:insert(Packet), {keep_state, State#state{dropped_message_count = State#state.dropped_message_count + DroppedCount}} end; @@ -148,10 +145,10 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S %% 将缓存中的数据推送到服务器端 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case cache_fetch_next() of + case efka_client_cache:fetch_next() of {ok, {Id, Packet}} -> ok = ssl:send(Socket, Packet), - ok = cache_delete(Id), + ok = efka_client_cache:delete(Id), {keep_state, State, [{next_event, info, flush_cache}]}; error -> {keep_state, State} @@ -207,7 +204,7 @@ handle_event(internal, {request, Ref, {container_request, #{action := remove, ta {keep_state, State}; handle_event(internal, {request, Ref, {container_request, #{action := config, target := Target, config := Config}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = update_container_config(container_target(Target), iolist_to_binary(Config)), + Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)), handle_container_response(Socket, Ref, Reply), {keep_state, State}; handle_event(internal, {request, Ref, {container_request, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> @@ -266,7 +263,7 @@ handle_event(info, Info, _, State = #state{}) -> -spec terminate(term(), atom(), #state{}) -> ok. terminate(_Reason, _StateName, _State = #state{socket = Socket}) -> disconnect(Socket), - close_cache_table(), + efka_client_cache:close(), ok. -spec code_change(term(), atom(), #state{}, term()) -> {ok, atom(), #state{}}. @@ -314,94 +311,6 @@ disconnect(Socket) -> schedule_reconnect() -> erlang:start_timer(5000, self(), create_transport). --spec open_cache_table() -> ok | {error, term()}. -open_cache_table() -> - {ok, DetsDir} = application:get_env(efka, dets_dir), - File = DetsDir ++ "cache.dets", - case dets:open_file(?CACHE_TAB, [{file, File}, {type, bag}, {keypos, 1}]) of - {ok, ?CACHE_TAB} -> - ok; - {error, Reason} -> - {error, Reason} - end. - --spec close_cache_table() -> ok. -close_cache_table() -> - case dets:close(?CACHE_TAB) of - ok -> - ok; - {error, not_owner} -> - ok - end. - --spec cache_insert(binary()) -> {ok, non_neg_integer()} | {error, term()}. -cache_insert(Data) when is_binary(Data) -> - case dets:insert(?CACHE_TAB, {generate_cache_id(), Data}) of - ok -> - trim_cache_limits(0); - {error, Reason} -> - {error, Reason} - end. - --spec cache_fetch_next() -> error | {ok, {integer(), binary()}}. -cache_fetch_next() -> - case dets:first(?CACHE_TAB) of - '$end_of_table' -> - error; - Key -> - case dets:lookup(?CACHE_TAB, Key) of - [Entry | _] -> - {ok, Entry}; - [] -> - cache_fetch_next() - end - end. - --spec cache_delete(integer()) -> ok | {error, term()}. -cache_delete(Id) when is_integer(Id) -> - dets:delete(?CACHE_TAB, Id). - --spec generate_cache_id() -> integer(). -generate_cache_id() -> - erlang:unique_integer([monotonic, positive]). - --spec cache_over_limit() -> boolean(). -cache_over_limit() -> - cache_item_count() > ?MAX_CACHE_ITEMS orelse cache_file_size() > ?MAX_CACHE_FILE_SIZE. - --spec cache_item_count() -> non_neg_integer(). -cache_item_count() -> - dets:info(?CACHE_TAB, size). - --spec cache_file_size() -> non_neg_integer(). -cache_file_size() -> - dets:info(?CACHE_TAB, file_size). - --spec trim_cache_limits(non_neg_integer()) -> {ok, non_neg_integer()} | {error, term()}. -trim_cache_limits(DroppedCount) -> - case cache_over_limit() of - true -> - case delete_oldest_cache_entry() of - ok -> - trim_cache_limits(DroppedCount + 1); - error -> - {ok, DroppedCount}; - {error, Reason} -> - {error, Reason} - end; - false -> - {ok, DroppedCount} - end. - --spec delete_oldest_cache_entry() -> ok | error | {error, term()}. -delete_oldest_cache_entry() -> - case dets:first(?CACHE_TAB) of - '$end_of_table' -> - error; - Key -> - dets:delete(?CACHE_TAB, Key) - end. - -spec handle_container_response(ssl:sslsocket(), reference(), term()) -> ok. handle_container_response(Socket, Ref, Reply) -> Packet = term_to_binary({response, Ref, {container_response, Reply}}), @@ -433,22 +342,4 @@ to_bool(1) -> to_bool(false) -> false; to_bool(0) -> - false. - --spec update_container_config(binary(), binary()) -> ok | {error, binary()}. -update_container_config(ContainerName, Config) when is_binary(ContainerName), is_binary(Config) -> - {ok, RootDir} = application:get_env(efka, root_dir), - case docker_helper:get_container_dir(RootDir, ContainerName) of - {ok, ContainerDir} -> - ConfigFile = docker_helper:get_config_file(ContainerDir), - case file:write_file(ConfigFile, Config, [write, binary]) of - ok -> - logger:warning("[efka_client] write config file: ~p success", [ConfigFile]), - ok; - {error, Reason} -> - logger:warning("[efka_client] write config file: ~p, get error: ~p", [ConfigFile, Reason]), - {error, <<"write config failed">>} - end; - error -> - {error, <<"error">>} - end. + false. \ No newline at end of file diff --git a/src/transport/efka_client_cache.erl b/src/transport/efka_client_cache.erl new file mode 100644 index 0000000..a352ebb --- /dev/null +++ b/src/transport/efka_client_cache.erl @@ -0,0 +1,103 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2026, +%%% @doc +%%% DETS backed outbound packet cache for efka_client. +%%% @end +%%%------------------------------------------------------------------- +-module(efka_client_cache). +-author("anlicheng"). + +-define(CACHE_TAB, cache). +-define(MAX_CACHE_ITEMS, 5000000). +-define(MAX_CACHE_FILE_SIZE, 1073741824). + +-export([open/0, close/0, insert/1, fetch_next/0, delete/1]). + +-spec open() -> ok | {error, term()}. +open() -> + {ok, DetsDir} = application:get_env(efka, dets_dir), + File = DetsDir ++ "cache.dets", + case dets:open_file(?CACHE_TAB, [{file, File}, {type, bag}, {keypos, 1}]) of + {ok, ?CACHE_TAB} -> + ok; + {error, Reason} -> + {error, Reason} + end. + +-spec close() -> ok. +close() -> + case dets:close(?CACHE_TAB) of + ok -> + ok; + {error, not_owner} -> + ok + end. + +-spec insert(binary()) -> {ok, non_neg_integer()} | {error, term()}. +insert(Data) when is_binary(Data) -> + case dets:insert(?CACHE_TAB, {generate_cache_id(), Data}) of + ok -> + trim_limits(0); + {error, Reason} -> + {error, Reason} + end. + +-spec fetch_next() -> error | {ok, {integer(), binary()}}. +fetch_next() -> + case dets:first(?CACHE_TAB) of + '$end_of_table' -> + error; + Key -> + case dets:lookup(?CACHE_TAB, Key) of + [Entry | _] -> + {ok, Entry}; + [] -> + fetch_next() + end + end. + +-spec delete(integer()) -> ok | {error, term()}. +delete(Id) when is_integer(Id) -> + dets:delete(?CACHE_TAB, Id). + +-spec generate_cache_id() -> integer(). +generate_cache_id() -> + erlang:unique_integer([monotonic, positive]). + +-spec over_limit() -> boolean(). +over_limit() -> + item_count() > ?MAX_CACHE_ITEMS orelse file_size() > ?MAX_CACHE_FILE_SIZE. + +-spec item_count() -> non_neg_integer(). +item_count() -> + dets:info(?CACHE_TAB, size). + +-spec file_size() -> non_neg_integer(). +file_size() -> + dets:info(?CACHE_TAB, file_size). + +-spec trim_limits(non_neg_integer()) -> {ok, non_neg_integer()} | {error, term()}. +trim_limits(DroppedCount) -> + case over_limit() of + true -> + case delete_oldest_entry() of + ok -> + trim_limits(DroppedCount + 1); + error -> + {ok, DroppedCount}; + {error, Reason} -> + {error, Reason} + end; + false -> + {ok, DroppedCount} + end. + +-spec delete_oldest_entry() -> ok | error | {error, term()}. +delete_oldest_entry() -> + case dets:first(?CACHE_TAB) of + '$end_of_table' -> + error; + Key -> + dets:delete(?CACHE_TAB, Key) + end.