From 0b52e3d8ad017ed4249a37a9141efcf6c95d9881 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sun, 26 Apr 2026 16:23:46 +0800 Subject: [PATCH] fix efka_client --- src/transport/efka_client.erl | 101 +++++++++++----------------------- 1 file changed, 33 insertions(+), 68 deletions(-) diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index 5aea791..5327d57 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -98,7 +98,7 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so Packet = term_to_binary({message, {data, #{route_key => RouteKey, metric => Metric}}}), case StateName of ?STATE_ACTIVATED -> - send_packet(Socket, Packet), + ok = ssl:send(Socket, Packet), {keep_state, State}; _ -> {ok, DroppedCount} = cache_insert(Packet), @@ -109,12 +109,12 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]), Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => Type, stream => Stream}}}), - send_packet(Socket, Packet), + ok = ssl:send(Socket, Packet), {keep_state, State}; handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => <<"close">>, stream => Reason}}}), - send_packet(Socket, Packet), + ok = ssl:send(Socket, Packet), {keep_state, State}; %% 其他情况下直接忽略 @@ -133,7 +133,7 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = case connect_socket() of {ok, Socket} -> AuthPacket = auth_packet(PacketId), - send_packet(Socket, AuthPacket), + ok = ssl:send(Socket, AuthPacket), {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_packet_id = PacketId, next_packet_id = PacketId + 1}, [{state_timeout, 5000, auth_timeout}]}; {error, _Reason} -> @@ -152,7 +152,7 @@ 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 {ok, {Id, Packet}} -> - send_packet(Socket, Packet), + ok = ssl:send(Socket, Packet), ok = cache_delete(Id), {keep_state, State, [{next_event, info, flush_cache}]}; error -> @@ -179,79 +179,47 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) -> %% 容器管理请求 handle_event(internal, {request, PacketId, {container_request, #{action := list}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case docker_commands:get_containers() of - {ok, Containers} -> - send_result_reply(Socket, PacketId, Containers); - {error, Reason} -> - send_error_reply(Socket, PacketId, Reason) - end, + Reply = docker_commands:get_containers(), + handle_container_reply(Socket, PacketId, Reply), {keep_state, State}; handle_event(internal, {request, PacketId, {container_request, #{action := deploy, task_id := TaskId, params := Params}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case docker_deploy_manager:deploy(TaskId, Params) of - ok -> - send_result_reply(Socket, PacketId, <<"ok">>); - {error, Reason} -> - send_error_reply(Socket, PacketId, Reason) - end, + Reply = docker_deploy_manager:deploy(TaskId, Params), + handle_container_reply(Socket, PacketId, Reply), {keep_state, State}; handle_event(internal, {request, PacketId, {container_request, #{action := start, target := Target}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case docker_commands:start_container(container_target(Target)) of - ok -> - send_result_reply(Socket, PacketId, <<"ok">>); - {error, Reason} -> - send_error_reply(Socket, PacketId, Reason) - end, + Reply = docker_commands:start_container(container_target(Target)), + handle_container_reply(Socket, PacketId, Reply), {keep_state, State}; handle_event(internal, {request, PacketId, {container_request, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case docker_commands:stop_container(container_target(Target), TimeoutSeconds) of - ok -> - send_result_reply(Socket, PacketId, <<"ok">>); - {error, Reason} -> - send_error_reply(Socket, PacketId, Reason) - end, + Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds), + handle_container_reply(Socket, PacketId, Reply), {keep_state, State}; handle_event(internal, {request, PacketId, {container_request, #{action := kill, target := Target, signal := Signal}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case docker_commands:kill_container(container_target(Target), to_binary(Signal)) of - ok -> - send_result_reply(Socket, PacketId, <<"ok">>); - {error, Reason} -> - send_error_reply(Socket, PacketId, Reason) - end, + Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)), + handle_container_reply(Socket, PacketId, Reply), {keep_state, State}; handle_event(internal, {request, PacketId, {container_request, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)) of - ok -> - send_result_reply(Socket, PacketId, <<"ok">>); - {error, Reason} -> - send_error_reply(Socket, PacketId, Reason) - end, + Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)), + handle_container_reply(Socket, PacketId, Reply), {keep_state, State}; handle_event(internal, {request, PacketId, {container_request, #{action := config, target := Target, config := Config}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case update_container_config(container_target(Target), iolist_to_binary(Config)) of - ok -> - send_result_reply(Socket, PacketId, <<"ok">>); - {error, Reason} -> - send_error_reply(Socket, PacketId, Reason) - end, + Reply = update_container_config(container_target(Target), iolist_to_binary(Config)), + handle_container_reply(Socket, PacketId, Reply), {keep_state, State}; -handle_event(internal, {request, PacketId, _Body}, - ?STATE_RESTRICTED, State = #state{socket = Socket}) -> - send_error_reply(Socket, PacketId, <<"agent restricted">>), - {keep_state, State}; -handle_event(internal, {request, PacketId, _Body}, - _StateName, State = #state{socket = Socket}) -> - send_error_reply(Socket, PacketId, <<"agent state invalid">>), +handle_event(internal, {request, PacketId, {container_request, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> + logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]), + handle_container_reply(Socket, PacketId, {error, <<"agent restricted">>}), {keep_state, State}; +%% 处理response handle_event(internal, {response, AuthPacketId, {auth_response, {ok, Message}}}, ?STATE_AUTH, State = #state{auth_packet_id = AuthPacketId}) -> - logger:debug("[efka_client] auth success, message: ~p", [Message]), {next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]}; handle_event(internal, {response, AuthPacketId, {auth_response, {error, 1, Message}}}, @@ -278,7 +246,7 @@ handle_event(internal, {message, {auth_control, Cmd}}, {keep_state, State}; {activate, _} -> AuthPacket = auth_packet(PacketId), - send_packet(Socket, AuthPacket), + ok = ssl:send(Socket, AuthPacket), {next_state, ?STATE_AUTH, State#state{auth_packet_id = PacketId, next_packet_id = PacketId + 1}, [{state_timeout, 5000, auth_timeout}]}; {deactivate, _} -> @@ -338,10 +306,6 @@ connect_socket() -> ], ssl:connect(Host, Port, SslOptions, 5000). --spec send_packet(ssl:sslsocket(), erlang:iodata()) -> ok. -send_packet(Socket, Packet) -> - ok = ssl:send(Socket, Packet). - -spec disconnect(undefined | ssl:sslsocket()) -> ok. disconnect(undefined) -> ok; @@ -441,15 +405,16 @@ delete_oldest_cache_entry() -> dets:delete(?CACHE_TAB, Key) end. --spec send_result_reply(ssl:sslsocket(), integer(), term()) -> ok. -send_result_reply(Socket, PacketId, Payload) -> - Packet = term_to_binary({response, PacketId, {container_response, {ok, Payload}}}), - send_packet(Socket, Packet). - --spec send_error_reply(ssl:sslsocket(), integer(), binary()) -> ok. -send_error_reply(Socket, PacketId, Reason) when is_binary(Reason) -> +-spec handle_container_reply(ssl:sslsocket(), integer(), term()) -> ok. +handle_container_reply(Socket, PacketId, ok) -> + Packet = term_to_binary({response, PacketId, {container_response, {ok, <<"ok">>}}}), + ok = ssl:send(Socket, Packet); +handle_container_reply(Socket, PacketId, {ok, Response}) -> + Packet = term_to_binary({response, PacketId, {container_response, {ok, Response}}}), + ok = ssl:send(Socket, Packet); +handle_container_reply(Socket, PacketId, {error, Reason}) -> Packet = term_to_binary({response, PacketId, {container_response, {error, -1, Reason}}}), - send_packet(Socket, Packet). + ok = ssl:send(Socket, Packet). -spec container_target(map()) -> binary(). container_target(Target) when is_map(Target) ->