fix

config/sys.config

@@ -22,9 +22,7 @@
 
     {auth, [
       {uuid, "qbxmjyzrkpntfgswaevodhluicqzxplkm"},
-      {username, "test"},
-      {salt, "salt2345"},
-      {token, "token124"}
+      {token, "zpxlkvmqwnbghytrujsdieofazxcvbnm"}
     ]}
 
   ]},

include/message_pb.hrl

@@ -1,277 +0,0 @@
-%% -*- coding: utf-8 -*-
-%% Automatically generated, do not edit
-%% Generated by gpb_compile version 4.21.7
-
--ifndef(message_pb).
--define(message_pb, true).
-
--define(message_pb_gpb_version, "4.21.7").
-
-
--ifndef('REQUESTFRAME_PB_H').
--define('REQUESTFRAME_PB_H', true).
--record('RequestFrame',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         body                   :: {auth_request, message_pb:'AuthRequest'()} | {container_request, message_pb:'ContainerRequest'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('AUTHREQUEST_PB_H').
--define('AUTHREQUEST_PB_H', true).
--record('AuthRequest',
-        {uuid = <<>>            :: iodata() | undefined, % = 1, optional
-         username = <<>>        :: iodata() | undefined, % = 2, optional
-         salt = <<>>            :: iodata() | undefined, % = 3, optional
-         token = <<>>           :: iodata() | undefined, % = 4, optional
-         timestamp = 0          :: integer() | undefined % = 5, optional, 32 bits
-        }).
--endif.
-
--ifndef('REPLYFRAME.ERROR_PB_H').
--define('REPLYFRAME.ERROR_PB_H', true).
--record('ReplyFrame.Error',
-        {code = 0               :: integer() | undefined, % = 1, optional, 32 bits
-         message = <<>>         :: unicode:chardata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('REPLYFRAME_PB_H').
--define('REPLYFRAME_PB_H', true).
--record('ReplyFrame',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         reply                  :: {result, iodata()} | {error, message_pb:'ReplyFrame.Error'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('CASTFRAME.PUB_PB_H').
--define('CASTFRAME.PUB_PB_H', true).
--record('CastFrame.Pub',
-        {topic = <<>>           :: iodata() | undefined, % = 1, optional
-         qos = 0                :: integer() | undefined, % = 2, optional, 32 bits
-         content = <<>>         :: iodata() | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('CASTFRAME.COMMAND_PB_H').
--define('CASTFRAME.COMMAND_PB_H', true).
--record('CastFrame.Command',
-        {command                :: {auth, message_pb:'CastFrame.Command.Authorization'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('CASTFRAME.COMMAND.AUTHORIZATION_PB_H').
--define('CASTFRAME.COMMAND.AUTHORIZATION_PB_H', true).
--record('CastFrame.Command.Authorization',
-        {cmd = 'DEACTIVATE'     :: 'DEACTIVATE' | 'ACTIVATE' | integer() | undefined % = 1, optional, enum CastFrame.Command.Authorization.CommandType
-        }).
--endif.
-
--ifndef('CASTFRAME.DATA_PB_H').
--define('CASTFRAME.DATA_PB_H', true).
--record('CastFrame.Data',
-        {route_key = <<>>       :: iodata() | undefined, % = 1, optional
-         metric = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('CASTFRAME.TASKEVENT_PB_H').
--define('CASTFRAME.TASKEVENT_PB_H', true).
--record('CastFrame.TaskEvent',
-        {task_id = 0            :: integer() | undefined, % = 1, optional, 32 bits
-         type = <<>>            :: iodata() | undefined, % = 2, optional
-         stream = <<>>          :: iodata() | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('CASTFRAME_PB_H').
--define('CASTFRAME_PB_H', true).
--record('CastFrame',
-        {body                   :: {pub, message_pb:'CastFrame.Pub'()} | {command, message_pb:'CastFrame.Command'()} | {data, message_pb:'CastFrame.Data'()} | {task_event, message_pb:'CastFrame.TaskEvent'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('CONTAINERREF_PB_H').
--define('CONTAINERREF_PB_H', true).
--record('ContainerRef',
-        {id = <<>>              :: unicode:chardata() | undefined, % = 1, optional
-         name = <<>>            :: unicode:chardata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST.LIST_PB_H').
--define('CONTAINERREQUEST.LIST_PB_H', true).
--record('ContainerRequest.List',
-        {all = false            :: boolean() | 0 | 1 | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST.DEPLOY_PB_H').
--define('CONTAINERREQUEST.DEPLOY_PB_H', true).
--record('ContainerRequest.Deploy',
-        {task_id = 0            :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         params = undefined     :: message_pb:'ContainerDeployParams'() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST.START_PB_H').
--define('CONTAINERREQUEST.START_PB_H', true).
--record('ContainerRequest.Start',
-        {target = undefined     :: message_pb:'ContainerRef'() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST.STOP_PB_H').
--define('CONTAINERREQUEST.STOP_PB_H', true).
--record('ContainerRequest.Stop',
-        {target = undefined     :: message_pb:'ContainerRef'() | undefined, % = 1, optional
-         timeout_seconds = 0    :: non_neg_integer() | undefined % = 2, optional, 32 bits
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST.KILL_PB_H').
--define('CONTAINERREQUEST.KILL_PB_H', true).
--record('ContainerRequest.Kill',
-        {target = undefined     :: message_pb:'ContainerRef'() | undefined, % = 1, optional
-         signal = <<>>          :: unicode:chardata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST.REMOVE_PB_H').
--define('CONTAINERREQUEST.REMOVE_PB_H', true).
--record('ContainerRequest.Remove',
-        {target = undefined     :: message_pb:'ContainerRef'() | undefined, % = 1, optional
-         force = false          :: boolean() | 0 | 1 | undefined, % = 2, optional
-         remove_volumes = false :: boolean() | 0 | 1 | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST.CONFIG_PB_H').
--define('CONTAINERREQUEST.CONFIG_PB_H', true).
--record('ContainerRequest.Config',
-        {target = undefined     :: message_pb:'ContainerRef'() | undefined, % = 1, optional
-         config = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('CONTAINERREQUEST_PB_H').
--define('CONTAINERREQUEST_PB_H', true).
--record('ContainerRequest',
-        {action                 :: {list, message_pb:'ContainerRequest.List'()} | {deploy, message_pb:'ContainerRequest.Deploy'()} | {start, message_pb:'ContainerRequest.Start'()} | {stop, message_pb:'ContainerRequest.Stop'()} | {kill, message_pb:'ContainerRequest.Kill'()} | {remove, message_pb:'ContainerRequest.Remove'()} | {config, message_pb:'ContainerRequest.Config'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('CONTAINERDEPLOYPARAMS_PB_H').
--define('CONTAINERDEPLOYPARAMS_PB_H', true).
--record('ContainerDeployParams',
-        {container_name = <<>>  :: unicode:chardata() | undefined, % = 1, optional
-         container_dir = <<>>   :: unicode:chardata() | undefined, % = 2, optional
-         create = undefined     :: message_pb:'DockerCreateOptions'() | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('DOCKERCREATEOPTIONS_PB_H').
--define('DOCKERCREATEOPTIONS_PB_H', true).
--record('DockerCreateOptions',
-        {config = undefined     :: message_pb:'DockerContainerConfig'() | undefined, % = 1, optional
-         host_config = undefined :: message_pb:'DockerHostConfig'() | undefined, % = 2, optional
-         networking_config = undefined :: message_pb:'DockerNetworkingConfig'() | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('DOCKERCONTAINERCONFIG_PB_H').
--define('DOCKERCONTAINERCONFIG_PB_H', true).
--record('DockerContainerConfig',
-        {image = <<>>           :: unicode:chardata() | undefined, % = 1, optional
-         cmd = []               :: [unicode:chardata()] | undefined, % = 2, repeated
-         entrypoint = []        :: [unicode:chardata()] | undefined, % = 3, repeated
-         env = []               :: [unicode:chardata()] | undefined, % = 4, repeated
-         labels = []            :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 5
-         volumes = []           :: [unicode:chardata()] | undefined, % = 6, repeated
-         user = <<>>            :: unicode:chardata() | undefined, % = 7, optional
-         working_dir = <<>>     :: unicode:chardata() | undefined, % = 8, optional
-         hostname = <<>>        :: unicode:chardata() | undefined, % = 9, optional
-         exposed_ports = []     :: [message_pb:'DockerExposedPort'()] | undefined, % = 10, repeated
-         healthcheck = undefined :: message_pb:'Healthcheck'() | undefined % = 11, optional
-        }).
--endif.
-
--ifndef('DOCKERHOSTCONFIG_PB_H').
--define('DOCKERHOSTCONFIG_PB_H', true).
--record('DockerHostConfig',
-        {binds = []             :: [unicode:chardata()] | undefined, % = 1, repeated
-         network_mode = <<>>    :: unicode:chardata() | undefined, % = 2, optional
-         restart_policy = undefined :: message_pb:'RestartPolicy'() | undefined, % = 3, optional
-         privileged = false     :: boolean() | 0 | 1 | undefined, % = 4, optional
-         cap_add = []           :: [unicode:chardata()] | undefined, % = 5, repeated
-         cap_drop = []          :: [unicode:chardata()] | undefined, % = 6, repeated
-         devices = []           :: [message_pb:'DeviceMapping'()] | undefined, % = 7, repeated
-         memory = 0             :: non_neg_integer() | undefined, % = 8, optional, 64 bits
-         memory_reservation = 0 :: non_neg_integer() | undefined, % = 9, optional, 64 bits
-         nano_cpus = 0          :: non_neg_integer() | undefined, % = 10, optional, 64 bits
-         cpu_shares = 0         :: non_neg_integer() | undefined, % = 11, optional, 64 bits
-         ulimits = []           :: [message_pb:'Ulimit'()] | undefined, % = 12, repeated
-         tmpfs = []             :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 13
-         sysctls = []           :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 14
-         extra_hosts = []       :: [unicode:chardata()] | undefined % = 15, repeated
-        }).
--endif.
-
--ifndef('DOCKERNETWORKINGCONFIG_PB_H').
--define('DOCKERNETWORKINGCONFIG_PB_H', true).
--record('DockerNetworkingConfig',
-        {endpoints = []         :: [message_pb:'DockerNetworkEndpoint'()] | undefined % = 1, repeated
-        }).
--endif.
-
--ifndef('DOCKERNETWORKENDPOINT_PB_H').
--define('DOCKERNETWORKENDPOINT_PB_H', true).
--record('DockerNetworkEndpoint',
-        {name = <<>>            :: unicode:chardata() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('DOCKEREXPOSEDPORT_PB_H').
--define('DOCKEREXPOSEDPORT_PB_H', true).
--record('DockerExposedPort',
-        {container_port = 0     :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         protocol = <<>>        :: unicode:chardata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('HEALTHCHECK_PB_H').
--define('HEALTHCHECK_PB_H', true).
--record('Healthcheck',
-        {test = []              :: [unicode:chardata()] | undefined, % = 1, repeated
-         interval_ns = 0        :: non_neg_integer() | undefined, % = 2, optional, 64 bits
-         timeout_ns = 0         :: non_neg_integer() | undefined, % = 3, optional, 64 bits
-         retries = 0            :: non_neg_integer() | undefined % = 4, optional, 32 bits
-        }).
--endif.
-
--ifndef('RESTARTPOLICY_PB_H').
--define('RESTARTPOLICY_PB_H', true).
--record('RestartPolicy',
-        {name = <<>>            :: unicode:chardata() | undefined, % = 1, optional
-         maximum_retry_count = 0 :: non_neg_integer() | undefined % = 2, optional, 32 bits
-        }).
--endif.
-
--ifndef('DEVICEMAPPING_PB_H').
--define('DEVICEMAPPING_PB_H', true).
--record('DeviceMapping',
-        {path_on_host = <<>>    :: unicode:chardata() | undefined, % = 1, optional
-         path_in_container = <<>> :: unicode:chardata() | undefined, % = 2, optional
-         cgroup_permissions = <<>> :: unicode:chardata() | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('ULIMIT_PB_H').
--define('ULIMIT_PB_H', true).
--record('Ulimit',
-        {name = <<>>            :: unicode:chardata() | undefined, % = 1, optional
-         soft = 0               :: non_neg_integer() | undefined, % = 2, optional, 64 bits
-         hard = 0               :: non_neg_integer() | undefined % = 3, optional, 64 bits
-        }).
--endif.
-
--endif.

rebar.config

@@ -10,7 +10,7 @@
 {gpb_opts, [
     {i, "../proto"},              % 共享 proto 文件路径
     {src_dirs, ["../proto"]},     % 源码目录（必须）
-    {f, ["message.proto", "service.proto"]},
+    {f, ["service.proto"]},
     recursive,                     % 递归查找 proto 文件
     {module_name_suffix, "_pb"},   % 生成模块后缀
     {o_erl, "src/protobuf"},                % .erl 输出目录

src/docker/docker_commands.erl

@@ -8,7 +8,6 @@
 %%%-------------------------------------------------------------------
 -module(docker_commands).
 -author("anlicheng").
--include("message_pb.hrl").
 
 %% API
 -export([pull_image/1, check_image_exist/1]).
@@ -34,10 +33,10 @@ check_image_exist(Image) when is_binary(Image) ->
             false
     end.
 
--spec create_container(ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) ->
+-spec create_container(ContainerDir :: string(), Params :: map()) ->
     {ok, ContainerId :: binary()} | {error, Reason :: any()}.
-create_container(ContainerDir, #'ContainerDeployParams'{container_name = ContainerName0, create = CreateOpts })
-    when is_list(ContainerDir), ContainerName0 =/= undefined ->
+create_container(ContainerDir, #{container_name := ContainerName0, create := CreateOpts})
+    when is_list(ContainerDir), is_binary(ContainerName0), is_map(CreateOpts) ->
     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),

src/docker/docker_container_builder.erl

@@ -8,136 +8,107 @@
 %%%-------------------------------------------------------------------
 -module(docker_container_builder).
 -author("anlicheng").
--include("message_pb.hrl").
 
 %% API
 -export([build_options/3]).
 
--spec build_options(ContainerName :: binary(), ContainerDir :: string(),
-    Create0 :: message_pb:'DockerCreateOptions'() | undefined) -> map().
+-spec build_options(binary(), string(), map() | undefined) -> map().
 build_options(ContainerName, ContainerDir, Create0) when is_binary(ContainerName), is_list(ContainerDir) ->
     ConfigFile = list_to_binary(docker_helper:get_config_file(ContainerDir)),
     Create = patch_create_options(ContainerName, ConfigFile, Create0),
     build_create_options(Create).
 
--spec patch_create_options(ContainerName :: binary(), ConfigFile :: binary(),
-    Create0 :: message_pb:'DockerCreateOptions'() | undefined) -> message_pb:'DockerCreateOptions'().
+-spec patch_create_options(binary(), binary(), map() | undefined) -> map().
 patch_create_options(ContainerName, ConfigFile, undefined) ->
-    patch_create_options(ContainerName, ConfigFile, #'DockerCreateOptions'{});
-patch_create_options(ContainerName, ConfigFile, Create = #'DockerCreateOptions'{
-    config = Config0,
-    host_config = HostConfig0
-}) when is_binary(ContainerName), is_binary(ConfigFile) ->
-    Config = patch_container_config(ContainerName, ensure_container_config(Config0)),
-    HostConfig = patch_host_config(ConfigFile, ensure_host_config(HostConfig0)),
-    Create#'DockerCreateOptions'{config = Config, host_config = HostConfig}.
+    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))),
+    Create0#{
+        config => Config,
+        host_config => HostConfig
+    }.
 
--spec patch_container_config(ContainerName :: binary(),
-    Config :: message_pb:'DockerContainerConfig'()) -> message_pb:'DockerContainerConfig'().
-patch_container_config(ContainerName, Config = #'DockerContainerConfig'{env = Env0, volumes = Volumes0})
-  when is_binary(ContainerName) ->
+-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, [])],
     ConfigVolume = <<"/usr/local/etc/service.conf">>,
-    Envs0 = [<<"CONTAINER_NAME=", ContainerName/binary>>],
-    Envs = add_unique_front(Envs0, [to_binary(EnvItem) || EnvItem <- Env0]),
-    Volumes = add_unique_front([ConfigVolume], [to_binary(Volume) || Volume <- Volumes0]),
-    Config#'DockerContainerConfig'{env = Envs, volumes = Volumes}.
+    Envs = add_unique_front([<<"CONTAINER_NAME=", ContainerName/binary>>], Env0),
+    Volumes = add_unique_front([ConfigVolume], Volumes0),
+    Config0#{
+        env => Envs,
+        volumes => Volumes
+    }.
 
--spec patch_host_config(ConfigFile :: binary(),
-    HostConfig :: message_pb:'DockerHostConfig'()) -> message_pb:'DockerHostConfig'().
-patch_host_config(ConfigFile, HostConfig = #'DockerHostConfig'{binds = Binds0}) when is_binary(ConfigFile) ->
+-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, [])],
     ConfigBind = <<ConfigFile/binary, ":/usr/local/etc/service.conf">>,
-    Binds = add_unique_front([ConfigBind], [to_binary(Bind) || Bind <- Binds0]),
-    HostConfig#'DockerHostConfig'{binds = Binds}.
+    HostConfig0#{binds => add_unique_front([ConfigBind], Binds0)}.
 
--spec ensure_container_config(Config :: message_pb:'DockerContainerConfig'() | undefined) ->
-    message_pb:'DockerContainerConfig'().
+-spec ensure_container_config(map() | undefined) -> map().
 ensure_container_config(undefined) ->
-    #'DockerContainerConfig'{};
-ensure_container_config(Config = #'DockerContainerConfig'{}) ->
+    #{};
+ensure_container_config(Config) when is_map(Config) ->
     Config.
 
--spec ensure_host_config(HostConfig :: message_pb:'DockerHostConfig'() | undefined) ->
-    message_pb:'DockerHostConfig'().
+-spec ensure_host_config(map() | undefined) -> map().
 ensure_host_config(undefined) ->
-    #'DockerHostConfig'{};
-ensure_host_config(HostConfig = #'DockerHostConfig'{}) ->
+    #{};
+ensure_host_config(HostConfig) when is_map(HostConfig) ->
     HostConfig.
 
--spec build_create_options(Create :: message_pb:'DockerCreateOptions'()) -> map().
-build_create_options(#'DockerCreateOptions'{
-    config = #'DockerContainerConfig'{
-        image = Image,
-        cmd = Cmd,
-        entrypoint = Entrypoint,
-        env = Env,
-        labels = Labels,
-        volumes = Volumes,
-        user = User,
-        working_dir = WorkingDir,
-        hostname = Hostname,
-        exposed_ports = ExposedPorts,
-        healthcheck = Healthcheck
-    },
-    host_config = #'DockerHostConfig'{
-        binds = Binds,
-        network_mode = NetworkMode,
-        restart_policy = RestartPolicy,
-        privileged = Privileged,
-        cap_add = CapAdd,
-        cap_drop = CapDrop,
-        devices = Devices,
-        memory = Memory,
-        memory_reservation = MemoryReservation,
-        nano_cpus = NanoCpus,
-        cpu_shares = CpuShares,
-        ulimits = Ulimits,
-        tmpfs = Tmpfs,
-        sysctls = Sysctls,
-        extra_hosts = ExtraHosts
-    },
-    networking_config = NetworkingConfig
-}) ->
-    Endpoints = networking_config_endpoints(NetworkingConfig),
+-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)),
     #{
-        <<"Image">> => to_binary(Image),
-        <<"Cmd">> => [to_binary(CommandItem) || CommandItem <- Cmd],
-        <<"Entrypoint">> => [to_binary(EntrypointItem) || EntrypointItem <- Entrypoint],
-        <<"Env">> => [to_binary(EnvItem) || EnvItem <- Env],
-        <<"Labels">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- Labels]),
-        <<"Volumes">> => build_volumes(Volumes),
-        <<"User">> => to_binary(User),
-        <<"WorkingDir">> => to_binary(WorkingDir),
-        <<"Hostname">> => to_binary(Hostname),
-        <<"ExposedPorts">> => build_expose(ExposedPorts),
+        <<"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, [])),
         <<"NetworkingConfig">> => build_networking_config(Endpoints),
-        <<"Healthcheck">> => build_healthcheck(Healthcheck),
+        <<"Healthcheck">> => build_healthcheck(maps:get(healthcheck, Config, undefined)),
         <<"HostConfig">> => fold_merge([
-            build_binds(Binds),
-            build_network_mode(NetworkMode),
-            build_restart(RestartPolicy),
-            build_privileged(Privileged),
-            build_cap_add_drop(CapAdd, CapDrop),
-            build_devices(Devices),
-            build_resources(Memory, MemoryReservation, NanoCpus, CpuShares),
-            build_ulimits(Ulimits),
-            build_tmpfs(Tmpfs),
-            build_sysctls(Sysctls),
-            build_extra_hosts(ExtraHosts)
+            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_resources(
+                maps:get(memory, HostConfig, 0),
+                maps:get(memory_reservation, HostConfig, 0),
+                maps:get(nano_cpus, HostConfig, 0),
+                maps:get(cpu_shares, HostConfig, 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, []))
         ])
     }.
 
--spec networking_config_endpoints(NetworkingConfig :: message_pb:'DockerNetworkingConfig'() | undefined) ->
-    [message_pb:'DockerNetworkEndpoint'()].
+-spec networking_config_endpoints(map() | undefined) -> [map()].
 networking_config_endpoints(undefined) ->
     [];
-networking_config_endpoints(#'DockerNetworkingConfig'{endpoints = Endpoints}) ->
-    Endpoints.
+networking_config_endpoints(NetworkingConfig) when is_map(NetworkingConfig) ->
+    maps:get(endpoints, NetworkingConfig, []).
 
--spec fold_merge(List :: [map()]) -> map().
+-spec fold_merge([map()]) -> map().
 fold_merge(List) ->
     lists:foldl(fun maps:merge/2, #{}, List).
 
--spec build_expose(Ports :: [message_pb:'DockerExposedPort'()]) -> map().
+-spec build_expose([map()]) -> map().
 build_expose(Ports) when is_list(Ports) ->
     case Ports of
         [] ->
@@ -146,7 +117,7 @@ build_expose(Ports) when is_list(Ports) ->
             maps:from_list([{normalize_expose_port(Port), #{}} || Port <- Ports])
     end.
 
--spec build_volumes(Volumes :: [binary() | list() | atom()]) -> map().
+-spec build_volumes([binary() | list() | atom()]) -> map().
 build_volumes(Volumes) when is_list(Volumes) ->
     case Volumes of
         [] ->
@@ -155,7 +126,7 @@ build_volumes(Volumes) when is_list(Volumes) ->
             maps:from_list([{to_binary(ContainerPath), #{}} || ContainerPath <- Volumes])
     end.
 
--spec build_binds(Binds :: [binary() | list() | atom()]) -> map().
+-spec build_binds([binary() | list() | atom()]) -> map().
 build_binds(Binds) when is_list(Binds) ->
     case Binds of
         [] ->
@@ -164,58 +135,50 @@ build_binds(Binds) when is_list(Binds) ->
             #{<<"Binds">> => [to_binary(Bind) || Bind <- Binds]}
     end.
 
--spec build_networking_config(Endpoints :: [message_pb:'DockerNetworkEndpoint'()]) -> map().
+-spec build_networking_config([map()]) -> map().
 build_networking_config(Endpoints) when is_list(Endpoints) ->
     case Endpoints of
         [] ->
             #{};
         _ ->
-            NetCfg = maps:from_list([{to_binary(Name), #{}} || #'DockerNetworkEndpoint'{name = Name} <- Endpoints]),
+            NetCfg = maps:from_list([{to_binary(Name), #{}} || #{name := Name} <- Endpoints]),
             #{<<"EndpointsConfig">> => NetCfg}
     end.
 
--spec build_network_mode(NetworkMode :: binary() | list() | atom()) -> map().
+-spec build_network_mode(binary() | list() | atom()) -> map().
 build_network_mode(<<>>) ->
     #{};
 build_network_mode(NetworkMode) ->
     #{<<"NetworkMode">> => to_binary(NetworkMode)}.
 
--spec build_healthcheck(Healthcheck :: message_pb:'Healthcheck'() | undefined) -> map().
+-spec build_healthcheck(map() | undefined) -> map().
 build_healthcheck(undefined) ->
     #{};
-build_healthcheck(#'Healthcheck'{
-    test = Test,
-    interval_ns = IntervalNs,
-    timeout_ns = TimeoutNs,
-    retries = Retries
-}) ->
+build_healthcheck(Healthcheck) when is_map(Healthcheck) ->
     #{
-        <<"Test">> => [to_binary(Item) || Item <- Test],
-        <<"Interval">> => IntervalNs,
-        <<"Timeout">> => TimeoutNs,
-        <<"Retries">> => Retries
+        <<"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)
     }.
 
--spec build_restart(RestartPolicy :: message_pb:'RestartPolicy'() | undefined) -> map().
+-spec build_restart(map() | undefined) -> map().
 build_restart(undefined) ->
     #{};
-build_restart(#'RestartPolicy'{
-    name = Name,
-    maximum_retry_count = RetryCount
-}) ->
+build_restart(RestartPolicy0) when is_map(RestartPolicy0) ->
     RestartPolicy = #{
-        <<"Name">> => to_binary(Name)
+        <<"Name">> => to_binary(maps:get(name, RestartPolicy0, <<>>))
     },
-    case RetryCount of
+    case maps:get(maximum_retry_count, RestartPolicy0, 0) of
         0 ->
             #{<<"RestartPolicy">> => RestartPolicy};
-        _ ->
+        RetryCount ->
             #{<<"RestartPolicy">> => RestartPolicy#{
                 <<"MaximumRetryCount">> => RetryCount
             }}
     end.
 
--spec build_privileged(Privileged :: any()) -> map().
+-spec build_privileged(any()) -> map().
 build_privileged(Privileged) ->
     case to_bool(Privileged) of
         true ->
@@ -224,7 +187,7 @@ build_privileged(Privileged) ->
             #{}
     end.
 
--spec build_cap_add_drop(Add :: [binary() | list() | atom()], Drop :: [binary() | list() | atom()]) -> map().
+-spec build_cap_add_drop([binary() | list() | atom()], [binary() | list() | atom()]) -> map().
 build_cap_add_drop(Add, Drop) when is_list(Add), is_list(Drop) ->
     case {Add, Drop} of
         {[], []} ->
@@ -236,7 +199,7 @@ build_cap_add_drop(Add, Drop) when is_list(Add), is_list(Drop) ->
             }
     end.
 
--spec build_devices(Devices :: [message_pb:'DeviceMapping'()]) -> map().
+-spec build_devices([map()]) -> map().
 build_devices(Devices) when is_list(Devices) ->
     case Devices of
         [] ->
@@ -246,16 +209,15 @@ build_devices(Devices) when is_list(Devices) ->
                 <<"PathOnHost">> => to_binary(HostPath),
                 <<"PathInContainer">> => to_binary(ContainerPath),
                 <<"CgroupPermissions">> => device_permissions(Permissions)
-            } || #'DeviceMapping'{
-                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.
 
--spec build_resources(MemoryBytes :: integer(), ReservationBytes :: integer(),
-    NanoCpus :: integer(), CpuShares :: integer()) -> map().
+-spec build_resources(integer(), integer(), integer(), integer()) -> map().
 build_resources(MemoryBytes, ReservationBytes, NanoCpus, CpuShares) ->
     HostConfig0 = #{},
     HostConfig1 = case MemoryBytes of
@@ -283,7 +245,7 @@ build_resources(MemoryBytes, ReservationBytes, NanoCpus, CpuShares) ->
             HostConfig3#{<<"CpuShares">> => CpuShares}
     end.
 
--spec build_ulimits(Ulimits :: [message_pb:'Ulimit'()]) -> map().
+-spec build_ulimits([map()]) -> map().
 build_ulimits(Ulimits) when is_list(Ulimits) ->
     case Ulimits of
         [] ->
@@ -293,28 +255,28 @@ build_ulimits(Ulimits) when is_list(Ulimits) ->
                 <<"Name">> => to_binary(Name),
                 <<"Soft">> => Soft,
                 <<"Hard">> => Hard
-            } || #'Ulimit'{name = Name, soft = Soft, hard = Hard} <- Ulimits]}
+            } || #{name := Name, soft := Soft, hard := Hard} <- Ulimits]}
     end.
 
--spec build_sysctls(Sysctls :: [{binary() | list() | atom(), binary() | list() | atom()}]) -> map().
-build_sysctls(Sysctls) when is_list(Sysctls) ->
-    case Sysctls of
-        [] ->
+-spec build_sysctls(map()) -> map().
+build_sysctls(Sysctls) when is_map(Sysctls) ->
+    case maps:size(Sysctls) of
+        0 ->
             #{};
         _ ->
-            #{<<"Sysctls">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- Sysctls])}
+            #{<<"Sysctls">> => maps:from_list([{to_binary(Key), to_binary(Value)} || {Key, Value} <- maps:to_list(Sysctls)])}
     end.
 
--spec build_tmpfs(Tmpfs :: [{binary() | list() | atom(), binary() | list() | atom()}]) -> map().
-build_tmpfs(Tmpfs) when is_list(Tmpfs) ->
-    case Tmpfs of
-        [] ->
+-spec build_tmpfs(map()) -> map().
+build_tmpfs(Tmpfs) when is_map(Tmpfs) ->
+    case maps:size(Tmpfs) of
+        0 ->
             #{};
         _ ->
-            #{<<"Tmpfs">> => maps:from_list([{to_binary(Path), to_binary(Options)} || {Path, Options} <- Tmpfs])}
+            #{<<"Tmpfs">> => maps:from_list([{to_binary(Path), to_binary(Options)} || {Path, Options} <- maps:to_list(Tmpfs)])}
     end.
 
--spec build_extra_hosts(Hosts :: [binary() | list() | atom()]) -> map().
+-spec build_extra_hosts([binary() | list() | atom()]) -> map().
 build_extra_hosts(Hosts) when is_list(Hosts) ->
     case Hosts of
         [] ->
@@ -323,8 +285,8 @@ build_extra_hosts(Hosts) when is_list(Hosts) ->
             #{<<"ExtraHosts">> => [to_binary(Host) || Host <- Hosts]}
     end.
 
--spec normalize_expose_port(Port :: message_pb:'DockerExposedPort'()) -> binary().
-normalize_expose_port(#'DockerExposedPort'{container_port = Port, protocol = Protocol}) ->
+-spec normalize_expose_port(map()) -> binary().
+normalize_expose_port(#{container_port := Port, protocol := Protocol}) ->
     PortBin = integer_to_binary(Port),
     ProtocolBin = to_binary(Protocol),
     case ProtocolBin of
@@ -336,13 +298,13 @@ normalize_expose_port(#'DockerExposedPort'{container_port = Port, protocol = Pro
             <<PortBin/binary, "/", ProtocolBin/binary>>
     end.
 
--spec device_permissions(Permissions :: binary() | list() | atom()) -> binary().
+-spec device_permissions(binary() | list() | atom()) -> binary().
 device_permissions(<<>>) ->
     <<"rwm">>;
 device_permissions(Permissions) ->
     to_binary(Permissions).
 
--spec add_unique_front(Items :: [binary()], List :: [binary()]) -> [binary()].
+-spec add_unique_front([binary()], [binary()]) -> [binary()].
 add_unique_front([], List) ->
     List;
 add_unique_front([Item | Rest], List) ->
@@ -354,7 +316,7 @@ add_unique_front([Item | Rest], List) ->
     end,
     add_unique_front(Rest, NList).
 
--spec to_binary(Value :: binary() | list() | atom() | any()) -> binary().
+-spec to_binary(binary() | list() | atom() | any()) -> binary().
 to_binary(Value) when is_binary(Value) ->
     Value;
 to_binary(Value) when is_list(Value) ->
@@ -364,7 +326,7 @@ to_binary(Value) when is_atom(Value) ->
 to_binary(Value) ->
     iolist_to_binary(io_lib:format("~p", [Value])).
 
--spec to_bool(Value :: true | false | 0 | 1 | undefined) -> boolean().
+-spec to_bool(true | false | 0 | 1 | undefined) -> boolean().
 to_bool(true) ->
     true;
 to_bool(1) ->

src/docker/docker_container_service.erl

@@ -8,41 +8,40 @@
 %%%-------------------------------------------------------------------
 -module(docker_container_service).
 -author("anlicheng").
--include("message_pb.hrl").
 
 %% API
 -export([handle_request/1]).
 
--spec handle_request(message_pb:'ContainerRequest'()) -> ok | {ok, binary()} | {error, binary()}.
-handle_request(#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}) ->
+-spec handle_request(map()) -> ok | {ok, binary()} | {error, binary()}.
+handle_request(#{action := list}) ->
     case docker_commands:get_containers() of
         {ok, Containers} ->
             {ok, iolist_to_binary(json:encode(Containers))};
         {error, Reason} when is_binary(Reason) ->
             {error, Reason}
     end;
-handle_request(#'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{task_id = TaskId, params = Params}}}) ->
+handle_request(#{action := deploy, task_id := TaskId, params := Params}) ->
     docker_deploy_manager:deploy(TaskId, Params);
-handle_request(#'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = Target}}}) ->
+handle_request(#{action := start, target := Target}) ->
     ContainerTarget = container_target(Target),
     docker_commands:start_container(ContainerTarget);
-handle_request(#'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{target = Target, timeout_seconds = TimeoutSeconds}}}) ->
+handle_request(#{action := stop, target := Target, timeout_seconds := TimeoutSeconds}) ->
     ContainerTarget = container_target(Target),
     docker_commands:stop_container(ContainerTarget, TimeoutSeconds);
-handle_request(#'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{target = Target, signal = Signal}}}) ->
+handle_request(#{action := kill, target := Target, signal := Signal}) ->
     ContainerTarget = container_target(Target),
     docker_commands:kill_container(ContainerTarget, to_binary(Signal));
-handle_request(#'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}) ->
+handle_request(#{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}) ->
     ContainerTarget = container_target(Target),
     docker_commands:remove_container(ContainerTarget, to_bool(Force), to_bool(RemoveVolumes));
-handle_request(#'ContainerRequest'{action = {config, #'ContainerRequest.Config'{target = Target, config = Config}}}) ->
+handle_request(#{action := config, target := Target, config := Config}) ->
     ContainerTarget = container_target(Target),
     update_container_config(ContainerTarget, iolist_to_binary(Config)).
 
--spec container_target(message_pb:'ContainerRef'()) -> binary().
-container_target(#'ContainerRef'{name = Name, id = Id}) ->
-    NameBin = to_binary(Name),
-    IdBin = to_binary(Id),
+-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, <<>>)),
     case NameBin of
         <<>> ->
             true = IdBin =/= <<>>,

src/docker/docker_deploy_manager.erl

@@ -8,7 +8,6 @@
 %%%-------------------------------------------------------------------
 -module(docker_deploy_manager).
 -author("anlicheng").
--include("message_pb.hrl").
 
 -behaviour(gen_server).
 
@@ -33,8 +32,8 @@
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
--spec deploy(integer(), message_pb:'ContainerDeployParams'()) -> ok | {error, binary()}.
-deploy(TaskId, Params) when is_integer(TaskId), is_record(Params, 'ContainerDeployParams') ->
+-spec deploy(integer(), map()) -> ok | {error, binary()}.
+deploy(TaskId, Params) when is_integer(TaskId), is_map(Params) ->
     gen_server:call(?SERVER, {deploy, TaskId, Params}).
 
 %%%===================================================================
@@ -48,10 +47,10 @@ init([]) ->
     {ok, #state{root_dir = RootDir}}.
 
 -spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.
-handle_call({deploy, TaskId, Params = #'ContainerDeployParams'{
-    container_name = ContainerName,
-    container_dir = ContainerDir0
-}}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) ->
+handle_call({deploy, TaskId, Params}, _From, State = #state{root_dir = RootDir, task_map = TaskMap})
+  when is_map(Params) ->
+    ContainerName = maps:get(container_name, Params),
+    ContainerDir0 = maps:get(container_dir, Params, <<>>),
     {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName, ContainerDir0),
     {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

@@ -8,7 +8,6 @@
 %%%-------------------------------------------------------------------
 -module(docker_deployer).
 -author("anlicheng").
--include("message_pb.hrl").
 -dialyzer([{nowarn_function, normalize_image/1}]).
 
 %% API
@@ -22,10 +21,10 @@
 %%% API
 %%%===================================================================
 
--spec(start_monitor(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) ->
+-spec(start_monitor(TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
     {ok, {pid(), reference()}}).
 start_monitor(TaskId, ContainerDir, Params)
-  when is_integer(TaskId), is_list(ContainerDir), is_record(Params, 'ContainerDeployParams') ->
+  when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
     {ok, spawn_monitor(?MODULE, deploy, [TaskId, ContainerDir, Params])}.
 
 %%%===================================================================
@@ -42,8 +41,8 @@ start_monitor(TaskId, ContainerDir, Params)
 %   "command": ["nginx", "-g", "daemon off;"],
 %   "restart": "always"
 %}
--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') ->
+-spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: map()) -> ok.
+deploy(TaskId, ContainerDir, Params) when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
     ContainerName = deploy_container_name(Params),
     Image0 = deploy_image(Params),
     report_task_event(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
@@ -150,10 +149,10 @@ await_pull_image(TaskId, Ref, Pid, MRef) ->
             throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>})
     end.
 
--spec create_container_and_config(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) ->
+-spec create_container_and_config(TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
     {ok, binary()}.
 create_container_and_config(TaskId, ContainerDir, Params)
-  when is_integer(TaskId), is_list(ContainerDir), is_record(Params, 'ContainerDeployParams') ->
+  when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
     case docker_commands:create_container(ContainerDir, Params) of
         {ok, ContainerId} ->
             ok = create_config_file(TaskId, ContainerDir),
@@ -199,15 +198,14 @@ short_container_id(ContainerId) when is_binary(ContainerId), byte_size(Container
 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 ->
+-spec deploy_container_name(map()) -> binary().
+deploy_container_name(#{container_name := ContainerName}) when is_binary(ContainerName) ->
     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 ->
+-spec deploy_image(map()) -> binary().
+deploy_image(#{create := #{config := #{image := Image}}}) when is_binary(Image) ->
     to_binary(Image);
 deploy_image(_) ->
     throw({deploy_error, <<"invalid deploy params: image missing">>}).

src/tests/docker_commands_tests.erl

@@ -8,7 +8,6 @@
 %%%-------------------------------------------------------------------
 -module(docker_commands_tests).
 -author("anlicheng").
--include("message_pb.hrl").
 
 %% API
 -export([
@@ -107,20 +106,20 @@ test_create_container_without_create_options() ->
 test_create_container_patches_options() ->
     Name = test_container_name(<<"create-patch">>),
     ContainerDir = prepare_container_dir(Name),
-    Create = #'DockerCreateOptions'{
-        config = #'DockerContainerConfig'{
-            image = ?TEST_IMAGE,
-            cmd = ?TEST_CMD,
-            env = [<<"EXISTING_ENV=1">>],
-            volumes = [<<"/data">>]
+    Create = #{
+        config => #{
+            image => ?TEST_IMAGE,
+            cmd => ?TEST_CMD,
+            env => [<<"EXISTING_ENV=1">>],
+            volumes => [<<"/data">>]
         },
-        host_config = #'DockerHostConfig'{
-            binds = [<<"/tmp:/tmp">>]
+        host_config => #{
+            binds => [<<"/tmp:/tmp">>]
         }
     },
-    Params = #'ContainerDeployParams'{
-        container_name = Name,
-        create = Create
+    Params = #{
+        container_name => Name,
+        create => Create
     },
     Options = docker_container_builder:build_options(Name, ContainerDir, Create),
     assert_patched_default_options(Name, ContainerDir, Options),
@@ -254,14 +253,14 @@ with_started_container(Name, Fun, Cleanup) when is_binary(Name), is_function(Fun
         ok = Fun(ContainerDir, ContainerId)
     end, Cleanup).
 
--spec minimal_params(binary()) -> message_pb:'ContainerDeployParams'().
+-spec minimal_params(binary()) -> map().
 minimal_params(Name) when is_binary(Name) ->
-    #'ContainerDeployParams'{
-        container_name = Name,
-        create = #'DockerCreateOptions'{
-            config = #'DockerContainerConfig'{
-                image = ?TEST_IMAGE,
-                cmd = ?TEST_CMD
+    #{
+        container_name => Name,
+        create => #{
+            config => #{
+                image => ?TEST_IMAGE,
+                cmd => ?TEST_CMD
             }
         }
     }.

src/transport/cloud_wire.erl

@@ -0,0 +1,81 @@
+-module(cloud_wire).
+
+-export([encode/1, decode/1]).
+
+-spec encode(term()) -> binary().
+encode(Term) ->
+    term_to_binary(validate(Term)).
+
+-spec decode(binary()) -> term().
+decode(Bin) when is_binary(Bin) ->
+    validate(binary_to_term(Bin, [safe])).
+
+-spec validate(term()) -> term().
+validate({request, PacketId, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp} = Auth}})
+  when is_integer(PacketId), PacketId > 0, is_binary(UUID), is_binary(Token), is_integer(Timestamp) ->
+    {request, PacketId, {auth_request, Auth}};
+validate({request, PacketId, {container_request, Request}})
+  when is_integer(PacketId), PacketId > 0 ->
+    {request, PacketId, {container_request, validate_container_request(Request)}};
+validate({response, PacketId, {ok, Result}})
+  when is_integer(PacketId), PacketId > 0, is_binary(Result) ->
+    {response, PacketId, {ok, Result}};
+validate({response, PacketId, {error, Code, Message}})
+  when is_integer(PacketId), PacketId > 0, is_integer(Code), is_binary(Message) ->
+    {response, PacketId, {error, Code, Message}};
+validate({message, {data, #{route_key := RouteKey, metric := Metric} = Body}})
+  when is_binary(RouteKey), is_binary(Metric) ->
+    {message, {data, Body}};
+validate({message, {task_event, #{task_id := TaskId, type := Type, stream := Stream} = Body}})
+  when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
+    {message, {task_event, Body}};
+validate({message, {pub, #{topic := Topic, qos := Qos, content := Content} = Body}})
+  when is_binary(Topic), is_integer(Qos), is_binary(Content) ->
+    {message, {pub, Body}};
+validate({message, {auth_control, Command}})
+  when Command =:= activate; Command =:= deactivate ->
+    {message, {auth_control, Command}};
+validate(Term) ->
+    error({invalid_wire_term, Term}).
+
+-spec validate_container_request(term()) -> map().
+validate_container_request(#{action := list} = Request) ->
+    true = is_boolean(maps:get(all, Request, true)),
+    Request;
+validate_container_request(#{action := deploy, task_id := TaskId, params := Params} = Request)
+  when is_integer(TaskId), is_map(Params) ->
+    true = is_binary(maps:get(container_name, Params, <<>>)),
+    true = is_binary(maps:get(container_dir, Params, <<>>)),
+    true = is_map(maps:get(create, Params, #{})),
+    Request;
+validate_container_request(#{action := start, target := Target} = Request) ->
+    validate_target(Target),
+    Request;
+validate_container_request(#{action := stop, target := Target, timeout_seconds := TimeoutSeconds} = Request)
+  when is_integer(TimeoutSeconds), TimeoutSeconds >= 0 ->
+    validate_target(Target),
+    Request;
+validate_container_request(#{action := kill, target := Target, signal := Signal} = Request)
+  when is_binary(Signal) ->
+    validate_target(Target),
+    Request;
+validate_container_request(#{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes} = Request)
+  when is_boolean(Force), is_boolean(RemoveVolumes) ->
+    validate_target(Target),
+    Request;
+validate_container_request(#{action := config, target := Target, config := Config} = Request)
+  when is_binary(Config) ->
+    validate_target(Target),
+    Request;
+validate_container_request(Request) ->
+    error({invalid_container_request, Request}).
+
+-spec validate_target(term()) -> ok.
+validate_target(#{name := Name, id := Id}) when is_binary(Name), is_binary(Id) ->
+    ok;
+validate_target(#{name := Name}) when is_binary(Name) ->
+    ok;
+validate_target(#{id := Id}) when is_binary(Id) ->
+    ok;
+validate_target(Target) ->
+    error({invalid_container_target, Target}).

src/transport/efka_client.erl

@@ -8,8 +8,6 @@
 %%%-------------------------------------------------------------------
 -module(efka_client).
 -author("anlicheng").
--include("protocol.hrl").
--include("message_pb.hrl").
 -include("efka_tables.hrl").
 
 -behaviour(gen_statem).
@@ -97,32 +95,26 @@ callback_mode() ->
 %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
 -spec handle_event(term(), term(), atom(), #state{}) -> term().
 handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
-    CastFrame = message_pb:encode_msg(#'CastFrame'{
-        body = {data, #'CastFrame.Data'{route_key = RouteKey, metric = Metric}}
-    }),
+    Packet = cloud_wire:encode({message, {data, #{route_key => RouteKey, metric => Metric}}}),
     case StateName of
         ?STATE_ACTIVATED ->
-            send_packet(Socket, [?FRAME_CAST, CastFrame]),
+            send_packet(Socket, Packet),
             {keep_state, State};
         _ ->
-            {ok, DroppedCount} = cache_insert(<<?FRAME_CAST, CastFrame/binary>>),
+            {ok, DroppedCount} = cache_insert(Packet),
             {keep_state, State#state{dropped_message_count = State#state.dropped_message_count + DroppedCount}}
     end;
 
 %% 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]),
-    EventPacket = message_pb:encode_msg(#'CastFrame'{
-        body = {task_event, #'CastFrame.TaskEvent'{task_id = TaskId, type = Type, stream = Stream}}
-    }),
-    send_packet(Socket, [?FRAME_CAST, EventPacket]),
+    Packet = cloud_wire:encode({message, {task_event, #{task_id => TaskId, type => Type, stream => Stream}}}),
+    send_packet(Socket, Packet),
     {keep_state, State};
 
 handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    EventPacket = message_pb:encode_msg(#'CastFrame'{
-        body = {task_event, #'CastFrame.TaskEvent'{task_id = TaskId, type = <<"close">>, stream = Reason}}
-    }),
-    send_packet(Socket, [?FRAME_CAST, EventPacket]),
+    Packet = cloud_wire:encode({message, {task_event, #{task_id => TaskId, type => <<"close">>, stream => Reason}}}),
+    send_packet(Socket, Packet),
     {keep_state, State};
 
 %% 其他情况下直接忽略
@@ -141,7 +133,7 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State =
     case connect_socket() of
         {ok, Socket} ->
             AuthPacket = auth_packet(PacketId),
-            send_packet(Socket, [?FRAME_REQUEST, AuthPacket]),
+            send_packet(Socket, 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} ->
@@ -170,15 +162,9 @@ handle_event(info, flush_cache, _, State) ->
     {keep_state, State};
 
 %% 处理收到的ssl消息
-handle_event(info, {ssl, Socket, <<?FRAME_REPLY, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
-    ReplyFrame = message_pb:decode_msg(PacketBin, 'ReplyFrame'),
-    {keep_state, State, [{next_event, internal, {decoded_reply, ReplyFrame}}]};
-handle_event(info, {ssl, Socket, <<?FRAME_REQUEST, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
-    RequestFrame = message_pb:decode_msg(PacketBin, 'RequestFrame'),
-    {keep_state, State, [{next_event, internal, {decoded_request, RequestFrame}}]};
-handle_event(info, {ssl, Socket, <<?FRAME_CAST, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
-    CastFrame = message_pb:decode_msg(PacketBin, 'CastFrame'),
-    {keep_state, State, [{next_event, internal, {decoded_cast, CastFrame}}]};
+handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) ->
+    Packet = cloud_wire:decode(PacketBin),
+    {keep_state, State, [{next_event, internal, {decoded_packet, Packet}}]};
 handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) ->
     logger:debug("[efka_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
@@ -191,7 +177,7 @@ handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
 %%% 处理内部消息，ssl收到的消息会解析成protobuf的消息格式，并按照internal类型处理
 
 %% 微服务部署
-handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}},
+handle_event(internal, {decoded_packet, {request, PacketId, {container_request, Request}}},
     ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     case docker_container_service:handle_request(Request) of
         ok ->
@@ -202,53 +188,53 @@ handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, b
             send_error_reply(Socket, PacketId, Reason)
     end,
     {keep_state, State};
-handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
+handle_event(internal, {decoded_packet, {request, PacketId, _Body}},
     ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
     send_error_reply(Socket, PacketId, <<"agent restricted">>),
     {keep_state, State};
-handle_event(internal, {decoded_request, #'RequestFrame'{packet_id = PacketId, body = _Body}},
+handle_event(internal, {decoded_packet, {request, PacketId, _Body}},
     _StateName, State = #state{socket = Socket}) ->
     send_error_reply(Socket, PacketId, <<"agent state invalid">>),
     {keep_state, State};
 
-handle_event(internal, {decoded_reply, #'ReplyFrame'{packet_id = AuthPacketId, reply = {result, Message}}},
+handle_event(internal, {decoded_packet, {response, AuthPacketId, {ok, Message}}},
     ?STATE_AUTH, State = #state{auth_packet_id = AuthPacketId}) ->
 
     logger:debug("[efka_client] auth success, message: ~p", [Message]),
     {next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]};
-handle_event(internal, {decoded_reply, #'ReplyFrame'{packet_id = AuthPacketId, reply = {error, #'ReplyFrame.Error'{code = 1, message = Message}}}},
+handle_event(internal, {decoded_packet, {response, AuthPacketId, {error, 1, Message}}},
     ?STATE_AUTH, State = #state{auth_packet_id = AuthPacketId}) ->
     logger:debug("[efka_client] auth denied, message: ~p", [Message]),
     {next_state, ?STATE_RESTRICTED, State};
-handle_event(internal, {decoded_reply, #'ReplyFrame'{packet_id = AuthPacketId, reply = {error, #'ReplyFrame.Error'{message = Message}}}},
+handle_event(internal, {decoded_packet, {response, AuthPacketId, {error, _Code, Message}}},
     ?STATE_AUTH, State = #state{socket = Socket, auth_packet_id = AuthPacketId}) ->
     logger:debug("[efka_client] auth failed, message: ~p", [Message]),
     disconnect(Socket),
     schedule_reconnect(),
     {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
-handle_event(internal, {decoded_reply, ReplyFrame}, StateName, State) ->
-    logger:warning("[efka_client] ignore unexpected reply in state ~p: ~p", [StateName, ReplyFrame]),
+handle_event(internal, {decoded_packet, {response, _PacketId, Reply}}, StateName, State) ->
+    logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
 
 %% 处理命令
-handle_event(internal, {decoded_cast, #'CastFrame'{body = {command, #'CastFrame.Command'{command = {auth, #'CastFrame.Command.Authorization'{cmd = Cmd}}}}}},
+handle_event(internal, {decoded_packet, {message, {auth_control, Cmd}}},
     StateName, State = #state{socket = Socket, next_packet_id = PacketId}) ->
 
     logger:debug("[efka_client] auth cmd: ~p", [Cmd]),
     case {Cmd, StateName} of
-        {'ACTIVATE', ?STATE_ACTIVATED} ->
+        {activate, ?STATE_ACTIVATED} ->
             {keep_state, State};
-        {'ACTIVATE', _} ->
+        {activate, _} ->
             AuthPacket = auth_packet(PacketId),
-            send_packet(Socket, [?FRAME_REQUEST, AuthPacket]),
+            send_packet(Socket, AuthPacket),
             {next_state, ?STATE_AUTH, State#state{auth_packet_id = PacketId, next_packet_id = PacketId + 1},
                 [{state_timeout, 5000, auth_timeout}]};
-        {'DEACTIVATE', _} ->
+        {deactivate, _} ->
             {next_state, ?STATE_RESTRICTED, State}
     end;
 
 %% 处理Pub/Sub机制
-handle_event(internal, {decoded_cast, #'CastFrame'{body = {pub, #'CastFrame.Pub'{topic = Topic, qos = Qos, content = Content}}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, {decoded_packet, {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};
@@ -275,19 +261,14 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 auth_packet(PktId) when is_integer(PktId) ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
-    Username = proplists:get_value(username, AuthInfo),
-    Salt = proplists:get_value(salt, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
-    message_pb:encode_msg(#'RequestFrame'{
-        packet_id = PktId,
-        body = {auth_request, #'AuthRequest'{
-            uuid = unicode:characters_to_binary(UUID),
-            username = unicode:characters_to_binary(Username),
-            salt = unicode:characters_to_binary(Salt),
-            token = unicode:characters_to_binary(Token),
-            timestamp = efka_util:timestamp()
-        }}
-    }).
+
+    Timestamp = efka_util:timestamp(),
+    cloud_wire:encode({request, PktId, {auth_request, #{
+        uuid => list_to_binary(UUID),
+        token => list_to_binary(Token),
+        timestamp => Timestamp
+    }}}).
 
 -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
 connect_socket() ->
@@ -407,16 +388,10 @@ delete_oldest_cache_entry() ->
 
 -spec send_result_reply(ssl:sslsocket(), integer(), binary()) -> ok.
 send_result_reply(Socket, PacketId, Payload) when is_binary(Payload) ->
-    Packet = message_pb:encode_msg(#'ReplyFrame'{
-        packet_id = PacketId,
-        reply = {result, Payload}
-    }),
-    send_packet(Socket, [?FRAME_REPLY, Packet]).
+    Packet = cloud_wire:encode({response, PacketId, {ok, Payload}}),
+    send_packet(Socket, Packet).
 
 -spec send_error_reply(ssl:sslsocket(), integer(), binary()) -> ok.
 send_error_reply(Socket, PacketId, Reason) when is_binary(Reason) ->
-    Packet = message_pb:encode_msg(#'ReplyFrame'{
-        packet_id = PacketId,
-        reply = {error, #'ReplyFrame.Error'{code = -1, message = Reason}}
-    }),
-    send_packet(Socket, [?FRAME_REPLY, Packet]).
+    Packet = cloud_wire:encode({response, PacketId, {error, -1, Reason}}),
+    send_packet(Socket, Packet).