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

fix protocol

Browse Source
This commit is contained in:
anlicheng 2026-07-07 23:56:13 +08:00
parent 6a51ce8627
commit 0e7bee5188
1 changed files with 56 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
apps/efka/src/iot/efka_iot_client.erl
View File

@ -40,8 +40,8 @@
socket :: undefined | ssl:sslsocket(), socket :: undefined | ssl:sslsocket(),
outbox :: efka_iot_outbox:outbox(), outbox :: efka_iot_outbox:outbox(),
streams = #{}, streams = #{},
%% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系 %% 保存当前 auth 请求的 packet id，用来建立 auth 请求和响应的对应关系
auth_ref = undefined :: undefined | pos_integer(), auth_pkt_id = undefined :: undefined | pos_integer(),
next_pkt_id = 1 :: pos_integer(), next_pkt_id = 1 :: pos_integer(),
ping_timer_ref = undefined :: undefined | reference(), ping_timer_ref = undefined :: undefined | reference(),
dropped_message_count = 0 :: non_neg_integer() dropped_message_count = 0 :: non_neg_integer()
@ -195,7 +195,7 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State =
AuthPacket = auth_packet(PktId), AuthPacket = auth_packet(PktId),
ok = ssl:send(Socket, AuthPacket), ok = ssl:send(Socket, AuthPacket),
logger:debug("[efka_iot_client] send auth request, packet_id: ~p", [PktId]), logger:debug("[efka_iot_client] send auth request, packet_id: ~p", [PktId]),
{next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = PktId, next_pkt_id = PktId + 1}, [{state_timeout, 5000, auth_timeout}]}; {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_pkt_id = PktId, next_pkt_id = PktId + 1}, [{state_timeout, 5000, auth_timeout}]};
{error, _Reason} -> {error, _Reason} ->
schedule_reconnect(), schedule_reconnect(),
{keep_state, State#state{socket = undefined}} {keep_state, State#state{socket = undefined}}
@ -205,7 +205,7 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S
logger:debug("[efka_iot_client] auth request timeout"), logger:debug("[efka_iot_client] auth request timeout"),
disconnect(Socket), disconnect(Socket),
schedule_reconnect(), schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) -> handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
Packet = encode_message_frame(ping), Packet = encode_message_frame(ping),
@ -217,7 +217,7 @@ handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #sta
disconnect(Socket), disconnect(Socket),
schedule_reconnect(), schedule_reconnect(),
NState = close_all_streams({send_ping_failed, Reason}, State), NState = close_all_streams({send_ping_failed, Reason}, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}} {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
end; end;
handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) -> handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
{keep_state, State}; {keep_state, State};
@ -254,7 +254,7 @@ handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket})
cancel_ssl_ping(State), cancel_ssl_ping(State),
schedule_reconnect(), schedule_reconnect(),
NState = close_all_streams(bad_packet, State), NState = close_all_streams(bad_packet, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}} {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
end; end;
handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) -> handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
logger:debug("[efka_iot_client] ssl error: ~p", [Reason]), logger:debug("[efka_iot_client] ssl error: ~p", [Reason]),
@ -262,14 +262,14 @@ handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
cancel_ssl_ping(State), cancel_ssl_ping(State),
schedule_reconnect(), schedule_reconnect(),
NState = close_all_streams(ssl_error, State), NState = close_all_streams(ssl_error, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}; {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
handle_event(info, {ssl_closed, Socket}, _, State = #state{}) -> handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
logger:debug("[efka_iot_client] ssl closed"), logger:debug("[efka_iot_client] ssl closed"),
disconnect(Socket), disconnect(Socket),
cancel_ssl_ping(State), cancel_ssl_ping(State),
schedule_reconnect(), schedule_reconnect(),
NState = close_all_streams(ssl_closed, State), NState = close_all_streams(ssl_closed, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}; {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName, State = #state{streams = Streams}) -> handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName, State = #state{streams = Streams}) ->
case take_stream_by_monitor(MonitorRef, WorkerPid, Streams) of case take_stream_by_monitor(MonitorRef, WorkerPid, Streams) of
@ -282,7 +282,7 @@ handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName,
undefined -> undefined ->
{keep_state, State#state{streams = NStreams}}; {keep_state, State#state{streams = NStreams}};
Socket -> Socket ->
Packet = encode_stream_frame(StreamId, {reset, safe_term({worker_down, Reason})}), Packet = encode_stream_frame(StreamId, {reset, {worker_down, Reason}}),
ok = ssl:send(Socket, Packet), ok = ssl:send(Socket, Packet),
{keep_state, State#state{streams = NStreams}} {keep_state, State#state{streams = NStreams}}
end end
@ -291,25 +291,24 @@ handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName,
%%% 处理内部消息。TLS 收到的数据先经过 protobuf 解码，再由这里按本地事件分发。 %%% 处理内部消息。TLS 收到的数据先经过 protobuf 解码，再由这里按本地事件分发。
%% 容器管理命令由 iot 发起，使用 command/command_response 语义。 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
handle_container_command(Ref, ContainerCommand, Socket), handle_container_command(PktId, ContainerCommand, Socket),
{keep_state, State}; {keep_state, State};
handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) -> handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) ->
Request = decode_container_command(ContainerCommand), logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]), send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
{keep_state, State}; {keep_state, State};
%% 处理response %% 处理response
handle_event(internal, #'Response'{packet_id = AuthRef, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> handle_event(internal, #'Response'{packet_id = AuthPktId, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_pkt_id = AuthPktId}) ->
logger:debug("[efka_iot_client] auth success"), logger:debug("[efka_iot_client] auth success"),
State1 = schedule_ssl_ping(State#state{auth_ref = undefined}), State1 = schedule_ssl_ping(State#state{auth_pkt_id = undefined}),
{next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]}; {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
handle_event(internal, #'Response'{packet_id = AuthRef, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) -> handle_event(internal, #'Response'{packet_id = AuthPktId, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_pkt_id = AuthPktId}) ->
logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]), logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
disconnect(Socket), disconnect(Socket),
schedule_reconnect(), schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
handle_event(internal, #'Response'{} = Reply, StateName, State) -> handle_event(internal, #'Response'{} = Reply, StateName, State) ->
logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]), logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
{keep_state, State}; {keep_state, State};
@ -372,14 +371,14 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
%%%=================================================================== %%%===================================================================
-spec auth_packet(pos_integer()) -> binary(). -spec auth_packet(pos_integer()) -> binary().
auth_packet(Ref) when is_integer(Ref), Ref > 0 -> auth_packet(PktId) when is_integer(PktId), PktId > 0 ->
{ok, AuthInfo} = application:get_env(efka, auth), {ok, AuthInfo} = application:get_env(efka, auth),
UUID = proplists:get_value(uuid, AuthInfo), UUID = proplists:get_value(uuid, AuthInfo),
Token = proplists:get_value(token, AuthInfo), Token = proplists:get_value(token, AuthInfo),
Timestamp = efka_util:timestamp(), Timestamp = efka_util:timestamp(),
encode_transport_frame(?CLASS_REQUEST, #'Request'{ encode_transport_frame(?CLASS_REQUEST, #'Request'{
packet_id = Ref, packet_id = PktId,
body = {auth_request, #'Request.AuthRequest'{ body = {auth_request, #'Request.AuthRequest'{
uuid = list_to_binary(UUID), uuid = list_to_binary(UUID),
token = list_to_binary(Token), token = list_to_binary(Token),
@ -492,7 +491,7 @@ cancel_timer(TimerRef) ->
ok. ok.
-spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok. -spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
send_container_response(Socket, Ref, Reply) -> send_container_response(Socket, PktId, Reply) ->
Body = case Reply of Body = case Reply of
ok -> ok ->
{result, <<"ok">>}; {result, <<"ok">>};
@ -504,73 +503,45 @@ send_container_response(Socket, Ref, Reply) ->
{result, reply_to_binary(Other)} {result, reply_to_binary(Other)}
end, end,
Packet = encode_transport_frame(?CLASS_COMMAND_RESPONSE, #'CommandResponse'{ Packet = encode_transport_frame(?CLASS_COMMAND_RESPONSE, #'CommandResponse'{
packet_id = Ref, packet_id = PktId,
body = Body body = Body
}), }),
ok = ssl:send(Socket, Packet). ok = ssl:send(Socket, Packet).
-spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok. -spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
handle_container_command(Ref, ContainerCommand, Socket) -> handle_container_command(PktId, #'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}, Socket) ->
handle_container_request(Ref, decode_container_command(ContainerCommand), Socket).
-spec handle_container_request(pos_integer(), map(), ssl:sslsocket()) -> ok.
handle_container_request(Ref, #{<<"action">> := <<"list">>}, Socket) ->
Reply = docker_commands:get_containers(), Reply = docker_commands:get_containers(),
send_container_response(Socket, Ref, Reply), send_container_response(Socket, PktId, Reply),
ok; ok;
handle_container_request(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) -> handle_container_command(PktId, #'Command.Container'{action = {deploy, #'Command.Container.ContainerDeploy'{task_id = TaskId, params = Params}}}, Socket) ->
Reply = docker_deploy_manager:deploy(TaskId, Params), Reply = docker_deploy_manager:deploy(TaskId, decode_json_payload(Params)),
send_container_response(Socket, Ref, Reply), send_container_response(Socket, PktId, Reply),
ok; ok;
handle_container_request(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) -> handle_container_command(PktId, #'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}, Socket) ->
Reply = docker_commands:start_container(container_target(Target)), Reply = docker_commands:start_container(container_target(Target)),
send_container_response(Socket, Ref, Reply), send_container_response(Socket, PktId, Reply),
ok; ok;
handle_container_request(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) -> handle_container_command(PktId, #'Command.Container'{action = {stop, #'Command.Container.ContainerStop'{target = Target, timeout_seconds = TimeoutSeconds}}}, Socket) ->
Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds), Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
send_container_response(Socket, Ref, Reply), send_container_response(Socket, PktId, Reply),
ok; ok;
handle_container_request(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) -> handle_container_command(PktId, #'Command.Container'{action = {kill, #'Command.Container.ContainerKill'{target = Target, signal = Signal}}}, Socket) ->
Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)), Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
send_container_response(Socket, Ref, Reply), send_container_response(Socket, PktId, Reply),
ok; ok;
handle_container_request(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) -> handle_container_command(PktId, #'Command.Container'{action = {remove, #'Command.Container.ContainerRemove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}, Socket) ->
Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)), Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
send_container_response(Socket, Ref, Reply), send_container_response(Socket, PktId, Reply),
ok; ok;
handle_container_request(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) -> handle_container_command(PktId, #'Command.Container'{action = {config, #'Command.Container.ContainerConfig'{target = Target, config = Config}}}, Socket) ->
Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)), Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
send_container_response(Socket, Ref, Reply), send_container_response(Socket, PktId, Reply),
ok; ok;
handle_container_request(Ref, Request, Socket) -> handle_container_command(PktId, ContainerCommand, Socket) ->
logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]), logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
send_container_response(Socket, Ref, {error, <<"agent invalid">>}), send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
ok. ok.
-spec decode_container_command(message_pb:'Command.Container'()) -> map().
decode_container_command(#'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}) ->
#{<<"action">> => <<"list">>};
decode_container_command(#'Command.Container'{action = {deploy, #'Command.Container.ContainerDeploy'{task_id = TaskId, params = Params}}}) ->
#{<<"action">> => <<"deploy">>, <<"task_id">> => TaskId, <<"params">> => decode_json_payload(Params)};
decode_container_command(#'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}) ->
#{<<"action">> => <<"start">>, <<"target">> => decode_container_target(Target)};
decode_container_command(#'Command.Container'{action = {stop, #'Command.Container.ContainerStop'{target = Target, timeout_seconds = TimeoutSeconds}}}) ->
#{<<"action">> => <<"stop">>, <<"target">> => decode_container_target(Target), <<"timeout_seconds">> => TimeoutSeconds};
decode_container_command(#'Command.Container'{action = {kill, #'Command.Container.ContainerKill'{target = Target, signal = Signal}}}) ->
#{<<"action">> => <<"kill">>, <<"target">> => decode_container_target(Target), <<"signal">> => Signal};
decode_container_command(#'Command.Container'{action = {remove, #'Command.Container.ContainerRemove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}) ->
#{<<"action">> => <<"remove">>, <<"target">> => decode_container_target(Target), <<"force">> => Force, <<"remove_volumes">> => RemoveVolumes};
decode_container_command(#'Command.Container'{action = {config, #'Command.Container.ContainerConfig'{target = Target, config = Config}}}) ->
#{<<"action">> => <<"config">>, <<"target">> => decode_container_target(Target), <<"config">> => Config};
decode_container_command(Other) ->
#{<<"action">> => <<"invalid">>, <<"payload">> => safe_term(Other)}.
-spec decode_container_target(message_pb:'Command.Container.ContainerTarget'() | undefined) -> map().
decode_container_target(#'Command.Container.ContainerTarget'{name = Name, id = Id}) ->
#{<<"name">> => iolist_to_binary(Name), <<"id">> => iolist_to_binary(Id)};
decode_container_target(undefined) ->
#{<<"name">> => <<>>, <<"id">> => <<>>}.
-spec decode_json_payload(iodata()) -> term(). -spec decode_json_payload(iodata()) -> term().
decode_json_payload(Payload) -> decode_json_payload(Payload) ->
Bin = iolist_to_binary(Payload), Bin = iolist_to_binary(Payload),
@ -661,58 +632,45 @@ demonitor_stream(#stream_state{monitor_ref = MonitorRef}) ->
reply_to_binary(Value) when is_binary(Value) -> reply_to_binary(Value) when is_binary(Value) ->
Value; Value;
reply_to_binary(Value) -> reply_to_binary(Value) ->
SafeValue = safe_term(Value),
try iolist_to_binary(Value) of try iolist_to_binary(Value) of
Bin -> Bin ->
Bin Bin
catch catch
_:_ -> _:_ ->
try iolist_to_binary(json:encode(SafeValue)) of try iolist_to_binary(json:encode(Value)) of
JsonBin -> JsonBin ->
JsonBin JsonBin
catch catch
_:_ -> _:_ ->
term_to_binary(SafeValue) unicode:characters_to_binary(io_lib:format("~p", [Value]))
end end
end. end.
-spec reason_to_binary(term()) -> binary(). -spec reason_to_binary(term()) -> binary().
reason_to_binary(Reason) when is_binary(Reason) -> reason_to_binary(Reason) when is_binary(Reason) ->
Reason; Reason;
reason_to_binary(Reason) when is_list(Reason) ->
iolist_to_binary(Reason);
reason_to_binary(Reason) -> reason_to_binary(Reason) ->
term_to_binary(safe_term(Reason)). try iolist_to_binary(Reason) of
Bin ->
Bin
catch
_:_ ->
unicode:characters_to_binary(io_lib:format("~p", [Reason]))
end.
-spec safe_term(term()) -> term(). -spec container_target(message_pb:'Command.Container.ContainerTarget'() | undefined) -> binary().
safe_term(true) -> container_target(#'Command.Container.ContainerTarget'{name = Name, id = Id}) ->
true; NameBin = iolist_to_binary(Name),
safe_term(false) -> IdBin = iolist_to_binary(Id),
false;
safe_term(undefined) ->
undefined;
safe_term(Value) when is_atom(Value) ->
atom_to_binary(Value, utf8);
safe_term(Value) when is_map(Value) ->
maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]);
safe_term(Value) when is_list(Value) ->
[safe_term(Item) || Item <- Value];
safe_term(Value) when is_tuple(Value) ->
list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
safe_term(Value) ->
Value.
-spec container_target(map()) -> binary().
container_target(Target) when is_map(Target) ->
NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)),
IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)),
case NameBin of case NameBin of
<<>> -> <<>> ->
true = IdBin =/= <<>>, true = IdBin =/= <<>>,
IdBin; IdBin;
_ -> _ ->
NameBin NameBin
end. end;
container_target(undefined) ->
error(bad_container_target).
-spec to_binary(binary() | list()) -> binary(). -spec to_binary(binary() | list()) -> binary().
to_binary(Value) when is_binary(Value) -> to_binary(Value) when is_binary(Value) ->