fix

apps/efka/src/docker/docker_deployer.erl

@@ -82,10 +82,10 @@ deploy(Pid) when is_pid(Pid) ->
     gen_server:cast(Pid, deploy).
 
 %% @doc Spawns the server and registers the local name (unique)
--spec(start_link(TaskId :: integer(), RootDir :: string(), Config :: map()) ->
+-spec(start_link(TaskId :: integer(), ContainerDir :: string(), Config :: map()) ->
     {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
-start_link(TaskId, RootDir, Config) when is_integer(TaskId), is_list(RootDir), is_map(Config) ->
-    gen_server:start_link(?MODULE, [TaskId, RootDir, Config], []).
+start_link(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerDir), is_map(Config) ->
+    gen_server:start_link(?MODULE, [TaskId, ContainerDir, Config], []).
 
 %%%===================================================================
 %%% gen_server callbacks
@@ -96,8 +96,8 @@ start_link(TaskId, RootDir, Config) when is_integer(TaskId), is_list(RootDir), i
 -spec(init(Args :: term()) ->
     {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term()} | ignore).
-init([TaskId, RootDir, Config]) ->
-    {ok, #state{task_id = TaskId, root_dir = RootDir, config = Config}}.
+init([TaskId, ContainerDir, Config]) ->
+    {ok, #state{task_id = TaskId, root_dir = ContainerDir, config = Config}}.
 
 %% @private
 %% @doc Handling call messages
@@ -165,14 +165,13 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
 %   "command": ["nginx", "-g", "daemon off;"],
 %   "restart": "always"
 %}
--spec do_deploy(TaskId :: integer(), RootDir :: string(), Config :: map()) -> ok | {error, Reason :: any()}.
-do_deploy(TaskId, RootDir, Config) when is_integer(TaskId), is_list(RootDir), is_map(Config) ->
+-spec do_deploy(TaskId :: integer(), ContainerDir :: string(), Config :: map()) -> ok | {error, Reason :: any()}.
+do_deploy(TaskId, ContainerDir, Config) when is_integer(TaskId), is_list(ContainerDir), is_map(Config) ->
     %% 尝试拉取镜像
     ContainerName = maps:get(<<"container_name">>, Config),
     Image0 = maps:get(<<"image">>, Config),
     Image = normalize_image(Image0),
 
-    {ok, ContainerDir} = ensure_dirs(RootDir, ContainerName),
     case try_pull_image(Image) of
         ok ->
             %% 创建container
@@ -212,13 +211,6 @@ maybe_create_env_file(ContainerDir, Envs) when is_list(Envs)->
     ok = file:close(IoDevice),
     {ok, TargetFile}.
 
--spec ensure_dirs(RootDir :: string(), ContainerName :: binary()) -> {ok, ServerRootDir :: string()}.
-ensure_dirs(RootDir, ContainerName) when is_list(RootDir), is_binary(ContainerName) ->
-    %% 根目录
-    ServiceRootDir = RootDir ++ "/" ++ binary_to_list(ContainerName) ++ "/",
-    ok = filelib:ensure_dir(ServiceRootDir),
-    {ok, ServiceRootDir}.
-
 -spec normalize_image(binary()) -> binary().
 normalize_image(Image) when is_binary(Image) ->
     Parts = binary:split(Image, <<"/">>, [global]),

apps/efka/src/docker/docker_manager.erl

@@ -81,8 +81,8 @@ init([]) ->
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_call({deploy, TaskId, Config = #{<<"container_name">> := ContainerName}}, _From, State = #state{root_dir = RootDir, task_map = TaskMap}) ->
     %% 创建目录
-    {ok, ContainerDir} = docker_container_helper:ensure_dir(ContainerDir, ContainerName),
-    {ok, TaskPid} = docker_deployer:start_link(TaskId, RootDir, Config),
+    {ok, ContainerDir} = docker_container_helper:ensure_dir(RootDir, ContainerName),
+    {ok, TaskPid} = docker_deployer:start_link(TaskId, ContainerDir, Config),
     docker_deployer:deploy(TaskPid),
     lager:debug("[efka_inetd] start deploy task_id: ~p, config: ~p", [TaskId, Config]),
     {reply, ok, State#state{task_map = maps:put(TaskPid, TaskId, TaskMap)}};

apps/efka/src/efka_codec.erl

@@ -14,10 +14,14 @@
 -define(Bytes, 2).
 
 %% API
--export([encode/1, decode/1]).
+-export([encode/2, decode/1]).
 
--spec encode(Message :: any()) -> binary().
-encode(#auth_request{uuid = UUID, username = Username, salt = Salt, token = Token, timestamp = Timestamp}) ->
+-spec encode0(Message :: any()) -> binary().
+encode(PacketType, Packet) when is_integer(PacketType) ->
+    Bin = encode0(Packet),
+    <<PacketType, Bin/binary>>.
+
+encode0(#auth_request{uuid = UUID, username = Username, salt = Salt, token = Token, timestamp = Timestamp}) ->
     iolist_to_binary([
         marshal(?Bytes, UUID),
         marshal(?Bytes, Username),
@@ -25,40 +29,40 @@ encode(#auth_request{uuid = UUID, username = Username, salt = Salt, token = Toke
         marshal(?Bytes, Token),
         marshal(?I32, Timestamp)
     ]);
-encode(#auth_reply{code = Code, message = Message}) ->
+encode0(#auth_reply{code = Code, message = Message}) ->
     iolist_to_binary([
         marshal(?I32, Code),
         marshal(?Bytes, Message)
     ]);
-encode(#pub{topic = Topic, content = Content}) ->
+encode0(#pub{topic = Topic, content = Content}) ->
     iolist_to_binary([
         marshal(?Bytes, Topic),
         marshal(?Bytes, Content)
     ]);
-encode(#command{command_type = CommandType, command = Command}) ->
+encode0(#command{command_type = CommandType, command = Command}) ->
     iolist_to_binary([
         marshal(?I32, CommandType),
         marshal(?Bytes, Command)
     ]);
-encode(#rpc_deploy{task_id = TaskId, config = Config}) ->
+encode0(#rpc_deploy{task_id = TaskId, config = Config}) ->
     iolist_to_binary([
         marshal(?I32, TaskId),
         marshal(?Bytes, Config)
     ]);
-encode(#rpc_container{method = Method, container_name = ContainerName, params = Params}) ->
+encode0(#rpc_container{method = Method, container_name = ContainerName, params = Params}) ->
     iolist_to_binary([
         marshal(?Bytes, Method),
         marshal(?Bytes, ContainerName),
         marshal(?Bytes, Params)
     ]);
-encode(#data{service_id = ServiceId, device_uuid = DeviceUUID, route_key = RouteKey, metric = Metric}) ->
+encode0(#data{service_id = ServiceId, device_uuid = DeviceUUID, route_key = RouteKey, metric = Metric}) ->
     iolist_to_binary([
         marshal(?Bytes, ServiceId),
         marshal(?Bytes, DeviceUUID),
         marshal(?Bytes, RouteKey),
         marshal(?Bytes, Metric)
     ]);
-encode(#event{service_id = ServiceId, event_type = EventType, params = Params}) ->
+encode0(#event{service_id = ServiceId, event_type = EventType, params = Params}) ->
     iolist_to_binary([
         marshal(?Bytes, ServiceId),
         marshal(?I32, EventType),

apps/efka/src/efka_remote_agent.erl

@@ -8,7 +8,6 @@
 %%%-------------------------------------------------------------------
 -module(efka_remote_agent).
 -author("anlicheng").
--include("message_pb.hrl").
 -include("efka.hrl").
 -include("efka_tables.hrl").
 
@@ -84,35 +83,35 @@ callback_mode() ->
 
 %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
 handle_event(cast, {metric_data, ServiceId, DeviceUUID, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    Packet = message_pb:encode_msg(#data{
+    Packet = efka_codec:encode(?METHOD_DATA, #data{
         service_id = ServiceId,
         device_uuid = DeviceUUID,
         route_key = RouteKey,
         metric = Metric
     }),
-    efka_transport:send(TransportPid, ?METHOD_DATA, Packet),
+    efka_transport:send(TransportPid, Packet),
     {keep_state, State};
 
 handle_event(cast, {metric_data, ServiceId, DeviceUUID, LineProtocolData}, _, State) ->
-    Packet = message_pb:encode_msg(#data{
+    Packet = efka_codec:encode(?METHOD_DATA, #data{
         service_id = ServiceId,
         device_uuid = DeviceUUID,
         metric = LineProtocolData
     }),
-    ok = cache_model:insert(?METHOD_DATA, Packet),
+    ok = cache_model:insert(Packet),
     {keep_state, State};
 
 %% 异步发送事件
 handle_event(cast, {event, ServiceId, EventType, Params}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
-    EventPacket = message_pb:encode_msg(#event{
+    EventPacket = efka_codec:encode(?METHOD_EVENT, #event{
         service_id = ServiceId,
         event_type = EventType,
         params = Params
     }),
-    efka_transport:send(TransportPid, ?METHOD_EVENT, EventPacket),
+    efka_transport:send(TransportPid, EventPacket),
     {keep_state, State};
 handle_event(cast, {event, ServiceId, EventType, Params}, ?STATE_ACTIVATED, State) ->
-    EventPacket = message_pb:encode_msg(#event{
+    EventPacket = efka_codec:encode(?METHOD_EVENT, #event{
         service_id = ServiceId,
         event_type = EventType,
         params = Params
@@ -120,26 +119,26 @@ handle_event(cast, {event, ServiceId, EventType, Params}, ?STATE_ACTIVATED, Stat
     ok = cache_model:insert(?METHOD_EVENT, EventPacket),
     {keep_state, State};
 
-handle_event(cast, {ping, AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces}, ?STATE_ACTIVATED,
-    State = #state{transport_pid = TransportPid}) ->
-
-    Ping = message_pb:encode_msg(#ping{
-        adcode = AdCode,
-        boot_time = BootTime,
-        province = Province,
-        city = City,
-        efka_version = EfkaVersion,
-        kernel_arch = KernelArch,
-        ips = Ips,
-        cpu_core = CpuCore,
-        cpu_load = CpuLoad,
-        cpu_temperature = CpuTemperature,
-        disk = Disk,
-        memory = Memory,
-        interfaces = Interfaces
-    }),
-    efka_transport:send(TransportPid, ?METHOD_PING, Ping),
-    {keep_state, State};
+%handle_event(cast, {ping, AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces}, ?STATE_ACTIVATED,
+%    State = #state{transport_pid = TransportPid}) ->
+%
+%    Ping = message_pb:encode_msg(#ping{
+%        adcode = AdCode,
+%        boot_time = BootTime,
+%        province = Province,
+%        city = City,
+%        efka_version = EfkaVersion,
+%        kernel_arch = KernelArch,
+%        ips = Ips,
+%        cpu_core = CpuCore,
+%        cpu_load = CpuLoad,
+%        cpu_temperature = CpuTemperature,
+%        disk = Disk,
+%        memory = Memory,
+%        interfaces = Interfaces
+%    }),
+%    efka_transport:send(TransportPid, ?METHOD_PING, Ping),
+%    {keep_state, State};
 
 %% 异步建立到服务器的连接
 handle_event(info, {timeout, _, create_transport}, ?STATE_DENIED, State) ->
@@ -336,7 +335,7 @@ auth_request() ->
     Salt = proplists:get_value(salt, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
 
-    message_pb:encode_msg(#auth_request{
+    efka_codec:encode(?METHOD_AUTH, #auth_request{
         uuid = unicode:characters_to_binary(UUID),
         username = unicode:characters_to_binary(Username),
         salt = unicode:characters_to_binary(Salt),

apps/efka/src/efka_transport.erl

@@ -8,7 +8,6 @@
 %%%-------------------------------------------------------------------
 -module(efka_transport).
 -author("anlicheng").
--include("message_pb.hrl").
 -include("efka.hrl").
 
 -behaviour(gen_server).
@@ -115,7 +114,7 @@ handle_cast(connect, State = #state{host = Host, port = Port, parent_pid = Paren
 %% auth校验
 handle_cast({auth_request, AuthRequestBin}, State = #state{parent_pid = ParentPid, socket = Socket}) ->
     PacketId = 1,
-    ok = ssl:send(Socket, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH, AuthRequestBin/binary>>),
+    ok = ssl:send(Socket, <<?PACKET_REQUEST, PacketId:32, AuthRequestBin/binary>>),
     %% 需要等待auth返回的结果
     receive
         {ssl, Socket, <<?PACKET_RESPONSE, PacketId:32, ReplyBin/binary>>} ->