From 72030d55d64b69028e4564353ed3a4b0575d445f Mon Sep 17 00:00:00 2001
From: anlicheng <244108715@qq.com>
Date: Tue, 21 Apr 2026 15:59:40 +0800
Subject: [PATCH] =?UTF-8?q?fix=E6=B5=8B=E8=AF=95=E7=94=A8=E4=BE=8B?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 config/sys.config                   |   2 -
 docs/todo.md                        |  50 ++++
 src/efka_app.erl                    |  12 +-
 src/tests/docker_commands_tests.erl | 354 +++++++++++++++++++++-------
 4 files changed, 320 insertions(+), 98 deletions(-)
 create mode 100644 docs/todo.md

diff --git a/config/sys.config b/config/sys.config
index ce9ca95..ac03d12 100644
--- a/config/sys.config
+++ b/config/sys.config
@@ -4,8 +4,6 @@
 
     {dets_dir, "/usr/local/code/tmp/dets/"},
 
-    {upload_dir, "/usr/local/code/tmp/upload/"},
-
     {tcp_server, [
       {port, 18088}
     ]},
diff --git a/docs/todo.md b/docs/todo.md
new file mode 100644
index 0000000..27144f5
--- /dev/null
+++ b/docs/todo.md
@@ -0,0 +1,50 @@
+# TODO
+
+## 协议层
+
+### 高优先级
+
+- 将 `message.proto` 收敛为 `efka` 与 `iot` 共用的一份单一事实来源，避免协议定义漂移和双端重复维护。
+
+### 中优先级
+
+- 增强 protobuf 映射和 Docker JSON 生成相关的自动化契约测试。
+
+## Docker 层
+
+### 高优先级
+
+- 将 `docker_commands` 拆分为更清晰的层次：Docker HTTP 客户端、Docker JSON 构造层、以及 efka 本地补丁逻辑。
+- 明确 `docker_task_reporter` 的投递语义；当前只有内存缓冲，还缺少持久化能力和队列上限控制。
+
+### 中优先级
+
+- 丰富 `docker_deploy_manager` 的任务跟踪状态，把部署元数据、耗时、阶段、失败原因等信息纳入统一管理，避免后续再回头重构。
+- 统一 `docker_container_service` 的返回结构，避免调用方分支处理 `ok | {ok, binary()} | {error, binary()}` 这种混合形式。
+- 消除 `docker_commands` 中重复的 Docker HTTP 响应解析逻辑。
+
+### 低优先级
+
+- 对于像 `docker_events.erl` 这样当前未启用的模块，要么清理掉，要么补充说明其保留原因。
+
+## 订阅与通道层
+
+### 高优先级
+
+- 将 `efka_subscription` 当前基于列表扫描的匹配方式替换为更可扩展的索引结构，例如 trie 或基于 ETS 的索引。
+
+### 中优先级
+
+- 继续收敛 `efka_client` 的职责，将传输层状态管理与 request/cast 协议处理进一步拆开。
+- 评估并落实 `service_channel` 中 `subscribed_topics` 集合的用途；如果只是被动保存状态，则应简化。
+
+### 低优先级
+
+- 如果 `efka_client` 的缓存指标量继续增长，可以为缓存刷出增加批量发送或节流机制。
+
+## 基础设施层
+
+### 低优先级
+
+- 如果上传相关能力已经不再属于当前运行时，移除或补充说明残留的启动逻辑，例如 `ensure_upload_dir/0`。
+- 评估是否将 `efka_logger` 与 OTP `logger` 统一，避免长期维护两套日志链路。
diff --git a/src/efka_app.erl b/src/efka_app.erl
index 716a7c7..0e68ba2 100644
--- a/src/efka_app.erl
+++ b/src/efka_app.erl
@@ -11,7 +11,6 @@
 
 start(_StartType, _StartArgs) ->
     io:setopts([{encoding, unicode}]),
-    ensure_upload_dir(),
     %% 加速内存的回收
     erlang:system_flag(fullsweep_after, 16),
     start_http_server(),
