fix efka_client

src/efka_client.erl

@@ -24,7 +24,7 @@
 -define(SERVER, ?MODULE).
 
 %% 标记当前agent的状态，只有在 activated 状态下才可以正常的发送数据
--define(STATE_DENIED, denied).
+-define(STATE_DISCONNECTED, disconnected).
 -define(STATE_AUTH, auth).
 %% 不能推送消息到服务，但是可以接受服务器的部分指令
 -define(STATE_RESTRICTED, restricted).
@@ -64,17 +64,18 @@ start_link() ->
 
 init([]) ->
     erlang:start_timer(0, self(), create_transport),
-    {ok, ?STATE_DENIED, #state{socket = undefined}}.
+    {ok, ?STATE_DISCONNECTED, #state{socket = undefined}}.
 
 callback_mode() ->
     handle_event_function.
 
 %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
 handle_event(cast, {metric_data, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    Packet = message_pb:encode_msg(#'CastFrame'{
+    CastFrame = message_pb:encode_msg(#'CastFrame'{
         body = {data, #'Data'{route_key = RouteKey, metric = Metric}}
     }),
-    send_packet(Socket, <<?FRAME_CAST, Packet/binary>>),
+    Packet = <<?FRAME_CAST, CastFrame/binary>>,
+    send_packet(Socket, Packet),
     {keep_state, State};
 
 handle_event(cast, {metric_data, RouteKey, Metric}, _, State) ->
@@ -104,7 +105,7 @@ handle_event(cast, {task_event_stream, _TaskId, _Stream}, _, State = #state{}) -
     {keep_state, State};
 
 %% 异步建立到服务器的连接
-handle_event(info, {timeout, _, create_transport}, ?STATE_DENIED, State) ->
+handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
     case connect_socket() of
         {ok, Socket} ->
             AuthPacket = auth_packet(),
@@ -122,7 +123,7 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S
     logger:debug("[efka_client] auth request timeout"),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DENIED, State#state{socket = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
 
 %% 将缓存中的数据推送到服务器端
 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
@@ -138,11 +139,45 @@ handle_event(info, flush_cache, _, State) ->
     {keep_state, State};
 
 %% 处理收到的ssl消息
+handle_event(info, {ssl, Socket, <<?FRAME_RESPONSE, PacketBin/binary>>}, _, State = #state{socket = Socket}) ->
+    ResponseFrame = message_pb:decode_msg(PacketBin, 'ResponseFrame'),
+    {keep_state, State, [{next_event, internal, ResponseFrame}]};
+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, 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, CastFrame}]};
+handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) ->
+    logger:debug("[efka_client] ssl error: ~p", [Reason]),
+    disconnect(Socket),
+    schedule_reconnect(),
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
+handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
+    schedule_reconnect(),
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
 
-handle_event(info, {ssl, Socket, <<?FRAME_RESPONSE, PacketBin/binary>>}, ?STATE_AUTH, State = #state{socket = Socket}) ->
+%%% 处理内部消息，ssl收到的消息会解析成protobuf的消息格式，并按照internal类型处理
-    #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}} =
-        message_pb:decode_msg(PacketBin, 'ResponseFrame'),
 
+%% 微服务部署
+handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = {container_request, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    case handle_container_request(Request) of
+        ok ->
+            send_rpc_result_reply(Socket, PacketId, <<"ok">>);
+        {ok, Reply} ->
+            send_rpc_result_reply(Socket, PacketId, Reply);
+        {error, Reason} ->
+            send_rpc_error_reply(Socket, PacketId, Reason)
+    end,
+    {keep_state, State};
+handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = _Body}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
+    send_rpc_error_reply(Socket, PacketId, <<"agent restricted">>),
+    {keep_state, State};
+handle_event(internal, #'RequestFrame'{packet_id = PacketId, body = _Body}, _StateName, State = #state{socket = Socket}) ->
+    send_rpc_error_reply(Socket, PacketId, <<"agent state invalid">>),
+    {keep_state, State};
+
+handle_event(internal, #'ResponseFrame'{packet_id = 1, body = {auth_reply, #'AuthReply'{code = Code, payload = Message}}}, ?STATE_AUTH, State = #state{socket = Socket}) ->
     case Code of
         0 ->
             logger:debug("[efka_client] auth success, message: ~p", [Message]),
@@ -154,242 +189,16 @@ handle_event(info, {ssl, Socket, <<?FRAME_RESPONSE, PacketBin/binary>>}, ?STATE_
             logger:debug("[efka_client] auth failed, message: ~p", [Message]),
             disconnect(Socket),
             schedule_reconnect(),
-            {next_state, ?STATE_DENIED, State#state{socket = undefined}};
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}};
         _ ->
             logger:debug("[efka_client] auth failed, invalid message"),
             disconnect(Socket),
             schedule_reconnect(),
-            {next_state, ?STATE_DENIED, State#state{socket = undefined}}
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined}}
     end;
 
-handle_event(info, {ssl, Socket, <<?FRAME_REQUEST, PacketBin/binary>>}, StateName, State = #state{socket = Socket})
-    when StateName =:= ?STATE_ACTIVATED; StateName =:= ?STATE_RESTRICTED ->
-    #'RequestFrame'{packet_id = PacketId, body = Body} = message_pb:decode_msg(PacketBin, 'RequestFrame'),
-    case Body of
-        {rpc_request, Request} ->
-            {keep_state, State, [{next_event, info, {server_rpc, PacketId, Request}}]};
-        {container_request, Request} ->
-            {keep_state, State, [{next_event, info, {container_request, PacketId, Request}}]}
-    end;
-
-handle_event(info, {ssl, Socket, <<?FRAME_CAST, PacketBin/binary>>}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    CastFrame = message_pb:decode_msg(PacketBin, 'CastFrame'),
-    {keep_state, State, [{next_event, info, {server_cast, CastFrame}}]};
-
-%% 微服务部署
-handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}},
-  ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    case docker_manager:get_containers() of
-        {ok, Containers} ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Containers)}}
-                }}
-            }),
-            send_packet(Socket, Packet);
-        {error, Reason} when is_binary(Reason) ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-                }}
-            }),
-            send_packet(Socket, Packet)
-    end,
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{
-    task_id = TaskId,
-    params = Params
-}}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    case docker_manager:deploy(TaskId, Params) of
-        ok ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
-                }}
-            }),
-            send_packet(Socket, Packet);
-        {error, Reason} when is_binary(Reason) ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-                }}
-            }),
-            send_packet(Socket, Packet)
-    end,
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = Target}}}},
-  ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ContainerTarget = container_target(Target),
-    case docker_manager:start_container(ContainerTarget) of
-        ok ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
-                }}
-            }),
-            send_packet(Socket, Packet);
-        {error, Reason} when is_binary(Reason) ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-                }}
-            }),
-            send_packet(Socket, Packet)
-    end,
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{
-    target = Target,
-    timeout_seconds = TimeoutSeconds
-}}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ContainerTarget = container_target(Target),
-    case docker_manager:stop_container(ContainerTarget, TimeoutSeconds) of
-        ok ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
-                }}
-            }),
-            send_packet(Socket, Packet);
-        {error, Reason} when is_binary(Reason) ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-                }}
-            }),
-            send_packet(Socket, Packet)
-    end,
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{
-    target = Target,
-    signal = Signal
-}}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ContainerTarget = container_target(Target),
-    case docker_manager:kill_container(ContainerTarget, to_binary(Signal)) of
-        ok ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
-                }}
-            }),
-            send_packet(Socket, Packet);
-        {error, Reason} when is_binary(Reason) ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-                }}
-            }),
-            send_packet(Socket, Packet)
-    end,
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{
-    target = Target,
-    force = Force,
-    remove_volumes = RemoveVolumes
-}}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ContainerTarget = container_target(Target),
-    case docker_manager:remove_container(ContainerTarget, to_bool(Force), to_bool(RemoveVolumes)) of
-        ok ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
-                }}
-            }),
-            send_packet(Socket, Packet);
-        {error, Reason} when is_binary(Reason) ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-                }}
-            }),
-            send_packet(Socket, Packet)
-    end,
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, #'ContainerRequest'{action = {config, #'ContainerRequest.Config'{
-    target = Target,
-    config = Config
-}}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ContainerTarget = container_target(Target),
-    case docker_manager:config_container(ContainerTarget, iolist_to_binary(Config)) of
-        ok ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(<<"ok">>)}}
-                }}
-            }),
-            send_packet(Socket, Packet);
-        {error, Reason} ->
-            Packet = message_pb:encode_msg(#'ResponseFrame'{
-                packet_id = PacketId,
-                body = {rpc_reply, #'RpcReply'{
-                    reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
-                }}
-            }),
-            send_packet(Socket, Packet)
-    end,
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, _Request}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    Packet = message_pb:encode_msg(#'ResponseFrame'{
-        packet_id = PacketId,
-        body = {rpc_reply, #'RpcReply'{
-            reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"unsupported container request">>}}
-        }}
-    }),
-    send_packet(Socket, Packet),
-    {keep_state, State};
-
-handle_event(info, {container_request, PacketId, _Request}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
-    Packet = message_pb:encode_msg(#'ResponseFrame'{
-        packet_id = PacketId,
-        body = {rpc_reply, #'RpcReply'{
-            reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"agent restricted">>}}
-        }}
-    }),
-    send_packet(Socket, Packet),
-    {keep_state, State};
-
-handle_event(info, {server_rpc, PacketId, #'RpcRequest'{method = Method}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    Packet = message_pb:encode_msg(#'ResponseFrame'{
-        packet_id = PacketId,
-        body = {rpc_reply, #'RpcReply'{
-            reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"unsupported rpc request: ", Method/binary>>}}
-        }}
-    }),
-    send_packet(Socket, Packet),
-    {keep_state, State};
-
-handle_event(info, {server_rpc, PacketId, _Request}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
-    Packet = message_pb:encode_msg(#'ResponseFrame'{
-        packet_id = PacketId,
-        body = {rpc_reply, #'RpcReply'{
-            reply = {error, #'RpcReply.RpcError'{code = -1, message = <<"agent restricted">>}}
-        }}
-    }),
-    send_packet(Socket, Packet),
-    {keep_state, State};
-
 %% 处理命令
