From 4f6556c2a2868b01052ddd6ee5a202aa0c4ed579 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Wed, 17 Sep 2025 11:33:25 +0800 Subject: [PATCH] fix service --- apps/efka/src/efka_service.erl | 104 ++------------------------------- 1 file changed, 5 insertions(+), 99 deletions(-) diff --git a/apps/efka/src/efka_service.erl b/apps/efka/src/efka_service.erl index a26a4b2..e8e6500 100644 --- a/apps/efka/src/efka_service.erl +++ b/apps/efka/src/efka_service.erl @@ -17,7 +17,7 @@ %% API -export([start_link/2]). -export([get_name/1, get_pid/1, attach_channel/2]). --export([push_config/3, request_config/1, invoke/3]). +-export([invoke/3]). -export([metric_data/4, send_event/3]). %% gen_server callbacks @@ -27,14 +27,8 @@ service_id :: binary(), %% 通道id信息 channel_pid :: pid() | undefined, - %% 当前进程的port信息, OSPid = erlang:port_info(Port, os_pid) - port :: undefined | port(), - %% 系统对应的pid - os_pid :: undefined | integer(), - %% 配置信息 - manifest :: undefined | efka_manifest:manifest(), - inflight = #{}, + inflight = #{}, %% 映射关系: #{Ref => Fun} callbacks = #{} }). @@ -51,18 +45,10 @@ get_name(ServiceId) when is_binary(ServiceId) -> get_pid(ServiceId) when is_binary(ServiceId) -> whereis(get_name(ServiceId)). --spec push_config(Pid :: pid(), Ref :: reference(), ConfigJson :: binary()) -> no_return(). -push_config(Pid, Ref, ConfigJson) when is_pid(Pid), is_binary(ConfigJson) -> - gen_server:cast(Pid, {push_config, Ref, self(), ConfigJson}). - -spec invoke(Pid :: pid(), Ref :: reference(), Payload :: binary()) -> no_return(). invoke(Pid, Ref, Payload) when is_pid(Pid), is_reference(Ref), is_binary(Payload) -> gen_server:cast(Pid, {invoke, Ref, self(), Payload}). --spec request_config(Pid :: pid()) -> {ok, Config :: binary()}. -request_config(Pid) when is_pid(Pid) -> - gen_server:call(Pid, request_config). - -spec metric_data(Pid :: pid(), DeviceUUID :: binary(), RouteKey :: binary(), Metric :: binary()) -> no_return(). metric_data(Pid, DeviceUUID, RouteKey, Metric) when is_pid(Pid), is_binary(DeviceUUID), is_binary(RouteKey), is_binary(Metric) -> gen_server:cast(Pid, {metric_data, DeviceUUID, RouteKey, Metric}). @@ -93,28 +79,8 @@ start_link(Name, ServiceId) when is_atom(Name), is_binary(ServiceId) -> init([ServiceId]) -> %% supervisor进程通过exit(ChildPid, shutdown)调用的时候，确保terminate函数被调用 erlang:process_flag(trap_exit, true), - case service_model:get_service(ServiceId) of - error -> - lager:notice("[efka_service] service_id: ~p, not found", [ServiceId]), - ignore; - {ok, #service{root_dir = RootDir}} -> - %% 第一次启动，要求必须成功；只有第一次启动成功，后续的重启逻辑才有意义 - case efka_manifest:new(RootDir) of - {ok, Manifest} -> - case efka_manifest:startup(Manifest) of - {ok, Port} -> - {os_pid, OSPid} = erlang:port_info(Port, os_pid), - lager:debug("[efka_service] service: ~p, port: ~p, boot_service success os_pid: ~p", [ServiceId, Port, OSPid]), - {ok, #state{service_id = ServiceId, manifest = Manifest, port = Port, os_pid = OSPid}}; - {error, Reason} -> - lager:debug("[efka_service] service: ~p, boot_service get error: ~p", [ServiceId, Reason]), - {stop, Reason} - end; - {error, Reason} -> - lager:notice("[efka_service] service: ~p, read manifest.json get error: ~p", [ServiceId, Reason]), - ignore - end - end. + lager:debug("[efka_service] service_id: ~p, started", [ServiceId]), + {ok, #state{service_id = ServiceId}}. %% @private %% @doc Handling call messages @@ -137,15 +103,6 @@ handle_call({attach_channel, ChannelPid}, _From, State = #state{channel_pid = Ol {reply, {error, <<"channel exists">>}, State} end; -%% 请求参数项 done -handle_call(request_config, _From, State = #state{service_id = ServiceId}) -> - case service_model:get_config_json(ServiceId) of - {ok, ConfigJson} -> - {reply, {ok, ConfigJson}, State}; - error -> - {reply, {ok, <<>>}, State} - end; - handle_call(_Request, _From, State = #state{}) -> {reply, ok, State}. @@ -165,19 +122,6 @@ handle_cast({send_event, EventType, Params}, State = #state{service_id = Service lager:debug("[efka_service] send_event, service_id: ~p, event_type: ~p, params: ~p", [ServiceId, EventType, Params]), {noreply, State}; -%% 推送配置项目 -handle_cast({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{channel_pid = ChannelPid, service_id = ServiceId, inflight = Inflight, callbacks = Callbacks}) -> - case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of - true -> - gen_channel:push_config(ChannelPid, Ref, self(), ConfigJson), - %% 设置成功，需要更新微服务的配置 - CB = fun() -> service_model:set_config(ServiceId, ConfigJson) end, - {noreply, State#state{inflight = maps:put(Ref, ReceiverPid, Inflight), callbacks = maps:put(Ref, CB, Callbacks)}}; - false -> - ReceiverPid ! {service_reply, Ref, {error, <<"channel is not alive">>}}, - {noreply, State} - end; - %% 推送配置项目 handle_cast({invoke, Ref, ReceiverPid, Payload}, State = #state{channel_pid = ChannelPid, inflight = Inflight}) -> case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of @@ -198,18 +142,6 @@ handle_cast(_Request, State = #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). -%% 重启服务 -handle_info({timeout, _, reboot_service}, State = #state{service_id = ServiceId, manifest = Manifest}) -> - case efka_manifest:startup(Manifest) of - {ok, Port} -> - {os_pid, OSPid} = erlang:port_info(Port, os_pid), - lager:debug("[efka_service] service_id: ~p, reboot success, port: ~p, os_pid: ~p", [ServiceId, Port, OSPid]), - {noreply, State#state{port = Port, os_pid = OSPid}}; - {error, Reason} -> - lager:debug("[efka_service] service_id: ~p, boot_service get error: ~p", [ServiceId, Reason]), - try_reboot(), - {noreply, State} - end; %% 处理channel的回复 handle_info({channel_reply, Ref, Reply}, State = #state{inflight = Inflight, callbacks = Callbacks}) -> @@ -226,17 +158,6 @@ handle_info({Port, {data, Data}}, State = #state{service_id = ServiceId}) when i lager:debug("[efka_service] service_id: ~p, port data: ~p", [ServiceId, Data]), {noreply, State}; -%% 处理port的消息, Port的被动关闭会触发；因此这个时候的Port和State.port的值是相等的 -handle_info({Port, {exit_status, Code}}, State = #state{service_id = ServiceId}) when is_port(Port) -> - lager:debug("[efka_service] service_id: ~p, port: ~p, exit with code: ~p", [ServiceId, Port, Code]), - {noreply, State#state{port = undefined, os_pid = undefined}}; - -%% 处理port的退出消息 -handle_info({'EXIT', Port, Reason}, State = #state{service_id = ServiceId}) when is_port(Port) -> - lager:debug("[efka_service] service_id: ~p, port: ~p, exit with reason: ~p", [ServiceId, Port, Reason]), - try_reboot(), - {noreply, State#state{port = undefined, os_pid = undefined}}; - %% 处理channel进程的退出 handle_info({'DOWN', _Ref, process, ChannelPid, Reason}, State = #state{channel_pid = ChannelPid, service_id = ServiceId}) -> lager:debug("[efka_service] service_id: ~p, channel exited: ~p", [ServiceId, Reason]), @@ -249,9 +170,7 @@ handle_info({'DOWN', _Ref, process, ChannelPid, Reason}, State = #state{channel_ %% with Reason. The return value is ignored. -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), State :: #state{}) -> term()). -terminate(Reason, _State = #state{service_id = ServiceId, port = Port, os_pid = OSPid}) -> - erlang:is_port(Port) andalso erlang:port_close(Port), - catch kill_os_pid(OSPid), +terminate(Reason, _State = #state{service_id = ServiceId}) -> lager:debug("[efka_service] service_id: ~p, terminate with reason: ~p", [ServiceId, Reason]), ok. @@ -267,19 +186,6 @@ code_change(_OldVsn, State = #state{}, _Extra) -> %%% Internal functions %%%=================================================================== -%% 关闭系统进程 --spec kill_os_pid(port() | undefined) -> no_return(). -kill_os_pid(undefined) -> - ok; -kill_os_pid(OSPid) when is_integer(OSPid) -> - Cmd = lists:flatten(io_lib:format("kill -9 ~p", [OSPid])), - lager:debug("kill cmd is: ~p", [Cmd]), - os:cmd(Cmd). - --spec try_reboot() -> no_return(). -try_reboot() -> - erlang:start_timer(5000, self(), reboot_service). - -spec trigger_callback(Ref :: reference(), Callbacks :: map()) -> NewCallbacks :: map(). trigger_callback(Ref, Callbacks) -> case maps:take(Ref, Callbacks) of