anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初步完善传输层

Browse Source
This commit is contained in:
anlicheng 2026-04-20 12:50:58 +08:00
parent 5b9c4aa06d
commit 8fd58750aa
7 changed files with 6591 additions and 324 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
include/message.hrl
View File

@ -8,13 +8,6 @@
%%%-------------------------------------------------------------------
-author("anlicheng").
%% efka主动发起的消息体类型, 消息大类
-define(PACKET_REQUEST, 16#01).
-define(PACKET_RESPONSE, 16#02).
%% efka主动发起不需要返回的数据
-define(PACKET_CAST, 16#03).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%%% 二级分类定义
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
@ -24,11 +17,9 @@
-define(MESSAGE_AUTH_REPLY, 16#02).
-define(MESSAGE_COMMAND, 16#03).
-define(MESSAGE_DEPLOY, 16#04).
-define(MESSAGE_PUB, 16#05).
-define(MESSAGE_DATA, 16#06).
-define(MESSAGE_EVENT, 16#07).
%% efka主动上报的event-stream流, 单向消息，主要是: docker-create的实时处理逻辑上报
-define(MESSAGE_EVENT_STREAM, 16#08).
@ -39,48 +30,3 @@
%%%% 命令类型子分类, 不需要返回值
%% 授权
-define(COMMAND_AUTH, 16#08).
-record(auth_request, {
uuid :: binary(),
username :: binary(),
salt :: binary(),
token :: binary(),
timestamp :: integer()
}).
-record(auth_reply, {
code :: integer(),
payload :: binary()
}).
-record(pub, {
topic :: binary(),
qos = 0 :: integer(),
content :: binary()
}).
-record(command, {
command_type :: integer(),
command :: binary()
}).
-record(jsonrpc_request, {
method :: binary(),
params = <<>> :: any()
}).
-record(jsonrpc_reply, {
result :: any() | undefined,
error :: any() | undefined
}).
-record(data, {
route_key :: binary(),
metric :: binary()
}).
-record(task_event_stream, {
task_id :: integer(),
type :: binary(),
stream :: binary()
}).

296
include/message_pb.hrl Normal file
View File

@ -0,0 +1,296 @@
%% -*- coding: utf-8 -*-
%% Automatically generated, do not edit
%% Generated by gpb_compile version 4.21.7
-ifndef(message_pb).
-define(message_pb, true).
-define(message_pb_gpb_version, "4.21.7").
-ifndef('REQUESTFRAME_PB_H').
-define('REQUESTFRAME_PB_H', true).
-record('RequestFrame',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
body :: {auth_request, message_pb:'AuthRequest'()} | {rpc_request, message_pb:'RpcRequest'()} | {container_request, message_pb:'ContainerRequest'()} | undefined % oneof
}).
-endif.
-ifndef('RESPONSEFRAME_PB_H').
-define('RESPONSEFRAME_PB_H', true).
-record('ResponseFrame',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
body :: {auth_reply, message_pb:'AuthReply'()} | {rpc_reply, message_pb:'RpcReply'()} | undefined % oneof
}).
-endif.
-ifndef('CASTFRAME_PB_H').
-define('CASTFRAME_PB_H', true).
-record('CastFrame',
{body :: {pub, message_pb:'Pub'()} | {command, message_pb:'Command'()} | {data, message_pb:'Data'()} | {event_stream, message_pb:'TaskEventStream'()} | undefined % oneof
}).
-endif.
-ifndef('CONTAINERREF_PB_H').
-define('CONTAINERREF_PB_H', true).
-record('ContainerRef',
{id = <<>> :: unicode:chardata() | undefined, % = 1, optional
name = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.LIST_PB_H').
-define('CONTAINERREQUEST.LIST_PB_H', true).
-record('ContainerRequest.List',
{all = false :: boolean() | 0 | 1 | undefined % = 1, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.DEPLOY_PB_H').
-define('CONTAINERREQUEST.DEPLOY_PB_H', true).
-record('ContainerRequest.Deploy',
{task_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
params = undefined :: message_pb:'ContainerDeployParams'() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.START_PB_H').
-define('CONTAINERREQUEST.START_PB_H', true).
-record('ContainerRequest.Start',
{target = undefined :: message_pb:'ContainerRef'() | undefined % = 1, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.STOP_PB_H').
-define('CONTAINERREQUEST.STOP_PB_H', true).
-record('ContainerRequest.Stop',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
timeout_seconds = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
}).
-endif.
-ifndef('CONTAINERREQUEST.KILL_PB_H').
-define('CONTAINERREQUEST.KILL_PB_H', true).
-record('ContainerRequest.Kill',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
signal = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.REMOVE_PB_H').
-define('CONTAINERREQUEST.REMOVE_PB_H', true).
-record('ContainerRequest.Remove',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
force = false :: boolean() | 0 | 1 | undefined, % = 2, optional
remove_volumes = false :: boolean() | 0 | 1 | undefined % = 3, optional
}).
-endif.
-ifndef('CONTAINERREQUEST.CONFIG_PB_H').
-define('CONTAINERREQUEST.CONFIG_PB_H', true).
-record('ContainerRequest.Config',
{target = undefined :: message_pb:'ContainerRef'() | undefined, % = 1, optional
config = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('CONTAINERREQUEST_PB_H').
-define('CONTAINERREQUEST_PB_H', true).
-record('ContainerRequest',
{action :: {list, message_pb:'ContainerRequest.List'()} | {deploy, message_pb:'ContainerRequest.Deploy'()} | {start, message_pb:'ContainerRequest.Start'()} | {stop, message_pb:'ContainerRequest.Stop'()} | {kill, message_pb:'ContainerRequest.Kill'()} | {remove, message_pb:'ContainerRequest.Remove'()} | {config, message_pb:'ContainerRequest.Config'()} | undefined % oneof
}).
-endif.
-ifndef('CONTAINERDEPLOYPARAMS_PB_H').
-define('CONTAINERDEPLOYPARAMS_PB_H', true).
-record('ContainerDeployParams',
{container_name = <<>> :: unicode:chardata() | undefined, % = 1, optional
container_dir = <<>> :: unicode:chardata() | undefined, % = 2, optional
spec = undefined :: message_pb:'ContainerSpec'() | undefined % = 3, optional
}).
-endif.
-ifndef('CONTAINERSPEC_PB_H').
-define('CONTAINERSPEC_PB_H', true).
-record('ContainerSpec',
{image = <<>> :: unicode:chardata() | undefined, % = 1, optional
command = [] :: [unicode:chardata()] | undefined, % = 2, repeated
entrypoint = [] :: [unicode:chardata()] | undefined, % = 3, repeated
env = [] :: [unicode:chardata()] | undefined, % = 4, repeated
labels = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 5
volumes = [] :: [message_pb:'VolumeBind'()] | undefined, % = 6, repeated
user = <<>> :: unicode:chardata() | undefined, % = 7, optional
working_dir = <<>> :: unicode:chardata() | undefined, % = 8, optional
hostname = <<>> :: unicode:chardata() | undefined, % = 9, optional
expose = [] :: [message_pb:'PortExpose'()] | undefined, % = 10, repeated
networks = [] :: [unicode:chardata()] | undefined, % = 11, repeated
network_mode = <<>> :: unicode:chardata() | undefined, % = 12, optional
healthcheck = undefined :: message_pb:'Healthcheck'() | undefined, % = 13, optional
restart = undefined :: message_pb:'RestartPolicy'() | undefined, % = 14, optional
privileged = false :: boolean() | 0 | 1 | undefined, % = 15, optional
cap_add = [] :: [unicode:chardata()] | undefined, % = 16, repeated
cap_drop = [] :: [unicode:chardata()] | undefined, % = 17, repeated
devices = [] :: [message_pb:'DeviceMapping'()] | undefined, % = 18, repeated
resources = undefined :: message_pb:'ResourceLimits'() | undefined, % = 19, optional
ulimits = [] :: [message_pb:'Ulimit'()] | undefined, % = 20, repeated
tmpfs = [] :: [message_pb:'TmpfsMount'()] | undefined, % = 21, repeated
sysctls = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 22
extra_hosts = [] :: [unicode:chardata()] | undefined % = 23, repeated
}).
-endif.
-ifndef('VOLUMEBIND_PB_H').
-define('VOLUMEBIND_PB_H', true).
-record('VolumeBind',
{host_path = <<>> :: unicode:chardata() | undefined, % = 1, optional
container_path = <<>> :: unicode:chardata() | undefined, % = 2, optional
read_only = false :: boolean() | 0 | 1 | undefined % = 3, optional
}).
-endif.
-ifndef('PORTEXPOSE_PB_H').
-define('PORTEXPOSE_PB_H', true).
-record('PortExpose',
{container_port = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
protocol = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('HEALTHCHECK_PB_H').
-define('HEALTHCHECK_PB_H', true).
-record('Healthcheck',
{test = [] :: [unicode:chardata()] | undefined, % = 1, repeated
interval_ns = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
timeout_ns = 0 :: non_neg_integer() | undefined, % = 3, optional, 64 bits
retries = 0 :: non_neg_integer() | undefined % = 4, optional, 32 bits
}).
-endif.
-ifndef('RESTARTPOLICY_PB_H').
-define('RESTARTPOLICY_PB_H', true).
-record('RestartPolicy',
{name = <<>> :: unicode:chardata() | undefined, % = 1, optional
maximum_retry_count = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
}).
-endif.
-ifndef('DEVICEMAPPING_PB_H').
-define('DEVICEMAPPING_PB_H', true).
-record('DeviceMapping',
{host_path = <<>> :: unicode:chardata() | undefined, % = 1, optional
container_path = <<>> :: unicode:chardata() | undefined, % = 2, optional
cgroup_permissions = <<>> :: unicode:chardata() | undefined % = 3, optional
}).
-endif.
-ifndef('RESOURCELIMITS_PB_H').
-define('RESOURCELIMITS_PB_H', true).
-record('ResourceLimits',
{memory_bytes = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
memory_reservation_bytes = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
nano_cpus = 0 :: non_neg_integer() | undefined, % = 3, optional, 64 bits
cpu_shares = 0 :: non_neg_integer() | undefined % = 4, optional, 64 bits
}).
-endif.
-ifndef('ULIMIT_PB_H').
-define('ULIMIT_PB_H', true).
-record('Ulimit',
{name = <<>> :: unicode:chardata() | undefined, % = 1, optional
soft = 0 :: non_neg_integer() | undefined, % = 2, optional, 64 bits
hard = 0 :: non_neg_integer() | undefined % = 3, optional, 64 bits
}).
-endif.
-ifndef('TMPFSMOUNT_PB_H').
-define('TMPFSMOUNT_PB_H', true).
-record('TmpfsMount',
{path = <<>> :: unicode:chardata() | undefined, % = 1, optional
options = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('AUTHREQUEST_PB_H').
-define('AUTHREQUEST_PB_H', true).
-record('AuthRequest',
{uuid = <<>> :: iodata() | undefined, % = 1, optional
username = <<>> :: iodata() | undefined, % = 2, optional
salt = <<>> :: iodata() | undefined, % = 3, optional
token = <<>> :: iodata() | undefined, % = 4, optional
timestamp = 0 :: integer() | undefined % = 5, optional, 32 bits
}).
-endif.
-ifndef('AUTHREPLY_PB_H').
-define('AUTHREPLY_PB_H', true).
-record('AuthReply',
{code = 0 :: integer() | undefined, % = 1, optional, 32 bits
payload = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('PUB_PB_H').
-define('PUB_PB_H', true).
-record('Pub',
{topic = <<>> :: iodata() | undefined, % = 1, optional
qos = 0 :: integer() | undefined, % = 2, optional, 32 bits
content = <<>> :: iodata() | undefined % = 3, optional
}).
-endif.
-ifndef('COMMAND_PB_H').
-define('COMMAND_PB_H', true).
-record('Command',
{command_type = 0 :: integer() | undefined, % = 1, optional, 32 bits
command = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('RPCREQUEST_PB_H').
-define('RPCREQUEST_PB_H', true).
-record('RpcRequest',
{method = <<>> :: iodata() | undefined, % = 1, optional
params = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('RPCREPLY.RPCRESULT_PB_H').
-define('RPCREPLY.RPCRESULT_PB_H', true).
-record('RpcReply.RpcResult',
{data = <<>> :: iodata() | undefined % = 1, optional
}).
-endif.
-ifndef('RPCREPLY.RPCERROR_PB_H').
-define('RPCREPLY.RPCERROR_PB_H', true).
-record('RpcReply.RpcError',
{code = 0 :: integer() | undefined, % = 1, optional, 32 bits
message = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.
-ifndef('RPCREPLY_PB_H').
-define('RPCREPLY_PB_H', true).
-record('RpcReply',
{reply :: {result, message_pb:'RpcReply.RpcResult'()} | {error, message_pb:'RpcReply.RpcError'()} | undefined % oneof
}).
-endif.
-ifndef('DATA_PB_H').
-define('DATA_PB_H', true).
-record('Data',
{route_key = <<>> :: iodata() | undefined, % = 1, optional
metric = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('TASKEVENTSTREAM_PB_H').
-define('TASKEVENTSTREAM_PB_H', true).
-record('TaskEventStream',
{task_id = 0 :: integer() | undefined, % = 1, optional, 32 bits
type = <<>> :: iodata() | undefined, % = 2, optional
stream = <<>> :: iodata() | undefined % = 3, optional
}).
-endif.
-endif.

20
include/protocol.hrl Normal file
View File

@ -0,0 +1,20 @@
%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2025, <COMPANY>
%%% @doc
%%% 传输层协议常量定义
%%% @end
%%%-------------------------------------------------------------------
-author("anlicheng").
%% 一级帧类型
%% REQUEST: 需要响应的请求帧
%% RESPONSE: 对 REQUEST 的响应帧
%% CAST: 单向消息帧，不需要响应
-define(FRAME_REQUEST, 16#01).
-define(FRAME_RESPONSE, 16#02).
-define(FRAME_CAST, 16#03).
%%%% 命令类型子分类, 不需要返回值
%% 授权
-define(COMMAND_AUTH, 16#08).

255
src/efka_remote_agent.erl
View File

@ -88,36 +88,30 @@ callback_mode() ->
%% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
handle_event(cast, {metric_data, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
Packet = message_codec:encode(?MESSAGE_DATA, #'Data'{
route_key = RouteKey,
metric = Metric
Packet = message_pb:encode_msg(#'CastFrame'{
body = {data, #'Data'{route_key = RouteKey, metric = Metric}}
}),
efka_transport:send(TransportPid, Packet),
{keep_state, State};
handle_event(cast, {metric_data, RouteKey, Metric}, _, State) ->
Packet = message_codec:encode(?MESSAGE_DATA, #'Data'{
route_key = RouteKey,
metric = Metric
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{transport_pid = TransportPid}) ->
logger:debug("[efka_remote_agent] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
EventPacket = message_codec:encode(?MESSAGE_EVENT_STREAM, #'TaskEventStream'{
task_id = TaskId,
type = Type,
stream = Stream
EventPacket = message_pb:encode_msg(#'CastFrame'{
body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream}}
}),
efka_transport:send(TransportPid, EventPacket),
{keep_state, State};
handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
EventPacket = message_codec:encode(?MESSAGE_EVENT_STREAM, #'TaskEventStream'{
task_id = TaskId,
type = <<"close">>,
stream = Reason
EventPacket = message_pb:encode_msg(#'CastFrame'{
body = {event_stream, #'TaskEventStream'{task_id = TaskId, type = <<"close">>, stream = Reason}}
}),
efka_transport:send(TransportPid, EventPacket),
{keep_state, State};
@ -144,7 +138,7 @@ handle_event(cast, {task_event_stream, _TaskId, _Stream}, _, State = #state{}) -
% memory = Memory,
% interfaces = Interfaces
% }),
% efka_transport:send(TransportPid, ?METHOD_PING, Ping),
% efka_transport:send(TransportPid, Ping),
% {keep_state, State};
%% 异步建立到服务器的连接
@ -160,44 +154,40 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DENIED, State) ->
handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{transport_pid = TransportPid}) ->
case Reply of
ok ->
AuthRequest = auth_request(),
efka_transport:auth_request(TransportPid, AuthRequest),
{next_state, ?STATE_AUTH, State};
AuthPacket = auth_packet(),
efka_transport:send(TransportPid, AuthPacket),
{next_state, ?STATE_AUTH, State, [{state_timeout, 5000, auth_timeout}]};
{error, Reason} ->
logger:debug("[efka_remote_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
end;
handle_event(info, {auth_reply, Reply}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
case Reply of
{ok, #'AuthReply'{code = Code, payload = Message}} ->
case Code of
0 ->
logger:debug("[efka_remote_agent] auth success, message: ~p", [Message]),
{next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]};
1 ->
%% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent
%% socket的连接状态需要维持
logger:debug("[efka_remote_agent] auth denied, message: ~p", [Message]),
{next_state, ?STATE_RESTRICTED, State};
2 ->
% 其他类型的错误，需要间隔时间重试
logger:debug("[efka_remote_agent] auth failed, message: ~p", [Message]),
efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
_ ->
% 其他类型的错误，需要间隔时间重试
logger:debug("[efka_remote_agent] auth failed, invalid message"),
efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
end;
{error, Reason} ->
logger:debug("[efka_remote_agent] auth_request failed, error: ~p", [Reason]),
handle_event(info, {server_packet, PacketBin}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
#'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_remote_agent] auth success, message: ~p", [Message]),
{next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]};
1 ->
logger:debug("[efka_remote_agent] auth denied, message: ~p", [Message]),
{next_state, ?STATE_RESTRICTED, State};
2 ->
logger:debug("[efka_remote_agent] auth failed, message: ~p", [Message]),
efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
_ ->
logger:debug("[efka_remote_agent] auth failed, invalid message"),
efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
end;
handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
logger:debug("[efka_remote_agent] auth request timeout"),
efka_transport:stop(TransportPid),
{next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
%% 将缓存中的数据推送到服务器端
handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
case cache_model:fetch_next() of
@ -211,6 +201,30 @@ handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{transport_pid =
handle_event(info, flush_cache, _, State) ->
{keep_state, State};
handle_event(info, {server_packet, <<8, _/binary>> = PacketBin}, ?STATE_ACTIVATED, State) ->
#'RequestFrame'{packet_id = PacketId, body = {rpc_request, Request}} =
message_pb:decode_msg(PacketBin, 'RequestFrame'),
true = is_integer(PacketId) andalso PacketId > 0,
{keep_state, State, [{next_event, info, {server_rpc, PacketId, Request}}]};
handle_event(info, {server_packet, <<10, _/binary>> = PacketBin}, ?STATE_ACTIVATED, State) ->
#'CastFrame'{body = {pub, Pub}} = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, info, {server_cast, Pub}}]};
handle_event(info, {server_packet, <<18, _/binary>> = PacketBin}, ?STATE_ACTIVATED, State) ->
#'CastFrame'{body = {command, Command}} = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, info, {server_cast, Command}}]};
handle_event(info, {server_packet, <<8, _/binary>> = PacketBin}, ?STATE_RESTRICTED, State) ->
#'RequestFrame'{packet_id = PacketId, body = {rpc_request, Request}} =
message_pb:decode_msg(PacketBin, 'RequestFrame'),
true = is_integer(PacketId) andalso PacketId > 0,
{keep_state, State, [{next_event, info, {server_rpc, PacketId, Request}}]};
handle_event(info, {server_packet, <<10, _/binary>> = PacketBin}, ?STATE_RESTRICTED, State) ->
#'CastFrame'{body = {pub, Pub}} = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, info, {server_cast, Pub}}]};
handle_event(info, {server_packet, <<18, _/binary>> = PacketBin}, ?STATE_RESTRICTED, State) ->
#'CastFrame'{body = {command, Command}} = message_pb:decode_msg(PacketBin, 'CastFrame'),
{keep_state, State, [{next_event, info, {server_cast, Command}}]};
%% 云端服务器推送了消息
%% 激活消息
@ -219,9 +233,21 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"get_containe
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:get_containers() of
{ok, Containers} ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(Containers));
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Containers)}}
}}
}),
efka_transport:send(TransportPid, Packet);
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
}}
}),
efka_transport:send(TransportPid, Packet)
end,
{keep_state, State};
@ -231,9 +257,21 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"deploy">>, p
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:deploy(TaskId, Config) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
}}
}),
efka_transport:send(TransportPid, Packet);
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
}}
}),
efka_transport:send(TransportPid, Packet)
end,
{keep_state, State};
@ -243,9 +281,21 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"start_contai
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:start_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
}}
}),
efka_transport:send(TransportPid, Packet);
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
}}
}),
efka_transport:send(TransportPid, Packet)
end,
{keep_state, State};
@ -255,9 +305,21 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"stop_contain
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:stop_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
}}
}),
efka_transport:send(TransportPid, Packet);
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
}}
}),
efka_transport:send(TransportPid, Packet)
end,
{keep_state, State};
@ -266,9 +328,21 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"kill_contain
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:kill_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
}}
}),
efka_transport:send(TransportPid, Packet);
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
}}
}),
efka_transport:send(TransportPid, Packet)
end,
{keep_state, State};
@ -277,9 +351,21 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"remove_conta
%% 短暂的等待，efka_inetd收到消息后就立即返回了
case docker_manager:remove_container(ContainerName) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
}}
}),
efka_transport:send(TransportPid, Packet);
{error, Reason} when is_binary(Reason) ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
}}
}),
efka_transport:send(TransportPid, Packet)
end,
{keep_state, State};
@ -288,9 +374,21 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"config_conta
#{<<"container_name">> := ContainerName, <<"config">> := Config} = decode_rpc_payload(ParamsBin),
case docker_manager:config_container(ContainerName, Config) of
ok ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_success(<<"ok">>));
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
}}
}),
efka_transport:send(TransportPid, Packet);
{error, Reason} ->
efka_transport:rpc_reply(TransportPid, PacketId, reply_error(-1, Reason))
Packet = message_pb:encode_msg(#'ResponseFrame'{
packet_id = PacketId,
body = {rpc_reply, #'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
}}
}),
efka_transport:send(TransportPid, Packet)
end,
{keep_state, State};
@ -306,7 +404,7 @@ handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = <<"config_conta
% false ->
% #async_call_reply{code = 1, result = <<"[]">>}
% end,
% efka_transport:async_call_reply(TransportPid, PacketId, message_pb:encode_msg(Reply)),
% efka_transport:send(TransportPid, message_pb:encode_msg(Reply)),
%
% {keep_state, State};
@ -318,9 +416,9 @@ handle_event(info, {server_cast, #'Command'{command_type = ?COMMAND_AUTH, comman
{keep_state, State};
{1, ?STATE_DENIED} ->
%% 重新激活, 需要重新校验
AuthRequest = auth_request(),
efka_transport:auth_request(TransportPid, AuthRequest),
{next_state, ?STATE_AUTH, State};
AuthPacket = auth_packet(),
efka_transport:send(TransportPid, AuthPacket),
{next_state, ?STATE_AUTH, State, [{state_timeout, 5000, auth_timeout}]};
{0, _} ->
%% 这个时候的主机应该是受限制的状态，不允许发送消息；但是能够接受服务器推送的消息
{next_state, ?STATE_RESTRICTED, State}
@ -362,33 +460,24 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
%%% Internal functions
%%%===================================================================
-spec auth_request() -> message_pb:'AuthRequest'().
auth_request() ->
-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),
#'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 reply_success(Result :: any()) -> message_pb:'RpcReply'().
reply_success(Result) ->
#'RpcReply'{
reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Result)}}
}.
-spec reply_error(Code :: integer(), Message :: binary()) -> message_pb:'RpcReply'().
reply_error(Code, Message) when is_integer(Code), is_binary(Message) ->
#'RpcReply'{
reply = {error, #'RpcReply.RpcError'{code = Code, message = Message}}
}.
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 decode_rpc_payload(binary()) -> any().
decode_rpc_payload(<<>>) ->

