代码逻辑调整

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)).

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.

src/transport/efka_client_cache.erl

@@ -0,0 +1,103 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2026, <COMPANY>
+%%% @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.