From 9d473d26fb156be90121deb591b54d8cabeebf85 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sun, 20 Apr 2025 17:04:14 +0800 Subject: [PATCH] add agent client --- apps/efka/src/efka.app.src | 2 + apps/efka/src/efka_agent.erl | 177 ++++++++++++++++++++++++++++++ apps/efka/src/efka_app.erl | 3 + apps/efka/src/efka_downloader.erl | 10 +- apps/efka/src/efka_inetd.erl | 30 ++++- apps/efka/src/efka_sup.erl | 9 ++ config/sys.config | 7 +- rebar.config | 3 +- rebar.lock | 34 +++--- 9 files changed, 252 insertions(+), 23 deletions(-) create mode 100644 apps/efka/src/efka_agent.erl diff --git a/apps/efka/src/efka.app.src b/apps/efka/src/efka.app.src index 32fda82..4415217 100644 --- a/apps/efka/src/efka.app.src +++ b/apps/efka/src/efka.app.src @@ -11,6 +11,8 @@ esockd, jiffy, gpb, + cowlib, + gun, mnesia, crypto, ssl, diff --git a/apps/efka/src/efka_agent.erl b/apps/efka/src/efka_agent.erl new file mode 100644 index 0000000..e2f4463 --- /dev/null +++ b/apps/efka/src/efka_agent.erl @@ -0,0 +1,177 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2025, +%%% @doc +%%% +%%% @end +%%% Created : 20. 4月 2025 16:25 +%%%------------------------------------------------------------------- +-module(efka_agent). +-author("anlicheng"). + +-behaviour(gen_server). + +%% API +-export([start_link/0]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). + +-define(DISCONNECTED, 0). +-define(CONNECTING, 1). +-define(CONNECTED, 2). + +-record(state, { + conn_pid :: pid() | undefined, + host :: string(), + port :: integer(), + status = ?DISCONNECTED +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +%% @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, Props} = application:get_env(efka, wss_server), + Host = proplists:get_value(host, Props), + Port = proplists:get_value(port, Props), + + %% 异步建立到服务器的链接 + erlang:start_timer(0, self(), create_connection), + + {ok, #state{host = Host, port = Port}}. + +%% @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(_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({timeout, _, create_connection}, State = #state{host = Host, port = Port}) -> + case connect(Host, Port) of + {ok, ConnPid} -> + {noreply, State#state{conn_pid = ConnPid, status = ?CONNECTING}}; + {error, Reason} -> + lager:debug("[efka_agent] connect host: ~p, port: ~p, get error: ~p", [Host, Port, Reason]), + retry_connect(), + {noreply, State#state{conn_pid = undefined, status = ?DISCONNECTED}} + end; + +%% upgrade逻辑处理 +handle_info({gun_upgrade, ConnPid, _StreamRef, [<<"websocket">>], Headers}, State = #state{conn_pid = ConnPid}) -> + lager:debug("[efka_agent] upgrade success, headers: ~p", [Headers]), + {noreply, State#state{status = ?CONNECTED}}; +handle_info({gun_response, ConnPid, _, _, Status, Headers}, State = #state{conn_pid = ConnPid}) -> + lager:debug("[efka_agent] upgrade failed, status: ~p, headers: ~p", [Status, Headers]), + retry_connect(), + gun:close(ConnPid), + {noreply, State#state{conn_pid = undefined, status = ?DISCONNECTED}}; + +%% 处理接受到的消息 +handle_info({gun_ws, ConnPid, {binary, Bin}}, State = #state{conn_pid = ConnPid}) -> + lager:debug("[efka_agent] get binary message: ~p", [Bin]), + {noreply, State}; + +handle_info({gun_ws, ConnPid, {text, Msg}}, State = #state{conn_pid = ConnPid}) -> + lager:debug("[efka_agent] get text message: ~p", [Msg]), + {noreply, State}; + +%% websocket链接关闭 +handle_info({gun_ws, ConnPid, close}, State = #state{conn_pid = ConnPid}) -> + lager:debug("[efka_agent] ws close"), + retry_connect(), + {noreply, State#state{conn_pid = undefined}}; + +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 +%%%=================================================================== + +retry_connect() -> + erlang:start_timer(5000, self(), create_connection). + +-spec connect(Host :: string(), Port :: integer()) -> {ok, ConnPid :: pid()} | {error, Reason :: any()}. +connect(Host, Port) when is_list(Host), is_integer(Port) -> + Opts = #{ + protocols => [http], + transport => tcp + }, + case gun:open(Host, Port, Opts) of + {ok, ConnPid} -> + %% 等待连接建立 + receive + {gun_up, ConnPid, http} -> + %% 发起 WebSocket 握手请求 + gun:ws_upgrade(ConnPid, "/", [ + {<<"connection">>, <<"upgrade">>}, + {<<"upgrade">>, <<"websocket">>}, + {<<"sec-websocket-version">>, <<"13">>}, + {<<"sec-websocket-key">>, base64:encode(crypto:strong_rand_bytes(16))} + ]), + {ok, ConnPid} + after 5000 -> + {error, connect_timeout} + end; + {error, Reason} -> + {error, Reason} + end. diff --git a/apps/efka/src/efka_app.erl b/apps/efka/src/efka_app.erl index b4845aa..53bb27d 100644 --- a/apps/efka/src/efka_app.erl +++ b/apps/efka/src/efka_app.erl @@ -19,6 +19,9 @@ start(_StartType, _StartArgs) -> %% 启动tcp的服务 start_tcp_server(), + %% 仓库的基础url + application:set_env(efka, repository_url, "http://118.178.229.213:3000/anlicheng/ekfa/"), + efka_sup:start_link(). stop(_State) -> diff --git a/apps/efka/src/efka_downloader.erl b/apps/efka/src/efka_downloader.erl index cc2693a..7789089 100644 --- a/apps/efka/src/efka_downloader.erl +++ b/apps/efka/src/efka_downloader.erl @@ -88,11 +88,16 @@ handle_cast({download, ReceiverPid, Ref, Url, TargetDir}, State = #state{}) -> ]}, TargetFile = get_filename_from_url(Url), + StartTs = os:timestamp(), case hackney:request(get, Url, [], <<>>, [async, {stream_to, self()}, {pool, false}, SSLOpts]) of {ok, ClientRef} -> case receive_data(ClientRef, TargetDir, TargetFile) of ok -> - ReceiverPid ! {download_response, Ref, ok}; + EndTs = os:timestamp(), + %% 计算操作的时间，单位为毫秒 + CostMs = timer:now_diff(EndTs, StartTs) div 1000, + lager:debug("download cost: ~p", [CostMs]), + ReceiverPid ! {download_response, Ref, {ok, CostMs}}; {error, Reason} -> ReceiverPid ! {download_response, Ref, {error, Reason}} end; @@ -186,4 +191,5 @@ extra_filename(Headers, Default) when is_list(Headers), is_list(Default) -> get_filename_from_url(Url) when is_list(Url) -> URIMap = uri_string:parse(Url), Path = maps:get(path, URIMap), - filename:basename(Path). \ No newline at end of file + filename:basename(Path). + diff --git a/apps/efka/src/efka_inetd.erl b/apps/efka/src/efka_inetd.erl index 908bdef..200927d 100644 --- a/apps/efka/src/efka_inetd.erl +++ b/apps/efka/src/efka_inetd.erl @@ -64,11 +64,16 @@ init([]) -> {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 + %% 创建目录 + {ok, ServerRootDir} = ensure_dirs(RootDir, ServerId, From), + case check_lock(ServerRootDir) of true -> ok; false -> + DownloadUrl = make_download_url(From), + {ok, TaskPid} = efka_downloader:start_link(), + efka_downloader:download(TaskPid, DownloadUrl, ServerRootDir), + ok end, @@ -117,7 +122,7 @@ code_change(_OldVsn, State = #state{}, _Extra) -> %%% Internal functions %%%=================================================================== --spec ensure_dirs(RootDir :: string(), ServerId :: binary(), From :: binary()) -> string(). +-spec ensure_dirs(RootDir :: string(), ServerId :: binary(), From :: binary()) -> {ok, ServerRootDir :: 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] -> @@ -125,9 +130,14 @@ ensure_dirs(RootDir, ServerId, From) when is_list(RootDir), is_binary(ServerId), [Name0, Branch0 | _] -> {Name0, Branch0} end, - DirName = RootDir ++ Name ++ "/" ++ Branch ++ "/" ++ binary_to_list(ServerId) ++ "/", - ok = filelib:ensure_dir(DirName), - DirName. + %% 根目录 + ServerRootDir = RootDir ++ Name ++ "/" ++ Branch ++ "/" ++ binary_to_list(ServerId) ++ "/", + ok = filelib:ensure_dir(ServerRootDir), + %% 工作逻辑，压缩文件需要解压到工作目录 + WorkDir = ServerRootDir ++ "/work_dir/", + ok = filelib:ensure_dir(WorkDir), + + {ok, ServerRootDir}. -spec check_lock(DirName :: string()) -> boolean(). check_lock(DirName) when is_list(DirName) -> @@ -137,3 +147,11 @@ check_lock(DirName) when is_list(DirName) -> %% 下载文件 download_file(From) when is_binary(From) -> ok. + +%% From的格式: dianbiao:1.0 +-spec make_download_url(From :: binary()) -> string(). +make_download_url(From) when is_binary(From) -> + {ok, BaseUrl} = application:get_env(efka, repository_url), + From1 = binary_to_list(From), + Basename = lists:flatten(string:replace(From1, ":", "-")) ++ ".tar.gz", + BaseUrl ++ Basename. \ No newline at end of file diff --git a/apps/efka/src/efka_sup.erl b/apps/efka/src/efka_sup.erl index 15b6828..ef350f9 100644 --- a/apps/efka/src/efka_sup.erl +++ b/apps/efka/src/efka_sup.erl @@ -28,6 +28,15 @@ start_link() -> init([]) -> SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600}, ChildSpecs = [ + #{ + id => 'efka_agent', + start => {'efka_agent', start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => ['efka_agent'] + }, + #{ id => 'efka_server_sup', start => {'efka_server_sup', start_link, []}, diff --git a/config/sys.config b/config/sys.config index daf0749..3a14d74 100644 --- a/config/sys.config +++ b/config/sys.config @@ -1,6 +1,11 @@ [ {efka, [ - {root_dir, "/tmp/efka/"} + {root_dir, "/tmp/efka/"}, + + {wss_server, [ + {host, "localhost"}, + {port, 18080} + ]} ]}, %% 系统日志配置，系统日志为lager, 支持日志按日期自动分割 diff --git a/rebar.config b/rebar.config index fa04a83..5de6a1a 100644 --- a/rebar.config +++ b/rebar.config @@ -4,7 +4,8 @@ {sync, ".*", {git, "https://github.com/rustyio/sync.git", {branch, "master"}}}, {esockd, ".*", {git, "https://github.com/emqx/esockd.git", {tag, "v5.8.0"}}}, {jiffy, ".*", {git, "https://github.com/davisp/jiffy.git", {tag, "1.1.2"}}}, - {gpb, ".*", {git, "https://github.com/tomas-abrahamsson/gpb.git", {tag, "4.21.1"}}}, + {gpb, ".*", {git, "https://github.com/tomas-abrahamsson/gpb.git", {tag, "4.20.0"}}}, + {gun, ".*", {git, "https://github.com/ninenines/gun.git", {tag, "2.2.0"}}}, {parse_trans, ".*", {git, "https://github.com/uwiger/parse_trans", {tag, "3.0.0"}}}, {lager, ".*", {git,"https://github.com/erlang-lager/lager.git", {tag, "3.9.2"}}} ]}. diff --git a/rebar.lock b/rebar.lock index e2973bf..4492df8 100644 --- a/rebar.lock +++ b/rebar.lock @@ -1,23 +1,31 @@ {"1.2.0", -[{<<"certifi">>,{pkg,<<"certifi">>,<<"2.5.2">>},1}, +[{<<"certifi">>,{pkg,<<"certifi">>,<<"2.5.3">>},1}, + {<<"cowlib">>, + {git,"https://github.com/ninenines/cowlib", + {ref,"f8d0ad7f19b5dddd33cbfb089ebd2e2be2a81a5d"}}, + 1}, {<<"esockd">>, {git,"https://github.com/emqx/esockd.git", - {ref,"d9ce4024cc42a65e9a05001997031e743442f955"}}, + {ref,"9b959fc11a1c398a589892f335235be6c5b4a454"}}, 0}, {<<"fs">>,{pkg,<<"fs">>,<<"6.1.1">>},1}, {<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},1}, {<<"gpb">>, {git,"https://github.com/tomas-abrahamsson/gpb.git", - {ref,"a53bc4909b3dc5a78b996263d92db38fed9d4782"}}, + {ref,"edda1006d863a09509673778c455d33d88e6edbc"}}, + 0}, + {<<"gun">>, + {git,"https://github.com/ninenines/gun.git", + {ref,"627b8f9ed65da255afaddd166b1b9d102e0fa512"}}, 0}, {<<"hackney">>, {git,"https://github.com/benoitc/hackney.git", - {ref,"f3e9292db22c807e73f57a8422402d6b423ddf5f"}}, + {ref,"f642ee0eccaff9aa15707ae3a3effa29f2920e41"}}, 0}, - {<<"idna">>,{pkg,<<"idna">>,<<"6.0.1">>},1}, + {<<"idna">>,{pkg,<<"idna">>,<<"6.1.1">>},1}, {<<"jiffy">>, {git,"https://github.com/davisp/jiffy.git", - {ref,"9ea1b35b6e60ba21dfd4adbd18e7916a831fd7d4"}}, + {ref,"50daa80a62a97ffb6dd46ea9cb8ccb4930cbf1ae"}}, 0}, {<<"lager">>, {git,"https://github.com/erlang-lager/lager.git", @@ -34,24 +42,24 @@ {git,"https://github.com/rustyio/sync.git", {ref,"7dc303ed4ce8d26db82e171dbbd7c41067852c65"}}, 0}, - {<<"unicode_util_compat">>,{pkg,<<"unicode_util_compat">>,<<"0.5.0">>},2}]}. + {<<"unicode_util_compat">>,{pkg,<<"unicode_util_compat">>,<<"0.7.0">>},1}]}. [ {pkg_hash,[ - {<<"certifi">>, <<"B7CFEAE9D2ED395695DD8201C57A2D019C0C43ECAF8B8BCB9320B40D6662F340">>}, + {<<"certifi">>, <<"70BDD7E7188C804F3A30EE0E7C99655BC35D8AC41C23E12325F36AB449B70651">>}, {<<"fs">>, <<"9D147B944D60CFA48A349F12D06C8EE71128F610C90870BDF9A6773206452ED0">>}, {<<"goldrush">>, <<"F06E5D5F1277DA5C413E84D5A2924174182FB108DABB39D5EC548B27424CD106">>}, - {<<"idna">>, <<"1D038FB2E7668CE41FBF681D2C45902E52B3CB9E9C77B55334353B222C2EE50C">>}, + {<<"idna">>, <<"8A63070E9F7D0C62EB9D9FCB360A7DE382448200FBBD1B106CC96D3D8099DF8D">>}, {<<"metrics">>, <<"25F094DEA2CDA98213CECC3AEFF09E940299D950904393B2A29D191C346A8486">>}, {<<"mimerl">>, <<"67E2D3F571088D5CFD3E550C383094B47159F3EEE8FFA08E64106CDF5E981BE3">>}, {<<"ssl_verify_fun">>, <<"CF344F5692C82D2CD7554F5EC8FD961548D4FD09E7D22F5B62482E5AEAEBD4B0">>}, - {<<"unicode_util_compat">>, <<"8516502659002CEC19E244EBD90D312183064BE95025A319A6C7E89F4BCCD65B">>}]}, + {<<"unicode_util_compat">>, <<"BC84380C9AB48177092F43AC89E4DFA2C6D62B40B8BD132B1059ECC7232F9A78">>}]}, {pkg_hash_ext,[ - {<<"certifi">>, <<"3B3B5F36493004AC3455966991EAF6E768CE9884693D9968055AEEEB1E575040">>}, + {<<"certifi">>, <<"ED516ACB3929B101208A9D700062D520F3953DA3B6B918D866106FFA980E1C10">>}, {<<"fs">>, <<"EF94E95FFE79916860649FED80AC62B04C322B0BB70F5128144C026B4D171F8B">>}, {<<"goldrush">>, <<"99CB4128CFFCB3227581E5D4D803D5413FA643F4EB96523F77D9E6937D994CEB">>}, - {<<"idna">>, <<"A02C8A1C4FD601215BB0B0324C8A6986749F807CE35F25449EC9E69758708122">>}, + {<<"idna">>, <<"92376EB7894412ED19AC475E4A86F7B413C1B9FBB5BD16DCCD57934157944CEA">>}, {<<"metrics">>, <<"69B09ADDDC4F74A40716AE54D140F93BEB0FB8978D8636EADED0C31B6F099F16">>}, {<<"mimerl">>, <<"F278585650AA581986264638EBF698F8BB19DF297F66AD91B18910DFC6E19323">>}, {<<"ssl_verify_fun">>, <<"BDB0D2471F453C88FF3908E7686F86F9BE327D065CC1EC16FA4540197EA04680">>}, - {<<"unicode_util_compat">>, <<"D48D002E15F5CC105A696CF2F1BBB3FC72B4B770A184D8420C8DB20DA2674B38">>}]} + {<<"unicode_util_compat">>, <<"25EEE6D67DF61960CF6A794239566599B09E17E668D3700247BC498638152521">>}]} ].