%%%------------------------------------------------------------------- %%% @author anlicheng %%% @copyright (C) 2025, %%% @doc %%% %%% @end %%% Created : 20. 4月 2026 00:00 %%%------------------------------------------------------------------- -module(efka_client). -author("anlicheng"). -include("protocol.hrl"). -include("message_pb.hrl"). -include("efka_tables.hrl"). -behaviour(gen_statem). %% API -export([start_link/0]). -export([metric_data/2, ping/13, task_event_stream/3, close_task_event_stream/2]). %% gen_statem callbacks -export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). -define(SERVER, ?MODULE). %% 标记当前agent的状态，只有在 activated 状态下才可以正常的发送数据 -define(STATE_DENIED, denied). -define(STATE_AUTH, auth). %% 不能推送消息到服务，但是可以接受服务器的部分指令 -define(STATE_RESTRICTED, restricted). %% 激活状态下 -define(STATE_ACTIVATED, activated). -record(state, { socket :: undefined | ssl:sslsocket() }). %%%=================================================================== %%% API %%%=================================================================== %% 发送数据 -spec metric_data(RouteKey :: binary(), Metric :: binary()) -> no_return(). metric_data(RouteKey, Metric) when is_binary(RouteKey), is_binary(Metric) -> gen_statem:cast(?SERVER, {metric_data, RouteKey, Metric}). -spec task_event_stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> no_return(). task_event_stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) -> gen_statem:cast(?SERVER, {task_event_stream, TaskId, Type, Stream}). -spec close_task_event_stream(TaskId :: integer(), Reason :: binary()) -> no_return(). close_task_event_stream(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) -> gen_statem:cast(?SERVER, {close_task_event_stream, TaskId, Reason}). ping(AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces) -> gen_statem:cast(?SERVER, {ping, AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces}). start_link() -> gen_statem:start_link({local, ?SERVER}, ?MODULE, [], []). %%%=================================================================== %%% gen_statem callbacks %%%=================================================================== init([]) -> erlang:start_timer(0, self(), create_transport), {ok, ?STATE_DENIED, #state{socket = undefined}}. callback_mode() -> handle_event_function. %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia handle_event(cast, {metric_data, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> Packet = message_pb:encode_msg(#'CastFrame'{ body = {data, #'Data'{route_key = RouteKey, metric = Metric}} }), send_packet(Socket, Packet), {keep_state, State}; handle_event(cast, {metric_data, RouteKey, Metric}, _, State) -> Packet = message_pb:encode_msg(#'CastFrame'{ body = {data, #'Data'{route_key = RouteKey, metric = Metric}} }), ok = cache_model:insert(Packet), {keep_state, State}; handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]), EventPacket = message_pb:encode_msg(#'CastFrame'{ body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream}} }), send_packet(Socket, EventPacket), {keep_state, State}; handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> EventPacket = message_pb:encode_msg(#'CastFrame'{ body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = <<"close">>, stream = Reason}} }), send_packet(Socket, EventPacket), {keep_state, State}; %% 其他情况下直接忽略 handle_event(cast, {task_event_stream, _TaskId, _Stream}, _, State = #state{}) -> {keep_state, State}; %% 异步建立到服务器的连接 handle_event(info, {timeout, _, create_transport}, ?STATE_DENIED, State) -> case connect_socket() of {ok, Socket} -> AuthPacket = auth_packet(), send_packet(Socket, AuthPacket), {next_state, ?STATE_AUTH, State#state{socket = Socket}, [{state_timeout, 5000, auth_timeout}]}; {error, Reason} -> logger:debug("[efka_client] connect failed, error: ~p", [Reason]), schedule_reconnect(), {keep_state, State#state{socket = undefined}} end; handle_event(info, {timeout, _, create_transport}, _, State) -> {keep_state, State}; handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket}) -> logger:debug("[efka_client] auth request timeout"), disconnect(Socket), schedule_reconnect(), {next_state, ?STATE_DENIED, State#state{socket = undefined}}; %% 将缓存中的数据推送到服务器端 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> case cache_model:fetch_next() of {ok, {Id, Packet}} -> send_packet(Socket, Packet), cache_model:delete(Id), {keep_state, State, [{next_event, info, flush_cache}]}; error -> {keep_state, State} end; handle_event(info, flush_cache, _, State) -> {keep_state, State}; %% 处理收到的ssl消息 handle_event(info, {ssl, Socket, PacketBin}, ?STATE_AUTH, State = #state{socket = Socket}) -> #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}} = message_pb:decode_msg(PacketBin, 'ResponseFrame'), case Code of 0 -> logger:debug("[efka_client] auth success, message: ~p", [Message]), {next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]}; 1 -> logger:debug("[efka_client] auth denied, message: ~p", [Message]), {next_state, ?STATE_RESTRICTED, State}; 2 -> logger:debug("[efka_client] auth failed, message: ~p", [Message]), disconnect(Socket), schedule_reconnect(), {next_state, ?STATE_DENIED, State#state{socket = undefined}}; _ -> logger:debug("[efka_client] auth failed, invalid message"), disconnect(Socket), schedule_reconnect(), {next_state, ?STATE_DENIED, State#state{socket = undefined}} end; handle_event(info, {ssl, Socket, <<8, _/binary>> = PacketBin}, StateName, State = #state{socket = Socket}) when StateName =:= ?STATE_ACTIVATED; StateName =:= ?STATE_RESTRICTED -> #'RequestFrame'{packet_id = PacketId, body = Body} = message_pb:decode_msg(PacketBin, 'RequestFrame'), true = is_integer(PacketId) andalso PacketId > 0, case Body of {rpc_request, Request} -> {keep_state, State, [{next_event, info, {server_rpc, PacketId, Request}}]}; {container_request, Request} -> {keep_state, State, [{next_event, info, {container_request, PacketId, Request}}]} end; handle_event(info, {ssl, Socket, <<10, _/binary>> = PacketBin}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> #'CastFrame'{body = {pub, Pub}} = message_pb:decode_msg(PacketBin, 'CastFrame'), {keep_state, State, [{next_event, info, {server_cast, Pub}}]}; handle_event(info, {ssl, Socket, <<18, _/binary>> = PacketBin}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> #'CastFrame'{body = {command, Command}} = message_pb:decode_msg(PacketBin, 'CastFrame'), {keep_state, State, [{next_event, info, {server_cast, Command}}]}; handle_event(info, {ssl, Socket, <<10, _/binary>> = PacketBin}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> #'CastFrame'{body = {pub, Pub}} = message_pb:decode_msg(PacketBin, 'CastFrame'), {keep_state, State, [{next_event, info, {server_cast, Pub}}]}; handle_event(info, {ssl, Socket, <<18, _/binary>> = PacketBin}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> #'CastFrame'{body = {command, Command}} = message_pb:decode_msg(PacketBin, 'CastFrame'), {keep_state, State, [{next_event, info, {server_cast, Command}}]}; %% 微服务部署 handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> case docker_manager:get_containers() of {ok, Containers} -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Containers)}} }} }), send_packet(Socket, Packet); {error, Reason} when is_binary(Reason) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}} }} }), send_packet(Socket, Packet) end, {keep_state, State}; handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{ task_id = TaskId, params = Params }}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> case docker_manager:deploy(TaskId, Params) of ok -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}} }} }), send_packet(Socket, Packet); {error, Reason} when is_binary(Reason) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}} }} }), send_packet(Socket, Packet) end, {keep_state, State}; handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = Target}}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> ContainerTarget = container_target(Target), case docker_manager:start_container(ContainerTarget) of ok -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}} }} }), send_packet(Socket, Packet); {error, Reason} when is_binary(Reason) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}} }} }), send_packet(Socket, Packet) end, {keep_state, State}; handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{ target = Target, timeout_seconds = TimeoutSeconds }}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> ContainerTarget = container_target(Target), case docker_manager:stop_container(ContainerTarget, TimeoutSeconds) of ok -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}} }} }), send_packet(Socket, Packet); {error, Reason} when is_binary(Reason) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}} }} }), send_packet(Socket, Packet) end, {keep_state, State}; handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{ target = Target, signal = Signal }}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> ContainerTarget = container_target(Target), case docker_manager:kill_container(ContainerTarget, to_binary(Signal)) of ok -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}} }} }), send_packet(Socket, Packet); {error, Reason} when is_binary(Reason) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}} }} }), send_packet(Socket, Packet) end, {keep_state, State}; handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{ target = Target, force = Force, remove_volumes = RemoveVolumes }}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> ContainerTarget = container_target(Target), case docker_manager:remove_container(ContainerTarget, to_bool(Force), to_bool(RemoveVolumes)) of ok -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}} }} }), send_packet(Socket, Packet); {error, Reason} when is_binary(Reason) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}} }} }), send_packet(Socket, Packet) end, {keep_state, State}; handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {config, #'ContainerRequest.Config'{ target = Target, config = Config }}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> ContainerTarget = container_target(Target), case docker_manager:config_container(ContainerTarget, iolist_to_binary(Config)) of ok -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}} }} }), send_packet(Socket, Packet); {error, Reason} -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}} }} }), send_packet(Socket, Packet) end, {keep_state, State}; handle_event(info, {container_request, PacketId, _Request}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"unsupported container request">>}} }} }), send_packet(Socket, Packet), {keep_state, State}; handle_event(info, {container_request, PacketId, _Request}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"agent restricted">>}} }} }), send_packet(Socket, Packet), {keep_state, State}; handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = Method}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"unsupported rpc request: ", Method/binary>>}} }} }), send_packet(Socket, Packet), {keep_state, State}; handle_event(info, {server_rpc, PacketId, _Request}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> Packet = message_pb:encode_msg(#'ResponseFrame'{ packet_id = PacketId, body = {rpc_reply, #'RpcReply'{ reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"agent restricted">>}} }} }), send_packet(Socket, Packet), {keep_state, State}; %% 处理命令 handle_event(info, {server_cast, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}, StateName, State = #state{socket = Socket}) -> Auth = binary_to_integer(Auth0), case {Auth, StateName} of {1, ?STATE_ACTIVATED} -> {keep_state, State}; {1, _} -> AuthPacket = auth_packet(), send_packet(Socket, AuthPacket), {next_state, ?STATE_AUTH, State, [{state_timeout, 5000, auth_timeout}]}; {0, _} -> {next_state, ?STATE_RESTRICTED, State} end; %% 处理Pub/Sub机制 handle_event(info, {server_cast, #'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(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) -> logger:debug("[efka_client] ssl error: ~p", [Reason]), disconnect(Socket), schedule_reconnect(), {next_state, ?STATE_DENIED, State#state{socket = undefined}}; handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) -> schedule_reconnect(), {next_state, ?STATE_DENIED, State#state{socket = undefined}}; handle_event(info, Info, _, State = #state{}) -> logger:notice("[efka_client] get unknown info: ~p", [Info]), {keep_state, State}. terminate(_Reason, _StateName, _State = #state{socket = Socket}) -> disconnect(Socket), ok. code_change(_OldVsn, StateName, State = #state{}, _Extra) -> {ok, StateName, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== -spec auth_packet() -> binary(). auth_packet() -> {ok, AuthInfo} = application:get_env(efka, auth), UUID = proplists:get_value(uuid, AuthInfo), Username = proplists:get_value(username, AuthInfo), Salt = proplists:get_value(salt, AuthInfo), Token = proplists:get_value(token, AuthInfo), message_pb:encode_msg(#'RequestFrame'{ packet_id = 1, body = {auth_request, #'AuthRequest'{ uuid = unicode:characters_to_binary(UUID), username = unicode:characters_to_binary(Username), salt = unicode:characters_to_binary(Salt), token = unicode:characters_to_binary(Token), timestamp = efka_util:timestamp() }} }). -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}. connect_socket() -> {ok, Props} = application:get_env(efka, tls_server), Host = proplists:get_value(host, Props), Port = proplists:get_value(port, Props), SslOptions = [ binary, {active, true}, {packet, 4}, {verify, verify_none} ], ssl:connect(Host, Port, SslOptions, 5000). -spec send_packet(ssl:sslsocket(), binary()) -> ok. send_packet(Socket, Packet) when is_binary(Packet) -> ok = ssl:send(Socket, Packet). -spec disconnect(undefined | ssl:sslsocket()) -> ok. disconnect(undefined) -> ok; disconnect(Socket) -> catch ssl:close(Socket), ok. -spec schedule_reconnect() -> reference(). schedule_reconnect() -> erlang:start_timer(5000, self(), create_transport). -spec encode_rpc_payload(any()) -> binary(). encode_rpc_payload(Payload) -> jiffy:encode(Payload, [force_utf8]). 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.