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

fix spec

Browse Source
This commit is contained in:
anlicheng 2026-04-21 16:24:59 +08:00
parent ac4fabad1d
commit 5cb386c363
18 changed files with 117 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/docker/docker_commands.erl
View File

@ -232,16 +232,19 @@ display_options(Options) when is_map(Options) ->
logger:debug("deploy options: ~p", [jiffy:encode(Options, [force_utf8])]),
lists:foreach(fun({K, V}) -> logger:debug("~p => ~p", [K, V]) end, maps:to_list(Options)).
-spec build_stop_container_url(ContainerName :: binary(), TimeoutSeconds :: non_neg_integer()) -> string().
build_stop_container_url(ContainerName, 0) ->
lists:flatten(io_lib:format("/containers/~s/stop", [binary_to_list(ContainerName)]));
build_stop_container_url(ContainerName, TimeoutSeconds) ->
lists:flatten(io_lib:format("/containers/~s/stop?t=~B", [binary_to_list(ContainerName), TimeoutSeconds])).
-spec build_kill_container_url(ContainerName :: binary(), Signal :: binary()) -> string().
build_kill_container_url(ContainerName, <<>>) ->
lists:flatten(io_lib:format("/containers/~s/kill", [binary_to_list(ContainerName)]));
build_kill_container_url(ContainerName, Signal) ->
lists:flatten(io_lib:format("/containers/~s/kill?signal=~s", [binary_to_list(ContainerName), binary_to_list(Signal)])).
-spec build_remove_container_url(ContainerName :: binary(), Force :: boolean(), RemoveVolumes :: boolean()) -> string().
build_remove_container_url(ContainerName, Force, RemoveVolumes) ->
ForceValue = boolean_to_query_value(Force),
RemoveVolumesValue = boolean_to_query_value(RemoveVolumes),
@ -251,6 +254,7 @@ build_remove_container_url(ContainerName, Force, RemoveVolumes) ->
RemoveVolumesValue
])).
-spec boolean_to_query_value(boolean()) -> string().
boolean_to_query_value(true) ->
"true";
boolean_to_query_value(false) ->

3
src/docker/docker_container_service.erl
View File