@@ -41,13 +40,4 @@ start_http_server() ->
     ],
     {ok, Pid} = cowboy:start_clear(ws_listener, TransOpts, #{env => #{dispatch => Dispatcher}}),
 
-    logger:debug("[efka_app] websocket server start at: ~p, pid is: ~p", [Port, Pid]).
-
-ensure_upload_dir() ->
-    {ok, UploadDir} = application:get_env(efka, upload_dir),
-    case filelib:is_dir(UploadDir) of
-        true ->
-            ok;
-        false ->
-            ok = file:make_dir(UploadDir)
-    end.
+    logger:debug("[efka_app] websocket server start at: ~p, pid is: ~p", [Port, Pid]).
\ No newline at end of file
diff --git a/src/tests/docker_commands_tests.erl b/src/tests/docker_commands_tests.erl
index 7533912..de92a61 100644
--- a/src/tests/docker_commands_tests.erl
+++ b/src/tests/docker_commands_tests.erl
@@ -11,101 +11,285 @@
 -include("message_pb.hrl").
 
 %% API
--export([test_pull/0, test_commands/0, test_create_container/0]).
+-export([
+    test_all/0,
+    test_pull/0,
+    test_check_image_exist/0,
+    test_check_image_not_exist/0,
+    test_create_container/0,
+    test_create_container_without_create_options/0,
+    test_create_container_patches_options/0,
+    test_check_container_exist/0,
+    test_check_container_not_exist/0,
+    test_is_container_running/0,
+    test_start_container/0,
+    test_stop_container/0,
+    test_stop_container_with_timeout/0,
+    test_kill_container/0,
+    test_kill_container_with_signal/0,
+    test_remove_container/0,
+    test_remove_container_with_options/0,
+    test_get_containers/0
+]).
+
+-define(TEST_IMAGE, <<"docker.1ms.run/library/busybox:latest">>).
+
+test_all() ->
+    ok = test_pull(),
+    ok = test_check_image_exist(),
+    ok = test_check_image_not_exist(),
+    ok = test_create_container(),
+    ok = test_create_container_without_create_options(),
+    ok = test_create_container_patches_options(),
+    ok = test_check_container_exist(),
+    ok = test_check_container_not_exist(),
+    ok = test_is_container_running(),
+    ok = test_start_container(),
+    ok = test_stop_container(),
+    ok = test_stop_container_with_timeout(),
+    ok = test_kill_container(),
+    ok = test_kill_container_with_signal(),
+    ok = test_remove_container(),
+    ok = test_remove_container_with_options(),
+    ok = test_get_containers().
 
 test_pull() ->
-    Image = <<"docker.1ms.run/library/nginx:latest">>,
-    docker_commands:pull_image(Image, fun(Msg) -> logger:debug("msg is: ~p", [Msg]) end).
+    ok = docker_commands:pull_image(?TEST_IMAGE, fun(Msg) -> logger:debug("msg is: ~p", [Msg]) end).
 
-test_commands() ->
-    Id = <<"redpanda-console">>,
-    StopRes = docker_commands:stop_container(Id),
-    logger:debug("stop res: ~p", [StopRes]),
-    StartRes = docker_commands:start_container(Id),
-    logger:debug("start res: ~p", [StartRes]).
+test_check_image_exist() ->
+    ok = test_pull(),
+    true = docker_commands:check_image_exist(?TEST_IMAGE),
+    ok.
+
+test_check_image_not_exist() ->
+    false = docker_commands:check_image_exist(<<"docker.1ms.run/library/not-exists-for-efka-tests:latest">>),
+    ok.
 
 test_create_container() ->
+    Name = test_container_name(<<"create">>),
+    ContainerDir = prepare_container_dir(Name),
+    try
+        ok = test_pull(),
+        {ok, ContainerId} = docker_commands:create_container(ContainerDir, minimal_params(Name)),
+        true = is_binary(ContainerId),
+        ok
+    after
+        cleanup_container(Name)
+    end.
+
+test_create_container_without_create_options() ->
+    Name = test_container_name(<<"create-default">>),
+    ContainerDir = prepare_container_dir(Name),
+    try
+        ok = test_pull(),
+        {ok, ContainerId} = docker_commands:create_container(ContainerDir, #'ContainerDeployParams'{
+            container_name = Name
+        }),
+        true = is_binary(ContainerId),
+        Inspect = inspect_container_json(Name),
+        assert_patched_defaults(Name, ContainerDir, Inspect),
+        ok
+    after
+        cleanup_container(Name)
+    end.
+
+test_create_container_patches_options() ->
+    Name = test_container_name(<<"create-patch">>),
+    ContainerDir = prepare_container_dir(Name),
     Params = #'ContainerDeployParams'{
