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

fix docker command

Browse Source
This commit is contained in:
anlicheng 2025-09-16 11:40:25 +08:00
parent 0165e52ccb
commit 29ff05684a
6 changed files with 560 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
apps/efka/src/docker/efka_docker_container.erl → apps/efka/src/docker/efka_container.erl
View File

@ -8,16 +8,16 @@
%%% @end
%%% Created : 18. 4月 2025 16:50
%%%-------------------------------------------------------------------
-module(efka_docker_container).
-module(efka_container).
-author("anlicheng").
-include("efka_tables.hrl").
-behaviour(gen_server).
%% API
-export([start_link/2]).
-export([get_name/1, get_pid/1, attach_channel/2]).
-export([push_config/3, request_config/1, invoke/3]).
-export([start_link/3]).
-export([get_name/1, get_pid/1, attach_channel/3]).
-export([push_envs/3, invoke/3]).
-export([metric_data/4, send_event/3]).
%% gen_server callbacks
@ -28,6 +28,8 @@
container_id :: binary(),
%% 通道id信息
channel_pid :: pid() | undefined,
%% 数据上传的时候，用来管理容器和微服务
meta :: binary(),
%% 当前进程的port信息, OSPid = erlang:port_info(Port, os_pid)
port :: undefined | port(),
%% 系统对应的pid
@ -44,26 +46,22 @@
%%% API
%%%===================================================================
-spec get_name(ServiceId :: binary()) -> atom().
get_name(ServiceId) when is_binary(ServiceId) ->
list_to_atom("efka_service:" ++ binary_to_list(ServiceId)).
-spec get_name(ContainerId :: binary()) -> atom().
get_name(ContainerId) when is_binary(ContainerId) ->
list_to_atom("efka_container:" ++ binary_to_list(ContainerId)).
-spec get_pid(ServiceId :: binary()) -> undefined | pid().
get_pid(ServiceId) when is_binary(ServiceId) ->
whereis(get_name(ServiceId)).
-spec get_pid(ContainerId :: binary()) -> undefined | pid().
get_pid(ContainerId) when is_binary(ContainerId) ->
whereis(get_name(ContainerId)).
-spec push_config(Pid :: pid(), Ref :: reference(), ConfigJson :: binary()) -> no_return().
push_config(Pid, Ref, ConfigJson) when is_pid(Pid), is_binary(ConfigJson) ->
gen_server:cast(Pid, {push_config, Ref, self(), ConfigJson}).
-spec push_envs(Pid :: pid(), Ref :: reference(), Envs :: binary()) -> no_return().
push_envs(Pid, Ref, Envs) when is_pid(Pid), is_binary(Envs) ->
gen_server:cast(Pid, {push_envs, Ref, self(), Envs}).
-spec invoke(Pid :: pid(), Ref :: reference(), Payload :: binary()) -> no_return().
invoke(Pid, Ref, Payload) when is_pid(Pid), is_reference(Ref), is_binary(Payload) ->
gen_server:cast(Pid, {invoke, Ref, self(), Payload}).
-spec request_config(Pid :: pid()) -> {ok, Config :: binary()}.
request_config(Pid) when is_pid(Pid) ->
gen_server:call(Pid, request_config).
-spec metric_data(Pid :: pid(), DeviceUUID :: binary(), RouteKey :: binary(), Metric :: binary()) -> no_return().
metric_data(Pid, DeviceUUID, RouteKey, Metric) when is_pid(Pid), is_binary(DeviceUUID), is_binary(RouteKey), is_binary(Metric) ->
gen_server:cast(Pid, {metric_data, DeviceUUID, RouteKey, Metric}).
@ -72,15 +70,15 @@ metric_data(Pid, DeviceUUID, RouteKey, Metric) when is_pid(Pid), is_binary(Devic
send_event(Pid, EventType, Params) when is_pid(Pid), is_integer(EventType), is_binary(Params) ->
gen_server:cast(Pid, {send_event, EventType, Params}).
-spec attach_channel(pid(), pid()) -> ok | {error, Reason :: binary()}.
attach_channel(Pid, ChannelPid) when is_pid(Pid), is_pid(ChannelPid) ->
gen_server:call(Pid, {attach_channel, ChannelPid}).
-spec attach_channel(Pid :: pid(), ChannelPid :: pid(), Meta :: binary()) -> ok | {error, Reason :: binary()}.
attach_channel(Pid, ChannelPid, Meta) when is_pid(Pid), is_pid(ChannelPid), is_binary(Meta) ->
gen_server:call(Pid, {attach_channel, ChannelPid, Meta}).
%% @doc Spawns the server and registers the local name (unique)
-spec(start_link(Name :: atom(), ContainerId :: binary()) ->
-spec(start_link(Name :: atom(), ContainerId :: binary(), Args :: [binary()]) ->
{ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link(Name, ContainerId) when is_atom(Name), is_binary(ContainerId) ->
gen_server:start_link({local, Name}, ?MODULE, [ContainerId], []).
start_link(Name, ContainerId, Args) when is_atom(Name), is_binary(ContainerId), is_list(Args) ->
gen_server:start_link({local, Name}, ?MODULE, [ContainerId, Args], []).
%%%===================================================================
%%% gen_server callbacks
@ -91,11 +89,11 @@ start_link(Name, ContainerId) when is_atom(Name), is_binary(ContainerId) ->
-spec(init(Args :: term()) ->
{ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
{stop, Reason :: term()} | ignore).
init([ContainerId]) ->
init([ContainerId, Args]) ->
%% supervisor进程通过exit(ChildPid, shutdown)调用的时候，确保terminate函数被调用
erlang:process_flag(trap_exit, true),
case startup(<<>>, []) of
case startup(<<>>, Args) of
{ok, Port} ->
{os_pid, OSPid} = erlang:port_info(Port, os_pid),
lager:debug("[efka_service] service: ~p, port: ~p, boot_service success os_pid: ~p", [ContainerId, Port, OSPid]),
@ -116,12 +114,12 @@ init([ContainerId]) ->
{stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
{stop, Reason :: term(), NewState :: #state{}}).
%% 绑定channel
handle_call({attach_channel, ChannelPid}, _From, State = #state{channel_pid = OldChannelPid, container_id = ContainerId}) ->
handle_call({attach_channel, ChannelPid, Meta}, _From, State = #state{channel_pid = OldChannelPid, container_id = ContainerId}) ->
case is_pid(OldChannelPid) andalso is_process_alive(OldChannelPid) of
false ->
erlang:monitor(process, ChannelPid),
lager:debug("[efka_service] service_id: ~p, channel attched", [ContainerId]),
{reply, ok, State#state{channel_pid = ChannelPid}};
{reply, ok, State#state{channel_pid = ChannelPid, meta = Meta}};
true ->
{reply, {error, <<"channel exists">>}, State}
end;
@ -135,23 +133,25 @@ handle_call(_Request, _From, State = #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_cast({metric_data, DeviceUUID, RouteKey, Metric}, State = #state{service_id = ServiceId}) ->
lager:debug("[efka_service] metric_data service_id: ~p, device_uuid: ~p, route_key: ~p, metric data: ~p", [ServiceId, DeviceUUID, RouteKey, Metric]),
efka_remote_agent:metric_data(ServiceId, DeviceUUID, RouteKey, Metric),
handle_cast({metric_data, DeviceUUID, RouteKey, Metric}, State = #state{container_id = ContainerId, meta = Meta}) ->
lager:debug("[efka_service] container_id: ~p, meta: ~p, device_uuid: ~p, route_key: ~p, metric data: ~p",
[ContainerId, Meta, DeviceUUID, RouteKey, Metric]),
%% 这里的数据需要转换成和meta相关的数据, todo container_id的数据是否需要加上
efka_remote_agent:metric_data(ContainerId, Meta, DeviceUUID, RouteKey, Metric),
{noreply, State};
handle_cast({send_event, EventType, Params}, State = #state{service_id = ServiceId}) ->
efka_remote_agent:event(ServiceId, EventType, Params),
lager:debug("[efka_service] send_event, service_id: ~p, event_type: ~p, params: ~p", [ServiceId, EventType, Params]),
handle_cast({send_event, EventType, Params}, State = #state{container_id = ContainerId, meta = Meta}) ->
efka_remote_agent:event(ContainerId, Meta, EventType, Params),
lager:debug("[efka_service] send_event, container_id: ~p, meta: ~p, event_type: ~p, params: ~p", [ContainerId, Meta, EventType, Params]),
{noreply, State};
%% 推送配置项目
handle_cast({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{channel_pid = ChannelPid, service_id = ServiceId, inflight = Inflight, callbacks = Callbacks}) ->
handle_cast({push_envs, Ref, ReceiverPid, Envs}, State = #state{channel_pid = ChannelPid, container_id = ContainerId, inflight = Inflight, callbacks = Callbacks}) ->
case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of
true ->
gen_channel:push_config(ChannelPid, Ref, self(), ConfigJson),
gen_channel:push_config(ChannelPid, Ref, self(), Envs),
%% 设置成功，需要更新微服务的配置
CB = fun() -> service_model:set_config(ServiceId, ConfigJson) end,
CB = fun() -> service_model:set_config(ContainerId, Envs) end,
{noreply, State#state{inflight = maps:put(Ref, ReceiverPid, Inflight), callbacks = maps:put(Ref, CB, Callbacks)}};
false ->
ReceiverPid ! {service_reply, Ref, {error, <<"channel is not alive">>}},

34
apps/efka/src/docker/efka_docker_container_sup.erl → apps/efka/src/docker/efka_container_sup.erl
View File

@ -6,7 +6,7 @@
%%% @end
%%% Created : 18. 4月 2025 16:42
%%%-------------------------------------------------------------------
-module(efka_docker_container_sup).
-module(efka_container_sup).
-author("anlicheng").
-include("efka_tables.hrl").
@ -14,7 +14,7 @@
%% API
-export([start_link/0]).
-export([start_service/1, stop_service/1]).
-export([start_container/1, stop_container/1]).
%% Supervisor callbacks
-export([init/1]).
@ -42,11 +42,11 @@ start_link() ->
init([]) ->
SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
%% 简化逻辑，只启动需要运行的微服务
{ok, Services} = service_model:get_running_services(),
ServiceIds = lists:map(fun(#service{service_id = ServiceId}) -> ServiceId end, Services),
lager:debug("[efka_service_sup] will start services: ~p", [ServiceIds]),
Specs = lists:map(fun(ServiceId) -> child_spec(ServiceId) end, Services),
%{ok, Services} = service_model:get_running_services(),
%ServiceIds = lists:map(fun(#service{service_id = ServiceId}) -> ServiceId end, Services),
%lager:debug("[efka_service_sup] will start services: ~p", [ServiceIds]),
%Specs = lists:map(fun(ServiceId) -> child_spec(ServiceId) end, Services),
Specs = [],
{ok, {SupFlags, Specs}}.
@ -54,9 +54,9 @@ init([]) ->
%%% Internal functions
%%%===================================================================
-spec start_service(ServiceId :: binary()) -> {ok, Pid :: pid()} | {error, Reason :: any()}.
start_service(ServiceId) when is_binary(ServiceId) ->
case supervisor:start_child(?MODULE, child_spec(ServiceId)) of
-spec start_container(ContainerId :: binary()) -> {ok, Pid :: pid()} | {error, Reason :: any()}.
start_container(ContainerId) when is_binary(ContainerId) ->
case supervisor:start_child(?MODULE, child_spec(ContainerId)) of
{ok, Pid} when is_pid(Pid) ->
{ok, Pid};
{error, {'already_started', Pid}} when is_pid(Pid) ->
@ -65,21 +65,21 @@ start_service(ServiceId) when is_binary(ServiceId) ->
{error, Error}
end.
-spec stop_service(ServiceId :: binary()) -> ok.
stop_service(ServiceId) when is_binary(ServiceId) ->
ChildId = efka_service:get_name(ServiceId),
-spec stop_container(ContainerId :: binary()) -> ok.
stop_container(ContainerId) when is_binary(ContainerId) ->
ChildId = efka_container:get_name(ContainerId),
supervisor:terminate_child(?MODULE, ChildId),
supervisor:delete_child(?MODULE, ChildId).
child_spec(#service{service_id = ServiceId}) when is_binary(ServiceId) ->
child_spec(ServiceId);
child_spec(ServiceId) when is_binary(ServiceId) ->
Name = efka_service:get_name(ServiceId),
child_spec(ContainerId) when is_binary(ContainerId) ->
Name = efka_service:get_name(ContainerId),
#{
id => Name,
start => {efka_service, start_link, [Name, ServiceId]},
start => {efka_container, start_link, [Name, ContainerId]},
restart => permanent,
shutdown => 5000,
type => worker,
modules => ['efka_service']
modules => ['efka_container']
}.

325
apps/efka/src/docker/efka_docker_command.erl Normal file
View File

@ -0,0 +1,325 @@
%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2025, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 15. 9月 2025 16:11
%%%-------------------------------------------------------------------
-module(efka_docker_command).
-author("anlicheng").
%% API
-export([pull_image/1, check_image_exist/1]).
-export([test/0]).
test() ->
M = #{ <<"image">> => <<"nginx:latest">>,
<<"container_name">> => <<"my_nginx">>,
<<"command">> => [<<"nginx">>,<<"-g">>,<<"daemon off;">>],
<<"entrypoint">> => [<<"/docker-entrypoint.sh">>],
<<"environment">> => [<<"ENV1=val1">>, <<"ENV2=val2">>],
<<"env_file">> => [<<"./env.list">>],
<<"ports">> => [<<"8080:80">>, <<"443:443">>],
<<"expose">> => [<<"80">>, <<"443">>],
<<"volumes">> => [<<"/host/data:/data">>, <<"/host/log:/var/log">>],
<<"networks">> => [<<"mynet">>],
<<"labels">> => #{ <<"role">> => <<"web">>, <<"env">> => <<"prod">> },
<<"restart">> => <<"always">>,
<<"user">> => <<"www-data">>,
<<"working_dir">> => <<"/app">>,
<<"hostname">> => <<"myhost">>,
<<"privileged">> => true,
<<"cap_add">> => [<<"NET_ADMIN">>],
<<"cap_drop">> => [<<"MKNOD">>],
<<"devices">> => [<<"/dev/snd:/dev/snd">>],
<<"mem_limit">> => <<"512m">>,
<<"mem_reservation">> => <<"256m">>,
<<"cpu_shares">> => 512,
<<"cpus">> => 1.5,
<<"ulimits">> => #{ <<"nofile">> => <<"1024:2048">> },
<<"sysctls">> => #{ <<"net.ipv4.ip_forward">> => <<"1">> },
<<"tmpfs">> => [<<"/tmp">>],
<<"extra_hosts">> => [<<"host1:192.168.0.1">>],
<<"healthcheck">> => #{
%<<"test">> => [<<"CMD-SHELL">>, <<"curl -f http://localhost || exit 1">>],
<<"test">> => [<<"CMD">>, <<"ls">>, <<"-l">>],
<<"interval">> => <<"30s">>,
<<"timeout">> => <<"10s">>,
<<"retries">> => 3
}
},
L = json_to_create_args(M),
io:format("~p~n", [L]).
-spec pull_image(Image :: binary()) -> ok | {error, Reason :: any()}.
pull_image(Image) when is_binary(Image) ->
%% todo 重定向错误流 {stderr_to_stdout, true}
PortSettings = [stream, exit_status, use_stdio, binary],
ExecCmd = "docker pull " ++ binary_to_list(Image),
lager:debug("cmd : ~p", [ExecCmd]),
case catch erlang:open_port({spawn, ExecCmd}, PortSettings) of
Port when is_port(Port) ->
case gather_output(Port) of
{0, Output} ->
lager:debug("docker pull output: ~p", [Output]),
ok;
{ExitCode, Error} ->
lager:debug("call me here: ~p", [Error]),
{error, {ExitCode, Error}}
end;
Error ->
lager:debug("error: ~p", [Error]),
{error, <<"exec command startup failed">>}
end.
-spec check_image_exist(Image :: binary()) -> boolean().
check_image_exist(Image) when is_binary(Image) ->
PortSettings = [stream, exit_status, use_stdio, binary],
ExecCmd = "docker images -q " ++ binary_to_list(Image),
lager:debug("cmd : ~p", [ExecCmd]),
case catch erlang:open_port({spawn, ExecCmd}, PortSettings) of
Port when is_port(Port) ->
case gather_output(Port) of
{0, <<>>} ->
false;
{0, <<_:12/binary, $\n>>} ->
true;
{_ExitCode, _Error} ->
false
end;
_Error ->
false
end.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
gather_output(Port) ->
gather_output(Port, <<>>).
gather_output(Port, Acc) ->
receive
{Port, {data, Data}} ->
gather_output(Port, [Acc, Data]);
{Port, {exit_status, Status}} ->
{Status, iolist_to_binary(Acc)}
end.
extract_sha256(Output) when is_binary(Output) ->
Parts = binary:split(Output, <<$\n>>, [global]),
lager:debug("parts: ~p", [Parts]),
case lists:search(fun(Line) -> starts_with(Line, <<"Digest:">>) end, Parts) of
{value, Digest} ->
Sha256 = lists:last(binary:split(Digest, <<":">>, [global])),
{ok, Sha256};
false ->
error
end.
starts_with(Binary, Prefix) when is_binary(Binary), is_binary(Prefix) ->
PrefixSize = byte_size(Prefix),
case Binary of
<<Prefix:PrefixSize/binary, _Rest/binary>> -> true;
_ -> false
end.
%% json_to_create_args/1
%% 输入: map() JSON 配置
%% 输出: [docker create 参数列表]，可以直接传给 open_port
%% 主函数
json_to_create_args(Config) when is_map(Config) ->
Image = maps:get(<<"image">>, Config),
Cmd = maps:get(<<"command">>, Config, []),
Options = build_options(Config),
Args = lists:flatten([Image | Options ++ Cmd]),
%lager:debug("args: ~p", [Args]),
iolist_to_binary(lists:join(<<" ">>, Args)).
%% 构建所有参数
build_options(Config) ->
lists:flatten([
build_name(Config),
build_entrypoint(Config),
build_ports(Config),
build_expose(Config),
build_volumes(Config),
build_env(Config),
build_env_file(Config),
build_networks(Config),
build_labels(Config),
build_restart(Config),
build_user(Config),
build_working_dir(Config),
build_hostname(Config),
build_privileged(Config),
build_cap_add_drop(Config),
build_devices(Config),
build_memory(Config),
build_cpu(Config),
build_ulimits(Config),
build_sysctls(Config),
build_tmpfs(Config),
build_extra_hosts(Config),
build_healthcheck(Config)
]).
%% Helper 函数示例
build_name(Config) ->
case maps:get(<<"container_name">>, Config, undefined) of
undefined -> [];
Name -> [<<"--name">>, Name]
end.
build_entrypoint(Config) ->
case maps:get(<<"entrypoint">>, Config, []) of
[] -> [];
EP -> [<<"--entrypoint">> | EP]
end.
build_ports(Config) ->
Ports = maps:get(<<"ports">>, Config, []),
lists:map(fun(P) -> [<<"-p">>, P] end, Ports).
build_expose(Config) ->
Ports = maps:get(<<"expose">>, Config, []),
lists:map(fun(P) -> [<<"--expose">>, P] end, Ports).
build_volumes(Config) ->
Vols = maps:get(<<"volumes">>, Config, []),
lists:map(fun(V) -> [<<"-v">>, V] end, Vols).
build_env(Config) ->
Envs = maps:get(<<"environment">>, Config, []),
lists:map(fun(E) -> [<<"-e">>, E] end, Envs).
build_env_file(Config) ->
Files = maps:get(<<"env_file">>, Config, []),
lists:map(fun(F) -> [<<"--env-file">>, F] end, Files).
build_networks(Config) ->
Nets = maps:get(<<"networks">>, Config, []),
lists:map(fun(Net) -> [<<"--network">>, Net] end, Nets).
build_labels(Config) ->
case maps:get(<<"labels">>, Config, #{}) of
#{} ->
[];
Labels ->
lists:map(fun({K, V}) -> [<<"--label">>, <<K/binary, "=", V/binary>>] end, maps:to_list(Labels))
end.
build_restart(Config) ->
case maps:get(<<"restart">>, Config, undefined) of
undefined -> [];
Policy -> [<<"--restart">>, Policy]
end.
build_user(Config) ->
case maps:get(<<"user">>, Config, undefined) of
undefined -> [];
U -> [<<"--user">>, U]
end.
build_working_dir(Config) ->
case maps:get(<<"working_dir">>, Config, undefined) of
undefined -> [];
D -> [<<"--workdir">>, D]
end.
build_hostname(Config) ->
case maps:get(<<"hostname">>, Config, undefined) of
undefined -> [];
H -> [<<"--hostname">>, H]
end.
build_privileged(Config) ->
case maps:get(<<"privileged">>, Config, false) of
true -> [<<"--privileged">>];
_ -> []
end.
build_cap_add_drop(Config) ->
Add = maps:get(<<"cap_add">>, Config, []),
Drop = maps:get(<<"cap_drop">>, Config, []),
lists:map(fun(C) -> [<<"--cap-add">>, C] end, Add) ++ lists:map(fun(C0) -> [<<"--cap-drop">>, C0] end, Drop).
build_devices(Config) ->
Devs = maps:get(<<"devices">>, Config, []),
lists:map(fun(D) -> [<<"--device">>, D] end, Devs).
build_memory(Config) ->
Mem = maps:get(<<"mem_limit">>, Config, undefined),
MemRes = maps:get(<<"mem_reservation">>, Config, undefined),
Res1 = if Mem /= undefined -> [<<"--memory">>, Mem]; true -> [] end,
Res2 = if MemRes /= undefined -> [<<"--memory-reservation">>, MemRes]; true -> [] end,
Res1 ++ Res2.
build_cpu(Config) ->
CPU = maps:get(<<"cpus">>, Config, undefined),
Shares = maps:get(<<"cpu_shares">>, Config, undefined),
Res1 = if
CPU /= undefined ->
Bin = iolist_to_binary(io_lib:format("~p", [CPU])),
[<<"--cpus">>, Bin];
true ->
[]
end,
Res2 = if
Shares /= undefined ->
Bin1 = iolist_to_binary(io_lib:format("~p", [Shares])),
[<<"--cpu-shares">>, Bin1];
true ->
[]
end,
Res1 ++ Res2.
build_ulimits(Config) ->
UL = maps:get(<<"ulimits">>, Config, #{}),
lists:map(fun({K, V}) -> [<<"--ulimit">>, <<K/binary, "=", V/binary>>] end, maps:to_list(UL)).
build_sysctls(Config) ->
SC = maps:get(<<"sysctls">>, Config, #{}),
lists:map(fun({K, V}) -> [<<"--sysctl ">>, <<K/binary, "=", V/binary>>] end, maps:to_list(SC)).
build_tmpfs(Config) ->
Tmp = maps:get(<<"tmpfs">>, Config, []),
lists:map(fun(T) -> [<<"--tmpfs">>, T] end, Tmp).
build_extra_hosts(Config) ->
Hosts = maps:get(<<"extra_hosts">>, Config, []),
lists:map(fun(H) -> [<<"--add-host">>, H] end, Hosts).
build_healthcheck(Config) ->
HC = maps:get(<<"healthcheck">>, Config, #{}),
lists:map(fun({K, V}) ->
case K of
<<"test">> ->
case V of
%% Test 是 ["CMD-SHELL", Cmd]
[<<"CMD-SHELL">>, Cmd] ->
[<<"--health-cmd">>, <<$", Cmd/binary, $">>];
%% Test 是 ["CMD", Arg1, Arg2...]
[<<"CMD">> | CmdList] ->
CmdArgs = iolist_to_binary(lists:join(<<" ">>, CmdList)),
[<<"--health-cmd">>, <<$", CmdArgs/binary, $">>];
%% Test 是 <<"NONE">>
[<<"NONE">>] ->
[<<"--no-healthcheck">>];
_ ->
[]
end;
<<"interval">> ->
[<<"--health-interval">>, V];
<<"timeout">> ->
[<<"--health-timeout">>, V];
<<"retries">> ->
[<<"--health-retries">>, io_lib:format("~p", [V])];
_ ->
[]
end
end, maps:to_list(HC)).

77
apps/efka/src/efka_inetd.erl
View File

@ -33,9 +33,9 @@
%%% API
%%%===================================================================
-spec deploy(TaskId :: integer(), ServerId :: binary(), TarUrl :: binary()) -> ok | {error, Reason :: binary()}.
deploy(TaskId, ServerId, TarUrl) when is_integer(TaskId), is_binary(ServerId), is_binary(TarUrl) ->
gen_server:call(?SERVER, {deploy, TaskId, ServerId, TarUrl}).
-spec deploy(TaskId :: integer(), ImageUrl :: binary(), Args :: [binary()]) -> {ok, ContainerId :: binary()} | {error, Reason :: binary()}.
deploy(TaskId, ImageUrl, Args) when is_integer(TaskId), is_binary(ImageUrl), is_list(Args) ->
gen_server:call(?SERVER, {deploy, TaskId, ImageUrl, Args}).
-spec start_service(ServiceId :: binary()) -> ok | {error, Reason :: term()}.
start_service(ServiceId) when is_binary(ServiceId) ->
@ -75,53 +75,39 @@ init([]) ->
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_call({deploy, TaskId, ServiceId, TarUrl}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) ->
handle_call({deploy, TaskId, ImageUrl, Args}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) ->
%% 创建目录
{ok, ServiceRootDir} = ensure_dirs(RootDir, ServiceId),
ServicePid = efka_service:get_pid(ServiceId),
case is_pid(ServicePid) of
true ->
{reply, {error, <<"the service is running, stop first">>}, State};
false ->
case check_download_url(TarUrl) of
ok ->
{ok, TaskPid} = efka_inetd_task:start_link(TaskId, ServiceRootDir, ServiceId, TarUrl),
efka_inetd_task:deploy(TaskPid),
lager:debug("[efka_inetd] start task_id: ~p, tar_url: ~p", [TaskId, TarUrl]),
{reply, ok, State#state{task_map = maps:put(TaskPid, {TaskId, ServiceId}, TaskMap)}};
{error, Reason} ->
lager:debug("[efka_inetd] check_download_url: ~p, get error: ~p", [TarUrl, Reason]),
{reply, {error, <<"download url error">>}, State}
end
end;
{ok, TaskPid} = efka_inetd_task:start_link(TaskId, RootDir, ImageUrl, Args),
efka_inetd_task:deploy(TaskPid),
lager:debug("[efka_inetd] start task_id: ~p, tar_url: ~p", [TaskId, ImageUrl]),
%% todo
{reply, ok, State#state{task_map = maps:put(TaskPid, {TaskId, xx}, TaskMap)}};
%% 启动服务: 当前服务如果正常运行，则不允许重启
handle_call({start_service, ServiceId}, _From, State) ->
case efka_service:get_pid(ServiceId) of
handle_call({start_container, ContainerId}, _From, State) ->
case efka_container:get_pid(ContainerId) of
undefined ->
case efka_service_sup:start_service(ServiceId) of
case efka_container_sup:start_container(ContainerId) of
{ok, _} ->
%% 更新数据库状态, 状态是为了保证下次efka重启的时候，服务能够启动
ok = service_model:change_status(ServiceId, 1),
ok = service_model:change_status(ContainerId, 1),
{reply, ok, State};
{error, Reason} ->
{reply, {error, Reason}, State}
end;
ServicePid when is_pid(ServicePid) ->
ContainerPid when is_pid(ContainerPid) ->
{reply, {error, <<"service is running">>}, State}
end;
%% 停止服务, 主动停止的时候会改变服务配置的status字段
handle_call({stop_service, ServiceId}, _From, State = #state{}) ->
case efka_service:get_pid(ServiceId) of
handle_call({stop_container, ContainerId}, _From, State = #state{}) ->
case efka_container:get_pid(ContainerId) of
undefined ->
{reply, {error, <<"service not running">>}, State};
ServicePid when is_pid(ServicePid) ->
efka_service_sup:stop_service(ServiceId),
ContainerPid when is_pid(ContainerPid) ->
efka_container_sup:stop_container(ContainerPid),
%% 主动停止的服务，需要更新数据库状态, 状态是为了保证下次efka重启的时候，不自动启动服务
ok = service_model:change_status(ServiceId, 0),
ok = service_model:change_status(ContainerId, 0),
{reply, ok, State}
end;
@ -184,28 +170,3 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec ensure_dirs(RootDir :: string(), ServerId :: binary()) -> {ok, ServerRootDir :: string()}.
ensure_dirs(RootDir, ServerId) when is_list(RootDir), is_binary(ServerId) ->
%% 根目录
ServiceRootDir = RootDir ++ "/" ++ binary_to_list(ServerId) ++ "/",
ok = filelib:ensure_dir(ServiceRootDir),
{ok, ServiceRootDir}.
%% 通过head请求先判定下载地址是否正确
-spec check_download_url(Url :: string() | binary()) -> ok | {error, Reason :: term()}.
check_download_url(Url) when is_binary(Url) ->
check_download_url(binary_to_list(Url));
check_download_url(Url) when is_list(Url) ->
SslOpts = [
{ssl, [
% 完全禁用证书验证
{verify, verify_none}
]}
],
case httpc:request(head, {Url, []}, SslOpts, [{sync, true}]) of
{ok, {{_, 200, "OK"}, _Headers, _}} ->
ok;
{error, Reason} ->
{error, Reason}
end.

281
apps/efka/src/efka_inetd_task.erl
View File

@ -16,18 +16,23 @@
-export([start_link/4]).
-export([deploy/1]).
-export([docker_pull/1, test/0]).
%% 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, {
service_root_dir :: string(),
root_dir :: string(),
task_id :: integer(),
service_id :: binary(),
tar_url :: binary()
image_url :: binary(),
args = [] :: list()
}).
test() ->
docker_pull(<<"docker.1ms.run/library/nginxyx">>).
%%%===================================================================
%%% API
%%%===================================================================
@ -37,10 +42,10 @@ deploy(Pid) when is_pid(Pid) ->
gen_server:cast(Pid, deploy).
%% @doc Spawns the server and registers the local name (unique)
-spec(start_link(TaskId :: integer(), ServiceRootDir :: string(), ServiceId :: binary(), TarUrl :: binary()) ->
-spec(start_link(TaskId :: integer(), RootDir :: string(), ImageUrl :: binary(), Args :: map()) ->
{ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link(TaskId, ServiceRootDir, ServiceId, TarUrl) when is_integer(TaskId), is_list(ServiceRootDir), is_binary(ServiceId), is_binary(TarUrl) ->
gen_server:start_link(?MODULE, [TaskId, ServiceRootDir, ServiceId, TarUrl], []).
start_link(TaskId, RootDir, ImageUrl, Args) when is_integer(TaskId), is_list(RootDir), is_binary(ImageUrl), is_map(Args) ->
gen_server:start_link(?MODULE, [TaskId, RootDir, ImageUrl, Args], []).
%%%===================================================================
%%% gen_server callbacks
@ -51,8 +56,8 @@ start_link(TaskId, ServiceRootDir, ServiceId, TarUrl) when is_integer(TaskId), i
-spec(init(Args :: term()) ->
{ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
{stop, Reason :: term()} | ignore).
init([TaskId, ServiceRootDir, ServiceId, TarUrl]) ->
{ok, #state{task_id = TaskId, service_root_dir = ServiceRootDir, service_id = ServiceId, tar_url = TarUrl}}.
init([TaskId, RootDir, ImageUrl, Args]) ->
{ok, #state{task_id = TaskId, root_dir = RootDir, image_url = ImageUrl, args = Args}}.
%% @private
%% @doc Handling call messages
@ -73,8 +78,8 @@ handle_call(_Request, _From, State = #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_cast(deploy, State = #state{task_id = TaskId, service_root_dir = ServiceRootDir, service_id = ServiceId, tar_url = TarUrl}) ->
do_deploy(TaskId, ServiceRootDir, ServiceId, TarUrl),
handle_cast(deploy, State = #state{task_id = TaskId, root_dir = RootDir, image_url = ImageUrl, args = Args}) ->
do_deploy(TaskId, RootDir, ImageUrl, Args),
{stop, normal, State};
handle_cast(_Request, State) ->
{stop, normal, State}.
@ -110,132 +115,154 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
%%% Internal functions
%%%===================================================================
%{
% "image": "nginx:latest",
% "container_name": "my_nginx",
% "ports": ["8080:80", "443:443"],
% "volumes": ["/host/data:/data", "/host/log:/var/log"],
% "envs": ["ENV1=val1", "ENV2=val2"],
% "entrypoint": ["/docker-entrypoint.sh"],
% "command": ["nginx", "-g", "daemon off;"],
% "restart": "always"
%}
-spec do_deploy(TaskId :: integer(), ServiceRootDir :: string(), ServiceId :: binary(), TarUrl :: binary()) -> no_return().
do_deploy(TaskId, ServiceRootDir, ServiceId, TarUrl) when is_integer(TaskId), is_list(ServiceRootDir), is_binary(ServiceId), is_binary(TarUrl) ->
case download(binary_to_list(TarUrl), ServiceRootDir) of
{ok, TarFile, CostTs} ->
Log = io_lib:format("download: ~p completed, cost time: ~p(ms)", [binary_to_list(TarUrl), CostTs]),
efka_inetd_task_log:stash(TaskId, list_to_binary(Log)),
do_deploy(TaskId, RootDir, Image, Args) when is_integer(TaskId), is_list(RootDir), is_binary(Image), is_map(Args) ->
%% 尝试拉取镜像
ContainerName = maps:get(<<"container_name">>, Args),
{ok, ContainerDir} = ensure_dirs(RootDir, ContainerName),
%% 创建工作目录
WorkDir = ServiceRootDir ++ "/work_dir/",
case filelib:ensure_dir(WorkDir) of
ok ->
%% 清理目录下的文件
catch delete_directory(WorkDir),
case tar_extract(TarFile, WorkDir) of
ok ->
%% 更新数据
ok = service_model:insert(#service{
service_id = ServiceId,
tar_url = TarUrl,
%% 工作目录
root_dir = ServiceRootDir,
config_json = <<"">>,
%% 状态： 0: 停止, 1: 运行中
status = 0
}),
efka_inetd_task_log:stash(TaskId, <<"deploy success">>);
{error, Reason} ->
TarLog = io_lib:format("tar decompression: ~p, error: ~p", [filename:basename(TarFile), Reason]),
efka_inetd_task_log:stash(TaskId, list_to_binary(TarLog))
end;
{error, Reason} ->
DownloadLog = io_lib:format("make work_dir error: ~p", [Reason]),
efka_inetd_task_log:stash(TaskId, list_to_binary(DownloadLog))
end;
{error, Reason} ->
DownloadLog = io_lib:format("download: ~p, error: ~p", [binary_to_list(TarUrl), Reason]),
efka_inetd_task_log:stash(TaskId, list_to_binary(DownloadLog))
end.
case try_pull_image(Image) of
ok ->
%% 创建container
%% 如果存在envs参数，则生成环境变量参数文件
maybe_create_env_file(ContainerDir, maps:get(<<"envs">>, Args, [])),
%% 递归删除目录下的问题
-spec delete_directory(string()) -> ok | {error, term()}.
delete_directory(Dir) when is_list(Dir) ->
% 递归删除目录内容
case file:list_dir(Dir) of
{ok, Files} ->
lists:foreach(fun(File) ->
FullPath = filename:join(Dir, File),
case filelib:is_dir(FullPath) of
true ->
delete_directory(FullPath);
false ->
file:delete(FullPath)
end
end, Files),
% 删除空目录
file:del_dir(Dir);
{error, enoent} ->
ok;
{error, Reason} ->
{error, Reason}
end.
end,
%% 解压文件到指定目录
-spec tar_extract(string(), string()) -> ok | {error, term()}.
tar_extract(TarFile, TargetDir) when is_list(TarFile), is_list(TargetDir) ->
%% 判断文件的后缀名来判断, options: verbose
erl_tar:extract(TarFile, [compressed, {cwd, TargetDir}]).
%case download(binary_to_list(ImageUrl), RootDir) of
% {ok, TarFile, CostTs} ->
% Log = io_lib:format("download: ~p completed, cost time: ~p(ms)", [binary_to_list(TarUrl), CostTs]),
% efka_inetd_task_log:stash(TaskId, list_to_binary(Log)),
%% 下载文件
-spec download(Url :: string(), TargetDir :: string()) ->
{ok, TarFile :: string(), CostTs :: integer()} | {error, Reason :: term()}.
download(Url, TargetDir) when is_list(Url), is_list(TargetDir) ->
SslOpts = [
{ssl, [
% 完全禁用证书验证
{verify, verify_none}
]}
],
% %% 创建工作目录
% WorkDir = ServiceRootDir ++ "/work_dir/",
% case filelib:ensure_dir(WorkDir) of
% ok ->
% %% 清理目录下的文件
% catch delete_directory(WorkDir),
% case tar_extract(TarFile, WorkDir) of
% ok ->
% %% 更新数据
% ok = service_model:insert(#service{
% service_id = ServiceId,
% tar_url = TarUrl,
% %% 工作目录
% root_dir = ServiceRootDir,
% config_json = <<"">>,
% %% 状态： 0: 停止, 1: 运行中
% status = 0
% }),
% efka_inetd_task_log:stash(TaskId, <<"deploy success">>);
% {error, Reason} ->
% TarLog = io_lib:format("tar decompression: ~p, error: ~p", [filename:basename(TarFile), Reason]),
% efka_inetd_task_log:stash(TaskId, list_to_binary(TarLog))
% end;
% {error, Reason} ->
% DownloadLog = io_lib:format("make work_dir error: ~p", [Reason]),
% efka_inetd_task_log:stash(TaskId, list_to_binary(DownloadLog))
% end;
% {error, Reason} ->
% DownloadLog = io_lib:format("download: ~p, error: ~p", [binary_to_list(TarUrl), Reason]),
% efka_inetd_task_log:stash(TaskId, list_to_binary(DownloadLog))
%end.
ok.
TargetFile = get_filename_from_url(Url),
FullFilename = TargetDir ++ TargetFile,
StartTs = os:timestamp(),
case httpc:request(get, {Url, []}, SslOpts, [{sync, false}, {stream, self}]) of
{ok, RequestId} ->
case receive_data(RequestId, FullFilename) of
ok ->
EndTs = os:timestamp(),
%% 计算操作的时间，单位为毫秒
CostMs = timer:now_diff(EndTs, StartTs) div 1000,
{ok, FullFilename, CostMs};
{error, Reason} ->
%% 出错需要删除掉文件
file:delete(FullFilename),
{error, Reason}
end;
{error, Reason} ->
{error, Reason}
end.
%% 处理头部信息, 解析可能的文件名
receive_data(RequestId, FullFilename) ->
receive
{http, {RequestId, stream_start, _Headers}} ->
{ok, File} = file:open(FullFilename, [write, binary]),
receive_data0(RequestId, File);
{http, {RequestId, {{_, 404, Status}, _Headers, Body}}} ->
lager:debug("[efka_downloader] http_status: ~p, body: ~p", [Status, Body]),
{error, Status}
end.
%% 接受文件数据
receive_data0(RequestId, File) ->
receive
{http, {RequestId, {error, Reason}}} ->
ok = file:close(File),
{error, Reason};
{http, {RequestId, stream_end, _Headers}} ->
ok = file:close(File),
-spec try_pull_image(Image :: binary()) -> ok | {error, Reason :: any()}.
try_pull_image(Image) when is_binary(Image) ->
case efka_docker_command:check_image_exist(Image) of
true ->
ok;
{http, {RequestId, stream, Data}} ->
file:write(File, Data),
receive_data0(RequestId, File)
false ->
efka_docker_command:pull_image(Image)
end.
-spec get_filename_from_url(Url :: string()) -> string().
get_filename_from_url(Url) when is_list(Url) ->
URIMap = uri_string:parse(Url),
Path = maps:get(path, URIMap),
filename:basename(Path).
maybe_create_env_file(_ContainerDir, []) ->
ok;
maybe_create_env_file(ContainerDir, Envs) when is_list(Envs)->
TargetFile = ContainerDir ++ "env",
{ok, IoDevice} = file:open(TargetFile, [write, binary]),
lists:foreach(fun(Env) -> file:write(IoDevice, <<Env/binary, $\n>>) end, Envs),
ok = file:close(IoDevice),
{ok, TargetFile}.
docker_pull(ImageUrl) when is_binary(ImageUrl) ->
%% todo 重定向错误流 {stderr_to_stdout, true}
PortSettings = [stream, exit_status, use_stdio, binary],
ImageUrl1 = normalize_image(ImageUrl),
ExecCmd = "docker pull " ++ binary_to_list(ImageUrl1),
lager:debug("cmd : ~p", [ExecCmd]),
case catch erlang:open_port({spawn, ExecCmd}, PortSettings) of
Port when is_port(Port) ->
case gather_output(Port) of
{0, Output} ->
extract_sha256(Output);
{ExitCode, Error} ->
lager:debug("call me here: ~p", [Error]),
{error, {ExitCode, Error}}
end;
Error ->
lager:debug("error: ~p", [Error]),
{error, <<"exec command startup failed">>}
end.
gather_output(Port) ->
gather_output(Port, <<>>).
gather_output(Port, Acc) ->
receive
{Port, {data, Data}} ->
gather_output(Port, [Acc, Data]);
{Port, {exit_status, Status}} ->
{Status, iolist_to_binary(Acc)}
end.
extract_sha256(Output) when is_binary(Output) ->
Parts = binary:split(Output, <<$\n>>, [global]),
lager:debug("parts: ~p", [Parts]),
case lists:search(fun(Line) -> starts_with(Line, <<"Digest:">>) end, Parts) of
{value, Digest} ->
Sha256 = lists:last(binary:split(Digest, <<":">>, [global])),
{ok, Sha256};
false ->
error
end.
starts_with(Binary, Prefix) when is_binary(Binary), is_binary(Prefix) ->
PrefixSize = byte_size(Prefix),
case Binary of
<<Prefix:PrefixSize/binary, _Rest/binary>> -> true;
_ -> false
end.
-spec normalize_image(binary()) -> binary().
normalize_image(Image) when is_binary(Image) ->
Parts = binary:split(Image, <<"/">>, [global]),
{PrefixParts, [Last]} = lists:split(length(Parts) - 1, Parts),
NormalizedLast = case binary:split(Last, <<":">>, [global]) of
[_Name] -> <<Last/binary, ":latest">>;
[_Name, _Tag] -> Last
end,
iolist_to_binary(lists:join(<<"/">>, PrefixParts ++ [NormalizedLast])).
-spec ensure_dirs(RootDir :: string(), ContainerName :: binary()) -> {ok, ServerRootDir :: string()}.
ensure_dirs(RootDir, ContainerName) when is_list(RootDir), is_binary(ContainerName) ->
%% 根目录
ServiceRootDir = RootDir ++ "/" ++ binary_to_list(ContainerName) ++ "/",
ok = filelib:ensure_dir(ServiceRootDir),
{ok, ServiceRootDir}.

16
apps/efka/src/efka_sup.erl
View File

@ -64,14 +64,14 @@ init([]) ->
modules => ['efka_inetd']
},
#{
id => 'efka_remote_agent',
start => {'efka_remote_agent', start_link, []},
restart => permanent,
shutdown => 2000,
type => worker,
modules => ['efka_remote_agent']
},
%#{
% id => 'efka_remote_agent',
% start => {'efka_remote_agent', start_link, []},
% restart => permanent,
% shutdown => 2000,
% type => worker,
% modules => ['efka_remote_agent']
%},
#{
id => 'tcp_sup',