@ -39,6 +39,7 @@ handle_request(#'ContainerRequest'{action = {config, #'ContainerRequest.Config'{
ContainerTarget = container_target(Target),
update_container_config(ContainerTarget, iolist_to_binary(Config)).
-spec container_target(message_pb:'ContainerRef'()) -> binary().
container_target(#'ContainerRef'{name = Name, id = Id}) ->
NameBin = to_binary(Name),
IdBin = to_binary(Id),
@ -50,11 +51,13 @@ container_target(#'ContainerRef'{name = Name, id = Id}) ->
NameBin
end.
-spec to_binary(binary() | list()) -> binary().
to_binary(Value) when is_binary(Value) ->
Value;
to_binary(Value) when is_list(Value) ->
unicode:characters_to_binary(Value).
-spec to_bool(true | false | 0 | 1) -> boolean().
to_bool(true) ->
true;
to_bool(1) ->

6
src/docker/docker_deploy_manager.erl
View File

@ -41,11 +41,13 @@ deploy(TaskId, Params) when is_integer(TaskId), is_record(Params, 'ContainerDepl
%%% gen_server callbacks
%%%===================================================================
-spec init(list()) -> {ok, #state{}}.
init([]) ->
erlang:process_flag(trap_exit, true),
{ok, RootDir} = application:get_env(efka, root_dir),
{ok, #state{root_dir = RootDir}}.
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.
handle_call({deploy, TaskId, Params = #'ContainerDeployParams'{
container_name = ContainerName,
container_dir = ContainerDir0
@ -57,9 +59,11 @@ handle_call({deploy, TaskId, Params = #'ContainerDeployParams'{
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 ->
@ -79,8 +83,10 @@ handle_info({'DOWN', _Ref, process, TaskPid, Reason}, State = #state{task_map =
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}.

2
src/docker/docker_events.erl
View File

@ -130,6 +130,7 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec handle_event(term(), #state{}) -> ok.
handle_event(#{<<"Type">> := <<"container">>, <<"status">> := Status, <<"id">> := Id}, #state{monitors = Monitors}) ->
case maps:find(Id, Monitors) of
error ->
@ -147,5 +148,6 @@ handle_event(#{<<"Type">> := <<"container">>, <<"status">> := Status, <<"id">> :
handle_event(_, _) ->
ok.
-spec try_attach_events(non_neg_integer()) -> reference().
try_attach_events(Timeout) ->
erlang:start_timer(Timeout, self(), attach_docker_events).

4
src/docker/docker_helper.erl
View File

@ -49,17 +49,21 @@ get_container_dir(RootDir, ContainerName) when is_list(RootDir), is_binary(Conta
error
end.
-spec default_container_dir(RootDir :: string(), ContainerName :: binary()) -> string().
default_container_dir(RootDir, ContainerName) ->
normalize_container_dir(RootDir ++ "/" ++ binary_to_list(ContainerName)).
-spec container_dir_pointer_file(ContainerDir :: string()) -> string().
container_dir_pointer_file(ContainerDir) ->
ContainerDir ++ ".container_dir".
-spec resolve_container_dir(RootDir :: string(), ContainerName :: binary(), ContainerDir :: binary()) -> string().
resolve_container_dir(RootDir, ContainerName, <<>>) ->
default_container_dir(RootDir, ContainerName);
resolve_container_dir(_RootDir, _ContainerName, ContainerDir) ->
normalize_container_dir(binary_to_list(ContainerDir)).
-spec normalize_container_dir(ContainerDir :: string()) -> string().
normalize_container_dir(ContainerDir) ->
case lists:last(ContainerDir) of
$/ ->

6
src/docker/docker_http.erl
View File

@ -22,6 +22,8 @@ request(Method, Path, Body, Headers) when is_list(Method), is_list(Path), is_bin
{error, Reason}
end.
-spec receive_response(pid(), reference()) ->
{ok, integer(), proplists:proplist(), binary()} | {error, any()}.
receive_response(ConnPid, StreamRef) ->
receive
{gun_response, ConnPid, StreamRef, nofin, Status, Headers} ->
@ -33,6 +35,8 @@ receive_response(ConnPid, StreamRef) ->
after 5000 ->
{error, timeout}
end.
-spec receive_body(pid(), reference(), integer(), proplists:proplist(), binary()) ->
{ok, integer(), proplists:proplist(), binary()} | {error, timeout22}.
receive_body(ConnPid, StreamRef, Status, Headers, Acc) ->
receive
{gun_data, ConnPid, StreamRef, fin, Data} ->
@ -63,6 +67,7 @@ stream_request(Callback, Method, Path, Body, Headers) when is_list(Method), is_l
{error, Reason}
end.
-spec receive_response(fun((term()) -> any()), pid(), reference()) -> ok | {error, any()}.
receive_response(Callback, ConnPid, StreamRef) ->
receive
{gun_response, ConnPid, StreamRef, nofin, _Status, _Headers} ->
@ -74,6 +79,7 @@ receive_response(Callback, ConnPid, StreamRef) ->
Callback({error, <<"处理超时"/utf8>>}),
{error, timeout}
end.
-spec receive_body(fun((term()) -> any()), pid(), reference()) -> ok.
receive_body(Callback, ConnPid, StreamRef) ->
receive
{gun_data, ConnPid, StreamRef, fin, Data} ->

10
src/docker/docker_task_reporter.erl
View File

@ -46,12 +46,15 @@ start_link() ->
%%% 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)};
@ -61,15 +64,18 @@ handle_cast({close, TaskId, Reason}, State0 = #state{pending = Pending0}) ->
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}.
@ -77,12 +83,14 @@ code_change(_OldVsn, State, _Extra) ->
%%% 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, _} ->
@ -96,11 +104,13 @@ flush_pending(State = #state{pending = Pending0}) ->
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_client:task_event_stream(TaskId, Type, Stream) end);
send_event({close, TaskId, Reason}) ->
maybe_send(fun() -> efka_client:close_task_event_stream(TaskId, Reason) end).
-spec maybe_send(fun(() -> any())) -> ok | not_ready.
maybe_send(SendFun) ->
case catch efka_client:is_activated() of
true ->

3
src/efka_app.erl
View File

@ -9,6 +9,7 @@
-export([start/2, stop/1]).
-spec start(term(), term()) -> {ok, pid()} | {error, term()}.
start(_StartType, _StartArgs) ->
io:setopts([{encoding, unicode}]),
%% 加速内存的回收
@ -17,10 +18,12 @@ start(_StartType, _StartArgs) ->
efka_sup:start_link().
-spec stop(term()) -> ok.
stop(_State) ->
ok.
%% 微服务和efka之间通过websocket协议通讯
-spec start_http_server() -> ok.
start_http_server() ->
{ok, Props} = application:get_env(efka, http_server),
Acceptors = proplists:get_value(acceptors, Props, 50),

4
src/efka_logger.erl
View File

@ -33,6 +33,7 @@
write(Data) when is_binary(Data) ->
gen_server:cast(?SERVER, {write, Data}).
-spec write_lines(Lines :: [binary()]) -> ok.
write_lines(Lines) when is_list(Lines) ->
gen_server:cast(?SERVER, {write_lines, Lines}).
@ -137,11 +138,13 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
%%% Internal functions
%%%===================================================================
-spec format(binary() | [iodata()]) -> binary().
format(Data) when is_binary(Data) ->
iolist_to_binary(Data);
format(Items) when is_list(Items) ->
iolist_to_binary(lists:join(<<"\t">>, Items)).
-spec time_prefix() -> binary().
time_prefix() ->
{{Y, M, D}, {H, I, S}} = calendar:local_time(),
iolist_to_binary(io_lib:format("[~b-~2..0b-~2..0b ~2..0b:~2..0b:~2..0b]", [Y, M, D, H, I, S])).
@ -153,6 +156,7 @@ make_file(LogFile) when is_list(LogFile) ->
RootDir = code:root_dir() ++ "/log/",
lists:flatten(RootDir ++ LogFile ++ "." ++ Suffix).
-spec ensure_dir() -> ok | {error, term()}.
ensure_dir() ->
RootDir = code:root_dir() ++ "/log/",
case filelib:is_dir(RootDir) of

2
src/efka_subscription.erl
View File

@ -225,6 +225,7 @@ match_components(_, _, _) ->
of_components(Topic) when is_binary(Topic) ->
binary:split(Topic, <<$/>>, [global]).
-spec is_valid_components([binary()]) -> boolean().
is_valid_components([]) ->
true;
is_valid_components([<<$+>>|T]) ->
@ -244,6 +245,7 @@ order_num([<<$+>>|_]) ->
order_num([_|Tail]) ->
order_num(Tail).
-spec broadcast(binary(), binary(), [#subscriber{}]) -> ok.
broadcast(Topic, Content, MatchedSubscribers) ->
lists:foreach(fun(#subscriber{subscriber_pid = SubscriberPid}) ->
SubscriberPid ! {topic_broadcast, Topic, Content}

2
src/efka_sup.erl
View File

@ -13,6 +13,7 @@
-define(SERVER, ?MODULE).
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
supervisor:start_link({local, ?SERVER}, ?MODULE, []).
@ -25,6 +26,7 @@ start_link() ->
%% shutdown => shutdown(), % optional
%% type => worker(), % optional
%% modules => modules()} % optional
-spec init(list()) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
init([]) ->
SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
ChildSpecs = [

13
src/efka_util.erl
View File

@ -16,25 +16,30 @@
-export([json_data/1, json_error/2]).
-export([starts_with/2, file_md5/1]).
-spec get_file_md5(string()) -> string().
get_file_md5(FilePath) when is_list(FilePath) ->
{ok, FileData} = file:read_file(FilePath),
Md5Binary = crypto:hash(md5, FileData),
string:lowercase(lists:flatten([io_lib:format("~2.16.0b", [X]) || X <- binary_to_list(Md5Binary)])).
%% 时间，精确到毫秒
-spec timestamp_ms() -> integer().
timestamp_ms() ->
{Mega, Seconds, Micro} = os:timestamp(),
(Mega * 1000000 + Seconds) * 1000 + Micro div 1000.
-spec timestamp() -> integer().
timestamp() ->
{Mega, Seconds, _Micro} = os:timestamp(),
Mega * 1000000 + Seconds.
-spec number_format(integer() | float(), integer()) -> integer() | float().
number_format(Num, _Decimals) when is_integer(Num) ->
Num;
number_format(Float, Decimals) when is_float(Float) ->
list_to_float(float_to_list(Float, [{decimals, Decimals}, compact])).
-spec int_format(integer(), pos_integer()) -> integer().
int_format(Num, Len) when is_integer(Num), Len > 0 ->
S = integer_to_list(Num),
case length(S) > Len of
@ -50,6 +55,8 @@ chunks(List, Size) when is_list(List), is_integer(Size), Size > 0, length(List)
[List];
chunks(List, Size) when is_list(List), is_integer(Size), Size > 0 ->
chunks0(List, Size, Size, [], []).
-spec chunks0(list(), integer(), integer(), list(), [list()]) -> [list()].
chunks0([], _, _, [], AccTarget) ->
lists:reverse(AccTarget);
chunks0([], _, _, Target, AccTarget) ->
@ -59,11 +66,13 @@ chunks0(List, Size, 0, Target, AccTarget) ->
chunks0([Hd | Tail], Size, Num, Target, AccTarget) ->
chunks0(Tail, Size, Num - 1, [Hd | Target], AccTarget).
-spec json_data(term()) -> binary().
json_data(Data) ->
jiffy:encode(#{
<<"result">> => Data
}, [force_utf8]).
-spec json_error(integer(), binary()) -> binary().
json_error(ErrCode, ErrMessage) when is_integer(ErrCode), is_binary(ErrMessage) ->
jiffy:encode(#{
<<"error">> => #{
@ -72,6 +81,7 @@ json_error(ErrCode, ErrMessage) when is_integer(ErrCode), is_binary(ErrMessage)
}
}, [force_utf8]).
-spec uuid() -> string().
uuid() ->
rand_bytes(16).
@ -86,6 +96,7 @@ rand_bytes(Size) when is_integer(Size), Size > 0 ->
md5(Str) when is_binary(Str) ->
list_to_binary(lists:flatten([hex(X) || <<X:4>> <= erlang:md5(Str)])).
-spec hex(0..15) -> byte().
hex(N) when N < 10 ->
$0 + N;
hex(N) ->
@ -120,6 +131,7 @@ file_md5(FilePath) when is_list(FilePath) ->
file:close(F),
lists:flatten(io_lib:format("~32.16.0b", [binary:decode_unsigned(Digest)])).
-spec md5_loop(file:io_device(), crypto:hash_state()) -> binary().
md5_loop(F, Context) ->
%% 每次读取 1MB，可调整块大小
case file:read(F, 1024 * 1024) of
@ -128,4 +140,3 @@ md5_loop(F, Context) ->
{ok, Bin} ->
md5_loop(F, crypto:hash_update(Context, Bin))
end.

2
src/micro_service/efka_service_sup.erl
View File

@ -39,6 +39,7 @@ start_link() ->
%% this function is called by the new process to find out about
%% restart strategy, maximum restart frequency and child
%% specifications.
-spec init(list()) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
init([]) ->
SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
{ok, {SupFlags, []}}.
@ -64,6 +65,7 @@ stop_service(ServiceId) when is_binary(ServiceId) ->
supervisor:terminate_child(?MODULE, ChildId),
supervisor:delete_child(?MODULE, ChildId).
-spec child_spec(binary()) -> supervisor:child_spec().
child_spec(ServiceId) when is_binary(ServiceId) ->
Name = efka_service:get_name(ServiceId),
#{

2
src/models/cache_model.erl
View File

@ -33,9 +33,11 @@
insert(Data) when is_binary(Data) ->
gen_server:call(?SERVER, {insert, {generate_id(), Data}}).
-spec fetch_next() -> error | {ok, {integer(), binary()}}.
fetch_next() ->
gen_server:call(?SERVER, fetch_next).
-spec delete(integer()) -> ok.
delete(Id) when is_integer(Id) ->
gen_server:call(?SERVER, {delete, Id}).

2
src/models/service_model.erl
View File

@ -30,9 +30,11 @@
%%% API
%%%===================================================================
-spec insert(#service{}) -> ok.
insert(Service = #service{}) ->
gen_server:call(?SERVER, {insert, Service}).
-spec change_status(binary(), integer()) -> ok | {error, binary()}.
change_status(ServiceId, NewStatus) when is_binary(ServiceId), is_integer(NewStatus) ->
gen_server:call(?SERVER, {change_status, ServiceId, NewStatus}).

31
src/tests/docker_commands_tests.erl
View File

@ -34,6 +34,7 @@
-define(TEST_IMAGE, <<"docker.1ms.run/library/busybox:latest">>).
-spec test_all() -> ok.
test_all() ->
ok = test_pull(),
ok = test_check_image_exist(),
@ -53,18 +54,22 @@ test_all() ->
ok = test_remove_container_with_options(),
ok = test_get_containers().
-spec test_pull() -> ok.
test_pull() ->
ok = docker_commands:pull_image(?TEST_IMAGE, fun(Msg) -> logger:debug("msg is: ~p", [Msg]) end).
-spec test_check_image_exist() -> ok.
test_check_image_exist() ->
ok = test_pull(),
true = docker_commands:check_image_exist(?TEST_IMAGE),
ok.
-spec test_check_image_not_exist() -> ok.
test_check_image_not_exist() ->
false = docker_commands:check_image_exist(<<"docker.1ms.run/library/not-exists-for-efka-tests:latest">>),
ok.
-spec test_create_container() -> ok.
test_create_container() ->
Name = test_container_name(<<"create">>),
ContainerDir = prepare_container_dir(Name),
@ -77,6 +82,7 @@ test_create_container() ->
cleanup_container(Name)
end.
-spec test_create_container_without_create_options() -> ok.
test_create_container_without_create_options() ->
Name = test_container_name(<<"create-default">>),
ContainerDir = prepare_container_dir(Name),
@ -93,6 +99,7 @@ test_create_container_without_create_options() ->
cleanup_container(Name)
end.
-spec test_create_container_patches_options() -> ok.
test_create_container_patches_options() ->
Name = test_container_name(<<"create-patch">>),
ContainerDir = prepare_container_dir(Name),
@ -125,6 +132,7 @@ test_create_container_patches_options() ->
cleanup_container(Name)
end.
-spec test_check_container_exist() -> ok.
test_check_container_exist() ->
Name = test_container_name(<<"exist">>),
with_created_container(Name, fun(_ContainerDir, _ContainerId) ->
@ -132,10 +140,12 @@ test_check_container_exist() ->
ok
end).
-spec test_check_container_not_exist() -> ok.
test_check_container_not_exist() ->
false = docker_commands:check_container_exist(test_container_name(<<"missing">>)),
ok.
-spec test_is_container_running() -> ok.
test_is_container_running() ->
Name = test_container_name(<<"running">>),
with_created_container(Name, fun(_ContainerDir, ContainerId) ->
@ -147,6 +157,7 @@ test_is_container_running() ->
ok
end).
-spec test_start_container() -> ok.
test_start_container() ->
Name = test_container_name(<<"start">>),
with_created_container(Name, fun(_ContainerDir, ContainerId) ->
@ -155,6 +166,7 @@ test_start_container() ->
ok
end).
-spec test_stop_container() -> ok.
test_stop_container() ->
Name = test_container_name(<<"stop">>),
with_started_container(Name, fun(_ContainerDir, ContainerId) ->
@ -163,6 +175,7 @@ test_stop_container() ->
ok
end).
-spec test_stop_container_with_timeout() -> ok.
test_stop_container_with_timeout() ->
Name = test_container_name(<<"stop-timeout">>),
with_started_container(Name, fun(_ContainerDir, ContainerId) ->
@ -171,6 +184,7 @@ test_stop_container_with_timeout() ->
ok
end).
-spec test_kill_container() -> ok.
test_kill_container() ->
Name = test_container_name(<<"kill">>),
with_started_container(Name, fun(_ContainerDir, ContainerId) ->
@ -180,6 +194,7 @@ test_kill_container() ->
ok
end).
-spec test_kill_container_with_signal() -> ok.
test_kill_container_with_signal() ->
Name = test_container_name(<<"kill-signal">>),
with_started_container(Name, fun(_ContainerDir, ContainerId) ->
@ -189,6 +204,7 @@ test_kill_container_with_signal() ->
ok
end).
-spec test_remove_container() -> ok.
test_remove_container() ->
Name = test_container_name(<<"remove">>),
with_created_container(Name, fun(_ContainerDir, _ContainerId) ->
@ -197,6 +213,7 @@ test_remove_container() ->
ok
end, false).
-spec test_remove_container_with_options() -> ok.
test_remove_container_with_options() ->
Name = test_container_name(<<"remove-opts">>),
with_started_container(Name, fun(_ContainerDir, _ContainerId) ->
@ -205,6 +222,7 @@ test_remove_container_with_options() ->
ok
end, false).
-spec test_get_containers() -> ok.
test_get_containers() ->
Name = test_container_name(<<"list">>),
with_created_container(Name, fun(_ContainerDir, ContainerId) ->
@ -214,9 +232,11 @@ test_get_containers() ->
ok
end).
-spec with_created_container(binary(), fun((string(), binary()) -> ok)) -> ok.
with_created_container(Name, Fun) ->
with_created_container(Name, Fun, true).
-spec with_created_container(binary(), fun((string(), binary()) -> ok), boolean()) -> ok.
with_created_container(Name, Fun, Cleanup) when is_binary(Name), is_function(Fun, 2), is_boolean(Cleanup) ->
ContainerDir = prepare_container_dir(Name),
try
@ -232,15 +252,18 @@ with_created_container(Name, Fun, Cleanup) when is_binary(Name), is_function(Fun
end
end.
-spec with_started_container(binary(), fun((string(), binary()) -> ok)) -> ok.
with_started_container(Name, Fun) ->
with_started_container(Name, Fun, true).
-spec with_started_container(binary(), fun((string(), binary()) -> ok), boolean()) -> ok.
with_started_container(Name, Fun, Cleanup) when is_binary(Name), is_function(Fun, 2), is_boolean(Cleanup) ->
with_created_container(Name, fun(ContainerDir, ContainerId) ->
ok = docker_commands:start_container(Name),
ok = Fun(ContainerDir, ContainerId)
end, Cleanup).
-spec minimal_params(binary()) -> message_pb:'ContainerDeployParams'().
minimal_params(Name) when is_binary(Name) ->
#'ContainerDeployParams'{
container_name = Name,
@ -252,21 +275,25 @@ minimal_params(Name) when is_binary(Name) ->
}
}.
-spec prepare_container_dir(binary()) -> string().
prepare_container_dir(Name) when is_binary(Name) ->
Dir = lists:flatten(io_lib:format("/tmp/efka_docker_tests/~ts/", [Name])),
ok = filelib:ensure_dir(Dir ++ "placeholder"),
ok = file:write_file(Dir ++ "service.conf", <<>>, [write]),
Dir.
-spec cleanup_container(binary()) -> ok.
cleanup_container(Name) when is_binary(Name) ->
_ = docker_commands:remove_container(Name, true, false),
ok.
-spec inspect_container_json(binary()) -> map().
inspect_container_json(Name) when is_binary(Name) ->
Url = lists:flatten(io_lib:format("/containers/~s/json", [binary_to_list(Name)])),
{ok, 200, _Headers, Resp} = docker_http:request("GET", Url, <<>>, []),
jiffy:decode(Resp, [return_maps]).
-spec assert_patched_defaults(binary(), string(), map()) -> ok.
assert_patched_defaults(Name, ContainerDir, Inspect)
when is_binary(Name), is_list(ContainerDir), is_map(Inspect) ->
ConfigFile = list_to_binary(docker_helper:get_config_file(ContainerDir)),
@ -278,18 +305,22 @@ assert_patched_defaults(Name, ContainerDir, Inspect)
true = lists:member(ExpectedBind, Binds),
ok.
-spec contains_container(binary(), binary(), [map()]) -> boolean().
contains_container(Name, ContainerId, Containers) when is_binary(Name), is_binary(ContainerId), is_list(Containers) ->
lists:any(fun(Container) -> container_matches(Name, ContainerId, Container) end, Containers).
-spec container_matches(binary(), binary(), map()) -> boolean().
container_matches(Name, ContainerId, #{<<"Id">> := Id, <<"Names">> := Names}) when is_binary(Id), is_list(Names) ->
lists:member(<<"/", Name/binary>>, Names) orelse has_id_prefix(ContainerId, Id);
container_matches(_Name, _ContainerId, _Container) ->
false.
-spec has_id_prefix(binary(), binary()) -> boolean().
has_id_prefix(ExpectedId, ActualId) when is_binary(ExpectedId), is_binary(ActualId) ->
PrefixLen = erlang:min(byte_size(ExpectedId), byte_size(ActualId)),
binary:part(ExpectedId, 0, PrefixLen) =:= binary:part(ActualId, 0, PrefixLen).
-spec test_container_name(binary()) -> binary().
test_container_name(Prefix) when is_binary(Prefix) ->
Suffix = integer_to_binary(erlang:unique_integer([positive])),
<<"efka-test-", Prefix/binary, "-", Suffix/binary>>.

7
src/transport/efka_client.erl
View File

@ -60,9 +60,11 @@ close_task_event_stream(TaskId, Reason) when is_integer(TaskId), is_binary(Reaso
is_activated() ->
gen_statem:call(?SERVER, is_activated).
-spec ping(term(), term(), term(), term(), term(), term(), term(), term(), term(), term(), term(), term(), term()) -> ok.
ping(AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces) ->
gen_statem:cast(?SERVER, {ping, AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces}).
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
gen_statem:start_link({local, ?SERVER}, ?MODULE, [], []).
@ -70,14 +72,17 @@ start_link() ->
%%% gen_statem callbacks
%%%===================================================================
-spec init(list()) -> {ok, atom(), #state{}}.
init([]) ->
erlang:start_timer(0, self(), create_transport),
{ok, ?STATE_DISCONNECTED, #state{socket = undefined}}.
-spec callback_mode() -> handle_event_function.
callback_mode() ->
handle_event_function.
%% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
-spec handle_event(term(), term(), atom(), #state{}) -> term().
handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
CastFrame = message_pb:encode_msg(#'CastFrame'{
body = {data, #'Data'{route_key = RouteKey, metric = Metric}}
@ -237,10 +242,12 @@ handle_event(info, Info, _, State = #state{}) ->
logger:notice("[efka_client] get unknown info: ~p", [Info]),
{keep_state, State}.
-spec terminate(term(), atom(), #state{}) -> ok.
terminate(_Reason, _StateName, _State = #state{socket = Socket}) ->
disconnect(Socket),
ok.
-spec code_change(term(), atom(), #state{}, term()) -> {ok, atom(), #state{}}.
code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
{ok, StateName, State}.

9
src/transport/service_channel.erl
View File

@ -27,14 +27,18 @@
%% 逻辑处理方法
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec init(term(), term()) -> {cowboy_websocket, term(), term()}.
init(Req, Opts) ->
{cowboy_websocket, Req, Opts}.
-spec websocket_init(term()) -> {ok, #state{}}.
websocket_init(_State) ->
logger:debug("[service_channel] get a new connection"),
%% 初始状态为true
{ok, #state{}}.
-spec websocket_handle(term(), #state{}) ->
{reply, term(), #state{}} | {ok, #state{}}.
websocket_handle(ping, State) ->
{reply, pong, State};
@ -52,6 +56,8 @@ websocket_handle(Info, State) ->
{ok, 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'{
body = {topic_event, #'ServiceCast.TopicEvent'{topic = Topic, content = Content}}
@ -75,6 +81,7 @@ websocket_info(Info, State) ->
{ok, State}.
%% 进程关闭事件
-spec terminate(term(), term(), #state{}) -> ok.
terminate(Reason, _Req, State = #state{service_id = ServiceId, is_registered = IsRegistered}) ->
ok = efka_subscription:unsubscribe_all(self()),
case IsRegistered of
@ -91,6 +98,7 @@ terminate(Reason, _Req, State = #state{service_id = ServiceId, is_registered = I
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% 注册, 要建立程序和容器之间的关系
-spec handle_request(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
@ -129,6 +137,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{}}.
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),