diff --git a/src/docker/docker_commands.erl b/src/docker/docker_commands.erl index e061150..5924a66 100644 --- a/src/docker/docker_commands.erl +++ b/src/docker/docker_commands.erl @@ -35,7 +35,7 @@ check_image_exist(Image) when is_binary(Image) -> -spec create_container(ContainerDir :: string(), Params :: map()) -> {ok, ContainerId :: binary()} | {error, Reason :: any()}. -create_container(ContainerDir, #{container_name := ContainerName0, create := CreateOpts}) +create_container(ContainerDir, #{<<"container_name">> := ContainerName0, <<"create">> := CreateOpts}) when is_list(ContainerDir), is_binary(ContainerName0), is_map(CreateOpts) -> Url = lists:flatten(io_lib:format("/containers/create?name=~s", [binary_to_list(ContainerName0)])), Options = docker_container_builder:build_options(ContainerName0, ContainerDir, CreateOpts), diff --git a/src/docker/docker_container_builder.erl b/src/docker/docker_container_builder.erl index 0d50669..c13cb9d 100644 --- a/src/docker/docker_container_builder.erl +++ b/src/docker/docker_container_builder.erl @@ -23,30 +23,30 @@ patch_create_options(ContainerName, ConfigFile, undefined) -> patch_create_options(ContainerName, ConfigFile, #{}); patch_create_options(ContainerName, ConfigFile, Create0) when is_binary(ContainerName), is_binary(ConfigFile), is_map(Create0) -> - Config = patch_container_config(ContainerName, ensure_container_config(maps:get(config, Create0, undefined))), - HostConfig = patch_host_config(ConfigFile, ensure_host_config(maps:get(host_config, Create0, undefined))), + Config = patch_container_config(ContainerName, ensure_container_config(field(Create0, <<"config">>, undefined))), + HostConfig = patch_host_config(ConfigFile, ensure_host_config(field(Create0, <<"host_config">>, undefined))), Create0#{ - config => Config, - host_config => HostConfig + <<"config">> => Config, + <<"host_config">> => HostConfig }. -spec patch_container_config(binary(), map()) -> map(). patch_container_config(ContainerName, Config0) when is_binary(ContainerName), is_map(Config0) -> - Env0 = [to_binary(EnvItem) || EnvItem <- maps:get(env, Config0, [])], - Volumes0 = [to_binary(Volume) || Volume <- maps:get(volumes, Config0, [])], + Env0 = [to_binary(EnvItem) || EnvItem <- field(Config0, <<"env">>, [])], + Volumes0 = [to_binary(Volume) || Volume <- field(Config0, <<"volumes">>, [])], ConfigVolume = <<"/usr/local/etc/service.conf">>, Envs = add_unique_front([<<"CONTAINER_NAME=", ContainerName/binary>>], Env0), Volumes = add_unique_front([ConfigVolume], Volumes0), Config0#{ - env => Envs, - volumes => Volumes + <<"env">> => Envs, + <<"volumes">> => Volumes }. -spec patch_host_config(binary(), map()) -> map(). patch_host_config(ConfigFile, HostConfig0) when is_binary(ConfigFile), is_map(HostConfig0) -> - Binds0 = [to_binary(Bind) || Bind <- maps:get(binds, HostConfig0, [])], + Binds0 = [to_binary(Bind) || Bind <- field(HostConfig0, <<"binds">>, [])], ConfigBind = <>, - HostConfig0#{binds => add_unique_front([ConfigBind], Binds0)}. + HostConfig0#{<<"binds">> => add_unique_front([ConfigBind], Binds0)}. -spec ensure_container_config(map() | undefined) -> map(). ensure_container_config(undefined) -> @@ -62,39 +62,39 @@ ensure_host_config(HostConfig) when is_map(HostConfig) -> -spec build_create_options(map()) -> map(). build_create_options(Create0) when is_map(Create0) -> - Config = maps:get(config, Create0, #{}), - HostConfig = maps:get(host_config, Create0, #{}), - Endpoints = networking_config_endpoints(maps:get(networking_config, Create0, undefined)), + Config = field(Create0, <<"config">>, #{}), + HostConfig = field(Create0, <<"host_config">>, #{}), + Endpoints = networking_config_endpoints(field(Create0, <<"networking_config">>, undefined)), #{ - <<"Image">> => to_binary(maps:get(image, Config, <<>>)), - <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- maps:get(cmd, Config, [])], - <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- maps:get(entrypoint, Config, [])], - <<"Env">> => [to_binary(EnvItem) || EnvItem <- maps:get(env, Config, [])], - <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(maps:get(labels, Config, #{}))]), - <<"Volumes">> => build_volumes(maps:get(volumes, Config, [])), - <<"User">> => to_binary(maps:get(user, Config, <<>>)), - <<"WorkingDir">> => to_binary(maps:get(working_dir, Config, <<>>)), - <<"Hostname">> => to_binary(maps:get(hostname, Config, <<>>)), - <<"ExposedPorts">> => build_expose(maps:get(exposed_ports, Config, [])), + <<"Image">> => to_binary(field(Config, <<"image">>, <<>>)), + <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- field(Config, <<"cmd">>, [])], + <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- field(Config, <<"entrypoint">>, [])], + <<"Env">> => [to_binary(EnvItem) || EnvItem <- field(Config, <<"env">>, [])], + <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(field(Config, <<"labels">>, #{}))]), + <<"Volumes">> => build_volumes(field(Config, <<"volumes">>, [])), + <<"User">> => to_binary(field(Config, <<"user">>, <<>>)), + <<"WorkingDir">> => to_binary(field(Config, <<"working_dir">>, <<>>)), + <<"Hostname">> => to_binary(field(Config, <<"hostname">>, <<>>)), + <<"ExposedPorts">> => build_expose(field(Config, <<"exposed_ports">>, [])), <<"NetworkingConfig">> => build_networking_config(Endpoints), - <<"Healthcheck">> => build_healthcheck(maps:get(healthcheck, Config, undefined)), + <<"Healthcheck">> => build_healthcheck(field(Config, <<"healthcheck">>, undefined)), <<"HostConfig">> => fold_merge([ - build_binds(maps:get(binds, HostConfig, [])), - build_network_mode(maps:get(network_mode, HostConfig, <<>>)), - build_restart(maps:get(restart_policy, HostConfig, undefined)), - build_privileged(maps:get(privileged, HostConfig, false)), - build_cap_add_drop(maps:get(cap_add, HostConfig, []), maps:get(cap_drop, HostConfig, [])), - build_devices(maps:get(devices, HostConfig, [])), + build_binds(field(HostConfig, <<"binds">>, [])), + build_network_mode(field(HostConfig, <<"network_mode">>, <<>>)), + build_restart(field(HostConfig, <<"restart_policy">>, undefined)), + build_privileged(field(HostConfig, <<"privileged">>, false)), + build_cap_add_drop(field(HostConfig, <<"cap_add">>, []), field(HostConfig, <<"cap_drop">>, [])), + build_devices(field(HostConfig, <<"devices">>, [])), build_resources( - maps:get(memory, HostConfig, 0), - maps:get(memory_reservation, HostConfig, 0), - maps:get(nano_cpus, HostConfig, 0), - maps:get(cpu_shares, HostConfig, 0) + field(HostConfig, <<"memory">>, 0), + field(HostConfig, <<"memory_reservation">>, 0), + field(HostConfig, <<"nano_cpus">>, 0), + field(HostConfig, <<"cpu_shares">>, 0) ), - build_ulimits(maps:get(ulimits, HostConfig, [])), - build_tmpfs(maps:get(tmpfs, HostConfig, #{})), - build_sysctls(maps:get(sysctls, HostConfig, #{})), - build_extra_hosts(maps:get(extra_hosts, HostConfig, [])) + build_ulimits(field(HostConfig, <<"ulimits">>, [])), + build_tmpfs(field(HostConfig, <<"tmpfs">>, #{})), + build_sysctls(field(HostConfig, <<"sysctls">>, #{})), + build_extra_hosts(field(HostConfig, <<"extra_hosts">>, [])) ]) }. @@ -102,7 +102,7 @@ build_create_options(Create0) when is_map(Create0) -> networking_config_endpoints(undefined) -> []; networking_config_endpoints(NetworkingConfig) when is_map(NetworkingConfig) -> - maps:get(endpoints, NetworkingConfig, []). + field(NetworkingConfig, <<"endpoints">>, []). -spec fold_merge([map()]) -> map(). fold_merge(List) -> @@ -141,7 +141,7 @@ build_networking_config(Endpoints) when is_list(Endpoints) -> [] -> #{}; _ -> - NetCfg = maps:from_list([{to_binary(Name), #{}} || #{name := Name} <- Endpoints]), + NetCfg = maps:from_list([{to_binary(Name), #{}} || #{<<"name">> := Name} <- Endpoints]), #{<<"EndpointsConfig">> => NetCfg} end. @@ -156,10 +156,10 @@ build_healthcheck(undefined) -> #{}; build_healthcheck(Healthcheck) when is_map(Healthcheck) -> #{ - <<"Test">> => [to_binary(Item) || Item <- maps:get(test, Healthcheck, [])], - <<"Interval">> => maps:get(interval_ns, Healthcheck, 0), - <<"Timeout">> => maps:get(timeout_ns, Healthcheck, 0), - <<"Retries">> => maps:get(retries, Healthcheck, 0) + <<"Test">> => [to_binary(Item) || Item <- field(Healthcheck, <<"test">>, [])], + <<"Interval">> => field(Healthcheck, <<"interval_ns">>, 0), + <<"Timeout">> => field(Healthcheck, <<"timeout_ns">>, 0), + <<"Retries">> => field(Healthcheck, <<"retries">>, 0) }. -spec build_restart(map() | undefined) -> map(). @@ -167,9 +167,9 @@ build_restart(undefined) -> #{}; build_restart(RestartPolicy0) when is_map(RestartPolicy0) -> RestartPolicy = #{ - <<"Name">> => to_binary(maps:get(name, RestartPolicy0, <<>>)) + <<"Name">> => to_binary(field(RestartPolicy0, <<"name">>, <<>>)) }, - case maps:get(maximum_retry_count, RestartPolicy0, 0) of + case field(RestartPolicy0, <<"maximum_retry_count">>, 0) of 0 -> #{<<"RestartPolicy">> => RestartPolicy}; RetryCount -> @@ -210,9 +210,9 @@ build_devices(Devices) when is_list(Devices) -> <<"PathInContainer">> => to_binary(ContainerPath), <<"CgroupPermissions">> => device_permissions(Permissions) } || #{ - path_on_host := HostPath, - path_in_container := ContainerPath, - cgroup_permissions := Permissions + <<"path_on_host">> := HostPath, + <<"path_in_container">> := ContainerPath, + <<"cgroup_permissions">> := Permissions } <- Devices], #{<<"Devices">> => DevObjs} end. @@ -255,7 +255,7 @@ build_ulimits(Ulimits) when is_list(Ulimits) -> <<"Name">> => to_binary(Name), <<"Soft">> => Soft, <<"Hard">> => Hard - } || #{name := Name, soft := Soft, hard := Hard} <- Ulimits]} + } || #{<<"name">> := Name, <<"soft">> := Soft, <<"hard">> := Hard} <- Ulimits]} end. -spec build_sysctls(map()) -> map(). @@ -286,7 +286,7 @@ build_extra_hosts(Hosts) when is_list(Hosts) -> end. -spec normalize_expose_port(map()) -> binary(). -normalize_expose_port(#{container_port := Port, protocol := Protocol}) -> +normalize_expose_port(#{<<"container_port">> := Port, <<"protocol">> := Protocol}) -> PortBin = integer_to_binary(Port), ProtocolBin = to_binary(Protocol), case ProtocolBin of @@ -316,6 +316,16 @@ add_unique_front([Item | Rest], List) -> end, add_unique_front(Rest, NList). +-spec field(map(), binary(), term()) -> term(). +field(Map, Key, Default) when is_map(Map), is_binary(Key) -> + safe_value(maps:get(Key, Map, Default)). + +-spec safe_value(term()) -> term(). +safe_value(<<"__undefined__">>) -> + undefined; +safe_value(Value) -> + Value. + -spec to_binary(binary() | list() | atom() | any()) -> binary(). to_binary(Value) when is_binary(Value) -> Value; diff --git a/src/docker/docker_deploy_manager.erl b/src/docker/docker_deploy_manager.erl index 9fb0ecc..0e4b4af 100644 --- a/src/docker/docker_deploy_manager.erl +++ b/src/docker/docker_deploy_manager.erl @@ -49,7 +49,7 @@ init([]) -> -spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}. handle_call({deploy, TaskId, Params}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) when is_map(Params) -> - ContainerName = maps:get(container_name, Params), + ContainerName = maps:get(<<"container_name">>, Params), {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName), {ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params), logger:debug("[docker_deploy_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]), diff --git a/src/docker/docker_deployer.erl b/src/docker/docker_deployer.erl index d8c3d43..3ab24b2 100644 --- a/src/docker/docker_deployer.erl +++ b/src/docker/docker_deployer.erl @@ -199,13 +199,13 @@ short_container_id(ContainerId) when is_binary(ContainerId) -> ContainerId. -spec deploy_container_name(map()) -> binary(). -deploy_container_name(#{container_name := ContainerName}) when is_binary(ContainerName) -> +deploy_container_name(#{<<"container_name">> := ContainerName}) when is_binary(ContainerName) -> ContainerName; deploy_container_name(_) -> throw({deploy_error, <<"invalid deploy params: container_name missing">>}). -spec deploy_image(map()) -> binary(). -deploy_image(#{create := #{config := #{image := Image}}}) when is_binary(Image) -> +deploy_image(#{<<"create">> := #{<<"config">> := #{<<"image">> := Image}}}) when is_binary(Image) -> Image; deploy_image(_) -> throw({deploy_error, <<"invalid deploy params: image missing">>}). diff --git a/src/transport/efka_client.erl b/src/transport/efka_client.erl index 3df9196..7c54e49 100644 --- a/src/transport/efka_client.erl +++ b/src/transport/efka_client.erl @@ -32,7 +32,7 @@ -record(state, { socket :: undefined | ssl:sslsocket(), %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系 - auth_ref = undefined :: undefined | reference(), + auth_ref = undefined :: undefined | binary(), dropped_message_count = 0 :: non_neg_integer() }). @@ -90,7 +90,7 @@ callback_mode() -> %% 异步发送数据, 连接存在时候直接发送；否则缓存到DETS -spec handle_event(term(), term(), atom(), #state{}) -> term(). handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) -> - Packet = term_to_binary({message, {data, #{route_key => RouteKey, metric => Metric}}}), + Packet = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}), case StateName of ?STATE_ACTIVATED -> ok = ssl:send(Socket, Packet), @@ -103,12 +103,12 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so %% Task的stream流，只做实时的 handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> logger:debug("[efka_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]), - Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => Type, stream => Stream}}}), + Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}), ok = ssl:send(Socket, Packet), {keep_state, State}; handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Packet = term_to_binary({message, {task_event, #{task_id => TaskId, type => <<"close">>, stream => Reason}}}), + Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}), ok = ssl:send(Socket, Packet), {keep_state, State}; @@ -127,9 +127,10 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) -> case connect_socket() of {ok, Socket} -> - Ref = make_ref(), + Ref = request_ref(), AuthPacket = auth_packet(Ref), ok = ssl:send(Socket, AuthPacket), + logger:debug("[efka_client] send auth request, ref: ~p", [Ref]), {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]}; {error, _Reason} -> schedule_reconnect(), @@ -157,8 +158,16 @@ handle_event(info, flush_cache, _, State) -> %% 处理收到的ssl消息 handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) -> - Packet = binary_to_term(PacketBin, [safe]), - {keep_state, State, [{next_event, internal, Packet}]}; + try binary_to_term(PacketBin, [safe]) of + Packet -> + {keep_state, State, [{next_event, internal, Packet}]} + catch + error:Error -> + logger:warning("[efka_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]), + disconnect(Socket), + schedule_reconnect(), + {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}} + end; handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) -> logger:debug("[efka_client] ssl error: ~p", [Reason]), disconnect(Socket), @@ -171,57 +180,32 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) -> %%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。 -handle_event(internal, {command, Ref, {container, #{action := list}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = docker_commands:get_containers(), - send_container_response(Socket, Ref, Reply), +handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> + handle_container_command(Ref, Request, Socket), {keep_state, State}; -handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = docker_deploy_manager:deploy(TaskId, Params), - send_container_response(Socket, Ref, Reply), - {keep_state, State}; -handle_event(internal, {command, Ref, {container, #{action := start, target := Target}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = docker_commands:start_container(container_target(Target)), - send_container_response(Socket, Ref, Reply), - {keep_state, State}; -handle_event(internal, {command, Ref, {container, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds), - send_container_response(Socket, Ref, Reply), - {keep_state, State}; -handle_event(internal, {command, Ref, {container, #{action := kill, target := Target, signal := Signal}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)), - send_container_response(Socket, Ref, Reply), - {keep_state, State}; -handle_event(internal, {command, Ref, {container, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)), - send_container_response(Socket, Ref, Reply), - {keep_state, State}; -handle_event(internal, {command, Ref, {container, #{action := config, target := Target, config := Config}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) -> - Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)), - send_container_response(Socket, Ref, Reply), - {keep_state, State}; -handle_event(internal, {command, Ref, {container, Request}}, _StateName, State = #state{socket = Socket}) -> +handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) -> logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]), send_container_response(Socket, Ref, {error, <<"agent invalid">>}), {keep_state, State}; %% 处理response -handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> +handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) -> logger:debug("[efka_client] auth success"), {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]}; -handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) -> +handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) -> logger:debug("[efka_client] auth failed, reason: ~p", [Reason]), disconnect(Socket), schedule_reconnect(), {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}}; -handle_event(internal, {response, _Ref, Reply}, StateName, State) -> +handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) -> logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]), {keep_state, State}; -handle_event(internal, {command_response, _Ref, Reply}, StateName, State) -> +handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) -> logger:warning("[efka_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]), {keep_state, State}; %% 处理Pub/Sub机制 -handle_event(internal, {message, {pub, #{topic := Topic, qos := Qos, content := Content}}}, ?STATE_ACTIVATED, State) -> +handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) -> logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]), efka_subscription:publish(Topic, Qos, Content), {keep_state, State}; @@ -233,10 +217,45 @@ handle_event(info, Info, _, State = #state{}) -> logger:notice("[efka_client] get unknown info: ~p", [Info]), {keep_state, State}. +-spec handle_container_command(binary(), term(), ssl:sslsocket()) -> ok. +handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) -> + Reply = docker_commands:get_containers(), + send_container_response(Socket, Ref, Reply), + ok; +handle_container_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) -> + Reply = docker_deploy_manager:deploy(TaskId, Params), + send_container_response(Socket, Ref, Reply), + ok; +handle_container_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) -> + Reply = docker_commands:start_container(container_target(Target)), + send_container_response(Socket, Ref, Reply), + ok; +handle_container_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) -> + Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds), + send_container_response(Socket, Ref, Reply), + ok; +handle_container_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) -> + Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)), + send_container_response(Socket, Ref, Reply), + ok; +handle_container_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) -> + Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)), + send_container_response(Socket, Ref, Reply), + ok; +handle_container_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) -> + Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)), + send_container_response(Socket, Ref, Reply), + ok; +handle_container_command(Ref, Request, Socket) -> + logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]), + send_container_response(Socket, Ref, {error, <<"agent invalid">>}), + ok. + -spec terminate(term(), atom(), #state{}) -> ok. -terminate(_Reason, _StateName, _State = #state{socket = Socket}) -> +terminate(Reason, _StateName, _State = #state{socket = Socket}) -> disconnect(Socket), efka_client_cache:close(), + logger:notice("[efka_client] terminate with reason: ~p", [Reason]), ok. -spec code_change(term(), atom(), #state{}, term()) -> {ok, atom(), #state{}}. @@ -247,17 +266,17 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) -> %%% Internal functions %%%=================================================================== --spec auth_packet(reference()) -> binary(). -auth_packet(Ref) when is_reference(Ref) -> +-spec auth_packet(binary()) -> binary(). +auth_packet(Ref) when is_binary(Ref) -> {ok, AuthInfo} = application:get_env(efka, auth), UUID = proplists:get_value(uuid, AuthInfo), Token = proplists:get_value(token, AuthInfo), Timestamp = efka_util:timestamp(), - term_to_binary({request, Ref, {auth_request, #{ - uuid => list_to_binary(UUID), - token => list_to_binary(Token), - timestamp => Timestamp + term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{ + <<"uuid">> => list_to_binary(UUID), + <<"token">> => list_to_binary(Token), + <<"timestamp">> => Timestamp }}}). -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}. @@ -284,15 +303,45 @@ disconnect(Socket) -> schedule_reconnect() -> erlang:start_timer(5000, self(), create_transport). --spec send_container_response(ssl:sslsocket(), reference(), term()) -> ok. +-spec request_ref() -> binary(). +request_ref() -> + crypto:strong_rand_bytes(16). + +-spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok. send_container_response(Socket, Ref, Reply) -> - Packet = term_to_binary({command_response, Ref, {container, Reply}}), + Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}), ok = ssl:send(Socket, Packet). +-spec safe_reply(term()) -> term(). +safe_reply(ok) -> + <<"ok">>; +safe_reply({ok, Result}) -> + {<<"ok">>, safe_term(Result)}; +safe_reply({error, Reason}) -> + {<<"error">>, safe_term(Reason)}. + +-spec safe_term(term()) -> term(). +safe_term(true) -> + true; +safe_term(false) -> + false; +safe_term(undefined) -> + <<"__undefined__">>; +safe_term(Value) when is_atom(Value) -> + atom_to_binary(Value, utf8); +safe_term(Value) when is_map(Value) -> + maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]); +safe_term(Value) when is_list(Value) -> + [safe_term(Item) || Item <- Value]; +safe_term(Value) when is_tuple(Value) -> + list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]); +safe_term(Value) -> + Value. + -spec container_target(map()) -> binary(). container_target(Target) when is_map(Target) -> - NameBin = to_binary(maps:get(name, Target, <<>>)), - IdBin = to_binary(maps:get(id, Target, <<>>)), + NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)), + IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)), case NameBin of <<>> -> true = IdBin =/= <<>>,