统一消息格式

This commit is contained in:
anlicheng 2026-04-27 11:25:24 +08:00
parent c9dd1cf90d
commit 6d6a9c4585

View File

@ -180,51 +180,51 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
handle_event(internal, {request, Ref, {container_request, #{action := list}}}, handle_event(internal, {request, Ref, {container_request, #{action := list}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:get_containers(), Reply = docker_commands:get_containers(),
handle_container_reply(Socket, Ref, Reply), handle_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, #{action := deploy, task_id := TaskId, params := Params}}}, handle_event(internal, {request, Ref, {container_request, #{action := deploy, task_id := TaskId, params := Params}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_deploy_manager:deploy(TaskId, Params), Reply = docker_deploy_manager:deploy(TaskId, Params),
handle_container_reply(Socket, Ref, Reply), handle_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, #{action := start, target := Target}}}, handle_event(internal, {request, Ref, {container_request, #{action := start, target := Target}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:start_container(container_target(Target)), Reply = docker_commands:start_container(container_target(Target)),
handle_container_reply(Socket, Ref, Reply), handle_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}}, handle_event(internal, {request, Ref, {container_request, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds), Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
handle_container_reply(Socket, Ref, Reply), handle_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, #{action := kill, target := Target, signal := Signal}}}, handle_event(internal, {request, Ref, {container_request, #{action := kill, target := Target, signal := Signal}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)), Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
handle_container_reply(Socket, Ref, Reply), handle_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}}, handle_event(internal, {request, Ref, {container_request, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)), Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
handle_container_reply(Socket, Ref, Reply), handle_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, #{action := config, target := Target, config := Config}}}, handle_event(internal, {request, Ref, {container_request, #{action := config, target := Target, config := Config}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) -> ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Reply = update_container_config(container_target(Target), iolist_to_binary(Config)), Reply = update_container_config(container_target(Target), iolist_to_binary(Config)),
handle_container_reply(Socket, Ref, Reply), handle_container_response(Socket, Ref, Reply),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> handle_event(internal, {request, Ref, {container_request, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]), logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]),
handle_container_reply(Socket, Ref, {error, <<"agent restricted">>}), handle_container_response(Socket, Ref, {error, <<"agent restricted">>}),
{keep_state, State}; {keep_state, State};
handle_event(internal, {request, Ref, {container_request, Request}}, _StateName, State = #state{socket = Socket}) -> handle_event(internal, {request, Ref, {container_request, Request}}, _StateName, State = #state{socket = Socket}) ->
logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]), logger:notice("[efka_client] get a invalid request: ~p, agent restricted", [Request]),
handle_container_reply(Socket, Ref, {error, <<"agent invalid">>}), handle_container_response(Socket, Ref, {error, <<"agent invalid">>}),
{keep_state, State}; {keep_state, State};
%% response %% response
handle_event(internal, {response, AuthRef, {auth_response, {ok, Message}}}, handle_event(internal, {response, AuthRef, {auth_response, ok}},
?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
logger:debug("[efka_client] auth success, message: ~p", [Message]), logger:debug("[efka_client] auth success"),
{next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]}; {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]};
handle_event(internal, {response, AuthRef, {auth_response, {error, {denied, Message}}}}, handle_event(internal, {response, AuthRef, {auth_response, {error, {denied, Message}}}},
?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
@ -410,15 +410,9 @@ delete_oldest_cache_entry() ->
dets:delete(?CACHE_TAB, Key) dets:delete(?CACHE_TAB, Key)
end. end.
-spec handle_container_reply(ssl:sslsocket(), reference(), term()) -> ok. -spec handle_container_response(ssl:sslsocket(), reference(), term()) -> ok.
handle_container_reply(Socket, Ref, ok) -> handle_container_response(Socket, Ref, Reply) ->
Packet = term_to_binary({response, Ref, {container_response, {ok, <<"ok">>}}}), Packet = term_to_binary({response, Ref, {container_response, Reply}}),
ok = ssl:send(Socket, Packet);
handle_container_reply(Socket, Ref, {ok, Response}) ->
Packet = term_to_binary({response, Ref, {container_response, {ok, Response}}}),
ok = ssl:send(Socket, Packet);
handle_container_reply(Socket, Ref, {error, Reason}) ->
Packet = term_to_binary({response, Ref, {container_response, {error, Reason}}}),
ok = ssl:send(Socket, Packet). ok = ssl:send(Socket, Packet).
-spec container_target(map()) -> binary(). -spec container_target(map()) -> binary().