From 5cb386c363cbf7d48131ae3681ae3b311b18b5d3 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Tue, 21 Apr 2026 16:24:59 +0800 Subject: [PATCH] fix spec --- src/docker/docker_commands.erl | 4 ++++ src/docker/docker_container_service.erl | 3 +++ src/docker/docker_deploy_manager.erl | 6 +++++ src/docker/docker_events.erl | 2 ++ src/docker/docker_helper.erl | 4 ++++ src/docker/docker_http.erl | 8 ++++++- src/docker/docker_task_reporter.erl | 10 ++++++++ src/efka_app.erl | 5 +++- src/efka_logger.erl | 6 ++++- src/efka_subscription.erl | 2 ++ src/efka_sup.erl | 2 ++ src/efka_util.erl | 13 ++++++++++- src/micro_service/efka_service_sup.erl | 4 +++- src/models/cache_model.erl | 4 +++- src/models/service_model.erl | 2 ++ src/tests/docker_commands_tests.erl | 31 +++++++++++++++++++++++++ src/transport/efka_client.erl | 9 ++++++- src/transport/service_channel.erl | 9 +++++++ 18 files changed, 117 insertions(+), 7 deletions(-) diff --git a/src/docker/docker_commands.erl b/src/docker/docker_commands.erl index 8a1ecaa..f84d8f1 100644 --- a/src/docker/docker_commands.erl +++ b/src/docker/docker_commands.erl @@ -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) -> diff --git a/src/docker/docker_container_service.erl b/src/docker/docker_container_service.erl index 7ed8cb9..1c908b2 100644 --- a/src/docker/docker_container_service.erl +++ b/src/docker/docker_container_service.erl @@ -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) -> diff --git a/src/docker/docker_deploy_manager.erl b/src/docker/docker_deploy_manager.erl index 84a0e00..2eadac7 100644 --- a/src/docker/docker_deploy_manager.erl +++ b/src/docker/docker_deploy_manager.erl @@ -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}. diff --git a/src/docker/docker_events.erl b/src/docker/docker_events.erl index f026a98..37a524d 100644 --- a/src/docker/docker_events.erl +++ b/src/docker/docker_events.erl @@ -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). diff --git a/src/docker/docker_helper.erl b/src/docker/docker_helper.erl index bfe10d7..6158101 100644 --- a/src/docker/docker_helper.erl +++ b/src/docker/docker_helper.erl @@ -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 $/ -> diff --git a/src/docker/docker_http.erl b/src/docker/docker_http.erl index 02babba..d51f3e7 100644 --- a/src/docker/docker_http.erl +++ b/src/docker/docker_http.erl @@ -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} -> @@ -82,4 +88,4 @@ receive_body(Callback, ConnPid, StreamRef) -> {gun_data, ConnPid, StreamRef, nofin, Data} -> Callback({message, Data}), receive_body(Callback, ConnPid, StreamRef) - end. \ No newline at end of file + end. diff --git a/src/docker/docker_task_reporter.erl b/src/docker/docker_task_reporter.erl index 5147fc9..6aecbe6 100644 --- a/src/docker/docker_task_reporter.erl +++ b/src/docker/docker_task_reporter.erl @@ -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 -> diff --git a/src/efka_app.erl b/src/efka_app.erl index 0e68ba2..035a008 100644 --- a/src/efka_app.erl +++ b/src/efka_app.erl @@ -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), @@ -40,4 +43,4 @@ start_http_server() -> ], {ok, Pid} = cowboy:start_clear(ws_listener, TransOpts, #{env => #{dispatch => Dispatcher}}), - logger:debug("[efka_app] websocket server start at: ~p, pid is: ~p", [Port, Pid]). \ No newline at end of file + logger:debug("[efka_app] websocket server start at: ~p, pid is: ~p", [Port, Pid]). diff --git a/src/efka_logger.erl b/src/efka_logger.erl index db3b69a..caf483a 100644 --- a/src/efka_logger.erl +++ b/src/efka_logger.erl @@ -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 @@ -172,4 +176,4 @@ get_date() -> -spec maybe_new_file(Date :: calendar:date()) -> boolean(). maybe_new_file({Y, M, D}) -> {{Y0, M0, D0}, _} = calendar:local_time(), - not (Y =:= Y0 andalso M =:= M0 andalso D =:= D0). \ No newline at end of file + not (Y =:= Y0 andalso M =:= M0 andalso D =:= D0). diff --git a/src/efka_subscription.erl b/src/efka_subscription.erl index b21a492..49a303b 100644 --- a/src/efka_subscription.erl +++ b/src/efka_subscription.erl @@ -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} diff --git a/src/efka_sup.erl b/src/efka_sup.erl index 18f1e77..a067abb 100644 --- a/src/efka_sup.erl +++ b/src/efka_sup.erl @@ -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 = [ diff --git a/src/efka_util.erl b/src/efka_util.erl index 1de8e78..769b54e 100644 --- a/src/efka_util.erl +++ b/src/efka_util.erl @@ -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) || <> <= 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. - diff --git a/src/micro_service/efka_service_sup.erl b/src/micro_service/efka_service_sup.erl index 495b899..bbf9f62 100644 --- a/src/micro_service/efka_service_sup.erl +++ b/src/micro_service/efka_service_sup.erl @@ -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), #{ @@ -73,4 +75,4 @@ child_spec(ServiceId) when is_binary(ServiceId) -> shutdown => 5000, type => worker, modules => ['efka_service'] - }. \ No newline at end of file + }. diff --git a/src/models/cache_model.erl b/src/models/cache_model.erl index 2644668..3191679 100644 --- a/src/models/cache_model.erl +++ b/src/models/cache_model.erl @@ -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}). @@ -139,4 +141,4 @@ code_change(_OldVsn, State = #state{}, _Extra) -> -spec generate_id() -> integer(). generate_id() -> - os:system_time(microsecond). \ No newline at end of file + os:system_time(microsecond). diff --git a/src/models/service_model.erl b/src/models/service_model.erl index c342ec0..3c17347 100644 --- a/src/models/service_model.erl +++ b/src/models/service_model.erl @@ -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}). diff --git a/src/tests/docker_commands_tests.erl b/src/tests/docker_commands_tests.erl index de92a61..77eabfb 100644 --- a/src/tests/docker_commands_tests.erl +++ b/src/tests/docker_commands_tests.erl @@ -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>>. diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index 1bdbfb2..d9bd5c7 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -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}. @@ -308,4 +315,4 @@ send_error_reply(Socket, PacketId, Reason) when is_binary(Reason) -> packet_id = PacketId, reply = {error, #'ReplyError'{code = -1, message = Reason}} }), - send_packet(Socket, Packet). \ No newline at end of file + send_packet(Socket, Packet). diff --git a/src/transport/service_channel.erl b/src/transport/service_channel.erl index efd545f..77a1041 100644 --- a/src/transport/service_channel.erl +++ b/src/transport/service_channel.erl @@ -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),