-        container_name = <<"my_nginx_new1">>,
+        container_name = Name,
         create = #'DockerCreateOptions'{
             config = #'DockerContainerConfig'{
-                image = <<"docker.1ms.run/library/nginx:latest">>,
-                cmd = [
-                    <<"nginx">>,
-                    <<"-g">>,
-                    <<"daemon off;">>
-                ],
-                entrypoint = [
-                    <<"/docker-entrypoint.sh">>
-                ],
-                env = [
-                    <<"ENV1=val1">>,
-                    <<"ENV2=val2">>
-                ],
-                volumes = [
-                    <<"/data">>,
-                    <<"/var/log">>
-                ],
-                labels = [
-                    {<<"role">>, <<"web">>},
-                    {<<"env">>, <<"prod">>}
-                ],
-                user = <<"www-data">>,
-                working_dir = <<"/app">>,
-                hostname = <<"myhost">>,
-                exposed_ports = [
-                    #'DockerExposedPort'{container_port = 80, protocol = <<"tcp">>},
-                    #'DockerExposedPort'{container_port = 443, protocol = <<"tcp">>}
-                ],
-                healthcheck = #'Healthcheck'{
-                    test = [
-                        <<"CMD-SHELL">>,
-                        <<"curl -f http://localhost || exit 1">>
-                    ],
-                    interval_ns = 30000000000,
-                    timeout_ns = 10000000000,
-                    retries = 3
-                }
+                image = ?TEST_IMAGE,
+                cmd = [<<"sleep">>, <<"300">>],
+                env = [<<"EXISTING_ENV=1">>],
+                volumes = [<<"/data">>]
             },
             host_config = #'DockerHostConfig'{
-                binds = [
-                    <<"/host/data:/data">>,
-                    <<"/host/log:/var/log">>
-                ],
-                network_mode = <<"bridge">>,
-                restart_policy = #'RestartPolicy'{name = <<"always">>},
-                privileged = true,
-                cap_add = [
-                    <<"NET_ADMIN">>
-                ],
-                cap_drop = [
-                    <<"MKNOD">>
-                ],
-                devices = [
-                    #'DeviceMapping'{host_path = <<"/dev/snd">>, container_path = <<"/dev/snd">>}
-                ],
-                memory = 512 * 1024 * 1024,
-                memory_reservation = 256 * 1024 * 1024,
-                cpu_shares = 512,
-                nano_cpus = 1500000000,
-                ulimits = [
-                    #'Ulimit'{name = <<"nofile">>, soft = 1024, hard = 2048}
-                ],
-                sysctls = [
-                    {<<"net.ipv4.ip_forward">>, <<"1">>}
-                ],
-                tmpfs = [
-                    {<<"/tmp">>, <<>>}
-                ],
-                extra_hosts = [
-                    <<"host1:192.168.0.1">>
-                ]
-            },
-            networking_config = #'DockerNetworkingConfig'{
-                endpoints = [
-                    #'DockerNetworkEndpoint'{name = <<"mynet">>}
-                ]
+                binds = [<<"/tmp:/tmp">>]
             }
         }
     },
