diff --git a/src/host/iot_container_task.erl b/src/host/iot_container_task.erl new file mode 100644 index 0000000..269834d --- /dev/null +++ b/src/host/iot_container_task.erl @@ -0,0 +1,147 @@ +%%%------------------------------------------------------------------- +%%% @doc Owns one container deployment task stream. +%%% @end +%%%------------------------------------------------------------------- +-module(iot_container_task). + +-behaviour(gen_server). + +%% API +-export([start_link/2]). +-export([get_pid/2, subscribe/3, stream/3, close/2]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(MAX_EVENTS, 200). +-define(STOP_AFTER_CLOSE, 300000). + +-record(state, { + uuid :: binary(), + task_id :: integer(), + status = pending :: pending | running | success | fail, + events = [] :: [{binary(), binary()}], + listeners = #{} :: #{pid() => reference()}, + close_reason = undefined :: undefined | binary(), + stop_ref = undefined :: undefined | reference() +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +-spec start_link(binary(), integer()) -> {ok, pid()} | ignore | {error, term()}. +start_link(UUID, TaskId) when is_binary(UUID), is_integer(TaskId) -> + gen_server:start_link({via, gproc, {n, l, name(UUID, TaskId)}}, ?MODULE, [UUID, TaskId], []). + +-spec get_pid(binary(), integer()) -> undefined | pid(). +get_pid(UUID, TaskId) when is_binary(UUID), is_integer(TaskId) -> + gproc:whereis_name({n, l, name(UUID, TaskId)}). + +-spec subscribe(binary(), integer(), pid()) -> ok | {error, term()}. +subscribe(UUID, TaskId, ListenerPid) when is_binary(UUID), is_integer(TaskId), is_pid(ListenerPid) -> + case iot_container_task_sup:ensure_started(UUID, TaskId) of + {ok, Pid} -> + gen_server:call(Pid, {subscribe, ListenerPid}); + {error, Reason} -> + {error, Reason} + end. + +-spec stream(pid(), binary(), binary()) -> ok. +stream(Pid, Type, Stream) when is_pid(Pid), is_binary(Type), is_binary(Stream) -> + gen_server:cast(Pid, {stream, Type, Stream}). + +-spec close(pid(), binary()) -> ok. +close(Pid, Reason) when is_pid(Pid), is_binary(Reason) -> + gen_server:cast(Pid, {close, Reason}). + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +-spec init([binary() | integer()]) -> {ok, #state{}}. +init([UUID, TaskId]) -> + ok = iot_log:set_metadata(), + {ok, #state{uuid = UUID, task_id = TaskId}}. + +-spec handle_call(term(), {pid(), term()}, #state{}) -> + {reply, term(), #state{}}. +handle_call({subscribe, ListenerPid}, _From, State0 = #state{task_id = TaskId, events = Events, listeners = Listeners, close_reason = CloseReason}) -> + MonRef = erlang:monitor(process, ListenerPid), + maps:get(ListenerPid, Listeners, undefined) =/= undefined andalso erlang:demonitor(maps:get(ListenerPid, Listeners), [flush]), + lists:foreach(fun({Type, Stream}) -> + ListenerPid ! {stream_data, TaskId, Type, Stream} + end, Events), + case CloseReason of + undefined -> + ok; + Reason -> + ListenerPid ! {stream_close, TaskId, Reason} + end, + {reply, ok, State0#state{listeners = maps:put(ListenerPid, MonRef, Listeners)}}; +handle_call(_Request, _From, State) -> + {reply, ok, State}. + +-spec handle_cast(term(), #state{}) -> {noreply, #state{}}. +handle_cast({stream, Type, Stream}, State0 = #state{task_id = TaskId, events = Events0, listeners = Listeners}) -> + Event = {Type, Stream}, + Events = trim_events(Events0 ++ [Event]), + maps:foreach(fun(ListenerPid, _MonRef) -> + ListenerPid ! {stream_data, TaskId, Type, Stream} + end, Listeners), + {noreply, State0#state{status = running, events = Events}}; +handle_cast({close, Reason}, State0 = #state{task_id = TaskId, listeners = Listeners}) -> + maps:foreach(fun(ListenerPid, _MonRef) -> + ListenerPid ! {stream_close, TaskId, Reason} + end, Listeners), + Status = close_status(Reason), + {noreply, ensure_stop_timer(State0#state{status = Status, close_reason = Reason})}; +handle_cast(_Request, State) -> + {noreply, State}. + +-spec handle_info(term(), #state{}) -> {noreply, #state{}} | {stop, normal, #state{}}. +handle_info({'DOWN', _Ref, process, ListenerPid, _Reason}, State = #state{listeners = Listeners}) -> + {noreply, State#state{listeners = maps:remove(ListenerPid, Listeners)}}; +handle_info(stop_after_close, State) -> + {stop, normal, State}; +handle_info(_Info, State) -> + {noreply, State}. + +-spec terminate(term(), #state{}) -> ok. +terminate(_Reason, _State) -> + ok. + +-spec code_change(term(), #state{}, term()) -> {ok, #state{}}. +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +-spec name(binary(), integer()) -> term(). +name(UUID, TaskId) -> + {iot_container_task, UUID, TaskId}. + +-spec trim_events([{binary(), binary()}]) -> [{binary(), binary()}]. +trim_events(Events) -> + Len = length(Events), + case Len > ?MAX_EVENTS of + true -> + lists:nthtail(Len - ?MAX_EVENTS, Events); + false -> + Events + end. + +-spec close_status(binary()) -> success | fail. +close_status(<<"success">>) -> + success; +close_status(_) -> + fail. + +-spec ensure_stop_timer(#state{}) -> #state{}. +ensure_stop_timer(State = #state{stop_ref = undefined}) -> + Ref = erlang:send_after(?STOP_AFTER_CLOSE, self(), stop_after_close), + State#state{stop_ref = Ref}; +ensure_stop_timer(State) -> + State. diff --git a/src/host/iot_container_task_sup.erl b/src/host/iot_container_task_sup.erl new file mode 100644 index 0000000..406217d --- /dev/null +++ b/src/host/iot_container_task_sup.erl @@ -0,0 +1,100 @@ +%%%------------------------------------------------------------------- +%%% @doc Dynamic supervisor for container deployment tasks. +%%% @end +%%%------------------------------------------------------------------- +-module(iot_container_task_sup). + +-behaviour(supervisor). + +%% API +-export([start_link/0]). +-export([ensure_started/2, stream/4, close/3, fail/3]). + +%% supervisor callbacks +-export([init/1]). + +-define(SERVER, ?MODULE). + +%%%=================================================================== +%%% API +%%%=================================================================== + +-spec start_link() -> {ok, pid()} | ignore | {error, term()}. +start_link() -> + supervisor:start_link({local, ?SERVER}, ?MODULE, []). + +-spec ensure_started(binary(), integer()) -> {ok, pid()} | {error, term()}. +ensure_started(UUID, TaskId) when is_binary(UUID), is_integer(TaskId) -> + case iot_container_task:get_pid(UUID, TaskId) of + undefined -> + case supervisor:start_child(?SERVER, [UUID, TaskId]) of + {ok, Pid} -> + {ok, Pid}; + {ok, Pid, _Info} -> + {ok, Pid}; + {error, {already_started, Pid}} -> + {ok, Pid}; + {error, {shutdown, {failed_to_start_child, _Id, {already_started, Pid}}}} -> + {ok, Pid}; + {error, Reason} -> + {error, Reason} + end; + Pid when is_pid(Pid) -> + {ok, Pid} + end. + +-spec stream(binary(), integer(), binary(), binary()) -> ok. +stream(UUID, TaskId, Type, Stream) + when is_binary(UUID), is_integer(TaskId), is_binary(Type), is_binary(Stream) -> + case ensure_started(UUID, TaskId) of + {ok, Pid} -> + iot_container_task:stream(Pid, Type, Stream); + {error, Reason} -> + logger:warning("[iot_container_task_sup] start task failed, uuid: ~p, task_id: ~p, reason: ~p", [UUID, TaskId, Reason]), + ok + end. + +-spec close(binary(), integer(), binary()) -> ok. +close(UUID, TaskId, Reason) when is_binary(UUID), is_integer(TaskId), is_binary(Reason) -> + case ensure_started(UUID, TaskId) of + {ok, Pid} -> + iot_container_task:close(Pid, Reason); + {error, StartReason} -> + logger:warning("[iot_container_task_sup] start task before close failed, uuid: ~p, task_id: ~p, reason: ~p", [UUID, TaskId, StartReason]), + ok + end. + +-spec fail(binary(), integer(), term()) -> ok. +fail(UUID, TaskId, Reason) when is_binary(UUID), is_integer(TaskId) -> + ReasonBin = reason_to_binary(Reason), + ok = stream(UUID, TaskId, <<"error">>, ReasonBin), + close(UUID, TaskId, <<"fail">>). + +%%%=================================================================== +%%% supervisor callbacks +%%%=================================================================== + +-spec init([]) -> {ok, {{simple_one_for_one, non_neg_integer(), pos_integer()}, [supervisor:child_spec()]}}. +init([]) -> + SupFlags = {simple_one_for_one, 10, 60}, + ChildSpec = #{ + id => iot_container_task, + start => {iot_container_task, start_link, []}, + restart => temporary, + shutdown => 5000, + type => worker, + modules => [iot_container_task] + }, + {ok, {SupFlags, [ChildSpec]}}. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +-spec reason_to_binary(term()) -> binary(). +reason_to_binary(Reason) when is_binary(Reason) -> + Reason; +reason_to_binary(timeout) -> + <<"timeout">>; +reason_to_binary(Reason) -> + unicode:characters_to_binary(io_lib:format("~p", [Reason])). diff --git a/src/host/iot_event_stream_observer.erl b/src/host/iot_event_stream_observer.erl deleted file mode 100644 index 0d55395..0000000 --- a/src/host/iot_event_stream_observer.erl +++ /dev/null @@ -1,129 +0,0 @@ -%%%------------------------------------------------------------------- -%%% @author anlicheng -%%% @copyright (C) 2025, -%%% @doc -%%% -%%% @end -%%% Created : 26. 9月 2025 12:19 -%%%------------------------------------------------------------------- --module(iot_event_stream_observer). --author("anlicheng"). - --behaviour(gen_server). - -%% API --export([start_link/0]). --export([add_listener/2, stream_data/3, stream_close/2]). - -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). - --define(SERVER, ?MODULE). - --record(state, { - listeners = #{} -}). - -%%%=================================================================== -%%% API -%%%=================================================================== - --spec add_listener(ListenerPid :: pid(), TaskId :: integer()) -> ok. -add_listener(ListenerPid, TaskId) when is_pid(ListenerPid), is_integer(TaskId) -> - gen_server:call(?SERVER, {add_listener, ListenerPid, TaskId}). - --spec stream_data(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> no_return(). -stream_data(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) -> - gen_server:cast(?SERVER, {stream_data, TaskId, Type, Stream}). - --spec stream_close(TaskId :: integer(), Reason :: binary()) -> no_return(). -stream_close(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) -> - gen_server:cast(?SERVER, {stream_close, TaskId, Reason}). - -%% @doc Spawns the server and registers the local name (unique) --spec(start_link() -> - {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). -start_link() -> - gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). - -%%%=================================================================== -%%% gen_server callbacks -%%%=================================================================== - -%% @private -%% @doc Initializes the server --spec(init(Args :: term()) -> - {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | - {stop, Reason :: term()} | ignore). -init([]) -> - ok = iot_log:set_metadata(), - {ok, #state{}}. - -%% @private -%% @doc Handling call messages --spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, - State :: #state{}) -> - {reply, Reply :: term(), NewState :: #state{}} | - {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_call({add_listener, ListenerPid, TaskId}, _From, State = #state{listeners = Listeners}) -> - erlang:monitor(process, ListenerPid), - {reply, ok, State#state{listeners = maps:put(TaskId, ListenerPid, Listeners)}}. - -%% @private -%% @doc Handling cast messages --spec(handle_cast(Request :: term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_cast({stream_data, TaskId, Type, Stream}, State = #state{listeners = Listeners}) -> - case maps:find(TaskId, Listeners) of - error -> - ok; - {ok, ListenerPid} -> - is_process_alive(ListenerPid) andalso ListenerPid ! {stream_data, TaskId, Type, Stream} - end, - {noreply, State}; -handle_cast({stream_close, TaskId, Reason}, State = #state{listeners = Listeners}) -> - case maps:find(TaskId, Listeners) of - error -> - ok; - {ok, ListenerPid} -> - is_process_alive(ListenerPid) andalso ListenerPid ! {stream_close, TaskId, Reason} - end, - {noreply, State}. - -%% @private -%% @doc Handling all non call/cast messages --spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_info({'DOWN', _Ref, process, Pid, _Reason}, State = #state{listeners = Listeners}) -> - NListeners = maps:filter(fun(_, ListenerPid) -> ListenerPid /= Pid end, Listeners), - {noreply, State#state{listeners = NListeners}}. - -%% @private -%% @doc This function is called by a gen_server when it is about to -%% terminate. It should be the opposite of Module:init/1 and do any -%% necessary cleaning up. When it returns, the gen_server terminates -%% with Reason. The return value is ignored. --spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), - State :: #state{}) -> term()). -terminate(_Reason, _State = #state{}) -> - ok. - -%% @private -%% @doc Convert process state when code is changed --spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{}, - Extra :: term()) -> - {ok, NewState :: #state{}} | {error, Reason :: term()}). -code_change(_OldVsn, State = #state{}, _Extra) -> - {ok, State}. - -%%%=================================================================== -%%% Internal functions -%%%=================================================================== diff --git a/src/iot_sup.erl b/src/iot_sup.erl index c84a665..0d15f30 100644 --- a/src/iot_sup.erl +++ b/src/iot_sup.erl @@ -29,12 +29,12 @@ init([]) -> Specs = [ #{ - id => 'iot_event_stream_observer', - start => {'iot_event_stream_observer', start_link, []}, + id => 'iot_container_task_sup', + start => {'iot_container_task_sup', start_link, []}, restart => permanent, shutdown => 2000, - type => worker, - modules => ['iot_event_stream_observer'] + type => supervisor, + modules => ['iot_container_task_sup'] }, #{ diff --git a/src/transport/http/container_handler.erl b/src/transport/http/container_handler.erl index 8f697e6..73af4c5 100644 --- a/src/transport/http/container_handler.erl +++ b/src/transport/http/container_handler.erl @@ -70,18 +70,11 @@ handle_request("POST", "/container/deploy", _, #{<<"uuid">> := UUID, <<"task_id" undefined -> {ok, 200, iot_util:json_error(404, <<"host not found">>)}; Pid when is_pid(Pid) -> - case iot_host:deploy_container(Pid, TaskId, Config) of - {ok, Ref} -> - case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of - ok -> - {ok, 200, request_success_response(<<"ok">>)}; - {ok, Result} -> - {ok, 200, request_success_response(Result)}; - {error, Reason} -> - request_error_http_response(Reason) - end; - {error, Reason} when is_binary(Reason) -> - {ok, 200, iot_util:json_error(400, Reason)} + case iot_container_task_sup:ensure_started(UUID, TaskId) of + {ok, _TaskPid} -> + handle_deploy_container(Pid, UUID, TaskId, Config); + {error, Reason} -> + {ok, 500, iot_util:json_error(500, reason_to_binary(Reason))} end end; @@ -225,3 +218,22 @@ decode_json_bytes(Data) when is_binary(Data) -> Decoded -> {ok, Decoded} end. + +-spec handle_deploy_container(pid(), binary(), integer(), map()) -> + {ok, integer(), iolist()}. +handle_deploy_container(Pid, UUID, TaskId, Config) -> + case iot_host:deploy_container(Pid, TaskId, Config) of + {ok, Ref} -> + case iot_host:await_reply(Pid, Ref, ?REQ_TIMEOUT) of + ok -> + {ok, 200, request_success_response(<<"ok">>)}; + {ok, Result} -> + {ok, 200, request_success_response(Result)}; + {error, Reason} -> + ok = iot_container_task_sup:fail(UUID, TaskId, Reason), + request_error_http_response(Reason) + end; + {error, Reason} when is_binary(Reason) -> + ok = iot_container_task_sup:fail(UUID, TaskId, Reason), + {ok, 200, iot_util:json_error(400, Reason)} + end. diff --git a/src/transport/http/event_stream_handler.erl b/src/transport/http/event_stream_handler.erl index 9c6c9a4..efdb612 100644 --- a/src/transport/http/event_stream_handler.erl +++ b/src/transport/http/event_stream_handler.erl @@ -19,20 +19,23 @@ init(Req0, Opts) -> GetParams0 = cowboy_req:parse_qs(Req0), GetParams = maps:from_list(GetParams0), - #{<<"task_id">> := TaskId0} = GetParams, - TaskId = binary_to_integer(TaskId0), - logger:debug("method: ~p, path: ~p, get: ~p", [Method, Path, GetParams]), - Req1 = cowboy_req:stream_reply(200, #{ - <<"Content-Type">> => <<"text/event-stream">>, - <<"Cache-Control">> => <<"no-cache">>, - <<"Connection">> => <<"keep-alive">> - }, Req0), + case parse_stream_params(GetParams) of + {ok, UUID, TaskId} -> + Req1 = cowboy_req:stream_reply(200, #{ + <<"Content-Type">> => <<"text/event-stream">>, + <<"Cache-Control">> => <<"no-cache">>, + <<"Connection">> => <<"keep-alive">> + }, Req0), - ok = iot_event_stream_observer:add_listener(self(), TaskId), - receiver_events(TaskId, Req1), + ok = iot_container_task:subscribe(UUID, TaskId, self()), + receiver_events(TaskId, Req1), - {ok, Req1, Opts}. + {ok, Req1, Opts}; + {error, Reason} -> + Req1 = cowboy_req:reply(400, #{<<"Content-Type">> => <<"application/json">>}, iot_util:json_error(400, Reason), Req0), + {ok, Req1, Opts} + end. receiver_events(TaskId, Req) -> receive @@ -45,4 +48,23 @@ receiver_events(TaskId, Req) -> {stream_close, TaskId, Reason} -> CloseFrame = iolist_to_binary([<<"event: close\n">>, <<"data: ", Reason/binary, "\n">>, <<"\n">>]), ok = cowboy_req:stream_body(CloseFrame, fin, Req) + after 30000 -> + ok = cowboy_req:stream_body(<<": heartbeat\n\n">>, nofin, Req), + receiver_events(TaskId, Req) end. + +-spec parse_stream_params(map()) -> {ok, binary(), integer()} | {error, binary()}. +parse_stream_params(#{<<"uuid">> := UUID, <<"task_id">> := TaskId0}) when is_binary(UUID), is_binary(TaskId0) -> + try binary_to_integer(TaskId0) of + TaskId when TaskId >= 0 -> + {ok, UUID, TaskId}; + _ -> + {error, <<"task_id must be non-negative">>} + catch + error:badarg -> + {error, <<"task_id must be integer">>} + end; +parse_stream_params(#{<<"task_id">> := _TaskId0}) -> + {error, <<"uuid required">>}; +parse_stream_params(_) -> + {error, <<"uuid and task_id required">>}. diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index c24f528..4d42a85 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -207,19 +207,22 @@ handle_request_frame(Ref, Body, State) -> handle_message_frame({data, #{route_key := RouteKey, metric := Metric}}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> iot_host:handle(HostPid, {data, RouteKey, Metric}), {noreply, State}; -handle_message_frame({task_event, Event}, State = #state{host_pid = HostPid}) when is_pid(HostPid) -> - handle_event_stream_frame(Event), +handle_message_frame({task_event, Event}, State = #state{uuid = UUID, host_pid = HostPid}) when is_binary(UUID), is_pid(HostPid) -> + handle_event_stream_frame(UUID, Event), {noreply, State}; handle_message_frame(Body, State) -> logger:warning("[ssl_channel] unsupported message body: ~p", [Body]), {noreply, State}. --spec handle_event_stream_frame(map()) -> any(). -handle_event_stream_frame(#{task_id := TaskId, type := <<"close">>, stream := Reason}) -> - iot_event_stream_observer:stream_close(TaskId, Reason); -handle_event_stream_frame(#{task_id := TaskId, type := Type, stream := Stream}) -> - logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]), - iot_event_stream_observer:stream_data(TaskId, Type, Stream). +-spec handle_event_stream_frame(binary(), map()) -> ok. +handle_event_stream_frame(UUID, #{task_id := TaskId, type := <<"close">>, stream := Reason}) -> + iot_container_task_sup:close(UUID, TaskId, Reason); +handle_event_stream_frame(UUID, #{task_id := TaskId, type := Type, stream := Stream}) -> + logger:debug("[ssl_channel] get uuid: ~p, task_id: ~p, type: ~ts, stream: ~ts", [UUID, TaskId, Type, Stream]), + iot_container_task_sup:stream(UUID, TaskId, Type, Stream); +handle_event_stream_frame(UUID, Event) -> + logger:warning("[ssl_channel] invalid task_event, uuid: ~p, event: ~p", [UUID, Event]), + ok. -spec handle_command_response_frame(reference(), tuple(), #state{}) -> {noreply, #state{}}.