-handle_event(info, {server_cast, {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}}, StateName,
+handle_event(internal, #'CastFrame'{body = {command, #'Command'{command_type = ?COMMAND_AUTH, command = Auth0}}}, StateName, State = #state{socket = Socket}) ->
-  State = #state{socket = Socket}) ->
     Auth = binary_to_integer(Auth0),
     case {Auth, StateName} of
         {1, ?STATE_ACTIVATED} ->
@@ -403,21 +212,11 @@ handle_event(info, {server_cast, {command, #'Command'{command_type = ?COMMAND_AU
     end;
 
 %% 处理Pub/Sub机制
-handle_event(info, {server_cast, {pub, #'Pub'{topic = Topic, qos = Qos, content = Content}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, #'CastFrame'{body = {pub, #'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};
 
-handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socket}) ->
-    logger:debug("[efka_client] ssl error: ~p", [Reason]),
-    disconnect(Socket),
-    schedule_reconnect(),
-    {next_state, ?STATE_DENIED, State#state{socket = undefined}};
-
-handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
-    schedule_reconnect(),
-    {next_state, ?STATE_DENIED, State#state{socket = undefined}};
-
 handle_event(info, Info, _, State = #state{}) ->
     logger:notice("[efka_client] get unknown info: ~p", [Info]),
     {keep_state, State}.
@@ -483,6 +282,52 @@ schedule_reconnect() ->
 encode_rpc_payload(Payload) ->
     jiffy:encode(Payload, [force_utf8]).
 
+-spec handle_container_request(message_pb:'ContainerRequest'()) -> ok.
+handle_container_request(#'ContainerRequest'{action = {list, #'ContainerRequest.List'{all = _All}}}) ->
+    case docker_manager:get_containers() of
+        {ok, Containers} ->
+            {ok, jiffy:encode(Containers, [force_utf8])};
+        {error, Reason} when is_binary(Reason) ->
+            {error, Reason}
+    end;
+handle_container_request(#'ContainerRequest'{action = {deploy, #'ContainerRequest.Deploy'{ task_id = TaskId, params = Params }}}) ->
+    docker_manager:deploy(TaskId, Params);
+handle_container_request(#'ContainerRequest'{action = {start, #'ContainerRequest.Start'{target = Target}}}) ->
+    ContainerTarget = container_target(Target),
+    docker_manager:start_container(ContainerTarget);
+handle_container_request(#'ContainerRequest'{action = {stop, #'ContainerRequest.Stop'{ target = Target, timeout_seconds = TimeoutSeconds }}}) ->
+    ContainerTarget = container_target(Target),
+    docker_manager:stop_container(ContainerTarget, TimeoutSeconds);
+handle_container_request(#'ContainerRequest'{action = {kill, #'ContainerRequest.Kill'{ target = Target, signal = Signal}}}) ->
+    ContainerTarget = container_target(Target),
+    docker_manager:kill_container(ContainerTarget, to_binary(Signal));
+handle_container_request(#'ContainerRequest'{action = {remove, #'ContainerRequest.Remove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}) ->
+    ContainerTarget = container_target(Target),
+    docker_manager:remove_container(ContainerTarget, to_bool(Force), to_bool(RemoveVolumes));
+handle_container_request(#'ContainerRequest'{action = {config, #'ContainerRequest.Config'{target = Target, config = Config}}}) ->
+    ContainerTarget = container_target(Target),
+    docker_manager:config_container(ContainerTarget, iolist_to_binary(Config)).
+
+-spec send_rpc_result_reply(ssl:sslsocket(), integer(), any()) -> ok.
+send_rpc_result_reply(Socket, PacketId, Payload) ->
+    Packet = message_pb:encode_msg(#'ResponseFrame'{
+        packet_id = PacketId,
+        body = {rpc_reply, #'RpcReply'{
+            reply = {result, #'RpcReply.RpcResult'{data = encode_rpc_payload(Payload)}}
+        }}
+    }),
+    send_packet(Socket, Packet).
+
+-spec send_rpc_error_reply(ssl:sslsocket(), integer(), binary()) -> ok.
+send_rpc_error_reply(Socket, PacketId, Reason) ->
+    Packet = message_pb:encode_msg(#'ResponseFrame'{
+        packet_id = PacketId,
+        body = {rpc_reply, #'RpcReply'{
+            reply = {error, #'RpcReply.RpcError'{code = -1, message = Reason}}
+        }}
+    }),
+    send_packet(Socket, Packet).
+
 container_target(#'ContainerRef'{name = Name, id = Id}) ->
     NameBin = to_binary(Name),
     IdBin = to_binary(Id),