fix binary_to_term

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),

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))),
+    Config = patch_container_config(ContainerName, ensure_container_config(field(Create0, <<"config">>, undefined))),
-    HostConfig = patch_host_config(ConfigFile, ensure_host_config(maps:get(host_config, Create0, undefined))),
+    HostConfig = patch_host_config(ConfigFile, ensure_host_config(field(Create0, <<"host_config">>, undefined))),
     Create0#{
-        config => Config,
+        <<"config">> => Config,
-        host_config => HostConfig
+        <<"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, [])],
+    Env0 = [to_binary(EnvItem) || EnvItem <- field(Config0, <<"env">>, [])],
-    Volumes0 = [to_binary(Volume) || Volume <- maps:get(volumes, Config0, [])],
+    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,
+        <<"env">> => Envs,
-        volumes => Volumes
+        <<"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 = <<ConfigFile/binary, ":/usr/local/etc/service.conf">>,
-    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, #{}),
+    Config = field(Create0, <<"config">>, #{}),
-    HostConfig = maps:get(host_config, Create0, #{}),
+    HostConfig = field(Create0, <<"host_config">>, #{}),
-    Endpoints = networking_config_endpoints(maps:get(networking_config, Create0, undefined)),
+    Endpoints = networking_config_endpoints(field(Create0, <<"networking_config">>, undefined)),
     #{
-        <<"Image">> => to_binary(maps:get(image, Config, <<>>)),
+        <<"Image">> => to_binary(field(Config, <<"image">>, <<>>)),
-        <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- maps:get(cmd, Config, [])],
+        <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- field(Config, <<"cmd">>, [])],
-        <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- maps:get(entrypoint, Config, [])],
+        <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- field(Config, <<"entrypoint">>, [])],
-        <<"Env">> => [to_binary(EnvItem) || EnvItem <- maps:get(env, Config, [])],
+        <<"Env">> => [to_binary(EnvItem) || EnvItem <- field(Config, <<"env">>, [])],
-        <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(maps:get(labels, Config, #{}))]),
+        <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(field(Config, <<"labels">>, #{}))]),
-        <<"Volumes">> => build_volumes(maps:get(volumes, Config, [])),
+        <<"Volumes">> => build_volumes(field(Config, <<"volumes">>, [])),
-        <<"User">> => to_binary(maps:get(user, Config, <<>>)),
+        <<"User">> => to_binary(field(Config, <<"user">>, <<>>)),
-        <<"WorkingDir">> => to_binary(maps:get(working_dir, Config, <<>>)),
+        <<"WorkingDir">> => to_binary(field(Config, <<"working_dir">>, <<>>)),
-        <<"Hostname">> => to_binary(maps:get(hostname, Config, <<>>)),
+        <<"Hostname">> => to_binary(field(Config, <<"hostname">>, <<>>)),
-        <<"ExposedPorts">> => build_expose(maps:get(exposed_ports, Config, [])),
+        <<"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_binds(field(HostConfig, <<"binds">>, [])),
-            build_network_mode(maps:get(network_mode, HostConfig, <<>>)),
+            build_network_mode(field(HostConfig, <<"network_mode">>, <<>>)),
-            build_restart(maps:get(restart_policy, HostConfig, undefined)),
+            build_restart(field(HostConfig, <<"restart_policy">>, undefined)),
-            build_privileged(maps:get(privileged, HostConfig, false)),
+            build_privileged(field(HostConfig, <<"privileged">>, false)),
-            build_cap_add_drop(maps:get(cap_add, HostConfig, []), maps:get(cap_drop, HostConfig, [])),
+            build_cap_add_drop(field(HostConfig, <<"cap_add">>, []), field(HostConfig, <<"cap_drop">>, [])),
-            build_devices(maps:get(devices, HostConfig, [])),
+            build_devices(field(HostConfig, <<"devices">>, [])),
             build_resources(
-                maps:get(memory, HostConfig, 0),
+                field(HostConfig, <<"memory">>, 0),
-                maps:get(memory_reservation, HostConfig, 0),
+                field(HostConfig, <<"memory_reservation">>, 0),
-                maps:get(nano_cpus, HostConfig, 0),
+                field(HostConfig, <<"nano_cpus">>, 0),
-                maps:get(cpu_shares, HostConfig, 0)
+                field(HostConfig, <<"cpu_shares">>, 0)
             ),
-            build_ulimits(maps:get(ulimits, HostConfig, [])),
+            build_ulimits(field(HostConfig, <<"ulimits">>, [])),
-            build_tmpfs(maps:get(tmpfs, HostConfig, #{})),
+            build_tmpfs(field(HostConfig, <<"tmpfs">>, #{})),
-            build_sysctls(maps:get(sysctls, HostConfig, #{})),
+            build_sysctls(field(HostConfig, <<"sysctls">>, #{})),
-            build_extra_hosts(maps:get(extra_hosts, HostConfig, []))
+            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, [])],
+        <<"Test">> => [to_binary(Item) || Item <- field(Healthcheck, <<"test">>, [])],
-        <<"Interval">> => maps:get(interval_ns, Healthcheck, 0),
+        <<"Interval">> => field(Healthcheck, <<"interval_ns">>, 0),
-        <<"Timeout">> => maps:get(timeout_ns, Healthcheck, 0),
+        <<"Timeout">> => field(Healthcheck, <<"timeout_ns">>, 0),
-        <<"Retries">> => maps:get(retries, Healthcheck, 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_on_host">> := HostPath,
-                path_in_container := ContainerPath,
+                <<"path_in_container">> := ContainerPath,
-                cgroup_permissions := Permissions
+                <<"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;

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]),

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">>}).

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]),
+    try binary_to_term(PacketBin, [safe]) of
-    {keep_state, State, [{next_event, internal, Packet}]};
+        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}) ->
+handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    Reply = docker_commands:get_containers(),
+    handle_container_command(Ref, Request, Socket),
-    send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, 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}) ->
     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().
+-spec auth_packet(binary()) -> binary().
-auth_packet(Ref) when is_reference(Ref) ->
+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, #{
+    term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
-        uuid => list_to_binary(UUID),
+        <<"uuid">> => list_to_binary(UUID),
-        token => list_to_binary(Token),
+        <<"token">> => list_to_binary(Token),
-        timestamp => Timestamp
+        <<"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, <<>>)),
+    NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)),
-    IdBin = to_binary(maps:get(id, Target, <<>>)),
+    IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)),
     case NameBin of
         <<>> ->
             true = IdBin =/= <<>>,