diff --git a/src/docker/docker_commands.erl b/src/docker/docker_commands.erl index bef2bdb..75d10a1 100644 --- a/src/docker/docker_commands.erl +++ b/src/docker/docker_commands.erl @@ -36,15 +36,14 @@ check_image_exist(Image) when is_binary(Image) -> -spec create_container(ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) -> {ok, ContainerId :: binary()} | {error, Reason :: any()}. -create_container(ContainerDir, #'ContainerDeployParams'{container_name = ContainerName, create = CreateOpts }) - when is_binary(ContainerName), is_list(ContainerDir) -> - +create_container(ContainerDir, #'ContainerDeployParams'{container_name = ContainerName0, create = CreateOpts }) + when is_list(ContainerDir), ContainerName0 =/= undefined -> + ContainerName = to_binary(ContainerName0), Url = lists:flatten(io_lib:format("/containers/create?name=~s", [binary_to_list(ContainerName)])), Options = docker_container_builder:build_options(ContainerName, ContainerDir, CreateOpts), display_options(Options), Body = iolist_to_binary(json:encode(Options)), - true = is_binary(Body), Headers = [ {<<"Content-Type">>, <<"application/json">>} @@ -67,7 +66,9 @@ create_container(ContainerDir, #'ContainerDeployParams'{container_name = Contain end; {error, Reason} -> {error, Reason} - end. + end; +create_container(_ContainerDir, _Params) -> + {error, <<"invalid container params">>}. -spec is_container_running(ContainerId :: binary()) -> boolean(). is_container_running(ContainerId) when is_binary(ContainerId) -> @@ -261,10 +262,11 @@ json_state_running(Json) when is_binary(Json) -> error end. --spec display_options(Options :: map()) -> no_return(). +-spec display_options(Options :: map()) -> ok. display_options(Options) when is_map(Options) -> logger:debug("deploy options: ~p", [iolist_to_binary(json:encode(Options))]), - lists:foreach(fun({K, V}) -> logger:debug("~p => ~p", [K, V]) end, maps:to_list(Options)). + lists:foreach(fun({K, V}) -> logger:debug("~p => ~p", [K, V]) end, maps:to_list(Options)), + ok. -spec build_stop_container_url(ContainerName :: binary(), TimeoutSeconds :: non_neg_integer()) -> string(). build_stop_container_url(ContainerName, 0) -> @@ -293,3 +295,9 @@ boolean_to_query_value(true) -> "true"; boolean_to_query_value(false) -> "false". + +-spec to_binary(unicode:chardata()) -> binary(). +to_binary(Value) when is_binary(Value) -> + Value; +to_binary(Value) when is_list(Value) -> + unicode:characters_to_binary(Value). diff --git a/src/docker/docker_container_builder.erl b/src/docker/docker_container_builder.erl index 7442fd5..f836c59 100644 --- a/src/docker/docker_container_builder.erl +++ b/src/docker/docker_container_builder.erl @@ -364,12 +364,10 @@ to_binary(Value) when is_atom(Value) -> to_binary(Value) -> iolist_to_binary(io_lib:format("~p", [Value])). --spec to_bool(Value :: any()) -> boolean(). +-spec to_bool(Value :: true | false | 0 | 1 | undefined) -> boolean(). to_bool(true) -> true; -to_bool(<<"true">>) -> - true; -to_bool("true") -> +to_bool(1) -> true; to_bool(_) -> false. diff --git a/src/docker/docker_deployer.erl b/src/docker/docker_deployer.erl index 7d53633..0d86291 100644 --- a/src/docker/docker_deployer.erl +++ b/src/docker/docker_deployer.erl @@ -42,11 +42,10 @@ start_monitor(TaskId, ContainerDir, Params) % "command": ["nginx", "-g", "daemon off;"], % "restart": "always" %} --spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) -> no_return(). -deploy(TaskId, ContainerDir, Params = #'ContainerDeployParams'{ - container_name = ContainerName, - create = #'DockerCreateOptions'{config = #'DockerContainerConfig'{image = Image0}} -}) when is_integer(TaskId), is_list(ContainerDir) -> +-spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) -> ok. +deploy(TaskId, ContainerDir, Params) when is_integer(TaskId), is_list(ContainerDir), is_record(Params, 'ContainerDeployParams') -> + ContainerName = deploy_container_name(Params), + Image0 = deploy_image(Params), report_task_event(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>), try ok = ensure_container_absent(TaskId, ContainerName), @@ -100,8 +99,8 @@ write_task_summary(TaskId, Status, ContainerName, Image, ContainerId) Fields = case ContainerId of undefined -> Fields0; - _ when is_binary(ContainerId) -> - Fields0 ++ [<<" container_id=">>, short_container_id(ContainerId)] + ContainerId0 when is_binary(ContainerId0) -> + Fields0 ++ [<<" container_id=">>, short_container_id(ContainerId0)] end, efka_logger:write(iolist_to_binary(Fields)). @@ -128,12 +127,7 @@ ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) -> case docker_commands:pull_image(Image) of {ok, Ref, Pid, MRef} -> await_pull_image(TaskId, Ref, Pid, MRef), - {ok, Image}; - {error, Reason} when is_binary(Reason) -> - throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>}); - {error, Reason} -> - Error = iolist_to_binary(io_lib:format("镜像拉取失败: ~p", [Reason])), - throw({deploy_error, Error}) + {ok, Image} end. -spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok. @@ -204,3 +198,22 @@ short_container_id(ContainerId) when is_binary(ContainerId), byte_size(Container binary:part(ContainerId, 0, 12); short_container_id(ContainerId) when is_binary(ContainerId) -> ContainerId. + +-spec deploy_container_name(message_pb:'ContainerDeployParams'()) -> binary(). +deploy_container_name(#'ContainerDeployParams'{container_name = ContainerName}) when ContainerName =/= undefined -> + to_binary(ContainerName); +deploy_container_name(_) -> + throw({deploy_error, <<"invalid deploy params: container_name missing">>}). + +-spec deploy_image(message_pb:'ContainerDeployParams'()) -> binary(). +deploy_image(#'ContainerDeployParams'{create = #'DockerCreateOptions'{config = #'DockerContainerConfig'{image = Image}}}) + when Image =/= undefined -> + to_binary(Image); +deploy_image(_) -> + throw({deploy_error, <<"invalid deploy params: image missing">>}). + +-spec to_binary(unicode:chardata()) -> binary(). +to_binary(Value) when is_binary(Value) -> + Value; +to_binary(Value) when is_list(Value) -> + unicode:characters_to_binary(Value). diff --git a/src/docker/docker_events.erl b/src/docker/docker_events.erl index 5c0fad1..62c1768 100644 --- a/src/docker/docker_events.erl +++ b/src/docker/docker_events.erl @@ -30,7 +30,7 @@ %%% API %%%=================================================================== --spec monitor_container(ReceiverPid :: pid(), ContainerId :: binary()) -> no_return(). +-spec monitor_container(ReceiverPid :: pid(), ContainerId :: binary()) -> ok. monitor_container(ReceiverPid, ContainerId) when is_pid(ReceiverPid), is_binary(ContainerId) -> gen_server:cast(?SERVER, {monitor_container, ReceiverPid, ContainerId}). diff --git a/src/efka_logger.erl b/src/efka_logger.erl index caf483a..74a8f4a 100644 --- a/src/efka_logger.erl +++ b/src/efka_logger.erl @@ -29,7 +29,7 @@ %%% API %%%=================================================================== --spec write(Data :: binary()) -> no_return(). +-spec write(Data :: binary()) -> ok. write(Data) when is_binary(Data) -> gen_server:cast(?SERVER, {write, Data}). diff --git a/src/efka_subscription.erl b/src/efka_subscription.erl index 49a303b..12b0d3f 100644 --- a/src/efka_subscription.erl +++ b/src/efka_subscription.erl @@ -47,7 +47,7 @@ subscribe(Topic, SubscriberPid) when is_binary(Topic), is_pid(SubscriberPid) -> gen_server:call(?SERVER, {subscribe, Topic, SubscriberPid}). --spec publish(Topic :: binary(), Qos :: integer(), Content :: binary()) -> no_return(). +-spec publish(Topic :: binary(), Qos :: integer(), Content :: binary()) -> ok. publish(Topic, Qos, Content) when is_binary(Topic), is_integer(Qos), is_binary(Content) -> gen_server:cast(?SERVER, {publish, Topic, Qos, Content}). diff --git a/src/micro_service/efka_service.erl b/src/micro_service/efka_service.erl index 0080063..bd3f87f 100644 --- a/src/micro_service/efka_service.erl +++ b/src/micro_service/efka_service.erl @@ -40,11 +40,11 @@ get_name(ServiceId) when is_binary(ServiceId) -> get_pid(ServiceId) when is_binary(ServiceId) -> whereis(get_name(ServiceId)). --spec metric_data(Pid :: pid(), RouteKey :: binary(), Metric :: binary()) -> no_return(). +-spec metric_data(Pid :: pid(), RouteKey :: binary(), Metric :: binary()) -> ok. metric_data(Pid, RouteKey, Metric) when is_pid(Pid), is_binary(RouteKey), is_binary(Metric) -> gen_server:cast(Pid, {metric_data, RouteKey, Metric}). --spec send_event(Pid :: pid(), EventType :: integer(), Params :: binary()) -> no_return(). +-spec send_event(Pid :: pid(), EventType :: integer(), Params :: binary()) -> ok. send_event(Pid, EventType, Params) when is_pid(Pid), is_integer(EventType), is_binary(Params) -> gen_server:cast(Pid, {send_event, EventType, Params}). diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index 303729a..b49b365 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -44,15 +44,15 @@ %%%=================================================================== %% 发送数据 --spec metric_data(RouteKey :: binary(), Metric :: binary()) -> no_return(). +-spec metric_data(RouteKey :: binary(), Metric :: binary()) -> ok. metric_data(RouteKey, Metric) when is_binary(RouteKey), is_binary(Metric) -> gen_statem:cast(?SERVER, {metric_data, RouteKey, Metric}). --spec task_event_stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> no_return(). +-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 close_task_event_stream(TaskId :: integer(), Reason :: binary()) -> no_return(). +-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}). @@ -128,8 +128,8 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = send_packet(Socket, [?FRAME_REQUEST, AuthPacket]), {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_packet_id = PacketId, next_packet_id = PacketId + 1}, [{state_timeout, 5000, auth_timeout}]}; - {error, Reason} -> - %logger:debug("[efka_client] connect failed, error: ~p", [Reason]), + {error, _Reason} -> + %logger:debug("[efka_client] connect failed"), schedule_reconnect(), {keep_state, State#state{socket = undefined}} end;