service管理

apps/efka/include/efka_tables.hrl

@@ -8,6 +8,9 @@
 %%%-------------------------------------------------------------------
 -author("anlicheng").
 
+-define(SERVICE_STOPPED, 0).
+-define(SERVICE_RUNNING, 1).
+
 %% 用来保存微服务
 -record(service, {
     service_id :: binary(),
@@ -15,5 +18,7 @@
     %% 配置信息, 微服务启动的时候自己注册的信息
     meta_data = #{} :: map(),
     %% 状态： 0: 停止, 1: 运行中
-    status = 0
+    status = 0,
+    create_ts  = 0 :: integer(),
+    update_ts  = 0 :: integer()
 }).

apps/efka/src/docker/docker_commands.erl

@@ -42,7 +42,7 @@ create_container(ContainerName, ContainerDir, Config) when is_binary(ContainerNa
     Volumes = [<<ConfigFile/binary, ":/usr/local/etc/service.conf">>|Volumes0],
     NewConfig = Config#{<<"volumes">> => Volumes},
 
-    Options = build_options(NewConfig),
+    Options = build_options(ContainerName, NewConfig),
     lists:foreach(fun({K, V}) -> lager:debug("~p => ~p", [K, V]) end, maps:to_list(Options)),
 
     Body = iolist_to_binary(jiffy:encode(Options, [force_utf8])),
@@ -201,12 +201,15 @@ inspect_container(ContainerId) when is_binary(ContainerId) ->
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 
 %% 构建最终 JSON Map
-build_options(Config) when is_map(Config) ->
+build_options(ContainerName, Config) when is_binary(ContainerName), is_map(Config) ->
+    %% 容器名称作为环境变量，注入到运行时环境中
+    Envs0 = maps:get(<<"envs">>, Config, []),
+    Envs = [<<"CONTAINER_NAME=", ContainerName>>|Envs0],
     #{
         <<"Image">> => maps:get(<<"image">>, Config, <<>>),
         <<"Cmd">> => maps:get(<<"command">>, Config, []),
         <<"Entrypoint">> => maps:get(<<"entrypoint">>, Config, []),
-        <<"Env">> => maps:get(<<"envs">>, Config, []),
+        <<"Env">> => Envs,
         <<"Labels">> => maps:get(<<"labels">>, Config, #{}),
         <<"Volumes">> => build_volumes(Config),
         <<"User">> => maps:get(<<"user">>, Config, <<>>),

apps/efka/src/models/service_model.erl

@@ -83,8 +83,18 @@ init([]) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_call({insert, Service}, _From, State = #state{}) ->
+handle_call({insert, Service = #service{service_id = ServiceId}}, _From, State = #state{}) ->
-    ok = dets:insert(?TAB, Service),
+    case dets:lookup(?TAB, ServiceId) of
+        [] ->
+            ok = dets:insert(?TAB, Service);
+        [OldService] ->
+            NewService = OldService#service{
+                meta_data = Service#service.meta_data,
+                container_name = Service#service.container_name,
+                update_ts = Service#service.update_ts
+            },
+            ok = dets:insert(?TAB, NewService)
+    end,
     {reply, ok, State};
 
 handle_call({change_status, ServiceId, NewStatus}, _From, State = #state{}) ->

apps/efka/src/ws_channel.erl

@@ -96,7 +96,15 @@ handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> :=
 
             %% 更新微服务的状态
             MetaData = maps:get(<<"meta_data">>, Params, #{}),
-            ok = service_model:insert(#service{service_id = ServiceId, status = 1, meta_data = MetaData}),
+            ContainerName = maps:get(<<"container_name">>, Params, <<>>),
+            ok = service_model:insert(#service{
+                service_id = ServiceId,
+                container_name = ContainerName,
+                status = ?SERVICE_RUNNING,
+                meta_data = MetaData,
+                create_ts = efka_util:timestamp(),
+                update_ts = efka_util:timestamp()
+            }),
 
             {reply, {text, Reply}, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}};
         {error, Error} ->