From 4168c5d4a834b00abe693c93d52909d42027cb4d Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 19 Apr 2025 17:09:52 +0800 Subject: [PATCH] fix --- apps/efka/src/efka.app.src | 3 + apps/efka/src/efka_downloader.erl | 148 ++++++++++++++++++++++++++++++ apps/efka/src/efka_inetd.erl | 139 ++++++++++++++++++++++++++++ apps/efka/src/efka_server.erl | 15 ++- apps/efka/src/efka_server_sup.erl | 6 +- apps/efka/src/efka_util.erl | 18 ++++ config/sys.config | 42 ++++++++- efkafile.json | 14 +++ 8 files changed, 379 insertions(+), 6 deletions(-) create mode 100644 apps/efka/src/efka_downloader.erl create mode 100644 apps/efka/src/efka_inetd.erl create mode 100644 apps/efka/src/efka_util.erl create mode 100644 efkafile.json diff --git a/apps/efka/src/efka.app.src b/apps/efka/src/efka.app.src index c4f5ccb..32fda82 100644 --- a/apps/efka/src/efka.app.src +++ b/apps/efka/src/efka.app.src @@ -6,12 +6,15 @@ {applications, [ sync, + hackney, lager, esockd, jiffy, gpb, mnesia, crypto, + ssl, + public_key, kernel, stdlib ]}, diff --git a/apps/efka/src/efka_downloader.erl b/apps/efka/src/efka_downloader.erl new file mode 100644 index 0000000..8c732cb --- /dev/null +++ b/apps/efka/src/efka_downloader.erl @@ -0,0 +1,148 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2025, +%%% @doc +%%% +%%% @end +%%% Created : 19. 4月 2025 16:48 +%%%------------------------------------------------------------------- +-module(efka_downloader). +-author("anlicheng"). + +-behaviour(gen_server). + +%% API +-export([start_link/2, download/1]). +-export([test/0]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-record(state, { + url :: string(), + target_file :: string() +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +test() -> + {ok, Pid} = start_link("https://codeload.github.com/genadyo/LivePhotoDemo/zip/refs/heads/master", "/tmp/test.tar"), + Ref = download(Pid), + receive + {download_response, Ref, Info} -> + lager:debug("info is: ~p", [Info]) + end, + ok. + +-spec download(Pid :: pid()) -> reference(). +download(Pid) when is_pid(Pid) -> + Ref = make_ref(), + ReceiverPid = self(), + gen_server:cast(Pid, {download, ReceiverPid, Ref}), + Ref. + +%% @doc Spawns the server and registers the local name (unique) +-spec(start_link(Url :: string(), TargetFile :: string()) -> + {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). +start_link(Url, TargetFile) when is_list(Url), is_list(TargetFile) -> + gen_server:start_link(?MODULE, [Url, TargetFile], []). + +%%%=================================================================== +%%% 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([Url, TargetFile]) -> + {ok, #state{url = Url, target_file = TargetFile}}. + +%% @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(_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({download, ReceiverPid, Ref}, State = #state{url = Url, target_file = TargetFile}) -> + case hackney:get(Url, [], <<>>, [async]) of + {ok, ClientRef} -> + {ok, File} = file:open(TargetFile, [write, binary]), + case receive_data(ClientRef, File) of + ok -> + ReceiverPid ! {download_response, Ref, ok}; + {error, Reason} -> + ReceiverPid ! {download_response, Ref, {error, Reason}} + end; + {error, Reason} -> + ReceiverPid ! {download_response, Ref, {error, Reason}} + end, + {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(_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 +%%%=================================================================== + +receive_data(ClientRef, File) -> + receive + {hackney_response, ClientRef, {headers, _Headers}} -> + receive_data(ClientRef, File); + {hackney_response, ClientRef, {status, 200, _Reason}} -> + receive_data(ClientRef, File); + {hackney_response, ClientRef, {status, StatusCode, _Reason}} -> + lager:debug("call me here"), + {error, {http_status, StatusCode}}; + {hackney_response, ClientRef, Data} -> + file:write(File, Data), + receive_data(ClientRef, File); + {hackney_response, ClientRef, done} -> + ok = file:close(File), + hackney:close(ClientRef), + ok; + {hackney_response, ClientRef, {error, Reason}} -> + ok = file:close(File), + {error, Reason} + end. \ No newline at end of file diff --git a/apps/efka/src/efka_inetd.erl b/apps/efka/src/efka_inetd.erl new file mode 100644 index 0000000..908bdef --- /dev/null +++ b/apps/efka/src/efka_inetd.erl @@ -0,0 +1,139 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2025, +%%% @doc +%%% 微服务守护进程 +%%% 1. 负责微服务的下载, 版本的管理 +%%% 2. 目录管理等 +%%% @end +%%% Created : 19. 4月 2025 14:55 +%%%------------------------------------------------------------------- +-module(efka_inetd). +-author("anlicheng"). + +-behaviour(gen_server). + +%% API +-export([start_link/0]). +-export([init_server/2]). +-export([ensure_dirs/3, check_lock/1]). + +%% 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() +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +init_server(ServerId, From) when is_binary(ServerId), is_binary(From) -> + gen_server:call(?SERVER, {init_server, ServerId, From}). + +%% @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([]) -> + {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({init_server, ServerId, From}, _From, State = #state{root_dir = RootDir}) -> + DirName = ensure_dirs(RootDir, ServerId, From), + case check_lock(DirName) of + true -> + ok; + false -> + ok + end, + + {reply, ok, State}; + +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(_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 +%%%=================================================================== + +-spec ensure_dirs(RootDir :: string(), ServerId :: binary(), From :: binary()) -> string(). +ensure_dirs(RootDir, ServerId, From) when is_list(RootDir), is_binary(ServerId), is_binary(From) -> + {Name, Branch} = case string:split(binary_to_list(From), ":") of + [Name0] -> + {Name0, "master"}; + [Name0, Branch0 | _] -> + {Name0, Branch0} + end, + DirName = RootDir ++ Name ++ "/" ++ Branch ++ "/" ++ binary_to_list(ServerId) ++ "/", + ok = filelib:ensure_dir(DirName), + DirName. + +-spec check_lock(DirName :: string()) -> boolean(). +check_lock(DirName) when is_list(DirName) -> + FileName = DirName ++ ".efka.lock", + filelib:is_file(FileName). + +%% 下载文件 +download_file(From) when is_binary(From) -> + ok. diff --git a/apps/efka/src/efka_server.erl b/apps/efka/src/efka_server.erl index 0194ae8..53e2242 100644 --- a/apps/efka/src/efka_server.erl +++ b/apps/efka/src/efka_server.erl @@ -2,7 +2,8 @@ %%% @author anlicheng %%% @copyright (C) 2025, %%% @doc -%%% +%%% 1. 需要管理服务的整个生命周期，包括: 启动，停止 +%%% 2. 需要监控服务的状态，通过port的方式 %%% @end %%% Created : 18. 4月 2025 16:50 %%%------------------------------------------------------------------- @@ -19,7 +20,11 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -record(state, { - server_id :: binary() + server_id :: binary(), + %% 通道id信息 + channel_pid :: pid() | undefined, + %% 当前服务的状态 + status = 0 }). %%%=================================================================== @@ -30,6 +35,12 @@ get_name(ServerId) when is_binary(ServerId) -> list_to_atom("efka_server:" ++ binary_to_list(ServerId)). +start_server(Pid) when is_pid(Pid) -> + gen_server:call(Pid, start_server). + +stop_server(Pid) when is_pid(Pid) -> + gen_server:call(Pid, stop_server). + %% @doc Spawns the server and registers the local name (unique) -spec(start_link(Name :: atom(), ServerId :: binary()) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). diff --git a/apps/efka/src/efka_server_sup.erl b/apps/efka/src/efka_server_sup.erl index 041f79f..0874791 100644 --- a/apps/efka/src/efka_server_sup.erl +++ b/apps/efka/src/efka_server_sup.erl @@ -13,6 +13,7 @@ %% API -export([start_link/0]). +-export([register_server/1, delete_server/1]). %% Supervisor callbacks -export([init/1]). @@ -51,8 +52,8 @@ init([]) -> %%% Internal functions %%%=================================================================== --spec start_server(ServerId :: binary()) -> {ok, Pid :: pid()} | {error, Reason :: any()}. -start_server(ServerId) when is_binary(ServerId) -> +-spec register_server(ServerId :: binary()) -> {ok, Pid :: pid()} | {error, Reason :: any()}. +register_server(ServerId) when is_binary(ServerId) -> case supervisor:start_child(?MODULE, child_spec(ServerId)) of {ok, Pid} when is_pid(Pid) -> {ok, Pid}; @@ -62,6 +63,7 @@ start_server(ServerId) when is_binary(ServerId) -> {error, Error} end. +-spec delete_server(ServerId :: binary()) -> ok. delete_server(ServerId) when is_binary(ServerId) -> ChildId = efka_server:get_name(ServerId), ok = supervisor:terminate_child(?MODULE, ChildId), diff --git a/apps/efka/src/efka_util.erl b/apps/efka/src/efka_util.erl new file mode 100644 index 0000000..9c2b389 --- /dev/null +++ b/apps/efka/src/efka_util.erl @@ -0,0 +1,18 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2025, +%%% @doc +%%% +%%% @end +%%% Created : 19. 4月 2025 16:33 +%%%------------------------------------------------------------------- +-module(efka_util). +-author("anlicheng"). + +%% API +-export([get_file_md5/1]). + +get_file_md5(FilePath) when is_list(FilePath) -> + {ok, FileData} = file:read_file(FilePath), + Md5Binary = crypto:hash(md5, FileData), + string:lowercase(lists:flatten([io_lib:format("~2.16.0b", [X]) || X <- binary_to_list(Md5Binary)])). \ No newline at end of file diff --git a/config/sys.config b/config/sys.config index 2c2e816..daf0749 100644 --- a/config/sys.config +++ b/config/sys.config @@ -1,3 +1,41 @@ [ - {efka, []} -]. + {efka, [ + {root_dir, "/tmp/efka/"} + ]}, + + %% 系统日志配置，系统日志为lager, 支持日志按日期自动分割 + {lager, [ + {colored, true}, + %% Whether to write a crash log, and where. Undefined means no crash logger. + {crash_log, "trade_hub.crash.log"}, + %% Maximum size in bytes of events in the crash log - defaults to 65536 + {crash_log_msg_size, 65536}, + %% Maximum size of the crash log in bytes, before its rotated, set + %% to 0 to disable rotation - default is 0 + {crash_log_size, 10485760}, + %% What time to rotate the crash log - default is no time + %% rotation. See the README for a description of this format. + {crash_log_date, "$D0"}, + %% Number of rotated crash logs to keep, 0 means keep only the + %% current one - default is 0 + {crash_log_count, 5}, + %% Whether to redirect error_logger messages into lager - defaults to true + {error_logger_redirect, true}, + + %% How big the gen_event mailbox can get before it is switched into sync mode + {async_threshold, 20}, + %% Switch back to async mode, when gen_event mailbox size decrease from `async_threshold' + %% to async_threshold - async_threshold_window + {async_threshold_window, 5}, + + {handlers, [ + %% debug | info | warning | error, 日志级别 + {lager_console_backend, debug}, + {lager_file_backend, [{file, "debug.log"}, {level, debug}, {size, 314572800}]}, + {lager_file_backend, [{file, "notice.log"}, {level, notice}, {size, 314572800}]}, + {lager_file_backend, [{file, "error.log"}, {level, error}, {size, 314572800}]}, + {lager_file_backend, [{file, "info.log"}, {level, info}, {size, 314572800}]} + ]} + + ]} +]. \ No newline at end of file diff --git a/efkafile.json b/efkafile.json new file mode 100644 index 0000000..671e58d --- /dev/null +++ b/efkafile.json @@ -0,0 +1,14 @@ +{ + "from": "dianbiao:1.0", + "server_id": "server_id:1234", + "args": { + + }, + "metrics": { + + }, + "env": { + + } + +} \ No newline at end of file