-    docker_commands:create_container("/usr/local/code/efka/", Params).
+    try
+        ok = test_pull(),
+        {ok, _ContainerId} = docker_commands:create_container(ContainerDir, Params),
+        Inspect = inspect_container_json(Name),
+        assert_patched_defaults(Name, ContainerDir, Inspect),
+        #{<<"Config">> := #{<<"Env">> := Env, <<"Volumes">> := Volumes},
+            <<"HostConfig">> := #{<<"Binds">> := Binds}} = Inspect,
+        true = lists:member(<<"EXISTING_ENV=1">>, Env),
+        true = maps:is_key(<<"/data">>, Volumes),
+        true = lists:member(<<"/tmp:/tmp">>, Binds),
+        ok
+    after
+        cleanup_container(Name)
+    end.
+
+test_check_container_exist() ->
+    Name = test_container_name(<<"exist">>),
+    with_created_container(Name, fun(_ContainerDir, _ContainerId) ->
+        true = docker_commands:check_container_exist(Name),
+        ok
+    end).
+
+test_check_container_not_exist() ->
+    false = docker_commands:check_container_exist(test_container_name(<<"missing">>)),
+    ok.
+
+test_is_container_running() ->
+    Name = test_container_name(<<"running">>),
+    with_created_container(Name, fun(_ContainerDir, ContainerId) ->
+        false = docker_commands:is_container_running(ContainerId),
+        ok = docker_commands:start_container(Name),
+        true = docker_commands:is_container_running(ContainerId),
+        ok = docker_commands:stop_container(Name),
+        false = docker_commands:is_container_running(ContainerId),
+        ok
+    end).
+
+test_start_container() ->
+    Name = test_container_name(<<"start">>),
+    with_created_container(Name, fun(_ContainerDir, ContainerId) ->
+        ok = docker_commands:start_container(Name),
+        true = docker_commands:is_container_running(ContainerId),
+        ok
+    end).
+
+test_stop_container() ->
+    Name = test_container_name(<<"stop">>),
+    with_started_container(Name, fun(_ContainerDir, ContainerId) ->
+        ok = docker_commands:stop_container(Name),
+        false = docker_commands:is_container_running(ContainerId),
+        ok
+    end).
+
+test_stop_container_with_timeout() ->
+    Name = test_container_name(<<"stop-timeout">>),
+    with_started_container(Name, fun(_ContainerDir, ContainerId) ->
+        ok = docker_commands:stop_container(Name, 1),
+        false = docker_commands:is_container_running(ContainerId),
+        ok
+    end).
+
+test_kill_container() ->
+    Name = test_container_name(<<"kill">>),
+    with_started_container(Name, fun(_ContainerDir, ContainerId) ->
+        ok = docker_commands:kill_container(Name),
+        timer:sleep(200),
+        false = docker_commands:is_container_running(ContainerId),
+        ok
+    end).
+
+test_kill_container_with_signal() ->
+    Name = test_container_name(<<"kill-signal">>),
+    with_started_container(Name, fun(_ContainerDir, ContainerId) ->
+        ok = docker_commands:kill_container(Name, <<"SIGKILL">>),
+        timer:sleep(200),
+        false = docker_commands:is_container_running(ContainerId),
+        ok
+    end).
+
+test_remove_container() ->
+    Name = test_container_name(<<"remove">>),
+    with_created_container(Name, fun(_ContainerDir, _ContainerId) ->
+        ok = docker_commands:remove_container(Name),
+        false = docker_commands:check_container_exist(Name),
+        ok
+    end, false).
+
+test_remove_container_with_options() ->
+    Name = test_container_name(<<"remove-opts">>),
+    with_started_container(Name, fun(_ContainerDir, _ContainerId) ->
+        ok = docker_commands:remove_container(Name, true, false),
+        false = docker_commands:check_container_exist(Name),
+        ok
+    end, false).
+
+test_get_containers() ->
+    Name = test_container_name(<<"list">>),
+    with_created_container(Name, fun(_ContainerDir, ContainerId) ->
+        {ok, Containers} = docker_commands:get_containers(),
+        true = is_list(Containers),
+        true = contains_container(Name, ContainerId, Containers),
+        ok
+    end).
+
+with_created_container(Name, Fun) ->
+    with_created_container(Name, Fun, true).
+
+with_created_container(Name, Fun, Cleanup) when is_binary(Name), is_function(Fun, 2), is_boolean(Cleanup) ->
+    ContainerDir = prepare_container_dir(Name),
+    try
+        ok = test_pull(),
+        {ok, ContainerId} = docker_commands:create_container(ContainerDir, minimal_params(Name)),
+        ok = Fun(ContainerDir, ContainerId)
+    after
+        case Cleanup of
+            true ->
+                cleanup_container(Name);
+            false ->
+                ok
+        end
+    end.
+
+with_started_container(Name, Fun) ->
+    with_started_container(Name, Fun, true).
+
+with_started_container(Name, Fun, Cleanup) when is_binary(Name), is_function(Fun, 2), is_boolean(Cleanup) ->
+    with_created_container(Name, fun(ContainerDir, ContainerId) ->
+        ok = docker_commands:start_container(Name),
+        ok = Fun(ContainerDir, ContainerId)
+    end, Cleanup).
+
+minimal_params(Name) when is_binary(Name) ->
+    #'ContainerDeployParams'{
+        container_name = Name,
+        create = #'DockerCreateOptions'{
+            config = #'DockerContainerConfig'{
+                image = ?TEST_IMAGE,
+                cmd = [<<"sleep">>, <<"300">>]
+            }
+        }
+    }.
+
+prepare_container_dir(Name) when is_binary(Name) ->
+    Dir = lists:flatten(io_lib:format("/tmp/efka_docker_tests/~ts/", [Name])),
+    ok = filelib:ensure_dir(Dir ++ "placeholder"),
+    ok = file:write_file(Dir ++ "service.conf", <<>>, [write]),
+    Dir.
+
+cleanup_container(Name) when is_binary(Name) ->
+    _ = docker_commands:remove_container(Name, true, false),
+    ok.
+
+inspect_container_json(Name) when is_binary(Name) ->
+    Url = lists:flatten(io_lib:format("/containers/~s/json", [binary_to_list(Name)])),
+    {ok, 200, _Headers, Resp} = docker_http:request("GET", Url, <<>>, []),
+    jiffy:decode(Resp, [return_maps]).
+
+assert_patched_defaults(Name, ContainerDir, Inspect)
+  when is_binary(Name), is_list(ContainerDir), is_map(Inspect) ->
+    ConfigFile = list_to_binary(docker_helper:get_config_file(ContainerDir)),
+    ExpectedBind = <<ConfigFile/binary, ":/usr/local/etc/service.conf">>,
+    #{<<"Config">> := #{<<"Env">> := Env, <<"Volumes">> := Volumes},
+        <<"HostConfig">> := #{<<"Binds">> := Binds}} = Inspect,
+    true = lists:member(<<"CONTAINER_NAME=", Name/binary>>, Env),
+    true = maps:is_key(<<"/usr/local/etc/service.conf">>, Volumes),
+    true = lists:member(ExpectedBind, Binds),
+    ok.
+
+contains_container(Name, ContainerId, Containers) when is_binary(Name), is_binary(ContainerId), is_list(Containers) ->
+    lists:any(fun(Container) -> container_matches(Name, ContainerId, Container) end, Containers).
+
+container_matches(Name, ContainerId, #{<<"Id">> := Id, <<"Names">> := Names}) when is_binary(Id), is_list(Names) ->
+    lists:member(<<"/", Name/binary>>, Names) orelse has_id_prefix(ContainerId, Id);
+container_matches(_Name, _ContainerId, _Container) ->
+    false.
+
+has_id_prefix(ExpectedId, ActualId) when is_binary(ExpectedId), is_binary(ActualId) ->
+    PrefixLen = erlang:min(byte_size(ExpectedId), byte_size(ActualId)),
+    binary:part(ExpectedId, 0, PrefixLen) =:= binary:part(ActualId, 0, PrefixLen).
+
+test_container_name(Prefix) when is_binary(Prefix) ->
+    Suffix = integer_to_binary(erlang:unique_integer([positive])),
+    <<"efka-test-", Prefix/binary, "-", Suffix/binary>>.