fix docker commands

apps/efka/src/docker/docker_http.erl

@@ -27,6 +27,8 @@ receive_response(ConnPid, StreamRef) ->
     receive
         {gun_response, ConnPid, StreamRef, nofin, Status, Headers} ->
             receive_body(ConnPid, StreamRef, Status, Headers, <<>>);
+        {gun_response, ConnPid, StreamRef, fin, Status, Headers} ->
+            {ok, Status, Headers, <<>>};
         {gun_down, ConnPid, _, Reason, _} ->
             {error, {http_closed, Reason}}
     after 5000 ->

apps/efka/src/docker/docker_uds_commands.erl

@@ -20,8 +20,10 @@ test() ->
 
 test1() ->
     Id = <<"redpanda-console">>,
-    is_container_running(Id).
-
+    StopRes = stop_container(Id),
+    lager:debug("stop res: ~p", [StopRes]),
+    StartRes = start_container(Id),
+    lager:debug("start res: ~p", [StartRes]).
 
 test_create_container() ->
     M = #{
@@ -161,52 +163,51 @@ is_container_running(ContainerId) when is_binary(ContainerId) ->
 
 -spec check_container_exist(ContainerName :: binary()) -> boolean().
 check_container_exist(ContainerName) when is_binary(ContainerName) ->
-    PortSettings = [stream, exit_status, use_stdio, binary],
-    ExecCmd = "docker inspect --type=container " ++ binary_to_list(ContainerName) ++ " >/dev/null 2>&1",
-    lager:debug("check_container_exist cmd : ~p", [ExecCmd]),
-    case catch erlang:open_port({spawn, ExecCmd}, PortSettings) of
-        Port when is_port(Port) ->
-            case gather_output(Port) of
-                {0, _} ->
-                    true;
-                {_ExitCode, _Error} ->
-                    false
-            end;
-        _Error ->
+    case inspect_container(ContainerName) of
+        {ok, #{<<"Id">> := Id}} when is_binary(Id) ->
+            true;
+        _ ->
             false
     end.
 
 -spec start_container(ContainerName :: binary()) -> ok | {error, Reason :: binary()}.
 start_container(ContainerName) when is_binary(ContainerName) ->
-    PortSettings = [stream, exit_status, use_stdio, binary],
-    ExecCmd = "docker start " ++ binary_to_list(ContainerName) ++ " 2>&1",
-    case catch erlang:open_port({spawn, ExecCmd}, PortSettings) of
-        Port when is_port(Port) ->
-            case gather_output(Port) of
-                {0, _} ->
-                    ok;
-                {_ExitCode, Error} ->
-                    {error, Error}
-            end;
-        _Error ->
-            {error, <<"启动失败"/utf8>>}
+    Url = lists:flatten(io_lib:format("/containers/~s/start", [binary_to_list(ContainerName)])),
+    Headers = [
+        {<<"Content-Type">>, <<"application/json">>}
+    ],
+    case docker_http:request("POST", Url, undefined, Headers) of
+        {ok, 204, _Headers, _} ->
+            ok;
+        {ok, 304, _Headers, _} ->
+            {error, <<"container already started">>};
+        {ok, _StatusCode, _Header, ErrorResp} ->
+            case catch jiffy:decode(ErrorResp) of
+                #{<<"message">> := Msg} ->
+                    {error, Msg};
+                _ ->
+                    {error, ErrorResp}
+            end
     end.
 
 -spec stop_container(ContainerName :: binary()) -> ok | {error, Reason :: binary()}.
 stop_container(ContainerName) when is_binary(ContainerName) ->
-    PortSettings = [stream, exit_status, use_stdio, binary],
-    ExecCmd = "docker stop " ++ binary_to_list(ContainerName) ++ " 2>&1",
-    lager:debug("[docker_commands] cmd : ~p", [ExecCmd]),
-    case catch erlang:open_port({spawn, ExecCmd}, PortSettings) of
-        Port when is_port(Port) ->
-            case gather_output(Port) of
-                {0, _} ->
-                    ok;
-                {_ExitCode, Error} ->
-                    {error, Error}
-            end;
-        _Error ->
-            false
+    Url = lists:flatten(io_lib:format("/containers/~s/stop", [binary_to_list(ContainerName)])),
+    Headers = [
+        {<<"Content-Type">>, <<"application/json">>}
+    ],
+    case docker_http:request("POST", Url, undefined, Headers) of
+        {ok, 204, _Headers, _} ->
+            ok;
+        {ok, 304, _Headers, _} ->
+            {error, <<"container already stopped">>};
+        {ok, _StatusCode, _Header, ErrorResp} ->
+            case catch jiffy:decode(ErrorResp) of
+                #{<<"message">> := Msg} ->
+                    {error, Msg};
+                _ ->
+                    {error, ErrorResp}
+            end
     end.
 
 -spec inspect_container(ContainerId :: binary()) -> {ok, Json :: map()} | {error, Error :: any()}.
@@ -232,16 +233,6 @@ inspect_container(ContainerId) when is_binary(ContainerId) ->
 %%% helper methods
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 
-gather_output(Port) ->
-    gather_output(Port, <<>>).
-gather_output(Port, Acc) ->
-    receive
-        {Port, {data, Data}} ->
-            gather_output(Port, [Acc, Data]);
-        {Port, {exit_status, Status}} ->
-            {Status, iolist_to_binary(Acc)}
-    end.
-
 %% 构建最终 JSON Map
 build_options(Config) when is_map(Config) ->
     #{

apps/efka/src/efka_sup.erl

@@ -37,14 +37,14 @@ init([]) ->
             modules => ['efka_service_sup']
         },
 
-        #{
-            id => 'docker_events',
-            start => {'docker_events', start_link, []},
-            restart => permanent,
-            shutdown => 2000,
-            type => worker,
-            modules => ['docker_events']
-        },
+        %#{
+        %    id => 'docker_events',
+        %    start => {'docker_events', start_link, []},
+        %    restart => permanent,
+        %    shutdown => 2000,
+        %    type => worker,
+        %    modules => ['docker_events']
+        %},
 
         #{
             id => 'efka_model_sup',