59
src/efka_transport.erl
View File

@ -8,13 +8,12 @@
%%%-------------------------------------------------------------------
-module(efka_transport).
-author("anlicheng").
-include("message.hrl").
-behaviour(gen_server).
%% API
-export([start_monitor/3]).
-export([connect/1, auth_request/2, send/2, rpc_reply/3, stop/1]).
-export([connect/1, send/2, stop/1]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@ -28,14 +27,6 @@
socket :: undefined | ssl:sslsocket()
}).
%%%===================================================================
%%% API
%%%===================================================================
-spec auth_request(Pid :: pid(), AuthRequest :: any()) -> no_return().
auth_request(Pid, AuthRequest) when is_pid(Pid) ->
gen_server:cast(Pid, {auth_request, AuthRequest}).
-spec connect(Pid :: pid()) -> no_return().
connect(Pid) when is_pid(Pid) ->
gen_server:cast(Pid, connect).
@ -44,12 +35,6 @@ connect(Pid) when is_pid(Pid) ->
send(Pid, Packet) when is_pid(Pid), is_binary(Packet) ->
gen_server:cast(Pid, {send, Packet}).
-spec rpc_reply(Pid :: pid() | undefined, PacketId :: integer(), Response :: any()) -> no_return().
rpc_reply(undefined, PacketId, _Response) when is_integer(PacketId) ->
ok;
rpc_reply(Pid, PacketId, Reply) when is_pid(Pid), is_integer(PacketId) ->
gen_server:cast(Pid, {rpc_reply, PacketId, Reply}).
%% 关闭的时候不一定能成功，可能关闭的时候；transport进程已经退出了
-spec stop(Pid :: pid() | undefined) -> ok.
stop(undefined) ->
@ -111,37 +96,7 @@ handle_cast(connect, State = #state{host = Host, port = Port, parent_pid = Paren
{noreply, State#state{socket = undefined}}
end;
%% auth校验
handle_cast({auth_request, AuthRequest}, State = #state{parent_pid = ParentPid, socket = Socket}) ->
PacketId = 1,
Packet = message_codec:encode_request(PacketId, ?MESSAGE_AUTH_REQUEST, AuthRequest),
ok = ssl:send(Socket, Packet),
%% 需要等待auth返回的结果
receive
{ssl, Socket, ReplyBin} ->
case message_codec:decode_response(ReplyBin) of
{ok, PacketId, Reply} ->
ParentPid ! {auth_reply, {ok, Reply}};
{ok, ReplyPacketId, _Reply} ->
logger:warning("[efka_transport] get unexpected auth_reply packet_id: ~p", [ReplyPacketId]),
ParentPid ! {auth_reply, {error, invalid_auth_reply}};
error ->
logger:warning("[efka_transport] get invalid auth_reply: ~p", [ReplyBin]),
ParentPid ! {auth_reply, {error, invalid_auth_reply}}
end,
{noreply, State}
after 5000 ->
ParentPid ! {auth_reply, {error, timeout}},
{noreply, State}
end;
handle_cast({send, Packet}, State = #state{socket = Socket}) ->
ok = ssl:send(Socket, Packet),
{noreply, State};
%% 服务push的消息的回复
handle_cast({rpc_reply, PacketId, Reply}, State = #state{socket = Socket}) ->
Packet = message_codec:encode_response(PacketId, ?MESSAGE_JSONRPC_REPLY, Reply),
ok = ssl:send(Socket, Packet),
{noreply, State}.
@ -153,17 +108,7 @@ handle_cast({rpc_reply, PacketId, Reply}, State = #state{socket = Socket}) ->
{stop, Reason :: term(), NewState :: #state{}}).
%% 服务器主动推送的数据
handle_info({ssl, Socket, PacketBin}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
case message_codec:decode_request(PacketBin) of
{ok, PacketId, Request} ->
ParentPid ! {server_rpc, PacketId, Request};
error ->
case message_codec:decode_cast(PacketBin) of
{ok, CastRequest} ->
ParentPid ! {server_cast, CastRequest};
error ->
logger:warning("[efka_transport] get invalid packet: ~p", [PacketBin])
end
end,
ParentPid ! {server_packet, PacketBin},
{noreply, State};
handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) ->

130
src/message/message_codec.erl
View File

@ -1,130 +0,0 @@
%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2025, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 17. 9月 2025 16:05
%%%-------------------------------------------------------------------
-module(message_codec).
-author("anlicheng").
-include("message.hrl").
-include("message_pb.hrl").
%% API
-export([encode/2, encode_request/3, encode_response/3]).
-export([decode_request/1, decode_response/1, decode_cast/1]).
-spec encode(MessageType :: integer(), Message :: any()) -> binary().
encode(MessageType, Message) when is_integer(MessageType) ->
case frame_for_type(MessageType, Message, 0) of
{request, Frame} ->
message_pb:encode_msg(Frame);
{response, Frame} ->
message_pb:encode_msg(Frame);
{cast, Frame} ->
message_pb:encode_msg(Frame)
end.
-spec encode_request(PacketId :: integer(), MessageType :: integer(), Message :: any()) -> binary().
encode_request(PacketId, MessageType, Message) when is_integer(PacketId), is_integer(MessageType) ->
case frame_for_type(MessageType, Message, PacketId) of
{request, Frame} ->
message_pb:encode_msg(Frame);
_ ->
erlang:error({unsupported_request_type, MessageType})
end.
-spec encode_response(PacketId :: integer(), MessageType :: integer(), Message :: any()) -> binary().
encode_response(PacketId, MessageType, Message) when is_integer(PacketId), is_integer(MessageType) ->
case frame_for_type(MessageType, Message, PacketId) of
{response, Frame} ->
message_pb:encode_msg(Frame);
_ ->
erlang:error({unsupported_response_type, MessageType})
end.
-spec decode_request(Bin :: binary()) -> {ok, PacketId :: integer(), Message :: any()} | error.
decode_request(Bin) when is_binary(Bin) ->
try
Frame = message_pb:decode_msg(Bin, 'RequestFrame'),
decode_request_frame(Frame)
catch
_:_ ->
error
end.
-spec decode_response(Bin :: binary()) -> {ok, PacketId :: integer(), Message :: any()} | error.
decode_response(Bin) when is_binary(Bin) ->
try
Frame = message_pb:decode_msg(Bin, 'ResponseFrame'),
decode_response_frame(Frame)
catch
_:_ ->
error
end.
-spec decode_cast(Bin :: binary()) -> {ok, Message :: any()} | error.
decode_cast(Bin) when is_binary(Bin) ->
try
Frame = message_pb:decode_msg(Bin, 'CastFrame'),
decode_cast_frame(Frame)
catch
_:_ ->
error
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
frame_for_type(?MESSAGE_AUTH_REQUEST, Message = #'AuthRequest'{}, PacketId) ->
{request, #'RequestFrame'{packet_id = PacketId, body = {auth_request, Message}}};
frame_for_type(?MESSAGE_JSONRPC_REQUEST, Message = #'RpcRequest'{}, PacketId) ->
{request, #'RequestFrame'{packet_id = PacketId, body = {rpc_request, Message}}};
frame_for_type(?MESSAGE_AUTH_REPLY, Message = #'AuthReply'{}, PacketId) ->
{response, #'ResponseFrame'{packet_id = PacketId, body = {auth_reply, Message}}};
frame_for_type(?MESSAGE_JSONRPC_REPLY, Message = #'RpcReply'{}, PacketId) ->
{response, #'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, Message}}};
frame_for_type(?MESSAGE_PUB, Message = #'Pub'{}, _PacketId) ->
{cast, #'CastFrame'{body = {pub, Message}}};
frame_for_type(?MESSAGE_COMMAND, Message = #'Command'{}, _PacketId) ->
{cast, #'CastFrame'{body = {command, Message}}};
frame_for_type(?MESSAGE_DATA, Message = #'Data'{}, _PacketId) ->
{cast, #'CastFrame'{body = {data, Message}}};
frame_for_type(?MESSAGE_EVENT_STREAM, Message = #'TaskEventStream'{}, _PacketId) ->
{cast, #'CastFrame'{body = {event_stream, Message}}};
frame_for_type(MessageType, Message, _PacketId) ->
erlang:error({unsupported_message_type, MessageType, Message}).
decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {auth_request, Message}})
when is_integer(PacketId), PacketId > 0 ->
{ok, PacketId, Message};
decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {rpc_request, Message}})
when is_integer(PacketId), PacketId > 0 ->
{ok, PacketId, Message};
decode_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, Message}})
when is_integer(PacketId), PacketId > 0 ->
{ok, PacketId, Message};
decode_request_frame(_) ->
error.
decode_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {auth_reply, Message}})
when is_integer(PacketId), PacketId > 0 ->
{ok, PacketId, Message};
decode_response_frame(#'ResponseFrame'{packet_id = PacketId, body = {rpc_reply, Message}})
when is_integer(PacketId), PacketId > 0 ->
{ok, PacketId, Message};
decode_response_frame(_) ->
error.
decode_cast_frame(#'CastFrame'{body = {pub, Message}}) ->
{ok, Message};
decode_cast_frame(#'CastFrame'{body = {command, Message}}) ->
{ok, Message};
decode_cast_frame(#'CastFrame'{body = {data, Message}}) ->
{ok, Message};
decode_cast_frame(#'CastFrame'{body = {event_stream, Message}}) ->
{ok, Message};
decode_cast_frame(_) ->
error.

6101
src/protobuf/message_pb.erl Normal file
View File

File diff suppressed because it is too large Load Diff