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

Compare commits

base: anlicheng:main
Branches Tags
anlicheng:main
anlicheng:protobuf_mst
anlicheng:docker_mst
anlicheng:dets_mst
anlicheng:v1.0
..
compare: anlicheng:v1.0
Branches Tags
anlicheng:main
anlicheng:protobuf_mst
anlicheng:docker_mst
anlicheng:dets_mst
anlicheng:v1.0

No commits in common. "main" and "v1.0" have entirely different histories.
main ... v1.0

19 changed files with 446 additions and 5165 deletions
Show Stats Expand all files Collapse all files

90
apps/docker/src/docker_deploy_manager.erl Normal file
View File

@ -0,0 +1,90 @@
%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2026, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 20. 4月 2026
%%%-------------------------------------------------------------------
-module(docker_deploy_manager).
-author("anlicheng").
-behaviour(gen_server).
%% API
-export([start_link/0, deploy/2]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-define(SERVER, ?MODULE).
-record(state, {
root_dir :: string(),
task_map = #{}
}).
%%%===================================================================
%%% API
%%%===================================================================
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
-spec deploy(integer(), map()) -> ok | {error, binary()}.
deploy(TaskId, Params) when is_integer(TaskId), is_map(Params) ->
gen_server:call(?SERVER, {deploy, TaskId, Params}).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
-spec init(list()) -> {ok, #state{}}.
init([]) ->
erlang:process_flag(trap_exit, true),
{ok, RootDir} = docker_helper:root_dir(),
{ok, #state{root_dir = RootDir}}.
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.
handle_call({deploy, TaskId, Params}, _From, State = #state{root_dir = RootDir, task_map = TaskMap})
when is_map(Params) ->
ContainerName = maps:get(<<"container_name">>, Params),
{ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName),
{ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params),
logger:debug("[docker_deploy_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]),
{reply, ok, State#state{task_map = maps:put(TaskPid, TaskId, TaskMap)}};
handle_call(_Request, _From, State) ->
{reply, ok, State}.
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast(_Request, State) ->
{noreply, State}.
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
handle_info({'DOWN', _Ref, process, TaskPid, Reason}, State = #state{task_map = TaskMap}) ->
case maps:take(TaskPid, TaskMap) of
error ->
{noreply, State};
{TaskId, NTaskMap} ->
case Reason of
normal ->
logger:debug("[docker_deploy_manager] task_id: ~p, exit normal", [TaskId]);
Error0 ->
Error = iolist_to_binary(io_lib:format("~p", [Error0])),
docker_task_reporter:stream(TaskId, <<"error">>, <<"任务失败: "/utf8, Error/binary>>),
docker_task_reporter:close(TaskId, <<"task exited">>),
logger:notice("[docker_deploy_manager] task_id: ~p, exit with error: ~p", [TaskId, Error])
end,
{noreply, State#state{task_map = NTaskMap}}
end;
handle_info(_Info, State) ->
{noreply, State}.
-spec terminate(term(), #state{}) -> ok.
terminate(_Reason, _State) ->
ok.
-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.

118
apps/docker/src/docker_deployer.erl
View File

@ -11,17 +11,26 @@
-dialyzer([{nowarn_function, normalize_image/1}]).
%% API
-export([deploy/4]).
-export([start_monitor/3]).
-export([deploy/3]).
-define(TASK_SUCCESS, <<"success">>).
-define(TASK_FAIL, <<"fail">>).
-type reporter() :: {stream, pos_integer()}.
%%%===================================================================
%%% API
%%%===================================================================
-spec(start_monitor(TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
{ok, {pid(), reference()}}).
start_monitor(TaskId, ContainerDir, Params)
when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
{ok, spawn_monitor(?MODULE, deploy, [TaskId, ContainerDir, Params])}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
%{
% "image": "nginx:latest",
% "container_name": "my_nginx",
@ -32,35 +41,34 @@
% "command": ["nginx", "-g", "daemon off;"],
% "restart": "always"
%}
-spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: map(), Reporter :: reporter()) -> ok.
deploy(TaskId, ContainerDir, Params, Reporter) when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
-spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: map()) -> ok.
deploy(TaskId, ContainerDir, Params) when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
ContainerName = deploy_container_name(Params),
Image0 = deploy_image(Params),
report_stream_event(Reporter, TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
report_task_event(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
try
ok = ensure_container_absent(Reporter, TaskId, ContainerName),
{ok, Image} = ensure_image_ready(Reporter, TaskId, Image0),
{ok, ContainerId} = create_container_and_config(Reporter, TaskId, ContainerDir, Params),
ok = ensure_container_absent(TaskId, ContainerName),
{ok, Image} = ensure_image_ready(TaskId, Image0),
{ok, ContainerId} = create_container_and_config(TaskId, ContainerDir, Params),
ShortContainerId = short_container_id(ContainerId),
report_stream_event(Reporter, TaskId, <<"container_id">>, ContainerId),
report_stream_event(Reporter, TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>),
report_stream_event(Reporter, TaskId, <<"info">>, <<"任务完成"/utf8>>),
report_task_event(TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>),
report_task_event(TaskId, <<"info">>, <<"任务完成"/utf8>>),
write_task_summary(TaskId, <<"success">>, ContainerName, Image, ContainerId),
close_task(Reporter, TaskId, ?TASK_SUCCESS)
docker_task_reporter:close(TaskId, ?TASK_SUCCESS)
catch
throw:{deploy_error, Reason} ->
report_stream_event(Reporter, TaskId, <<"error">>, Reason),
report_stream_event(Reporter, TaskId, <<"error">>, <<"任务失败"/utf8>>),
report_task_event(TaskId, <<"error">>, Reason),
report_task_event(TaskId, <<"error">>, <<"任务失败"/utf8>>),
write_task_summary(TaskId, <<"fail">>, ContainerName, Image0, undefined),
write_task_failure_reason(TaskId, Reason),
close_task(Reporter, TaskId, ?TASK_FAIL);
docker_task_reporter:close(TaskId, ?TASK_FAIL);
Class:Reason:Stacktrace ->
Error = iolist_to_binary(io_lib:format("deploy crashed: ~p:~p ~p", [Class, Reason, Stacktrace])),
report_stream_event(Reporter, TaskId, <<"error">>, Error),
report_stream_event(Reporter, TaskId, <<"error">>, <<"任务失败"/utf8>>),
report_task_event(TaskId, <<"error">>, Error),
report_task_event(TaskId, <<"error">>, <<"任务失败"/utf8>>),
write_task_summary(TaskId, <<"fail">>, ContainerName, Image0, undefined),
write_task_failure_reason(TaskId, Error),
close_task(Reporter, TaskId, ?TASK_FAIL)
docker_task_reporter:close(TaskId, ?TASK_FAIL)
end.
-spec normalize_image(binary()) -> binary().
@ -73,33 +81,9 @@ normalize_image(Image) when is_binary(Image) ->
end,
iolist_to_binary(lists:join(<<"/">>, PrefixParts ++ [NormalizedLast])).
-spec report_stream_event(reporter(), TaskId :: integer(), Level :: binary(), Msg :: binary()) -> ok.
report_stream_event({stream, StreamId}, _TaskId, Level, Msg) when is_integer(StreamId), is_binary(Level), is_binary(Msg) ->
efka_iot_client:send_stream(StreamId, {data, encode_stream_event(Level, Msg)}).
-spec close_task(reporter(), integer(), binary()) -> ok.
close_task({stream, StreamId}, _TaskId, Reason) when is_integer(StreamId), is_binary(Reason) ->
ok = efka_iot_client:send_stream(StreamId, {data, encode_close_event(Reason)}),
efka_iot_client:send_stream(StreamId, fin).
-spec encode_stream_event(binary(), binary()) -> binary().
encode_stream_event(<<"container_id">>, ContainerId) ->
iolist_to_binary(json:encode(#{
<<"type">> => <<"container_id">>,
<<"container_id">> => ContainerId
}));
encode_stream_event(Type, Msg) ->
iolist_to_binary(json:encode(#{
<<"type">> => Type,
<<"message">> => Msg
})).
-spec encode_close_event(binary()) -> binary().
encode_close_event(Reason) ->
iolist_to_binary(json:encode(#{
<<"type">> => <<"close">>,
<<"reason">> => Reason
})).
-spec report_task_event(TaskId :: integer(), Level :: binary(), Msg :: binary()) -> ok.
report_task_event(TaskId, Level, Msg) when is_integer(TaskId), is_binary(Level), is_binary(Msg) ->
docker_task_reporter:stream(TaskId, Level, Msg).
-spec write_task_summary(integer(), binary(), binary(), binary(), undefined | binary()) -> ok.
write_task_summary(TaskId, Status, ContainerName, Image, ContainerId)
@ -129,55 +113,55 @@ write_task_failure_reason(TaskId, Reason) when is_integer(TaskId), is_binary(Rea
]),
efka_logger:write(Info).
-spec ensure_container_absent(reporter(), TaskId :: integer(), ContainerName :: binary()) -> ok.
ensure_container_absent(Reporter, TaskId, ContainerName) when is_integer(TaskId), is_binary(ContainerName) ->
report_stream_event(Reporter, TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
-spec ensure_container_absent(TaskId :: integer(), ContainerName :: binary()) -> ok.
ensure_container_absent(TaskId, ContainerName) when is_integer(TaskId), is_binary(ContainerName) ->
report_task_event(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
ok.
-spec ensure_image_ready(reporter(), TaskId :: integer(), Image0 :: binary()) -> {ok, binary()}.
ensure_image_ready(Reporter, TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
-spec ensure_image_ready(TaskId :: integer(), Image0 :: binary()) -> {ok, binary()}.
ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
Image = normalize_image(Image0),
report_stream_event(Reporter, TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
report_task_event(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
case docker_commands:check_image_exist(Image) of
true ->
report_stream_event(Reporter, TaskId, <<"info">>, <<"本地镜像已存在，跳过拉取:"/utf8, Image/binary>>),
report_task_event(TaskId, <<"info">>, <<"本地镜像已存在，跳过拉取:"/utf8, Image/binary>>),
{ok, Image};
false ->
report_stream_event(Reporter, TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
report_task_event(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
case docker_commands:pull_image(Image) of
{ok, Ref, Pid, MRef} ->
await_pull_image(Reporter, TaskId, Ref, Pid, MRef),
await_pull_image(TaskId, Ref, Pid, MRef),
{ok, Image}
end
end.
-spec await_pull_image(reporter(), TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok.
await_pull_image(Reporter, TaskId, Ref, Pid, MRef) ->
-spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok.
await_pull_image(TaskId, Ref, Pid, MRef) ->
receive
{docker_client, Ref, {response, _Status, _Headers}} ->
await_pull_image(Reporter, TaskId, Ref, Pid, MRef);
await_pull_image(TaskId, Ref, Pid, MRef);
{docker_client, Ref, {data, Data}} ->
report_stream_event(Reporter, TaskId, <<"info">>, Data),
await_pull_image(Reporter, TaskId, Ref, Pid, MRef);
report_task_event(TaskId, <<"info">>, Data),
await_pull_image(TaskId, Ref, Pid, MRef);
{docker_client, Ref, done} ->
erlang:demonitor(MRef, [flush]),
ok;
{docker_client, Ref, {error, Reason}} ->
erlang:demonitor(MRef, [flush]),
report_stream_event(Reporter, TaskId, <<"error">>, Reason),
report_task_event(TaskId, <<"error">>, Reason),
throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>});
{'DOWN', MRef, process, Pid, Reason0} ->
Reason = iolist_to_binary(io_lib:format("~p", [Reason0])),
throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>})
end.
-spec create_container_and_config(reporter(), TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
-spec create_container_and_config(TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
{ok, binary()}.
create_container_and_config(Reporter, TaskId, ContainerDir, Params)
create_container_and_config(TaskId, ContainerDir, Params)
when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
case docker_commands:create_container(ContainerDir, Params) of
{ok, ContainerId} ->
ok = create_config_file(Reporter, TaskId, ContainerDir),
ok = create_config_file(TaskId, ContainerDir),
{ok, ContainerId};
{error, Reason} when is_binary(Reason) ->
throw({deploy_error, format_create_container_error(Reason)});
@ -186,8 +170,8 @@ create_container_and_config(Reporter, TaskId, ContainerDir, Params)
throw({deploy_error, Error})
end.
-spec create_config_file(reporter(), TaskId :: integer(), ContainerDir :: string()) -> ok.
create_config_file(Reporter, TaskId, ContainerDir) when is_integer(TaskId), is_list(ContainerDir) ->
-spec create_config_file(TaskId :: integer(), ContainerDir :: string()) -> ok.
create_config_file(TaskId, ContainerDir) when is_integer(TaskId), is_list(ContainerDir) ->
ConfigFile = docker_helper:get_config_file(ContainerDir),
case file:open(ConfigFile, [write, exclusive]) of
{ok, FD} ->
@ -196,7 +180,7 @@ create_config_file(Reporter, TaskId, ContainerDir) when is_integer(TaskId), is_l
ok;
{error, Reason} ->
ReasonBin = list_to_binary(io_lib:format("~p", [Reason])),
report_stream_event(Reporter, TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, ReasonBin/binary>>),
report_task_event(TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, ReasonBin/binary>>),
ok
end.

20
apps/docker/src/docker_sup.erl
View File

@ -19,6 +19,24 @@ start_link() ->
-spec init(list()) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
init([]) ->
SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
ChildSpecs = [],
ChildSpecs = [
#{
id => docker_task_reporter,
start => {docker_task_reporter, start_link, []},
restart => permanent,
shutdown => 2000,
type => worker,
modules => [docker_task_reporter]
},
#{
id => docker_deploy_manager,
start => {docker_deploy_manager, start_link, []},
restart => permanent,
shutdown => 2000,
type => worker,
modules => [docker_deploy_manager]
}
],
{ok, {SupFlags, ChildSpecs}}.

121
apps/docker/src/docker_task_reporter.erl Normal file
View File

@ -0,0 +1,121 @@
%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2026, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 20. 4月 2026
%%%-------------------------------------------------------------------
-module(docker_task_reporter).
-author("anlicheng").
-behaviour(gen_server).
%% API
-export([start_link/0]).
-export([stream/3, close/2]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-define(SERVER, ?MODULE).
-define(FLUSH_INTERVAL, 1000).
-record(state, {
pending = queue:new(),
flush_ref = undefined
}).
%%%===================================================================
%%% API
%%%===================================================================
-spec stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
gen_server:cast(?SERVER, {stream, TaskId, Type, Stream}).
-spec close(TaskId :: integer(), Reason :: binary()) -> ok.
close(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
gen_server:cast(?SERVER, {close, TaskId, Reason}).
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
-spec init(list()) -> {ok, #state{}}.
init([]) ->
{ok, #state{}}.
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, ok, #state{}}.
handle_call(_Request, _From, State) ->
{reply, ok, State}.
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast({stream, TaskId, Type, Stream}, State0 = #state{pending = Pending0}) ->
Pending = queue:in({stream, TaskId, Type, Stream}, Pending0),
{noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
handle_cast({close, TaskId, Reason}, State0 = #state{pending = Pending0}) ->
Pending = queue:in({close, TaskId, Reason}, Pending0),
{noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
handle_cast(_Request, State) ->
{noreply, State}.
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
handle_info(flush, State0 = #state{}) ->
State1 = State0#state{flush_ref = undefined},
{noreply, flush_pending(State1)};
handle_info(_Info, State) ->
{noreply, State}.
-spec terminate(term(), #state{}) -> ok.
terminate(_Reason, _State) ->
ok.
-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec ensure_flush_timer(#state{}, non_neg_integer()) -> #state{}.
ensure_flush_timer(State = #state{flush_ref = undefined}, Delay) when is_integer(Delay), Delay >= 0 ->
Ref = erlang:send_after(Delay, self(), flush),
State#state{flush_ref = Ref};
ensure_flush_timer(State = #state{}, _Delay) ->
State.
-spec flush_pending(#state{}) -> #state{}.
flush_pending(State = #state{pending = Pending0}) ->
case queue:out(Pending0) of
{empty, _} ->
State;
{{value, Event}, Pending1} ->
case send_event(Event) of
ok ->
flush_pending(State#state{pending = Pending1});
not_ready ->
ensure_flush_timer(State#state{pending = Pending0}, ?FLUSH_INTERVAL)
end
end.
-spec send_event({stream, integer(), binary(), binary()} | {close, integer(), binary()}) -> ok | not_ready.
send_event({stream, TaskId, Type, Stream}) ->
maybe_send(fun() -> efka_iot_client:task_event_stream(TaskId, Type, Stream) end);
send_event({close, TaskId, Reason}) ->
maybe_send(fun() -> efka_iot_client:close_task_event_stream(TaskId, Reason) end).
-spec maybe_send(fun(() -> any())) -> ok | not_ready.
maybe_send(SendFun) ->
case catch efka_iot_client:is_activated() of
true ->
_ = catch SendFun(),
ok;
_ ->
not_ready
end.

16
apps/efka/include/service_pb.hrl → apps/efka/include/efka_service_pb.hrl
View File

@ -2,10 +2,10 @@
%% Automatically generated, do not edit
%% Generated by gpb_compile version 4.21.7
-ifndef(service_pb).
-define(service_pb, true).
-ifndef(efka_service_pb).
-define(efka_service_pb, true).
-define(service_pb_gpb_version, "4.21.7").
-define(efka_service_pb_gpb_version, "4.21.7").
-ifndef('SERVICEREQUEST.REGISTER_PB_H').
@ -25,8 +25,8 @@
-ifndef('SERVICEREQUEST_PB_H').
-define('SERVICEREQUEST_PB_H', true).
-record('ServiceRequest',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
request :: {register, service_pb:'ServiceRequest.Register'()} | {subscribe, service_pb:'ServiceRequest.Subscribe'()} | undefined % oneof
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
request :: {register, efka_service_pb:'ServiceRequest.Register'()} | {subscribe, efka_service_pb:'ServiceRequest.Subscribe'()} | undefined % oneof
}).
-endif.
@ -41,8 +41,8 @@
-ifndef('SERVICEREPLY_PB_H').
-define('SERVICEREPLY_PB_H', true).
-record('ServiceReply',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
reply :: {result, iodata()} | {error, service_pb:'ServiceReply.Error'()} | undefined % oneof
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
reply :: {result, iodata()} | {error, efka_service_pb:'ServiceReply.Error'()} | undefined % oneof
}).
-endif.
@ -65,7 +65,7 @@
-ifndef('SERVICECAST_PB_H').
-define('SERVICECAST_PB_H', true).
-record('ServiceCast',
{body :: {topic_event, service_pb:'ServiceCast.TopicEvent'()} | {metric_data, service_pb:'ServiceCast.MetricData'()} | undefined % oneof
{body :: {topic_event, efka_service_pb:'ServiceCast.TopicEvent'()} | {metric_data, efka_service_pb:'ServiceCast.MetricData'()} | undefined % oneof
}).
-endif.

25
apps/efka/include/message.hrl
View File

@ -1,25 +0,0 @@
%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2026, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 07. 7月 2026 17:09
%%%-------------------------------------------------------------------
-author("anlicheng").
%%--------------------------------------------------------------------
%% Wire classes
%%--------------------------------------------------------------------
-define(CLASS_REQUEST, 1).
-define(CLASS_RESPONSE, 2).
-define(CLASS_COMMAND, 3).
-define(CLASS_COMMAND_RESPONSE, 4).
-define(CLASS_MESSAGE, 5).
-define(CLASS_STREAM, 6).
%%--------------------------------------------------------------------
%% Stream targets
%%--------------------------------------------------------------------
-define(STREAM_TARGET_MANAGER, 1).
-define(STREAM_TARGET_CONTAINER_DEPLOY, 2).

225
apps/efka/include/message_pb.hrl
View File

@ -1,225 +0,0 @@
%% -*- 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('REQUEST.AUTHREQUEST_PB_H').
-define('REQUEST.AUTHREQUEST_PB_H', true).
-record('Request.AuthRequest',
{uuid = <<>> :: iodata() | undefined, % = 1, optional
token = <<>> :: iodata() | undefined, % = 2, optional
timestamp = 0 :: non_neg_integer() | undefined % = 3, optional, 64 bits
}).
-endif.
-ifndef('REQUEST_PB_H').
-define('REQUEST_PB_H', true).
-record('Request',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
body :: {auth_request, message_pb:'Request.AuthRequest'()} | undefined % oneof
}).
-endif.
-ifndef('RESPONSE.ERROR_PB_H').
-define('RESPONSE.ERROR_PB_H', true).
-record('Response.Error',
{code = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
reason = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('RESPONSE.AUTHRESPONSE_PB_H').
-define('RESPONSE.AUTHRESPONSE_PB_H', true).
-record('Response.AuthResponse',
{
}).
-endif.
-ifndef('RESPONSE_PB_H').
-define('RESPONSE_PB_H', true).
-record('Response',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
body :: {auth_response, message_pb:'Response.AuthResponse'()} | {error, message_pb:'Response.Error'()} | undefined % oneof
}).
-endif.
-ifndef('COMMAND.CONTAINER_PB_H').
-define('COMMAND.CONTAINER_PB_H', true).
-record('Command.Container',
{action :: {list, message_pb:'Command.Container.ContainerList'()} | {start, message_pb:'Command.Container.ContainerStart'()} | {stop, message_pb:'Command.Container.ContainerStop'()} | {kill, message_pb:'Command.Container.ContainerKill'()} | {remove, message_pb:'Command.Container.ContainerRemove'()} | {config, message_pb:'Command.Container.ContainerConfig'()} | undefined % oneof
}).
-endif.
-ifndef('COMMAND.CONTAINER.CONTAINERCONFIG_PB_H').
-define('COMMAND.CONTAINER.CONTAINERCONFIG_PB_H', true).
-record('Command.Container.ContainerConfig',
{target = undefined :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
config = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('COMMAND.CONTAINER.CONTAINERREMOVE_PB_H').
-define('COMMAND.CONTAINER.CONTAINERREMOVE_PB_H', true).
-record('Command.Container.ContainerRemove',
{target = undefined :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
force = false :: boolean() | 0 | 1 | undefined, % = 2, optional
remove_volumes = false :: boolean() | 0 | 1 | undefined % = 3, optional
}).
-endif.
-ifndef('COMMAND.CONTAINER.CONTAINERKILL_PB_H').
-define('COMMAND.CONTAINER.CONTAINERKILL_PB_H', true).
-record('Command.Container.ContainerKill',
{target = undefined :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
signal = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('COMMAND.CONTAINER.CONTAINERSTOP_PB_H').
-define('COMMAND.CONTAINER.CONTAINERSTOP_PB_H', true).
-record('Command.Container.ContainerStop',
{target = undefined :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
timeout_seconds = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
}).
-endif.
-ifndef('COMMAND.CONTAINER.CONTAINERSTART_PB_H').
-define('COMMAND.CONTAINER.CONTAINERSTART_PB_H', true).
-record('Command.Container.ContainerStart',
{target = undefined :: message_pb:'Command.Container.ContainerTarget'() | undefined % = 1, optional
}).
-endif.
-ifndef('COMMAND.CONTAINER.CONTAINERLIST_PB_H').
-define('COMMAND.CONTAINER.CONTAINERLIST_PB_H', true).
-record('Command.Container.ContainerList',
{
}).
-endif.
-ifndef('COMMAND.CONTAINER.CONTAINERTARGET_PB_H').
-define('COMMAND.CONTAINER.CONTAINERTARGET_PB_H', true).
-record('Command.Container.ContainerTarget',
{name = <<>> :: iodata() | undefined, % = 1, optional
id = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('COMMAND_PB_H').
-define('COMMAND_PB_H', true).
-record('Command',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
body :: {container, message_pb:'Command.Container'()} | undefined % oneof
}).
-endif.
-ifndef('COMMANDRESPONSE.ERROR_PB_H').
-define('COMMANDRESPONSE.ERROR_PB_H', true).
-record('CommandResponse.Error',
{code = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
reason = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('COMMANDRESPONSE_PB_H').
-define('COMMANDRESPONSE_PB_H', true).
-record('CommandResponse',
{packet_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 64 bits
body :: {result, iodata()} | {error, message_pb:'CommandResponse.Error'()} | undefined % oneof
}).
-endif.
-ifndef('MESSAGE.PING_PB_H').
-define('MESSAGE.PING_PB_H', true).
-record('Message.Ping',
{
}).
-endif.
-ifndef('MESSAGE.PONG_PB_H').
-define('MESSAGE.PONG_PB_H', true).
-record('Message.Pong',
{
}).
-endif.
-ifndef('MESSAGE.PUB_PB_H').
-define('MESSAGE.PUB_PB_H', true).
-record('Message.Pub',
{topic = <<>> :: iodata() | undefined, % = 1, optional
qos = 0 :: non_neg_integer() | undefined, % = 2, optional, 32 bits
content = <<>> :: iodata() | undefined % = 3, optional
}).
-endif.
-ifndef('MESSAGE.METRICDATA_PB_H').
-define('MESSAGE.METRICDATA_PB_H', true).
-record('Message.MetricData',
{route_key = <<>> :: iodata() | undefined, % = 1, optional
metric = <<>> :: iodata() | undefined % = 2, optional
}).
-endif.
-ifndef('MESSAGE_PB_H').
-define('MESSAGE_PB_H', true).
-record('Message',
{body :: {ping, message_pb:'Message.Ping'()} | {pong, message_pb:'Message.Pong'()} | {pub, message_pb:'Message.Pub'()} | {metric_data, message_pb:'Message.MetricData'()} | undefined % oneof
}).
-endif.
-ifndef('STREAM.OPEN_PB_H').
-define('STREAM.OPEN_PB_H', true).
-record('Stream.Open',
{target = 0 :: non_neg_integer() | undefined % = 1, optional, 32 bits
}).
-endif.
-ifndef('STREAM.OPENED_PB_H').
-define('STREAM.OPENED_PB_H', true).
-record('Stream.Opened',
{
}).
-endif.
-ifndef('STREAM.OPENERROR_PB_H').
-define('STREAM.OPENERROR_PB_H', true).
-record('Stream.OpenError',
{reason = <<>> :: iodata() | undefined % = 1, optional
}).
-endif.
-ifndef('STREAM.DATA_PB_H').
-define('STREAM.DATA_PB_H', true).
-record('Stream.Data',
{bytes = <<>> :: iodata() | undefined % = 1, optional
}).
-endif.
-ifndef('STREAM.FIN_PB_H').
-define('STREAM.FIN_PB_H', true).
-record('Stream.Fin',
{
}).
-endif.
-ifndef('STREAM.RESET_PB_H').
-define('STREAM.RESET_PB_H', true).
-record('Stream.Reset',
{reason = <<>> :: iodata() | undefined % = 1, optional
}).
-endif.
-ifndef('STREAM_PB_H').
-define('STREAM_PB_H', true).
-record('Stream',
{stream_id = 0 :: non_neg_integer() | undefined, % = 1, optional, 32 bits
payload :: {open, message_pb:'Stream.Open'()} | {opened, message_pb:'Stream.Opened'()} | {open_error, message_pb:'Stream.OpenError'()} | {data, message_pb:'Stream.Data'()} | {fin, message_pb:'Stream.Fin'()} | {reset, message_pb:'Stream.Reset'()} | undefined % oneof
}).
-endif.
-endif.

179
apps/efka/proto/message.proto
View File

@ -1,179 +0,0 @@
syntax = "proto3";
// iot <-> efka protocol payload definitions.
//
// The transport frame keeps the first byte as the coarse message class:
//
// [CLASS_REQUEST][protobuf(Request)]
// [CLASS_RESPONSE][protobuf(Response)]
// [CLASS_COMMAND][protobuf(Command)]
// [CLASS_COMMAND_RESPONSE][protobuf(CommandResponse)]
// [CLASS_MESSAGE][protobuf(Message)]
// [CLASS_STREAM][protobuf(Stream)]
//
// HTTP proxy bytes carried by Stream.Data are transparent payload bytes.
message Request {
uint64 packet_id = 1;
message AuthRequest {
bytes uuid = 1;
bytes token = 2;
uint64 timestamp = 3;
}
oneof body {
AuthRequest auth_request = 10;
}
}
message Response {
uint64 packet_id = 1;
message Error {
uint32 code = 1;
bytes reason = 2;
}
message AuthResponse {
}
oneof body {
AuthResponse auth_response = 10;
Error error = 11;
}
}
message Command {
uint64 packet_id = 1;
message Container {
message ContainerTarget {
bytes name = 1;
bytes id = 2;
}
message ContainerList {
}
message ContainerStart {
ContainerTarget target = 1;
}
message ContainerStop {
ContainerTarget target = 1;
uint32 timeout_seconds = 2;
}
message ContainerKill {
ContainerTarget target = 1;
bytes signal = 2;
}
message ContainerRemove {
ContainerTarget target = 1;
bool force = 2;
bool remove_volumes = 3;
}
message ContainerConfig {
ContainerTarget target = 1;
bytes config = 2;
}
oneof action {
ContainerList list = 1;
ContainerStart start = 3;
ContainerStop stop = 4;
ContainerKill kill = 5;
ContainerRemove remove = 6;
ContainerConfig config = 7;
}
}
oneof body {
Container container = 10;
}
}
message CommandResponse {
uint64 packet_id = 1;
message Error {
uint32 code = 1;
bytes reason = 2;
}
oneof body {
bytes result = 10;
Error error = 11;
}
}
message Message {
message Ping {
}
message Pong {
}
message Pub {
bytes topic = 1;
uint32 qos = 2;
bytes content = 3;
}
message MetricData {
bytes route_key = 1;
bytes metric = 2;
}
oneof body {
Ping ping = 10;
Pong pong = 11;
Pub pub = 12;
MetricData metric_data = 13;
}
}
message Stream {
uint32 stream_id = 1;
message Open {
uint32 target = 1;
}
message Opened {
}
message OpenError {
bytes reason = 1;
}
message Data {
bytes bytes = 1;
}
message Fin {
}
message Reset {
bytes reason = 1;
}
oneof payload {
Open open = 10;
Opened opened = 11;
OpenError open_error = 12;
Data data = 13;
Fin fin = 14;
Reset reset = 15;
}
}

4
apps/efka/proto/service.proto
View File

@ -4,7 +4,7 @@ syntax = "proto3";
// 该协议用于替代当前基于 JSON 的请求/响应/推送格式。
message ServiceRequest {
uint64 packet_id = 1;
uint32 packet_id = 1;
message Register {
string service_id = 1;
@ -21,7 +21,7 @@ message ServiceRequest {
}
message ServiceReply {
uint64 packet_id = 1;
uint32 packet_id = 1;
message Error {
int32 code = 1;

4
apps/efka/rebar.config
View File

@ -2,9 +2,9 @@
{gpb_opts, [
{i, "proto"},
{f, ["service.proto", "message.proto"]},
{f, ["service.proto"]},
recursive,
{module_name_prefix, ""},
{module_name_prefix, "efka_"},
{module_name_suffix, "_pb"},
{o_erl, "src/protobuf"},
{o_hrl, "include"},

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

@ -9,15 +9,12 @@
-module(efka_iot_client).
-author("anlicheng").
-include("efka_tables.hrl").
-include("message.hrl").
-include("message_pb.hrl").
-behaviour(gen_statem).
%% API
-export([start_link/0]).
-export([metric_data/2, ping/13]).
-export([send_stream/2, stream_done/1]).
-export([metric_data/2, ping/13, task_event_stream/3, close_task_event_stream/2]).
-export([is_activated/0, dropped_message_count/0]).
%% gen_statem callbacks
@ -39,21 +36,12 @@
-record(state, {
socket :: undefined | ssl:sslsocket(),
outbox :: efka_iot_outbox:outbox(),
streams = #{},
%% 保存当前 auth 请求的 packet id，用来建立 auth 请求和响应的对应关系
auth_pkt_id = undefined :: undefined | pos_integer(),
next_pkt_id = 1 :: pos_integer(),
%% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
auth_ref = undefined :: undefined | binary(),
ping_timer_ref = undefined :: undefined | reference(),
dropped_message_count = 0 :: non_neg_integer()
}).
-record(stream_state, {
worker_pid :: pid(),
monitor_ref :: reference()
}).
-type stream_id() :: pos_integer().
%%%===================================================================
%%% API
%%%===================================================================
@ -63,13 +51,13 @@
metric_data(RouteKey, Metric) when is_binary(RouteKey), is_binary(Metric) ->
gen_statem:cast(?SERVER, {metric_data, RouteKey, Metric}).
-spec send_stream(StreamId :: stream_id(), Body :: term()) -> ok.
send_stream(StreamId, Body) when is_integer(StreamId), StreamId > 0 ->
gen_statem:cast(?SERVER, {send_stream, StreamId, Body}).
-spec task_event_stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
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 stream_done(StreamId :: stream_id()) -> ok.
stream_done(StreamId) when is_integer(StreamId), StreamId > 0 ->
gen_statem:cast(?SERVER, {stream_done, StreamId}).
-spec close_task_event_stream(TaskId :: integer(), Reason :: binary()) -> ok.
close_task_event_stream(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
gen_statem:cast(?SERVER, {close_task_event_stream, TaskId, Reason}).
-spec is_activated() -> boolean().
is_activated() ->
@ -118,7 +106,7 @@ outbox_options() ->
%% 异步发送数据，连接存在时直接发送；否则写入持久化 outbox。
-spec handle_event(term(), term(), atom(), #state{}) -> term().
handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
Packet = encode_message_frame({metric_data, RouteKey, Metric}),
Packet = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}),
case StateName of
?STATE_ACTIVATED ->
ok = ssl:send(Socket, Packet),
@ -139,23 +127,17 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so
end
end;
handle_event(cast, {send_stream, StreamId, Body}, ?STATE_ACTIVATED, State = #state{socket = Socket})
when is_integer(StreamId), StreamId > 0 ->
Packet = encode_stream_frame(StreamId, Body),
%% Task的stream流，只做实时的
handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
logger:debug("[efka_iot_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}),
ok = ssl:send(Socket, Packet),
{keep_state, State};
handle_event(cast, {send_stream, _StreamId, _Body}, _StateName, State) ->
{keep_state, State};
handle_event(cast, {stream_done, StreamId}, _StateName, State = #state{streams = Streams})
when is_integer(StreamId), StreamId > 0 ->
case maps:take(StreamId, Streams) of
error ->
handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
ok = ssl:send(Socket, Packet),
{keep_state, State};
{StreamState, NStreams} ->
demonitor_stream(StreamState),
{keep_state, State#state{streams = NStreams}}
end;
%% 其他情况下直接忽略
handle_event(cast, _, _, State = #state{}) ->
@ -169,13 +151,14 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro
{keep_state, State, [{reply, From, DroppedCount}]};
%% 异步建立到服务器的连接
handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = #state{next_pkt_id = PktId}) ->
handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
case connect_socket() of
{ok, Socket} ->
AuthPacket = auth_packet(PktId),
Ref = request_ref(),
AuthPacket = auth_packet(Ref),
ok = ssl:send(Socket, AuthPacket),
logger:debug("[efka_iot_client] send auth request, packet_id: ~p", [PktId]),
{next_state, ?STATE_AUTH, State#state{socket = Socket, auth_pkt_id = PktId, next_pkt_id = PktId + 1}, [{state_timeout, 5000, auth_timeout}]};
logger:debug("[efka_iot_client] send auth request, ref: ~p", [Ref]),
{next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
{error, _Reason} ->
schedule_reconnect(),
{keep_state, State#state{socket = undefined}}
@ -185,10 +168,10 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S
logger:debug("[efka_iot_client] auth request timeout"),
disconnect(Socket),
schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
Packet = encode_message_frame(ping),
Packet = term_to_binary({<<"message">>, <<"ping">>}),
case ssl:send(Socket, Packet) of
ok ->
{keep_state, schedule_ssl_ping(State)};
@ -196,8 +179,7 @@ handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #sta
logger:warning("[efka_iot_client] send ssl ping failed, reason: ~p", [Reason]),
disconnect(Socket),
schedule_reconnect(),
NState = close_all_streams({send_ping_failed, Reason}, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
end;
handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
{keep_state, State};
@ -225,123 +207,110 @@ handle_event(info, flush_cache, _, State) ->
%% 处理收到的ssl消息
handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
case decode_frame(PacketBin) of
{ok, Packet} ->
{keep_state, State, [{next_event, internal, Packet}]};
{error, Reason} ->
logger:warning("[efka_iot_client] decode packet failed: ~p, packet_size: ~p", [Reason, byte_size(PacketBin)]),
try binary_to_term(PacketBin, [safe]) of
Packet ->
{keep_state, State, [{next_event, internal, Packet}]}
catch
error:Error ->
logger:warning("[efka_iot_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
disconnect(Socket),
cancel_ssl_ping(State),
schedule_reconnect(),
NState = close_all_streams(bad_packet, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
end;
handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
logger:debug("[efka_iot_client] ssl error: ~p", [Reason]),
disconnect(Socket),
cancel_ssl_ping(State),
schedule_reconnect(),
NState = close_all_streams(ssl_error, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
logger:debug("[efka_iot_client] ssl closed"),
disconnect(Socket),
cancel_ssl_ping(State),
schedule_reconnect(),
NState = close_all_streams(ssl_closed, State),
{next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName, State = #state{streams = Streams}) ->
case take_stream_by_monitor(MonitorRef, WorkerPid, Streams) of
error ->
{keep_state, State};
{StreamId, StreamState, NStreams} ->
demonitor_stream(StreamState),
case Reason of
normal ->
{keep_state, State#state{streams = NStreams}};
_ ->
logger:warning("[efka_iot_client] stream worker down, stream_id: ~p, reason: ~p", [StreamId, Reason]),
case State#state.socket of
undefined ->
{keep_state, State#state{streams = NStreams}};
Socket ->
Packet = encode_stream_frame(StreamId, {reset, {worker_down, Reason}}),
ok = ssl:send(Socket, Packet),
{keep_state, State#state{streams = NStreams}}
end
end
end;
%%% 处理内部消息。TLS 收到的数据先经过 protobuf 解码，再由这里按本地事件分发。
%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
%% 容器管理命令由 iot 发起，使用 command/command_response 语义。
handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
handle_container_command(PktId, ContainerCommand, Socket),
handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
handle_container_command(Ref, Request, Socket),
{keep_state, State};
handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) ->
logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
{keep_state, State};
%% 处理response
handle_event(internal, #'Response'{packet_id = AuthPktId, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_pkt_id = AuthPktId}) ->
handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
logger:debug("[efka_iot_client] auth success"),
State1 = schedule_ssl_ping(State#state{auth_pkt_id = undefined}),
State1 = schedule_ssl_ping(State#state{auth_ref = undefined}),
{next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
handle_event(internal, #'Response'{packet_id = AuthPktId, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_pkt_id = AuthPktId}) ->
handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
disconnect(Socket),
schedule_reconnect(),
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
handle_event(internal, #'Response'{} = Reply, StateName, State) ->
{next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) ->
logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
{keep_state, State};
handle_event(internal, #'CommandResponse'{} = Reply, StateName, State) ->
handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) ->
logger:warning("[efka_iot_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
{keep_state, State};
%% 透明 TCP stream 多路复用。
handle_event(internal, #'Stream'{stream_id = StreamId, payload = Payload}, ?STATE_ACTIVATED, State) ->
Body = case Payload of
{open, #'Stream.Open'{target = Target}} ->
{open, Target};
{opened, #'Stream.Opened'{}} ->
opened;
{open_error, #'Stream.OpenError'{reason = Reason}} ->
{open_error, Reason};
{data, #'Stream.Data'{bytes = Data}} ->
{data, Data};
{fin, #'Stream.Fin'{}} ->
fin;
{reset, #'Stream.Reset'{reason = Reason}} ->
{reset, Reason}
end,
handle_stream_frame(StreamId, Body, State);
handle_event(internal, #'Stream'{stream_id = StreamId, payload = Payload}, StateName, State) ->
logger:warning("[efka_iot_client] ignore stream frame in state ~p, stream_id: ~p, body: ~p",
[StateName, StreamId, Payload]),
{keep_state, State};
%% 处理Pub/Sub机制
handle_event(internal, #'Message'{body = {pong, #'Message.Pong'{}}}, ?STATE_ACTIVATED, State) ->
handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
{keep_state, State};
handle_event(internal, #'Message'{body = {pub, #'Message.Pub'{topic = Topic, qos = Qos, content = Content}}}, ?STATE_ACTIVATED, State) ->
handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
logger:debug("[efka_iot_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
efka_subscription:publish(Topic, Qos, Content),
{keep_state, State};
handle_event(internal, Packet, StateName, State) ->
logger:warning("[efka_iot_client] ignore unknown packet: ~p, state_name: ~p", [Packet, StateName]),
handle_event(internal, Packet, _StateName, State) ->
logger:warning("[efka_iot_client] ignore unknown packet: ~p", [Packet]),
{keep_state, State};
handle_event(info, Info, _, State = #state{}) ->
logger:notice("[efka_iot_client] get unknown info: ~p", [Info]),
{keep_state, State}.
-spec handle_container_command(binary(), term(), ssl:sslsocket()) -> ok.
handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) ->
Reply = docker_commands:get_containers(),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
Reply = docker_deploy_manager:deploy(TaskId, Params),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
Reply = docker_commands:start_container(container_target(Target)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
send_container_response(Socket, Ref, Reply),
ok;
handle_container_command(Ref, Request, Socket) ->
logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
ok.
-spec terminate(term(), atom(), #state{}) -> ok.
terminate(Reason, _StateName, State = #state{socket = Socket, outbox = Outbox}) ->
cancel_ssl_ping(State),
_ = close_all_streams(Reason, State),
disconnect(Socket),
efka_iot_outbox:close(Outbox),
logger:notice("[efka_iot_client] terminate with reason: ~p", [Reason]),
@ -355,83 +324,18 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
%%% Internal functions
%%%===================================================================
-spec auth_packet(pos_integer()) -> binary().
auth_packet(PktId) when is_integer(PktId), PktId > 0 ->
-spec auth_packet(binary()) -> binary().
auth_packet(Ref) when is_binary(Ref) ->
{ok, AuthInfo} = application:get_env(efka, auth),
UUID = proplists:get_value(uuid, AuthInfo),
Token = proplists:get_value(token, AuthInfo),
Timestamp = efka_util:timestamp(),
encode_transport_frame(?CLASS_REQUEST, #'Request'{
packet_id = PktId,
body = {auth_request, #'Request.AuthRequest'{
uuid = list_to_binary(UUID),
token = list_to_binary(Token),
timestamp = Timestamp
}}
}).
-spec encode_message_frame(term()) -> binary().
encode_message_frame(ping) ->
encode_transport_frame(?CLASS_MESSAGE, #'Message'{body = {ping, #'Message.Ping'{}}});
encode_message_frame({metric_data, RouteKey, Metric}) ->
encode_transport_frame(?CLASS_MESSAGE, #'Message'{
body = {metric_data, #'Message.MetricData'{route_key = RouteKey, metric = Metric}}
});
encode_message_frame(Body) ->
error({unsupported_message_body, Body}).
-spec encode_stream_frame(stream_id(), term()) -> binary().
encode_stream_frame(StreamId, Body) ->
Payload = case Body of
open ->
{open, #'Stream.Open'{target = ?STREAM_TARGET_MANAGER}};
{open, Target} when is_integer(Target), Target > 0 ->
{open, #'Stream.Open'{target = Target}};
opened ->
{opened, #'Stream.Opened'{}};
{open_error, Reason} ->
{open_error, #'Stream.OpenError'{reason = reason_to_binary(Reason)}};
{data, Data} when is_binary(Data) ->
{data, #'Stream.Data'{bytes = Data}};
fin ->
{fin, #'Stream.Fin'{}};
{reset, Reason} ->
{reset, #'Stream.Reset'{reason = reason_to_binary(Reason)}}
end,
encode_transport_frame(?CLASS_STREAM, #'Stream'{stream_id = StreamId, payload = Payload}).
-spec encode_transport_frame(byte(), message_pb:'$msg'()) -> binary().
encode_transport_frame(Class, Msg) ->
Payload = message_pb:encode_msg(Msg),
<<Class, Payload/binary>>.
-spec decode_frame(binary()) -> {ok, tuple()} | {error, term()}.
decode_frame(<<?CLASS_RESPONSE, Payload/binary>>) ->
decode_pb_frame(Payload, 'Response');
decode_frame(<<?CLASS_COMMAND, Payload/binary>>) ->
decode_pb_frame(Payload, 'Command');
decode_frame(<<?CLASS_COMMAND_RESPONSE, Payload/binary>>) ->
decode_pb_frame(Payload, 'CommandResponse');
decode_frame(<<?CLASS_MESSAGE, Payload/binary>>) ->
decode_pb_frame(Payload, 'Message');
decode_frame(<<?CLASS_STREAM, Payload/binary>>) ->
decode_pb_frame(Payload, 'Stream');
decode_frame(<<?CLASS_REQUEST, _Payload/binary>>) ->
{error, unsupported_request_frame};
decode_frame(Other) ->
{error, {invalid_frame, Other}}.
-spec decode_pb_frame(binary(), message_pb:'$msg_name'()) ->
{ok, tuple()} | {error, term()}.
decode_pb_frame(Payload, MsgName) ->
try message_pb:decode_msg(Payload, MsgName) of
Msg ->
{ok, Msg}
catch
Class:Reason ->
{error, {bad_protobuf, MsgName, Class, Reason}}
end.
term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
<<"uuid">> => list_to_binary(UUID),
<<"token">> => list_to_binary(Token),
<<"timestamp">> => Timestamp
}}}).
-spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
connect_socket() ->
@ -473,190 +377,52 @@ cancel_timer(TimerRef) ->
_ = erlang:cancel_timer(TimerRef),
ok.
-spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
send_container_response(Socket, PktId, Reply) ->
Body = case Reply of
ok ->
{result, <<"ok">>};
{ok, Result} ->
{result, reply_to_binary(Result)};
{error, Reason} ->
{error, #'CommandResponse.Error'{code = 1, reason = reason_to_binary(Reason)}};
Other ->
{result, reply_to_binary(Other)}
end,
Packet = encode_transport_frame(?CLASS_COMMAND_RESPONSE, #'CommandResponse'{
packet_id = PktId,
body = Body
}),
-spec request_ref() -> binary().
request_ref() ->
crypto:strong_rand_bytes(16).
-spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
send_container_response(Socket, Ref, Reply) ->
Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
ok = ssl:send(Socket, Packet).
-spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
handle_container_command(PktId, #'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}, Socket) ->
Reply = docker_commands:get_containers(),
send_container_response(Socket, PktId, Reply),
ok;
handle_container_command(PktId, #'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}, Socket) ->
Reply = docker_commands:start_container(container_target(Target)),
send_container_response(Socket, PktId, Reply),
ok;
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),
send_container_response(Socket, PktId, Reply),
ok;
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)),
send_container_response(Socket, PktId, Reply),
ok;
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)),
send_container_response(Socket, PktId, Reply),
ok;
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)),
send_container_response(Socket, PktId, Reply),
ok;
handle_container_command(PktId, ContainerCommand, Socket) ->
logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
ok.
-spec safe_reply(term()) -> term().
safe_reply(ok) ->
<<"ok">>;
safe_reply({ok, Result}) ->
{<<"ok">>, safe_term(Result)};
safe_reply({error, Reason}) ->
{<<"error">>, safe_term(Reason)}.
%% iot 发起 open 时携带 target；efka 根据本地 stream_targets 配置建连。
-spec handle_stream_frame(term(), term(), #state{}) -> gen_statem:event_handler_result(atom(), #state{}).
handle_stream_frame(StreamId, {open, Target}, State = #state{streams = Streams})
when is_integer(StreamId), StreamId > 0, is_integer(Target), Target > 0 ->
case valid_iot_stream_id(StreamId) andalso not maps:is_key(StreamId, Streams) of
true ->
case start_stream_worker(StreamId, Target) of
{ok, {WorkerPid, MonitorRef}} ->
StreamState = #stream_state{worker_pid = WorkerPid, monitor_ref = MonitorRef},
{keep_state, State#state{streams = maps:put(StreamId, StreamState, Streams)}};
{error, Reason} ->
send_stream(StreamId, {open_error, Reason}),
{keep_state, State}
end;
false ->
send_stream(StreamId, {reset, <<"invalid stream open">>}),
{keep_state, State}
end;
handle_stream_frame(StreamId, {open, Target}, State)
when is_integer(StreamId), StreamId > 0 ->
logger:warning("[efka_iot_client] invalid stream target, stream_id: ~p, target: ~p", [StreamId, Target]),
send_stream(StreamId, {open_error, <<"invalid stream target">>}),
{keep_state, State};
handle_stream_frame(StreamId, Body, State = #state{streams = Streams})
when is_integer(StreamId), StreamId > 0 ->
case maps:get(StreamId, Streams, undefined) of
undefined ->
maybe_reset_unknown_stream(StreamId, Body),
{keep_state, State};
StreamState = #stream_state{worker_pid = WorkerPid} ->
WorkerPid ! {stream, StreamId, Body},
case Body of
{reset, _Reason} ->
demonitor_stream(StreamState),
{keep_state, State#state{streams = maps:remove(StreamId, Streams)}};
_ ->
{keep_state, State}
end
end;
handle_stream_frame(StreamId, Body, State) ->
logger:warning("[efka_iot_client] invalid stream frame, stream_id: ~p, body: ~p", [StreamId, Body]),
{keep_state, State}.
-spec safe_term(term()) -> term().
safe_term(true) ->
true;
safe_term(false) ->
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 start_stream_worker(stream_id(), pos_integer()) -> {ok, {pid(), reference()}} | {error, term()}.
start_stream_worker(StreamId, ?STREAM_TARGET_MANAGER) ->
efka_iot_stream:start_stream(StreamId, ?STREAM_TARGET_MANAGER);
start_stream_worker(StreamId, ?STREAM_TARGET_CONTAINER_DEPLOY) ->
efka_iot_deploy_stream:start_stream(StreamId);
start_stream_worker(_StreamId, Target) ->
{error, {unknown_stream_target, Target}}.
-spec maybe_reset_unknown_stream(stream_id(), term()) -> ok.
maybe_reset_unknown_stream(_StreamId, {reset, _Reason}) ->
ok;
maybe_reset_unknown_stream(_StreamId, fin) ->
ok;
maybe_reset_unknown_stream(_StreamId, {open_error, _Reason}) ->
ok;
maybe_reset_unknown_stream(StreamId, _Body) ->
send_stream(StreamId, {reset, <<"unknown stream">>}).
-spec valid_iot_stream_id(stream_id()) -> boolean().
valid_iot_stream_id(StreamId) ->
StreamId rem 2 =:= 1.
-spec take_stream_by_monitor(reference(), pid(), map()) ->
{stream_id(), #stream_state{}, map()} | error.
take_stream_by_monitor(MonitorRef, WorkerPid, Streams) ->
take_stream_by_monitor(MonitorRef, WorkerPid, maps:iterator(Streams), Streams).
take_stream_by_monitor(MonitorRef, WorkerPid, Iter, Streams) ->
case maps:next(Iter) of
none ->
error;
{StreamId, StreamState = #stream_state{worker_pid = WorkerPid, monitor_ref = MonitorRef}, _NextIter} ->
{StreamId, StreamState, maps:remove(StreamId, Streams)};
{_StreamId, _StreamState, NextIter} ->
take_stream_by_monitor(MonitorRef, WorkerPid, NextIter, Streams)
end.
-spec close_all_streams(term(), #state{}) -> #state{}.
close_all_streams(Reason, State = #state{streams = Streams}) ->
maps:foreach(fun(StreamId, StreamState = #stream_state{worker_pid = WorkerPid}) ->
demonitor_stream(StreamState),
WorkerPid ! {stream, StreamId, {reset, {channel_closed, Reason}}}
end, Streams),
State#state{streams = #{}}.
-spec demonitor_stream(#stream_state{}) -> ok.
demonitor_stream(#stream_state{monitor_ref = MonitorRef}) ->
erlang:demonitor(MonitorRef, [flush]),
ok.
-spec reply_to_binary(term()) -> binary().
reply_to_binary(Value) when is_binary(Value) ->
Value;
reply_to_binary(Value) ->
try iolist_to_binary(Value) of
Bin ->
Bin
catch
_:_ ->
try iolist_to_binary(json:encode(Value)) of
JsonBin ->
JsonBin
catch
_:_ ->
unicode:characters_to_binary(io_lib:format("~p", [Value]))
end
end.
-spec reason_to_binary(term()) -> binary().
reason_to_binary(Reason) when is_binary(Reason) ->
Reason;
reason_to_binary(Reason) ->
try iolist_to_binary(Reason) of
Bin ->
Bin
catch
_:_ ->
unicode:characters_to_binary(io_lib:format("~p", [Reason]))
end.
-spec container_target(message_pb:'Command.Container.ContainerTarget'() | undefined) -> binary().
container_target(#'Command.Container.ContainerTarget'{name = Name, id = Id}) ->
NameBin = iolist_to_binary(Name),
IdBin = iolist_to_binary(Id),
-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
<<>> ->
true = IdBin =/= <<>>,
IdBin;
_ ->
NameBin
end;
container_target(undefined) ->
error(bad_container_target).
end.
-spec to_binary(binary() | list()) -> binary().
to_binary(Value) when is_binary(Value) ->

119
apps/efka/src/iot/efka_iot_deploy_stream.erl
View File

@ -1,119 +0,0 @@
%%%-------------------------------------------------------------------
%%% @doc One container deploy feedback stream from iot.
%%% @end
%%%-------------------------------------------------------------------
-module(efka_iot_deploy_stream).
-export([start_stream/1]).
-export([run/1]).
-define(REQUEST_TIMEOUT, 10000).
-define(MAX_REQUEST_BYTES, 16 * 1024 * 1024).
-type stream_id() :: pos_integer().
-spec start_stream(StreamId :: stream_id()) -> {ok, {pid(), reference()}}.
start_stream(StreamId) when is_integer(StreamId), StreamId > 0 ->
{ok, spawn_monitor(?MODULE, run, [StreamId])}.
-spec run(StreamId :: stream_id()) -> ok.
run(StreamId) when is_integer(StreamId), StreamId > 0 ->
try run0(StreamId) of
ok ->
ok
catch
Class:Reason:Stack ->
logger:warning("[efka_iot_deploy_stream] stream_id: ~p crashed, class: ~p, reason: ~p, stack: ~p",
[StreamId, Class, Reason, Stack]),
send_error_and_close(StreamId, iolist_to_binary(io_lib:format("~p:~p", [Class, Reason]))),
ok
after
efka_iot_client:stream_done(StreamId)
end.
-spec run0(stream_id()) -> ok.
run0(StreamId) ->
efka_iot_client:send_stream(StreamId, opened),
case receive_request(StreamId, <<>>) of
{ok, Request} ->
handle_request(StreamId, Request);
{error, reset} ->
ok;
{error, Reason} ->
send_error_and_close(StreamId, reason_to_binary(Reason))
end.
-spec receive_request(stream_id(), binary()) -> {ok, binary()} | {error, term()}.
receive_request(StreamId, Acc) ->
receive
{stream, StreamId, {data, Data}} when is_binary(Data) ->
NAcc = <<Acc/binary, Data/binary>>,
case byte_size(NAcc) =< ?MAX_REQUEST_BYTES of
true ->
receive_request(StreamId, NAcc);
false ->
{error, request_too_large}
end;
{stream, StreamId, fin} ->
{ok, Acc};
{stream, StreamId, {reset, _Reason}} ->
{error, reset};
Info ->
logger:debug("[efka_iot_deploy_stream] stream_id: ~p ignore unknown info: ~p", [StreamId, Info]),
receive_request(StreamId, Acc)
after ?REQUEST_TIMEOUT ->
{error, request_timeout}
end.
-spec handle_request(stream_id(), binary()) -> ok.
handle_request(StreamId, RequestBin) ->
case decode_request(RequestBin) of
{ok, TaskId, Params} ->
deploy(StreamId, TaskId, Params);
{error, Reason} ->
send_error_and_close(StreamId, reason_to_binary(Reason))
end.
-spec decode_request(binary()) -> {ok, non_neg_integer(), map()} | {error, term()}.
decode_request(RequestBin) ->
try json:decode(RequestBin) of
#{<<"task_id">> := TaskId, <<"params">> := Params}
when is_integer(TaskId), TaskId >= 0, is_map(Params) ->
{ok, TaskId, Params};
_ ->
{error, invalid_deploy_request}
catch
Class:Reason ->
{error, {bad_json, Class, Reason}}
end.
-spec deploy(stream_id(), non_neg_integer(), map()) -> ok.
deploy(StreamId, TaskId, Params) ->
try
ContainerName = maps:get(<<"container_name">>, Params),
{ok, RootDir} = docker_helper:root_dir(),
{ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName),
docker_deployer:deploy(TaskId, ContainerDir, Params, {stream, StreamId})
catch
Class:Reason ->
Error = iolist_to_binary(io_lib:format("deploy stream failed: ~p:~p", [Class, Reason])),
send_error_and_close(StreamId, Error)
end.
-spec send_error_and_close(stream_id(), binary()) -> ok.
send_error_and_close(StreamId, Reason) ->
ok = efka_iot_client:send_stream(StreamId, {data, iolist_to_binary(json:encode(#{
<<"type">> => <<"error">>,
<<"message">> => Reason
}))}),
ok = efka_iot_client:send_stream(StreamId, {data, iolist_to_binary(json:encode(#{
<<"type">> => <<"close">>,
<<"reason">> => <<"fail">>
}))}),
efka_iot_client:send_stream(StreamId, fin).
-spec reason_to_binary(term()) -> binary().
reason_to_binary(Reason) when is_binary(Reason) ->
Reason;
reason_to_binary(Reason) ->
unicode:characters_to_binary(io_lib:format("~p", [Reason])).

156
apps/efka/src/iot/efka_iot_stream.erl
View File

@ -1,156 +0,0 @@
%%%-------------------------------------------------------------------
%%% @doc One transparent TCP stream from iot to the local manager service.
%%% The iot/efka protocol does not inspect HTTP; all HTTP bytes are carried
%%% in data frames.
%%% @end
%%%-------------------------------------------------------------------
-module(efka_iot_stream).
-include("message.hrl").
-export([start_stream/2]).
-export([run/2]).
-define(DEFAULT_CONNECT_TIMEOUT, 3000).
-define(DEFAULT_IDLE_TIMEOUT, 120000).
-type stream_id() :: pos_integer().
-type stream_target() :: pos_integer().
-spec start_stream(StreamId :: integer(), Target :: stream_target()) -> {ok, {pid(), reference()}}.
start_stream(StreamId, Target) when is_integer(StreamId), StreamId > 0, is_integer(Target), Target > 0 ->
{ok, spawn_monitor(?MODULE, run, [StreamId, Target])}.
-spec run(StreamId :: stream_id(), Target :: stream_target()) -> ok.
run(StreamId, Target) when is_integer(StreamId), StreamId > 0, is_integer(Target), Target > 0 ->
try run0(StreamId, Target) of
ok ->
ok
catch
Class:Reason:Stack ->
logger:warning("[efka_iot_stream] stream_id: ~p crashed, class: ~p, reason: ~p, stack: ~p",
[StreamId, Class, Reason, Stack]),
efka_iot_client:send_stream(StreamId, {reset, safe_term({Class, Reason})}),
ok
after
efka_iot_client:stream_done(StreamId)
end.
-spec run0(stream_id(), stream_target()) -> ok.
run0(StreamId, Target) ->
case open_target_socket(Target) of
{ok, Socket, IdleTimeout} ->
efka_iot_client:send_stream(StreamId, opened),
ok = inet:setopts(Socket, [{active, once}]),
loop(StreamId, Socket, IdleTimeout);
{error, Reason} ->
efka_iot_client:send_stream(StreamId, {open_error, safe_term(Reason)}),
ok
end.
-spec open_target_socket(stream_target()) -> {ok, gen_tcp:socket(), timeout()} | {error, term()}.
open_target_socket(Target) ->
case stream_target_props(Target) of
{ok, Props} ->
open_target_socket(Target, Props);
{error, Reason} ->
{error, Reason}
end.
-spec open_target_socket(stream_target(), proplists:proplist()) ->
{ok, gen_tcp:socket(), timeout()} | {error, term()}.
open_target_socket(Target, Props) ->
Host = proplists:get_value(host, Props),
Port = proplists:get_value(port, Props),
ConnectTimeout = proplists:get_value(connect_timeout, Props, ?DEFAULT_CONNECT_TIMEOUT),
IdleTimeout = proplists:get_value(idle_timeout, Props, ?DEFAULT_IDLE_TIMEOUT),
SocketOpts = [
binary,
{packet, raw},
{active, false},
{nodelay, true}
],
case gen_tcp:connect(Host, Port, SocketOpts, ConnectTimeout) of
{ok, Socket} ->
{ok, Socket, IdleTimeout};
{error, Reason} ->
{error, {connect_failed, Target, Reason}}
end.
-spec stream_target_props(stream_target()) -> {ok, proplists:proplist()} | {error, term()}.
stream_target_props(Target) ->
case application:get_env(efka, stream_targets) of
{ok, Targets} ->
case proplists:get_value(Target, Targets) of
undefined ->
{error, {unknown_stream_target, Target}};
Props ->
{ok, Props}
end;
undefined when Target =:= ?STREAM_TARGET_MANAGER ->
case application:get_env(efka, stream_target) of
{ok, Props} ->
{ok, Props};
undefined ->
{error, {unknown_stream_target, Target}}
end;
undefined ->
{error, {unknown_stream_target, Target}}
end.
-spec loop(stream_id(), gen_tcp:socket(), timeout()) -> ok.
loop(StreamId, Socket, IdleTimeout) ->
receive
{stream, StreamId, {data, Data}} when is_binary(Data) ->
case gen_tcp:send(Socket, Data) of
ok ->
loop(StreamId, Socket, IdleTimeout);
{error, Reason} ->
efka_iot_client:send_stream(StreamId, {reset, safe_term(Reason)}),
close_socket(Socket)
end;
{stream, StreamId, fin} ->
_ = gen_tcp:shutdown(Socket, write),
loop(StreamId, Socket, IdleTimeout);
{stream, StreamId, {reset, _Reason}} ->
close_socket(Socket);
{tcp, Socket, Data} ->
efka_iot_client:send_stream(StreamId, {data, Data}),
ok = inet:setopts(Socket, [{active, once}]),
loop(StreamId, Socket, IdleTimeout);
{tcp_closed, Socket} ->
efka_iot_client:send_stream(StreamId, fin),
close_socket(Socket);
{tcp_error, Socket, Reason} ->
efka_iot_client:send_stream(StreamId, {reset, safe_term(Reason)}),
close_socket(Socket);
Info ->
logger:debug("[efka_iot_stream] stream_id: ~p ignore unknown info: ~p", [StreamId, Info]),
loop(StreamId, Socket, IdleTimeout)
after IdleTimeout ->
efka_iot_client:send_stream(StreamId, {reset, <<"idle_timeout">>}),
close_socket(Socket)
end.
-spec close_socket(gen_tcp:socket()) -> ok.
close_socket(Socket) ->
catch gen_tcp:close(Socket),
ok.
-spec safe_term(term()) -> term().
safe_term(true) ->
true;
safe_term(false) ->
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.

30
apps/efka/src/protobuf/service_pb.erl → apps/efka/src/protobuf/efka_service_pb.erl
View File

@ -3,7 +3,7 @@
%% Automatically @generated, do not edit
%% Generated by gpb_compile version 4.21.7
%% Version source: file
-module(service_pb).
-module(efka_service_pb).
-export([encode_msg/1, encode_msg/2, encode_msg/3]).
-export([decode_msg/2, decode_msg/3]).
@ -46,7 +46,7 @@
-export([gpb_version_as_string/0, gpb_version_as_list/0]).
-export([gpb_version_source/0]).
-include("service_pb.hrl").
-include("efka_service_pb.hrl").
-include_lib("gpb/include/gpb.hrl").
%% enumerated types
@ -558,7 +558,7 @@ dg_read_field_def_ServiceRequest(<<>>, 0, 0, _, F@_1, F@_2, _) -> #'ServiceReque
d_field_ServiceRequest_packet_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceRequest_packet_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
d_field_ServiceRequest_packet_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
{NewFValue, RestF} = {id((X bsl N + Acc) band 18446744073709551615, TrUserData), Rest},
{NewFValue, RestF} = {id((X bsl N + Acc) band 4294967295, TrUserData), Rest},
dfp_read_field_def_ServiceRequest(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
d_field_ServiceRequest_register(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceRequest_register(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
@ -687,7 +687,7 @@ dg_read_field_def_ServiceReply(<<>>, 0, 0, _, F@_1, F@_2, _) -> #'ServiceReply'{
d_field_ServiceReply_packet_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceReply_packet_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
d_field_ServiceReply_packet_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
{NewFValue, RestF} = {id((X bsl N + Acc) band 18446744073709551615, TrUserData), Rest},
{NewFValue, RestF} = {id((X bsl N + Acc) band 4294967295, TrUserData), Rest},
dfp_read_field_def_ServiceReply(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
d_field_ServiceReply_result(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceReply_result(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
@ -1111,7 +1111,7 @@ verify_msg(Msg, MsgName, Opts) ->
-dialyzer({nowarn_function,v_msg_ServiceRequest/3}).
v_msg_ServiceRequest(#'ServiceRequest'{packet_id = F1, request = F2}, Path, TrUserData) ->
if F1 == undefined -> ok;
true -> v_type_uint64(F1, [packet_id | Path], TrUserData)
true -> v_type_uint32(F1, [packet_id | Path], TrUserData)
end,
case F2 of
undefined -> ok;
@ -1142,7 +1142,7 @@ v_msg_ServiceRequest(X, Path, _TrUserData) -> mk_type_error({expected_msg, 'Serv
-dialyzer({nowarn_function,v_msg_ServiceReply/3}).
v_msg_ServiceReply(#'ServiceReply'{packet_id = F1, reply = F2}, Path, TrUserData) ->
if F1 == undefined -> ok;
true -> v_type_uint64(F1, [packet_id | Path], TrUserData)
true -> v_type_uint32(F1, [packet_id | Path], TrUserData)
end,
case F2 of
undefined -> ok;
@ -1203,11 +1203,11 @@ v_type_int32(N, _Path, _TrUserData) when is_integer(N), -2147483648 =< N, N =< 2
v_type_int32(N, Path, _TrUserData) when is_integer(N) -> mk_type_error({value_out_of_range, int32, signed, 32}, N, Path);
v_type_int32(X, Path, _TrUserData) -> mk_type_error({bad_integer, int32, signed, 32}, X, Path).
-compile({nowarn_unused_function,v_type_uint64/3}).
-dialyzer({nowarn_function,v_type_uint64/3}).
v_type_uint64(N, _Path, _TrUserData) when is_integer(N), 0 =< N, N =< 18446744073709551615 -> ok;
v_type_uint64(N, Path, _TrUserData) when is_integer(N) -> mk_type_error({value_out_of_range, uint64, unsigned, 64}, N, Path);
v_type_uint64(X, Path, _TrUserData) -> mk_type_error({bad_integer, uint64, unsigned, 64}, X, Path).
-compile({nowarn_unused_function,v_type_uint32/3}).
-dialyzer({nowarn_function,v_type_uint32/3}).
v_type_uint32(N, _Path, _TrUserData) when is_integer(N), 0 =< N, N =< 4294967295 -> ok;
v_type_uint32(N, Path, _TrUserData) when is_integer(N) -> mk_type_error({value_out_of_range, uint32, unsigned, 32}, N, Path);
v_type_uint32(X, Path, _TrUserData) -> mk_type_error({bad_integer, uint32, unsigned, 32}, X, Path).
-compile({nowarn_unused_function,v_type_string/3}).
-dialyzer({nowarn_function,v_type_string/3}).
@ -1267,14 +1267,14 @@ get_msg_defs() ->
[{{msg, 'ServiceRequest.Register'}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}]},
{{msg, 'ServiceRequest.Subscribe'}, [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}]},
{{msg, 'ServiceRequest'},
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
#gpb_oneof{name = request, rnum = 3,
fields =
[#field{name = register, fnum = 10, rnum = 3, type = {msg, 'ServiceRequest.Register'}, occurrence = optional, opts = []}, #field{name = subscribe, fnum = 11, rnum = 3, type = {msg, 'ServiceRequest.Subscribe'}, occurrence = optional, opts = []}],
opts = []}]},
{{msg, 'ServiceReply.Error'}, [#field{name = code, fnum = 1, rnum = 2, type = int32, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
{{msg, 'ServiceReply'},
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
#gpb_oneof{name = reply, rnum = 3, fields = [#field{name = result, fnum = 10, rnum = 3, type = bytes, occurrence = optional, opts = []}, #field{name = error, fnum = 11, rnum = 3, type = {msg, 'ServiceReply.Error'}, occurrence = optional, opts = []}],
opts = []}]},
{{msg, 'ServiceCast.MetricData'}, [#field{name = route_key, fnum = 1, rnum = 2, type = bytes, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}]},
@ -1312,14 +1312,14 @@ fetch_enum_def(EnumName) -> erlang:error({no_such_enum, EnumName}).
find_msg_def('ServiceRequest.Register') -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}];
find_msg_def('ServiceRequest.Subscribe') -> [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}];
find_msg_def('ServiceRequest') ->
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
#gpb_oneof{name = request, rnum = 3,
fields =
[#field{name = register, fnum = 10, rnum = 3, type = {msg, 'ServiceRequest.Register'}, occurrence = optional, opts = []}, #field{name = subscribe, fnum = 11, rnum = 3, type = {msg, 'ServiceRequest.Subscribe'}, occurrence = optional, opts = []}],
opts = []}];
find_msg_def('ServiceReply.Error') -> [#field{name = code, fnum = 1, rnum = 2, type = int32, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
find_msg_def('ServiceReply') ->
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
[#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
#gpb_oneof{name = reply, rnum = 3, fields = [#field{name = result, fnum = 10, rnum = 3, type = bytes, occurrence = optional, opts = []}, #field{name = error, fnum = 11, rnum = 3, type = {msg, 'ServiceReply.Error'}, occurrence = optional, opts = []}],
opts = []}];
find_msg_def('ServiceCast.MetricData') -> [#field{name = route_key, fnum = 1, rnum = 2, type = bytes, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}];

3938
apps/efka/src/protobuf/message_pb.erl
View File

File diff suppressed because it is too large Load Diff

16
apps/efka/src/service/efka_service_channel.erl
View File

@ -9,7 +9,7 @@
-module(efka_service_channel).
-author("licheng5").
-include("efka_tables.hrl").
-include("service_pb.hrl").
-include("efka_service_pb.hrl").
%% 一级帧类型
%% REQUEST: 需要响应的请求帧
@ -50,11 +50,11 @@ websocket_handle(ping, State) ->
{reply, pong, State};
websocket_handle({binary, <<?FRAME_REQUEST, PacketBin/binary>>}, State) ->
Request = service_pb:decode_msg(PacketBin, 'ServiceRequest'),
Request = efka_service_pb:decode_msg(PacketBin, 'ServiceRequest'),
logger:debug("[efka_service_channel] get request: ~p", [Request]),
handle_request(Request, State);
websocket_handle({binary, <<?FRAME_CAST, PacketBin/binary>>}, State) ->
Cast = service_pb:decode_msg(PacketBin, 'ServiceCast'),
Cast = efka_service_pb:decode_msg(PacketBin, 'ServiceCast'),
logger:debug("[efka_service_channel] get cast: ~p", [Cast]),
handle_cast(Cast, State);
@ -66,7 +66,7 @@ websocket_handle(Info, State) ->
-spec websocket_info(term(), #state{}) ->
{reply, term(), #state{}} | {stop, #state{}} | {ok, #state{}}.
websocket_info({topic_broadcast, Topic, Content}, State = #state{}) ->
Packet = service_pb:encode_msg(#'ServiceCast'{
Packet = efka_service_pb:encode_msg(#'ServiceCast'{
body = {topic_event, #'ServiceCast.TopicEvent'{topic = Topic, content = Content}}
}),
logger:debug("[efka_service_channel] will publish topic: ~p", [Topic]),
@ -105,7 +105,7 @@ terminate(Reason, _Req, State = #state{service_id = ServiceId, is_registered = I
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% 注册, 要建立程序和容器之间的关系
-spec handle_request(service_pb:'ServiceRequest'(), #state{}) -> {reply, {binary, binary()}, #state{}}.
-spec handle_request(efka_service_pb:'ServiceRequest'(), #state{}) -> {reply, {binary, binary()}, #state{}}.
handle_request(#'ServiceRequest'{packet_id = PacketId, request = {register, #'ServiceRequest.Register'{service_id = ServiceId}}}, State) ->
{ok, ServicePid} = efka_service_sup:start_service(ServiceId),
case efka_service:attach_channel(ServicePid, self()) of
@ -144,7 +144,7 @@ handle_request(#'ServiceRequest'{packet_id = PacketId, request = {subscribe, #'S
handle_request(#'ServiceRequest'{packet_id = PacketId}, State) ->
{reply, {binary, error_reply_packet(PacketId, -1, <<"invalid request">>)}, State}.
-spec handle_cast(service_pb:'ServiceCast'(), #state{}) -> {ok, #state{}}.
-spec handle_cast(efka_service_pb:'ServiceCast'(), #state{}) -> {ok, #state{}}.
handle_cast(#'ServiceCast'{body = {metric_data, #'ServiceCast.MetricData'{route_key = RouteKey, metric = Metric}}},
State = #state{service_pid = ServicePid, is_registered = true}) ->
efka_service:metric_data(ServicePid, RouteKey, Metric),
@ -154,7 +154,7 @@ handle_cast(#'ServiceCast'{body = _Body}, State) ->
-spec result_reply_packet(integer(), binary()) -> binary().
result_reply_packet(PacketId, Result) when is_integer(PacketId), is_binary(Result) ->
Reply = service_pb:encode_msg(#'ServiceReply'{
Reply = efka_service_pb:encode_msg(#'ServiceReply'{
packet_id = PacketId,
reply = {result, Result}
}),
@ -162,7 +162,7 @@ result_reply_packet(PacketId, Result) when is_integer(PacketId), is_binary(Resul
-spec error_reply_packet(integer(), integer(), binary()) -> binary().
error_reply_packet(PacketId, Code, Message) when is_integer(PacketId), is_integer(Code), is_binary(Message) ->
Reply = service_pb:encode_msg(#'ServiceReply'{
Reply = efka_service_pb:encode_msg(#'ServiceReply'{
packet_id = PacketId,
reply = {error, #'ServiceReply.Error'{code = Code, message = Message}}
}),

10
config/sys.config.src
View File

@ -15,16 +15,6 @@
{udp_port, 24000}
]},
{stream_targets, [
{1, [
{name, manager},
{host, "127.0.0.1"},
{port, 81},
{connect_timeout, 3000},
{idle_timeout, 120000}
]}
]},
{heartbeat, [
{interval, 5000}
]},

46
docs/efka_iot_protocol.md
View File

@ -21,7 +21,6 @@
{<<"command">>, Ref, {Domain, Payload}}
{<<"command_response">>, Ref, {Domain, Reply}}
{<<"message">>, Body}
{<<"stream">>, StreamId, Body}
```
语义说明：
@ -33,56 +32,11 @@
| `{<<"command">>, Ref, {Domain, Payload}}` | iot -> efka | iot 下发命令，需要 efka 回复 |
| `{<<"command_response">>, Ref, {Domain, Reply}}` | efka -> iot | efka 对 iot command 的回复 |
| `{<<"message">>, Body}` | 双向 | 异步消息，不要求回复 |
| `{<<"stream">>, StreamId, Body}` | 双向 | 透明 TCP 字节流多路复用 |
`command` 和 `command_response` 的 `Domain` 表示业务域，目前支持：
- `<<"container">>`
## 透明 TCP Stream
`stream` 用于在同一条 TLS 长连接上复用多条透明 TCP 字节流。旧的 `request`、`response`、`command`、`command_response`、`message` 帧格式保持不变；`StreamId = 0` 只作为保留概念，不出现在新的 `stream` 帧里。
`StreamId` 规则：
- `0`：保留给现有控制流。
- 奇数：`iot` 发起。
- 偶数：`efka` 发起，当前预留。
- 同一条 TLS 连接内 `StreamId` 不复用。
`Body` 取值：
```erlang
<<"open">>
<<"opened">>
{<<"open_error">>, Reason}
{<<"data">>, Chunk}
<<"fin">>
{<<"reset">>, Reason}
```
语义：
- `open`：发起方请求打开一个透明 TCP stream。`open` 不携带参数；`efka` 端固定连接本机 `stream_target`，通常指向本机 manager/nginx 入口。
- `opened`：本地 TCP 连接建立成功。
- `open_error`：本地 TCP 连接建立失败，stream 结束。
- `data`：透明字节数据，`Chunk` 是 binary。HTTP 请求头、请求体、响应头、响应体都只是该 binary 的内容，iot/efka 中间协议不解析 HTTP。
- `fin`：半关闭，发送方后续不再发送 `data`，但仍可继续接收对端 `data`。
- `reset`：异常关闭，双方应立即释放该 `StreamId` 的资源。
典型 iot 发起访问 efka 本机监控服务的流程：
```erlang
iot -> efka: {<<"stream">>, 1, <<"open">>}
efka -> iot: {<<"stream">>, 1, <<"opened">>}
iot -> efka: {<<"stream">>, 1, {<<"data">>, RawHttpRequestBytes}}
iot -> efka: {<<"stream">>, 1, <<"fin">>}
efka -> iot: {<<"stream">>, 1, {<<"data">>, RawHttpResponseBytes}}
efka -> iot: {<<"stream">>, 1, <<"fin">>}
```
## 鉴权请求
初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。

4
rebar.config
View File

@ -16,8 +16,8 @@
[efka,
sasl]},
{mode, dev},
{include_erts, false},
{mode, prod},
{include_erts, true},
{sys_config_src, "./config/sys.config.src"},
{vm_args_src, "./config/vm.args.src"}