diff --git a/src/docker/docker_container_service.erl b/src/docker/docker_container_service.erl new file mode 100644 index 0000000..7ed8cb9 --- /dev/null +++ b/src/docker/docker_container_service.erl @@ -0,0 +1,83 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2026, +%%% @doc +%%% +%%% @end +%%% Created : 20. 4月 2026 +%%%------------------------------------------------------------------- +-module(docker_container_service). +-author("anlicheng"). +-include("message_pb.hrl"). + +%% API +-export([handle_request/1]). + +-spec handle_request(message_pb:'ContainerRequest'()) -> ok | {ok, binary()} | {error, binary()}. +handle_request(#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}) -> + case docker_commands:get_containers() of + {ok, Containers} -> + {ok, jiffy:encode(Containers, [force_utf8])}; + {error, Reason} when is_binary(Reason) -> + {error, Reason} + end; +handle_request(#'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{task_id = TaskId, params = Params}}}) -> + docker_deploy_manager:deploy(TaskId, Params); +handle_request(#'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = Target}}}) -> + ContainerTarget = container_target(Target), + docker_commands:start_container(ContainerTarget); +handle_request(#'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{target = Target, timeout_seconds = TimeoutSeconds}}}) -> + ContainerTarget = container_target(Target), + docker_commands:stop_container(ContainerTarget, TimeoutSeconds); +handle_request(#'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{target = Target, signal = Signal}}}) -> + ContainerTarget = container_target(Target), + docker_commands:kill_container(ContainerTarget, to_binary(Signal)); +handle_request(#'ContainerRequest'{action = {remove, #'ContainerRequest.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(#'ContainerRequest'{action = {config, #'ContainerRequest.Config'{target = Target, config = Config}}}) -> + ContainerTarget = container_target(Target), + update_container_config(ContainerTarget, iolist_to_binary(Config)). + +container_target(#'ContainerRef'{name = Name, id = Id}) -> + NameBin = to_binary(Name), + IdBin = to_binary(Id), + case NameBin of + <<>> -> + true = IdBin =/= <<>>, + IdBin; + _ -> + NameBin + end. + +to_binary(Value) when is_binary(Value) -> + Value; +to_binary(Value) when is_list(Value) -> + unicode:characters_to_binary(Value). + +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. diff --git a/src/docker/docker_deploy_manager.erl b/src/docker/docker_deploy_manager.erl new file mode 100644 index 0000000..8fd4bb5 --- /dev/null +++ b/src/docker/docker_deploy_manager.erl @@ -0,0 +1,86 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2026, +%%% @doc +%%% +%%% @end +%%% Created : 20. 4月 2026 +%%%------------------------------------------------------------------- +-module(docker_deploy_manager). +-author("anlicheng"). +-include("message_pb.hrl"). + +-behaviour(gen_server). + +%% API +-export([start_link/0, deploy/2]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). + +-record(state, { + root_dir :: string(), + task_map = #{} +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +-spec start_link() -> {ok, pid()} | ignore | {error, term()}. +start_link() -> + gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). + +-spec deploy(integer(), message_pb:'ContainerDeployParams'()) -> ok | {error, binary()}. +deploy(TaskId, Params) when is_integer(TaskId), is_record(Params, 'ContainerDeployParams') -> + gen_server:call(?SERVER, {deploy, TaskId, Params}). + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +init([]) -> + erlang:process_flag(trap_exit, true), + {ok, RootDir} = application:get_env(efka, root_dir), + {ok, #state{root_dir = RootDir}}. + +handle_call({deploy, TaskId, Params = #'ContainerDeployParams'{ + container_name = ContainerName, + container_dir = ContainerDir0 +}}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) -> + {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName, ContainerDir0), + {ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params), + logger:debug("[docker_deploy_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]), + {reply, ok, State#state{task_map = maps:put(TaskPid, TaskId, TaskMap)}}; +handle_call(_Request, _From, State) -> + {reply, ok, State}. + +handle_cast(_Request, State) -> + {noreply, State}. + +handle_info({'DOWN', _Ref, process, TaskPid, Reason}, State = #state{task_map = TaskMap}) -> + case maps:take(TaskPid, TaskMap) of + error -> + {noreply, State}; + {TaskId, NTaskMap} -> + case Reason of + normal -> + logger:debug("[docker_deploy_manager] task_id: ~p, exit normal", [TaskId]); + Error0 -> + Error = iolist_to_binary(io_lib:format("~p", [Error0])), + efka_task_reporter:stream(TaskId, <<"error">>, <<"任务失败: "/utf8, Error/binary>>), + efka_task_reporter:close(TaskId, <<"task exited">>), + logger:notice("[docker_deploy_manager] task_id: ~p, exit with error: ~p", [TaskId, Error]) + end, + {noreply, State#state{task_map = NTaskMap}} + end; +handle_info(_Info, State) -> + {noreply, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. diff --git a/src/docker/docker_deployer.erl b/src/docker/docker_deployer.erl index 58c27e6..889c1fd 100644 --- a/src/docker/docker_deployer.erl +++ b/src/docker/docker_deployer.erl @@ -12,7 +12,7 @@ -dialyzer([{nowarn_function, normalize_image/1}]). %% API --export([start_link/3]). +-export([start_monitor/3]). -export([deploy/3]). -define(TASK_SUCCESS, <<"success">>). @@ -22,12 +22,11 @@ %%% API %%%=================================================================== --spec(start_link(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) -> - {ok, pid()}). -start_link(TaskId, ContainerDir, Params) +-spec(start_monitor(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) -> + {ok, {pid(), reference()}}). +start_monitor(TaskId, ContainerDir, Params) when is_integer(TaskId), is_list(ContainerDir), is_record(Params, 'ContainerDeployParams') -> - Pid = spawn_link(?MODULE, deploy, [TaskId, ContainerDir, Params]), - {ok, Pid}. + {ok, spawn_monitor(?MODULE, deploy, [TaskId, ContainerDir, Params])}. %%%=================================================================== %%% Internal functions diff --git a/src/docker/docker_deployer_sup.erl b/src/docker/docker_deployer_sup.erl deleted file mode 100644 index 433a720..0000000 --- a/src/docker/docker_deployer_sup.erl +++ /dev/null @@ -1,57 +0,0 @@ -%%%------------------------------------------------------------------- -%%% @author anlicheng -%%% @copyright (C) 2026, -%%% @doc -%%% -%%% @end -%%% Created : 20. 4月 2026 -%%%------------------------------------------------------------------- --module(docker_deployer_sup). --author("anlicheng"). --include("message_pb.hrl"). - --behaviour(supervisor). - -%% API --export([start_link/0, start_deployer/3]). - -%% Supervisor callbacks --export([init/1]). - --define(SERVER, ?MODULE). - -%%%=================================================================== -%%% API -%%%=================================================================== - --spec start_link() -> {ok, pid()} | ignore | {error, term()}. -start_link() -> - supervisor:start_link({local, ?SERVER}, ?MODULE, []). - --spec start_deployer(integer(), string(), message_pb:'ContainerDeployParams'()) -> - {ok, pid()} | {error, term()}. -start_deployer(TaskId, ContainerDir, Params) - when is_integer(TaskId), is_list(ContainerDir), is_record(Params, 'ContainerDeployParams') -> - supervisor:start_child(?SERVER, child_spec(TaskId, ContainerDir, Params)). - -%%%=================================================================== -%%% Supervisor callbacks -%%%=================================================================== - -init([]) -> - SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600}, - {ok, {SupFlags, []}}. - -%%%=================================================================== -%%% Internal functions -%%%=================================================================== - -child_spec(TaskId, ContainerDir, Params) -> - #{ - id => make_ref(), - start => {docker_deployer, start_link, [TaskId, ContainerDir, Params]}, - restart => temporary, - shutdown => 5000, - type => worker, - modules => [docker_deployer] - }. diff --git a/src/docker/docker_manager.erl b/src/docker/docker_manager.erl deleted file mode 100644 index 5454ea0..0000000 --- a/src/docker/docker_manager.erl +++ /dev/null @@ -1,141 +0,0 @@ -%%%------------------------------------------------------------------- -%%% @author anlicheng -%%% @copyright (C) 2025, -%%% @doc -%%% 微服务守护进程 -%%% 1. 负责微服务的下载, 版本的管理 -%%% 2. 目录管理等 -%%% @end -%%% Created : 19. 4月 2025 14:55 -%%%------------------------------------------------------------------- --module(docker_manager). --author("anlicheng"). --include("message_pb.hrl"). - --behaviour(gen_server). - -%% API --export([start_link/0]). --export([deploy/2]). - -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). - --define(SERVER, ?MODULE). - --record(state, { - root_dir :: string(), - %% 建立任务到ref之间的映射, #{TaskPid => TaskId} - task_map = #{} -}). - -%%%=================================================================== -%%% API -%%%=================================================================== - --spec deploy(TaskId :: integer(), Params :: message_pb:'ContainerDeployParams'()) -> ok | {error, Reason :: binary()}. -deploy(TaskId, Params) when is_integer(TaskId), is_record(Params, 'ContainerDeployParams') -> - gen_server:call(?SERVER, {deploy, TaskId, Params}). - -%% @doc Spawns the server and registers the local name (unique) --spec(start_link() -> - {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). -start_link() -> - gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). - -%%%=================================================================== -%%% gen_server callbacks -%%%=================================================================== - -%% @private -%% @doc Initializes the server --spec(init(Args :: term()) -> - {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | - {stop, Reason :: term()} | ignore). -init([]) -> - erlang:process_flag(trap_exit, true), - {ok, RootDir} = application:get_env(efka, root_dir), - {ok, #state{root_dir = RootDir}}. - -%% @private -%% @doc Handling call messages --spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, - State :: #state{}) -> - {reply, Reply :: term(), NewState :: #state{}} | - {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_call({deploy, TaskId, Params = #'ContainerDeployParams'{ - container_name = ContainerName, - container_dir = ContainerDir0 -}}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) -> - %% 创建目录 - {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName, ContainerDir0), - {ok, TaskPid} = docker_deployer_sup:start_deployer(TaskId, ContainerDir, Params), - _Ref = erlang:monitor(process, TaskPid), - logger:debug("[docker_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]), - {reply, ok, State#state{task_map = maps:put(TaskPid, TaskId, TaskMap)}}; - -handle_call(_Request, _From, State = #state{}) -> - {reply, ok, State}. - -%% @private -%% @doc Handling cast messages --spec(handle_cast(Request :: term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_cast(_Request, State = #state{}) -> - {noreply, State}. - -%% @private -%% @doc Handling all non call/cast messages --spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_info({'DOWN', _Ref, process, TaskPid, Reason}, State = #state{task_map = TaskMap}) -> - case maps:take(TaskPid, TaskMap) of - error -> - {noreply, State}; - {TaskId, NTaskMap} -> - case Reason of - normal -> - logger:debug("[docker_manager] task_id: ~p, exit normal", [TaskId]), - ok; - Error0 -> - Error = iolist_to_binary(io_lib:format("~p", [Error0])), - efka_task_reporter:stream(TaskId, <<"error">>, <<"任务失败: "/utf8, Error/binary>>), - efka_task_reporter:close(TaskId, <<"task exited">>), - logger:notice("[docker_manager] task_id: ~p, exit with error: ~p", [TaskId, Error]), - ok - end, - {noreply, State#state{task_map = NTaskMap}} - end; - -handle_info(_Info, State = #state{}) -> - {noreply, State}. - -%% @private -%% @doc This function is called by a gen_server when it is about to -%% terminate. It should be the opposite of Module:init/1 and do any -%% necessary cleaning up. When it returns, the gen_server terminates -%% with Reason. The return value is ignored. --spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), - State :: #state{}) -> term()). -terminate(_Reason, _State = #state{}) -> - ok. - -%% @private -%% @doc Convert process state when code is changed --spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{}, - Extra :: term()) -> - {ok, NewState :: #state{}} | {error, Reason :: term()}). -code_change(_OldVsn, State = #state{}, _Extra) -> - {ok, State}. - -%%%=================================================================== -%%% Internal functions -%%%=================================================================== diff --git a/src/efka_client.erl b/src/efka_client.erl index adc1958..a568ec3 100644 --- a/src/efka_client.erl +++ b/src/efka_client.erl @@ -175,7 +175,8 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) -> %% 微服务部署 handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - case handle_container_request(Request) of + Result = docker_container_service:handle_request(Request), + case Result of ok -> send_result_reply(Socket, PacketId, <<"ok">>); {ok, Reply} -> @@ -298,32 +299,6 @@ disconnect(Socket) -> schedule_reconnect() -> erlang:start_timer(5000, self(), create_transport). --spec handle_container_request(message_pb:'ContainerRequest'()) -> ok. -handle_container_request(#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}) -> - case docker_commands:get_containers() of - {ok, Containers} -> - {ok, jiffy:encode(Containers, [force_utf8])}; - {error, Reason} when is_binary(Reason) -> - {error, Reason} - end; -handle_container_request(#'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{ task_id = TaskId, params = Params }}}) -> - docker_manager:deploy(TaskId, Params); -handle_container_request(#'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = Target}}}) -> - ContainerTarget = container_target(Target), - docker_commands:start_container(ContainerTarget); -handle_container_request(#'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{target = Target, timeout_seconds = TimeoutSeconds }}}) -> - ContainerTarget = container_target(Target), - docker_commands:stop_container(ContainerTarget, TimeoutSeconds); -handle_container_request(#'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{target = Target, signal = Signal}}}) -> - ContainerTarget = container_target(Target), - docker_commands:kill_container(ContainerTarget, to_binary(Signal)); -handle_container_request(#'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}) -> - ContainerTarget = container_target(Target), - docker_commands:remove_container(ContainerTarget, to_bool(Force), to_bool(RemoveVolumes)); -handle_container_request(#'ContainerRequest'{action = {config, #'ContainerRequest.Config'{target = Target, config = Config}}}) -> - ContainerTarget = container_target(Target), - update_container_config(ContainerTarget, iolist_to_binary(Config)). - -spec send_result_reply(ssl:sslsocket(), integer(), binary()) -> ok. send_result_reply(Socket, PacketId, Payload) when is_binary(Payload) -> Packet = message_pb:encode_msg(#'ReplyFrame'{ @@ -339,46 +314,3 @@ send_error_reply(Socket, PacketId, Reason) when is_binary(Reason) -> reply = {error, #'ReplyError'{code = -1, message = Reason}} }), send_packet(Socket, Packet). - -container_target(#'ContainerRef'{name = Name, id = Id}) -> - NameBin = to_binary(Name), - IdBin = to_binary(Id), - case NameBin of - <<>> -> - true = IdBin =/= <<>>, - IdBin; - _ -> - NameBin - end. - -to_binary(Value) when is_binary(Value) -> - Value; -to_binary(Value) when is_list(Value) -> - unicode:characters_to_binary(Value). - -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. diff --git a/src/efka_sup.erl b/src/efka_sup.erl index 69d5362..097d389 100644 --- a/src/efka_sup.erl +++ b/src/efka_sup.erl @@ -101,21 +101,12 @@ init([]) -> }, #{ - id => 'docker_deployer_sup', - start => {'docker_deployer_sup', start_link, []}, - restart => permanent, - shutdown => 2000, - type => supervisor, - modules => ['docker_deployer_sup'] - }, - - #{ - id => 'docker_manager', - start => {'docker_manager', start_link, []}, + id => 'docker_deploy_manager', + start => {'docker_deploy_manager', start_link, []}, restart => permanent, shutdown => 2000, type => worker, - modules => ['docker_manager'] + modules => ['docker_deploy_manager'] } ],