add agent client

2025-04-20 17:04:14 +08:00 · 2025-04-20 17:04:14 +08:00 · 9d473d26fb
commit 9d473d26fb
parent 80a34d271a
9 changed files with 252 additions and 23 deletions
--- 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,
--- a/apps/efka/src/efka_agent.erl
+++ b/apps/efka/src/efka_agent.erl
@ -0,0 +1,177 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @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.
--- 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) ->
--- 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).
+    filename:basename(Path).
+
--- 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.
--- 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, []},
--- 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, 支持日志按日期自动分割
--- 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"}}}
 ]}.
--- 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">>}]}
 ].