From 15ea812aaba5b385d77e67e2b97416d5a679906f Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Mon, 20 Apr 2026 16:11:49 +0800 Subject: [PATCH] fix efka_client --- src/efka_client.erl | 343 ++++++++++++-------------------------------- 1 file changed, 94 insertions(+), 249 deletions(-) diff --git a/src/efka_client.erl b/src/efka_client.erl index 8748e66..aa8ce31 100644 --- a/src/efka_client.erl +++ b/src/efka_client.erl @@ -24,7 +24,7 @@ -define(SERVER, ?MODULE). %% 标记当前agent的状态，只有在 activated 状态下才可以正常的发送数据 --define(STATE_DENIED, denied). +-define(STATE_DISCONNECTED, disconnected). -define(STATE_AUTH, auth). %% 不能推送消息到服务，但是可以接受服务器的部分指令 -define(STATE_RESTRICTED, restricted). @@ -64,17 +64,18 @@ start_link() -> init([]) -> erlang:start_timer(0, self(), create_transport), - {ok, ?STATE_DENIED, #state{socket = undefined}}. + {ok, ?STATE_DISCONNECTED, #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'{ + CastFrame = message_pb:encode_msg(#'CastFrame'{ body = {data, #'Data'{route_key = RouteKey, metric = Metric}} }), - send_packet(Socket, <>), + Packet = <>, + send_packet(Socket, Packet), {keep_state, State}; handle_event(cast, {metric_data, RouteKey, Metric}, _, State) -> @@ -104,7 +105,7 @@ handle_event(cast, {task_event_stream, _TaskId, _Stream}, _, State = #state{}) - {keep_state, State}; %% 异步建立到服务器的连接 -handle_event(info, {timeout, _, create_transport}, ?STATE_DENIED, State) -> +handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) -> case connect_socket() of {ok, Socket} -> AuthPacket = auth_packet(), @@ -122,7 +123,7 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S logger:debug("[efka_client] auth request timeout"), disconnect(Socket), schedule_reconnect(), - {next_state, ?STATE_DENIED, State#state{socket = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}}; %% 将缓存中的数据推送到服务器端 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> @@ -138,11 +139,45 @@ handle_event(info, flush_cache, _, State) -> {keep_state, State}; %% 处理收到的ssl消息 +handle_event(info, {ssl, Socket, <>}, _, State = #state{socket = Socket}) -> + ResponseFrame = message_pb:decode_msg(PacketBin, 'ResponseFrame'), + {keep_state, State, [{next_event, internal, ResponseFrame}]}; +handle_event(info, {ssl, Socket, <>}, _, State = #state{socket = Socket}) -> + RequestFrame = message_pb:decode_msg(PacketBin, 'RequestFrame'), + {keep_state, State, [{next_event, internal, RequestFrame}]}; +handle_event(info, {ssl, Socket, <>}, _, State = #state{socket = Socket}) -> + CastFrame = message_pb:decode_msg(PacketBin, 'CastFrame'), + {keep_state, State, [{next_event, internal, CastFrame}]}; +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_DISCONNECTED, State#state{socket = undefined}}; +handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) -> + schedule_reconnect(), + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}}; -handle_event(info, {ssl, Socket, <>}, ?STATE_AUTH, State = #state{socket = Socket}) -> - #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}} = - message_pb:decode_msg(PacketBin, 'ResponseFrame'), +%%% 处理内部消息，ssl收到的消息会解析成protobuf的消息格式，并按照internal类型处理 +%% 微服务部署 +handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> + case handle_container_request(Request) of + ok -> + send_rpc_result_reply(Socket, PacketId, <<"ok">>); + {ok, Reply} -> + send_rpc_result_reply(Socket, PacketId, Reply); + {error, Reason} -> + send_rpc_error_reply(Socket, PacketId, Reason) + end, + {keep_state, State}; +handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = _Body}, ?STATE_RESTRICTED, State = #state{socket = Socket}) -> + send_rpc_error_reply(Socket, PacketId, <<"agent restricted">>), + {keep_state, State}; +handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = _Body}, _StateName, State = #state{socket = Socket}) -> + send_rpc_error_reply(Socket, PacketId, <<"agent state invalid">>), + {keep_state, State}; + +handle_event(internal, #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}}, ?STATE_AUTH, State = #state{socket = Socket}) -> case Code of 0 -> logger:debug("[efka_client] auth success, message: ~p", [Message]), @@ -154,242 +189,16 @@ handle_event(info, {ssl, Socket, <>}, ?STATE_ logger:debug("[efka_client] auth failed, message: ~p", [Message]), disconnect(Socket), schedule_reconnect(), - {next_state, ?STATE_DENIED, State#state{socket = undefined}}; + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}}; _ -> logger:debug("[efka_client] auth failed, invalid message"), disconnect(Socket), schedule_reconnect(), - {next_state, ?STATE_DENIED, State#state{socket = undefined}} + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}} end; -handle_event(info, {ssl, Socket, <>}, StateName, State = #state{socket = Socket}) - when StateName =:= ?STATE_ACTIVATED; StateName =:= ?STATE_RESTRICTED -> - #'RequestFrame'{packet_id = PacketId, body = Body} = message_pb:decode_msg(PacketBin, 'RequestFrame'), - 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, <>}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - CastFrame = message_pb:decode_msg(PacketBin, 'CastFrame'), - {keep_state, State, [{next_event, info, {server_cast, CastFrame}}]}; - -%% 微服务部署 -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'{command_type = ?COMMAND_AUTH, command = Auth0}}}, StateName, - State = #state{socket = Socket}) -> +handle_event(internal, #'CastFrame'{body = {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}}, StateName, State = #state{socket = Socket}) -> Auth = binary_to_integer(Auth0), case {Auth, StateName} of {1, ?STATE_ACTIVATED} -> @@ -403,21 +212,11 @@ handle_event(info, {server_cast, {command, #'Command'{command_type = ?COMMAND_AU end; %% 处理Pub/Sub机制 -handle_event(info, {server_cast, {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}}, ?STATE_ACTIVATED, State) -> +handle_event(internal, #'CastFrame'{body = {pub, #'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}. @@ -483,6 +282,52 @@ schedule_reconnect() -> encode_rpc_payload(Payload) -> jiffy:encode(Payload, [force_utf8]). +-spec handle_container_request(message_pb:'ContainerRequest'()) -> ok. +handle_container_request(#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}) -> + case docker_manager: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_manager:start_container(ContainerTarget); +handle_container_request(#'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{ target = Target, timeout_seconds = TimeoutSeconds }}}) -> + ContainerTarget = container_target(Target), + docker_manager:stop_container(ContainerTarget, TimeoutSeconds); +handle_container_request(#'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{ target = Target, signal = Signal}}}) -> + ContainerTarget = container_target(Target), + docker_manager: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_manager: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), + docker_manager:config_container(ContainerTarget, iolist_to_binary(Config)). + +-spec send_rpc_result_reply(ssl:sslsocket(), integer(), any()) -> ok. +send_rpc_result_reply(Socket, PacketId, Payload) -> + Packet = message_pb:encode_msg(#'ResponseFrame'{ + packet_id = PacketId, + body = {rpc_reply, #'RpcReply'{ + reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Payload)}} + }} + }), + send_packet(Socket, Packet). + +-spec send_rpc_error_reply(ssl:sslsocket(), integer(), binary()) -> ok. +send_rpc_error_reply(Socket, PacketId, 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). + container_target(#'ContainerRef'{name = Name, id = Id}) -> NameBin = to_binary(Name), IdBin = to_binary(Id),