fix efka_client

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
+    Reply = docker_commands:get_containers(),
-        {ok, Containers} ->
+    handle_container_reply(Socket, PacketId, Reply),
-            send_result_reply(Socket, PacketId, Containers);
-        {error, Reason} ->
-            send_error_reply(Socket, PacketId, Reason)
-    end,
     {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
+    Reply = docker_deploy_manager:deploy(TaskId, Params),
-        ok ->
+    handle_container_reply(Socket, PacketId, Reply),
-            send_result_reply(Socket, PacketId, <<"ok">>);
-        {error, Reason} ->
-            send_error_reply(Socket, PacketId, Reason)
-    end,
     {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
+    Reply = docker_commands:start_container(container_target(Target)),
-        ok ->
+    handle_container_reply(Socket, PacketId, Reply),
-            send_result_reply(Socket, PacketId, <<"ok">>);
-        {error, Reason} ->
-            send_error_reply(Socket, PacketId, Reason)
-    end,
     {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
+    Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
-        ok ->
+    handle_container_reply(Socket, PacketId, Reply),
-            send_result_reply(Socket, PacketId, <<"ok">>);
-        {error, Reason} ->
-            send_error_reply(Socket, PacketId, Reason)
-    end,
     {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
+    Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
-        ok ->
+    handle_container_reply(Socket, PacketId, Reply),
-            send_result_reply(Socket, PacketId, <<"ok">>);
-        {error, Reason} ->
-            send_error_reply(Socket, PacketId, Reason)
-    end,
     {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
+    Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
-        ok ->
+    handle_container_reply(Socket, PacketId, Reply),
-            send_result_reply(Socket, PacketId, <<"ok">>);
-        {error, Reason} ->
-            send_error_reply(Socket, PacketId, Reason)
-    end,
     {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
+    Reply = update_container_config(container_target(Target), iolist_to_binary(Config)),
-        ok ->
+    handle_container_reply(Socket, PacketId, Reply),
-            send_result_reply(Socket, PacketId, <<"ok">>);
-        {error, Reason} ->
-            send_error_reply(Socket, PacketId, Reason)
-    end,
     {keep_state, State};
-handle_event(internal, {request, PacketId, _Body},
+handle_event(internal, {request, PacketId, {container_request, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
-    ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
+    logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]),
-    send_error_reply(Socket, PacketId, <<"agent restricted">>),
+    handle_container_reply(Socket, PacketId, {error, <<"agent restricted">>}),
-    {keep_state, State};
-handle_event(internal, {request, PacketId, _Body},
-    _StateName, State = #state{socket = Socket}) ->
-    send_error_reply(Socket, PacketId, <<"agent state invalid">>),
     {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.
+-spec handle_container_reply(ssl:sslsocket(), integer(), term()) -> ok.
-send_result_reply(Socket, PacketId, Payload) ->
+handle_container_reply(Socket, PacketId, ok) ->
-    Packet = term_to_binary({response, PacketId, {container_response, {ok, Payload}}}),
+    Packet = term_to_binary({response, PacketId, {container_response, {ok, <<"ok">>}}}),
-    send_packet(Socket, Packet).
+    ok = ssl:send(Socket, Packet);
-
+handle_container_reply(Socket, PacketId, {ok, Response}) ->
--spec send_error_reply(ssl:sslsocket(), integer(), binary()) -> ok.
+    Packet = term_to_binary({response, PacketId, {container_response, {ok, Response}}}),
-send_error_reply(Socket, PacketId, Reason) when is_binary(Reason) ->
+    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) ->