anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix efka

Browse Source
This commit is contained in:
anlicheng 2026-04-26 15:57:18 +08:00
parent 30069648a1
commit 78074303de
2 changed files with 118 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
src/docker/docker_container_service.erl
View File

@ -1,85 +0,0 @@
%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2026, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 20. 4月 2026
%%%-------------------------------------------------------------------
-module(docker_container_service).
-author("anlicheng").
%% API
-export([handle_request/1]).
-spec handle_request(map()) -> ok | {ok, binary()} | {error, binary()}.
handle_request(#{action := list}) ->
case docker_commands:get_containers() of
{ok, Containers} ->
{ok, iolist_to_binary(json:encode(Containers))};
{error, Reason} when is_binary(Reason) ->
{error, Reason}
end;
handle_request(#{action := deploy, task_id := TaskId, params := Params}) ->
docker_deploy_manager:deploy(TaskId, Params);
handle_request(#{action := start, target := Target}) ->
ContainerTarget = container_target(Target),
docker_commands:start_container(ContainerTarget);
handle_request(#{action := stop, target := Target, timeout_seconds := TimeoutSeconds}) ->
ContainerTarget = container_target(Target),
docker_commands:stop_container(ContainerTarget, TimeoutSeconds);
handle_request(#{action := kill, target := Target, signal := Signal}) ->
ContainerTarget = container_target(Target),
docker_commands:kill_container(ContainerTarget, to_binary(Signal));
handle_request(#{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}) ->
ContainerTarget = container_target(Target),
docker_commands:remove_container(ContainerTarget, to_bool(Force), to_bool(RemoveVolumes));
handle_request(#{action := config, target := Target, config := Config}) ->
ContainerTarget = container_target(Target),
update_container_config(ContainerTarget, iolist_to_binary(Config)).
-spec container_target(map()) -> binary().
container_target(Target) when is_map(Target) ->
NameBin = to_binary(maps:get(name, Target, <<>>)),
IdBin = to_binary(maps:get(id, Target, <<>>)),
case NameBin of
<<>> ->
true = IdBin =/= <<>>,
IdBin;
_ ->
NameBin
end.
-spec to_binary(binary() | list()) -> binary().
to_binary(Value) when is_binary(Value) ->
Value;
to_binary(Value) when is_list(Value) ->
unicode:characters_to_binary(Value).
-spec to_bool(true | false | 0 | 1) -> boolean().
to_bool(true) ->
true;
to_bool(1) ->
true;
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("[docker_container_service] write config file: ~p success", [ConfigFile]),
ok;
{error, Reason} ->
logger:warning("[docker_container_service] write config file: ~p, get error: ~p", [ConfigFile, Reason]),
{error, <<"write config failed">>}
end;
error ->
{error, <<"error">>}
end.

169
src/transport/efka_client.erl
View File

@ -162,26 +162,9 @@ handle_event(info, flush_cache, _, State) ->
{keep_state, State};
%% 处理收到的ssl消息
handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket})
when is_binary(PacketBin) ->
case binary_to_term(PacketBin, [safe]) of
{request, PacketId, {container_request, Request}} ->
{keep_state, State, [{next_event, internal, {decoded_request, PacketId, Request}}]};
{request, PacketId, Body} ->
{keep_state, State, [{next_event, internal, {decoded_request_invalid, PacketId, Body}}]};
{response, AuthPacketId, {ok, Message}} ->
{keep_state, State, [{next_event, internal, {decoded_auth_ok, AuthPacketId, Message}}]};
{response, AuthPacketId, {error, 1, Message}} ->
{keep_state, State, [{next_event, internal, {decoded_auth_denied, AuthPacketId, Message}}]};
{response, AuthPacketId, {error, Code, Message}} ->
{keep_state, State, [{next_event, internal, {decoded_auth_error, AuthPacketId, Code, Message}}]};
{message, {auth_control, Cmd}} ->
{keep_state, State, [{next_event, internal, {decoded_auth_control, Cmd}}]};
{message, {pub, #{topic := Topic, qos := Qos, content := Content}}} ->
{keep_state, State, [{next_event, internal, {decoded_pub, Topic, Qos, Content}}]};
Packet ->
{keep_state, State, [{next_event, internal, {decoded_unknown, Packet}}]}
end;
handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
Packet = binary_to_term(PacketBin, [safe]),
{keep_state, State, [{next_event, internal, Packet}]};
handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) ->
logger:debug("[efka_client] ssl error: ~p", [Reason]),
disconnect(Socket),
@ -191,64 +174,102 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
%%% 处理内部消息，ssl收到的消息会解析成protobuf的消息格式，并按照internal类型处理
%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
%% 微服务部署
handle_event(internal, {decoded_request, PacketId, Request},
%% 容器管理请求
handle_event(internal, {request, PacketId, {container_request, #{action := list}}},
?STATE_ACTIVATED, State = #state{socket = Socket}) ->
case docker_container_service:handle_request(Request) of
ok ->
send_result_reply(Socket, PacketId, <<"ok">>);
{ok, Reply} ->
send_result_reply(Socket, PacketId, Reply);
case docker_commands:get_containers() of
{ok, Containers} ->
send_result_reply(Socket, PacketId, iolist_to_binary(json:encode(Containers)));
{error, Reason} ->
send_error_reply(Socket, PacketId, Reason)
end,
{keep_state, State};
handle_event(internal, {decoded_request_invalid, PacketId, _Body},
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,
{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,
{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,
{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,
{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,
{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,
{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, {decoded_request_invalid, PacketId, _Body},
_StateName, State = #state{socket = Socket}) ->
send_error_reply(Socket, PacketId, <<"agent state invalid">>),
{keep_state, State};
handle_event(internal, {decoded_request, PacketId, _Request},
?STATE_RESTRICTED, State = #state{socket = Socket}) ->
send_error_reply(Socket, PacketId, <<"agent restricted">>),
{keep_state, State};
handle_event(internal, {decoded_request, PacketId, _Request},
handle_event(internal, {request, PacketId, _Body},
_StateName, State = #state{socket = Socket}) ->
send_error_reply(Socket, PacketId, <<"agent state invalid">>),
{keep_state, State};
handle_event(internal, {decoded_auth_ok, AuthPacketId, Message},
handle_event(internal, {response, AuthPacketId, {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, {decoded_auth_denied, AuthPacketId, Message},
handle_event(internal, {response, AuthPacketId, {error, 1, Message}},
?STATE_AUTH, State = #state{auth_packet_id = AuthPacketId}) ->
logger:debug("[efka_client] auth denied, message: ~p", [Message]),
{next_state, ?STATE_RESTRICTED, State};
handle_event(internal, {decoded_auth_error, AuthPacketId, _Code, Message},
handle_event(internal, {response, AuthPacketId, {error, _Code, Message}},
?STATE_AUTH, State = #state{socket = Socket, auth_packet_id = AuthPacketId}) ->
logger:debug("[efka_client] auth failed, message: ~p", [Message]),
disconnect(Socket),
schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
handle_event(internal, {decoded_auth_ok, _PacketId, Reply}, StateName, State) ->
logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
{keep_state, State};
handle_event(internal, {decoded_auth_denied, _PacketId, Reply}, StateName, State) ->
logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
{keep_state, State};
handle_event(internal, {decoded_auth_error, _PacketId, _Code, Reply}, StateName, State) ->
handle_event(internal, {response, _PacketId, Reply}, StateName, State) ->
logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
{keep_state, State};
%% 处理命令
handle_event(internal, {decoded_auth_control, Cmd},
handle_event(internal, {message, {auth_control, Cmd}},
StateName, State = #state{socket = Socket, next_packet_id = PacketId}) ->
logger:debug("[efka_client] auth cmd: ~p", [Cmd]),
@ -265,11 +286,11 @@ handle_event(internal, {decoded_auth_control, Cmd},
end;
%% 处理Pub/Sub机制
handle_event(internal, {decoded_pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State) ->
handle_event(internal, {message, {pub, #{topic := Topic, qos := Qos, content := Content}}}, ?STATE_ACTIVATED, State) ->
logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
efka_subscription:publish(Topic, Qos, Content),
{keep_state, State};
handle_event(internal, {decoded_unknown, Packet}, _StateName, State) ->
handle_event(internal, Packet, _StateName, State) ->
logger:warning("[efka_client] ignore unknown packet: ~p", [Packet]),
{keep_state, State};
@ -429,3 +450,49 @@ send_result_reply(Socket, PacketId, Payload) when is_binary(Payload) ->
send_error_reply(Socket, PacketId, Reason) when is_binary(Reason) ->
Packet = term_to_binary({response, PacketId, {error, -1, Reason}}),
send_packet(Socket, Packet).
-spec container_target(map()) -> binary().
container_target(Target) when is_map(Target) ->
NameBin = to_binary(maps:get(name, Target, <<>>)),
IdBin = to_binary(maps:get(id, Target, <<>>)),
case NameBin of
<<>> ->
true = IdBin =/= <<>>,
IdBin;
_ ->
NameBin
end.
-spec to_binary(binary() | list()) -> binary().
to_binary(Value) when is_binary(Value) ->
Value;
to_binary(Value) when is_list(Value) ->
unicode:characters_to_binary(Value).
-spec to_bool(true | false | 0 | 1) -> boolean().
to_bool(true) ->
true;
to_bool(1) ->
true;
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.