yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #9781 from olcai/delete-zip-file-from-trace-log-download 

fix(emqx_management): delete files after trace log download
This commit is contained in:
Erik Timan 2023-01-18 13:05:39 +01:00 committed by GitHub
parent 8e1475addb 42182279b7
commit 46fc69cd48
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 128 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
apps/emqx_management/src/emqx_mgmt_api_trace.erl
View File

@ -20,6 +20,7 @@
-include_lib("kernel/include/file.hrl"). -include_lib("kernel/include/file.hrl").
-include_lib("typerefl/include/types.hrl"). -include_lib("typerefl/include/types.hrl").
-include_lib("emqx/include/logger.hrl"). -include_lib("emqx/include/logger.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-export([ -export([
api_spec/0, api_spec/0,
@ -461,16 +462,31 @@ download_trace_log(get, #{bindings := #{name := Name}, query_string := Query}) -
case parse_node(Query, undefined) of case parse_node(Query, undefined) of
{ok, Node} -> {ok, Node} ->
TraceFiles = collect_trace_file(Node, TraceLog), TraceFiles = collect_trace_file(Node, TraceLog),
ZipDir = emqx_trace:zip_dir(), %% We generate a session ID so that we name files
%% with unique names. Then we won't cause
%% overwrites for concurrent requests.
SessionId = emqx_misc:gen_id(),
ZipDir = filename:join([emqx_trace:zip_dir(), SessionId]),
ok = file:make_dir(ZipDir),
%% Write files to ZipDir and create an in-memory zip file
Zips = group_trace_file(ZipDir, TraceLog, TraceFiles), Zips = group_trace_file(ZipDir, TraceLog, TraceFiles),
FileName = binary_to_list(Name) ++ ".zip", ZipName = binary_to_list(Name) ++ ".zip",
ZipFileName = filename:join([ZipDir, FileName]), Binary =
{ok, ZipFile} = zip:zip(ZipFileName, Zips, [{cwd, ZipDir}]), try
%% emqx_trace:delete_files_after_send(ZipFileName, Zips), {ok, {ZipName, Bin}} = zip:zip(ZipName, Zips, [memory, {cwd, ZipDir}]),
%% TODO use file replace file_binary.(delete file after send is not ready now). Bin
{ok, Binary} = file:read_file(ZipFile), after
ZipName = filename:basename(ZipFile), %% emqx_trace:delete_files_after_send(ZipFileName, Zips),
_ = file:delete(ZipFile), %% TODO use file replace file_binary.(delete file after send is not ready now).
ok = file:del_dir_r(ZipDir)
end,
?tp(trace_api_download_trace_log, #{
files => Zips,
name => Name,
session_id => SessionId,
zip_dir => ZipDir,
zip_name => ZipName
}),
Headers = #{ Headers = #{
<<"content-type">> => <<"application/x-zip">>, <<"content-type">> => <<"application/x-zip">>,
<<"content-disposition">> => iolist_to_binary( <<"content-disposition">> => iolist_to_binary(

15
apps/emqx_management/test/emqx_mgmt_api_test_util.erl
View File

@ -61,7 +61,8 @@ uri(Parts) ->
%% compatible_mode will return as same as 'emqx_dashboard_api_test_helpers:request' %% compatible_mode will return as same as 'emqx_dashboard_api_test_helpers:request'
request_api_with_body(Method, Url, Body) -> request_api_with_body(Method, Url, Body) ->
request_api(Method, Url, [], auth_header_(), Body, #{compatible_mode => true}). Opts = #{compatible_mode => true, httpc_req_opts => [{body_format, binary}]},
request_api(Method, Url, [], auth_header_(), Body, Opts).
request_api(Method, Url) -> request_api(Method, Url) ->
request_api(Method, Url, auth_header_()). request_api(Method, Url, auth_header_()).
@ -111,15 +112,9 @@ request_api(Method, Url, QueryParams, AuthOrHeaders, Body, Opts) when
do_request_api(Method, Request, Opts) -> do_request_api(Method, Request, Opts) ->
ReturnAll = maps:get(return_all, Opts, false), ReturnAll = maps:get(return_all, Opts, false),
CompatibleMode = maps:get(compatible_mode, Opts, false), CompatibleMode = maps:get(compatible_mode, Opts, false),
ReqOpts = HttpcReqOpts = maps:get(httpc_req_opts, Opts, []),
case CompatibleMode of ct:pal("Method: ~p, Request: ~p, Opts: ~p", [Method, Request, Opts]),
true -> case httpc:request(Method, Request, [], HttpcReqOpts) of
[{body_format, binary}];
_ ->
[]
end,
ct:pal("Method: ~p, Request: ~p", [Method, Request]),
case httpc:request(Method, Request, [], ReqOpts) of
{error, socket_closed_remotely} -> {error, socket_closed_remotely} ->
{error, socket_closed_remotely}; {error, socket_closed_remotely};
{ok, {{_, Code, _}, _Headers, Body}} when CompatibleMode -> {ok, {{_, Code, _}, _Headers, Body}} when CompatibleMode ->

179
apps/emqx_management/test/emqx_mgmt_api_trace_SUITE.erl
View File

@ -26,12 +26,6 @@
-include_lib("stdlib/include/zip.hrl"). -include_lib("stdlib/include/zip.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -include_lib("snabbkaffe/include/snabbkaffe.hrl").
-define(HOST, "http://127.0.0.1:18083/").
-define(API_VERSION, "v5").
-define(BASE_PATH, "api").
-import(emqx_dashboard_SUITE, [auth_header_/0]).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Setups %% Setups
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
@ -49,14 +43,14 @@ end_per_suite(_) ->
t_http_test(_Config) -> t_http_test(_Config) ->
emqx_trace:clear(), emqx_trace:clear(),
load(), load(),
Header = auth_header_(),
%% list %% list
{ok, Empty} = request_api(get, api_path("trace"), Header), {ok, Empty} = request_api(get, api_path("trace")),
?assertEqual([], json(Empty)), ?assertEqual([], json(Empty)),
%% create %% create
ErrorTrace = #{}, ErrorTrace = #{},
{error, {"HTTP/1.1", 400, "Bad Request"}, Body} = Opts = #{return_all => true},
request_api(post, api_path("trace"), Header, ErrorTrace), {error, {{"HTTP/1.1", 400, "Bad Request"}, _, Body}} =
emqx_mgmt_api_test_util:request_api(post, api_path("trace"), [], [], ErrorTrace, Opts),
?assertMatch(#{<<"code">> := <<"BAD_REQUEST">>}, json(Body)), ?assertMatch(#{<<"code">> := <<"BAD_REQUEST">>}, json(Body)),
Name = <<"test-name">>, Name = <<"test-name">>,
@ -66,15 +60,15 @@ t_http_test(_Config) ->
{<<"topic">>, <<"/x/y/z">>} {<<"topic">>, <<"/x/y/z">>}
], ],
{ok, Create} = request_api(post, api_path("trace"), Header, Trace), {ok, Create} = request_api(post, api_path("trace"), Trace),
?assertMatch(#{<<"name">> := Name}, json(Create)), ?assertMatch(#{<<"name">> := Name}, json(Create)),
{ok, List} = request_api(get, api_path("trace"), Header), {ok, List} = request_api(get, api_path("trace")),
[Data] = json(List), [Data] = json(List),
?assertEqual(Name, maps:get(<<"name">>, Data)), ?assertEqual(Name, maps:get(<<"name">>, Data)),
%% update %% update
{ok, Update} = request_api(put, api_path("trace/test-name/stop"), Header, #{}), {ok, Update} = request_api(put, api_path("trace/test-name/stop"), #{}),
?assertEqual( ?assertEqual(
#{ #{
<<"enable">> => false, <<"enable">> => false,
@ -84,10 +78,10 @@ t_http_test(_Config) ->
), ),
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 404, _}, _}, {error, {"HTTP/1.1", 404, _}},
request_api(put, api_path("trace/test-name-not-found/stop"), Header, #{}) request_api(put, api_path("trace/test-name-not-found/stop"), #{})
), ),
{ok, List1} = request_api(get, api_path("trace"), Header), {ok, List1} = request_api(get, api_path("trace")),
[Data1] = json(List1), [Data1] = json(List1),
Node = atom_to_binary(node()), Node = atom_to_binary(node()),
?assertMatch( ?assertMatch(
@ -104,11 +98,11 @@ t_http_test(_Config) ->
), ),
%% delete %% delete
{ok, Delete} = request_api(delete, api_path("trace/test-name"), Header), {ok, Delete} = request_api(delete, api_path("trace/test-name")),
?assertEqual(<<>>, Delete), ?assertEqual(<<>>, Delete),
{error, {"HTTP/1.1", 404, "Not Found"}, DeleteNotFound} = {error, {{"HTTP/1.1", 404, "Not Found"}, _, DeleteNotFound}} =
request_api(delete, api_path("trace/test-name"), Header), emqx_mgmt_api_test_util:request_api(delete, api_path("trace/test-name"), [], [], [], Opts),
?assertEqual( ?assertEqual(
#{ #{
<<"code">> => <<"NOT_FOUND">>, <<"code">> => <<"NOT_FOUND">>,
@ -117,14 +111,14 @@ t_http_test(_Config) ->
json(DeleteNotFound) json(DeleteNotFound)
), ),
{ok, List2} = request_api(get, api_path("trace"), Header), {ok, List2} = request_api(get, api_path("trace")),
?assertEqual([], json(List2)), ?assertEqual([], json(List2)),
%% clear %% clear
{ok, Create1} = request_api(post, api_path("trace"), Header, Trace), {ok, Create1} = request_api(post, api_path("trace"), Trace),
?assertMatch(#{<<"name">> := Name}, json(Create1)), ?assertMatch(#{<<"name">> := Name}, json(Create1)),
{ok, Clear} = request_api(delete, api_path("trace"), Header), {ok, Clear} = request_api(delete, api_path("trace")),
?assertEqual(<<>>, Clear), ?assertEqual(<<>>, Clear),
unload(), unload(),
@ -132,27 +126,26 @@ t_http_test(_Config) ->
t_create_failed(_Config) -> t_create_failed(_Config) ->
load(), load(),
Header = auth_header_(),
Trace = [{<<"type">>, <<"topic">>}, {<<"topic">>, <<"/x/y/z">>}], Trace = [{<<"type">>, <<"topic">>}, {<<"topic">>, <<"/x/y/z">>}],
BadName1 = {<<"name">>, <<"test/bad">>}, BadName1 = {<<"name">>, <<"test/bad">>},
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 400, _}, _}, {error, {"HTTP/1.1", 400, _}},
request_api(post, api_path("trace"), Header, [BadName1 | Trace]) request_api(post, api_path("trace"), [BadName1 | Trace])
), ),
BadName2 = {<<"name">>, list_to_binary(lists:duplicate(257, "t"))}, BadName2 = {<<"name">>, list_to_binary(lists:duplicate(257, "t"))},
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 400, _}, _}, {error, {"HTTP/1.1", 400, _}},
request_api(post, api_path("trace"), Header, [BadName2 | Trace]) request_api(post, api_path("trace"), [BadName2 | Trace])
), ),
%% already_exist %% already_exist
GoodName = {<<"name">>, <<"test-name-0">>}, GoodName = {<<"name">>, <<"test-name-0">>},
{ok, Create} = request_api(post, api_path("trace"), Header, [GoodName | Trace]), {ok, Create} = request_api(post, api_path("trace"), [GoodName | Trace]),
?assertMatch(#{<<"name">> := <<"test-name-0">>}, json(Create)), ?assertMatch(#{<<"name">> := <<"test-name-0">>}, json(Create)),
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 409, _}, _}, {error, {"HTTP/1.1", 409, _}},
request_api(post, api_path("trace"), Header, [GoodName | Trace]) request_api(post, api_path("trace"), [GoodName | Trace])
), ),
%% MAX Limited %% MAX Limited
@ -170,17 +163,18 @@ t_create_failed(_Config) ->
), ),
GoodName1 = {<<"name">>, <<"test-name-1">>}, GoodName1 = {<<"name">>, <<"test-name-1">>},
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 400, _}, _}, {error, {"HTTP/1.1", 400, _}},
request_api(post, api_path("trace"), Header, [GoodName1 | Trace]) request_api(post, api_path("trace"), [GoodName1 | Trace])
), ),
%% clear %% clear
?assertMatch({ok, _}, request_api(delete, api_path("trace"), Header, [])), ?assertMatch({ok, _}, request_api(delete, api_path("trace"), [])),
{ok, Create} = request_api(post, api_path("trace"), Header, [GoodName | Trace]), {ok, Create1} = request_api(post, api_path("trace"), [GoodName | Trace]),
?assertMatch(#{<<"name">> := <<"test-name-0">>}, json(Create1)),
%% new name but same trace %% new name but same trace
GoodName2 = {<<"name">>, <<"test-name-1">>}, GoodName2 = {<<"name">>, <<"test-name-1">>},
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 409, _}, _}, {error, {"HTTP/1.1", 409, _}},
request_api(post, api_path("trace"), Header, [GoodName2 | Trace]) request_api(post, api_path("trace"), [GoodName2 | Trace])
), ),
unload(), unload(),
@ -202,14 +196,13 @@ t_log_file(_Config) ->
|| _ <- lists:seq(1, 5) || _ <- lists:seq(1, 5)
], ],
ok = emqx_trace_handler_SUITE:filesync(Name, clientid), ok = emqx_trace_handler_SUITE:filesync(Name, clientid),
Header = auth_header_(),
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 404, "Not Found"}, _}, {error, {"HTTP/1.1", 404, "Not Found"}},
request_api(get, api_path("trace/test_client_not_found/log_detail"), Header) request_api(get, api_path("trace/test_client_not_found/log_detail"))
), ),
{ok, Detail} = request_api(get, api_path("trace/test_client_id/log_detail"), Header), {ok, Detail} = request_api(get, api_path("trace/test_client_id/log_detail")),
?assertMatch([#{<<"mtime">> := _, <<"size">> := _, <<"node">> := _}], json(Detail)), ?assertMatch([#{<<"mtime">> := _, <<"size">> := _, <<"node">> := _}], json(Detail)),
{ok, Binary} = request_api(get, api_path("trace/test_client_id/download"), Header), {ok, Binary} = request_api(get, api_path("trace/test_client_id/download")),
{ok, [ {ok, [
Comment, Comment,
#zip_file{ #zip_file{
@ -221,7 +214,7 @@ t_log_file(_Config) ->
ZipNamePrefix = lists:flatten(io_lib:format("~s-trace_~s", [node(), Name])), ZipNamePrefix = lists:flatten(io_lib:format("~s-trace_~s", [node(), Name])),
?assertNotEqual(nomatch, re:run(ZipName, [ZipNamePrefix])), ?assertNotEqual(nomatch, re:run(ZipName, [ZipNamePrefix])),
Path = api_path("trace/test_client_id/download?node=" ++ atom_to_list(node())), Path = api_path("trace/test_client_id/download?node=" ++ atom_to_list(node())),
{ok, Binary2} = request_api(get, Path, Header), {ok, Binary2} = request_api(get, Path),
?assertMatch( ?assertMatch(
{ok, [ {ok, [
Comment, Comment,
@ -232,25 +225,22 @@ t_log_file(_Config) ->
]}, ]},
zip:table(Binary2) zip:table(Binary2)
), ),
{error, {_, 400, _}, _} = {error, {_, 400, _}} =
request_api( request_api(
get, get,
api_path("trace/test_client_id/download?node=unknonwn_node"), api_path("trace/test_client_id/download?node=unknonwn_node")
Header
), ),
{error, {_, 400, _}, _} = {error, {_, 400, _}} =
request_api( request_api(
get, get,
% known atom but unknown node % known atom but unknown node
api_path("trace/test_client_id/download?node=undefined"), api_path("trace/test_client_id/download?node=undefined")
Header
), ),
?assertMatch( ?assertMatch(
{error, {"HTTP/1.1", 404, "Not Found"}, _}, {error, {"HTTP/1.1", 404, "Not Found"}},
request_api( request_api(
get, get,
api_path("trace/test_client_not_found/download?node=" ++ atom_to_list(node())), api_path("trace/test_client_not_found/download?node=" ++ atom_to_list(node()))
Header
) )
), ),
ok = emqtt:disconnect(Client), ok = emqtt:disconnect(Client),
@ -297,64 +287,87 @@ t_stream_log(_Config) ->
ct:pal("FileName: ~p", [File]), ct:pal("FileName: ~p", [File]),
{ok, FileBin} = file:read_file(File), {ok, FileBin} = file:read_file(File),
ct:pal("FileBin: ~p ~s", [byte_size(FileBin), FileBin]), ct:pal("FileBin: ~p ~s", [byte_size(FileBin), FileBin]),
Header = auth_header_(), {ok, Binary} = request_api(get, api_path("trace/test_stream_log/log?bytes=10")),
{ok, Binary} = request_api(get, api_path("trace/test_stream_log/log?bytes=10"), Header),
#{<<"meta">> := Meta, <<"items">> := Bin} = json(Binary), #{<<"meta">> := Meta, <<"items">> := Bin} = json(Binary),
?assertEqual(10, byte_size(Bin)), ?assertEqual(10, byte_size(Bin)),
?assertEqual(#{<<"position">> => 10, <<"bytes">> => 10}, Meta), ?assertEqual(#{<<"position">> => 10, <<"bytes">> => 10}, Meta),
Path = api_path("trace/test_stream_log/log?position=20&bytes=10"), Path = api_path("trace/test_stream_log/log?position=20&bytes=10"),
{ok, Binary1} = request_api(get, Path, Header), {ok, Binary1} = request_api(get, Path),
#{<<"meta">> := Meta1, <<"items">> := Bin1} = json(Binary1), #{<<"meta">> := Meta1, <<"items">> := Bin1} = json(Binary1),
?assertEqual(#{<<"position">> => 30, <<"bytes">> => 10}, Meta1), ?assertEqual(#{<<"position">> => 30, <<"bytes">> => 10}, Meta1),
?assertEqual(10, byte_size(Bin1)), ?assertEqual(10, byte_size(Bin1)),
{error, {_, 400, _}, _} = {error, {_, 400, _}} =
request_api( request_api(
get, get,
api_path("trace/test_stream_log/log?node=unknonwn_node"), api_path("trace/test_stream_log/log?node=unknonwn_node")
Header
), ),
{error, {_, 400, _}, _} = {error, {_, 400, _}} =
request_api( request_api(
get, get,
% known atom but not a node % known atom but not a node
api_path("trace/test_stream_log/log?node=undefined"), api_path("trace/test_stream_log/log?node=undefined")
Header
), ),
{error, {_, 404, _}, _} = {error, {_, 404, _}} =
request_api( request_api(
get, get,
api_path("trace/test_stream_log_not_found/log"), api_path("trace/test_stream_log_not_found/log")
Header
), ),
unload(), unload(),
ok. ok.
t_trace_files_are_deleted_after_download(_Config) ->
ClientId = <<"client-test-delete-after-download">>,
Now = erlang:system_time(second),
Name = <<"test_client_id">>,
load(),
create_trace(Name, ClientId, Now),
{ok, Client} = emqtt:start_link([{clean_start, true}, {clientid, ClientId}]),
{ok, _} = emqtt:connect(Client),
[
begin
_ = emqtt:ping(Client)
end
|| _ <- lists:seq(1, 5)
],
ok = emqtt:disconnect(Client),
ok = emqx_trace_handler_SUITE:filesync(Name, clientid),
%% Check that files have been removed after download and that zip
%% directories uses unique session ids
?check_trace(
begin
%% Download two zip files
Path = api_path(["trace/", binary_to_list(Name), "/download"]),
{ok, Binary1} = request_api(get, Path),
{ok, Binary2} = request_api(get, Path),
?assertMatch({ok, _}, zip:table(Binary1)),
?assertMatch({ok, _}, zip:table(Binary2))
end,
fun(Trace) ->
[
#{session_id := SessionId1, zip_dir := ZipDir1},
#{session_id := SessionId2, zip_dir := ZipDir2}
] = ?of_kind(trace_api_download_trace_log, Trace),
?assertEqual({error, enoent}, file:list_dir(ZipDir1)),
?assertEqual({error, enoent}, file:list_dir(ZipDir2)),
?assertNotEqual(SessionId1, SessionId2),
?assertNotEqual(ZipDir1, ZipDir2)
end
),
ok.
to_rfc3339(Second) -> to_rfc3339(Second) ->
list_to_binary(calendar:system_time_to_rfc3339(Second)). list_to_binary(calendar:system_time_to_rfc3339(Second)).
request_api(Method, Url, Auth) -> do_request_api(Method, {Url, [Auth]}). request_api(Method, Url) ->
request_api(Method, Url, []).
request_api(Method, Url, Auth, Body) -> request_api(Method, Url, Body) ->
Request = {Url, [Auth], "application/json", emqx_json:encode(Body)}, Opts = #{httpc_req_opts => [{body_format, binary}]},
do_request_api(Method, Request). emqx_mgmt_api_test_util:request_api(Method, Url, [], [], Body, Opts).
do_request_api(Method, Request) ->
ct:pal("Method: ~p, Request: ~p", [Method, Request]),
case httpc:request(Method, Request, [], [{body_format, binary}]) of
{error, socket_closed_remotely} ->
{error, socket_closed_remotely};
{error, {shutdown, server_closed}} ->
{error, server_closed};
{ok, {{"HTTP/1.1", Code, _}, _Headers, Return}} when
Code =:= 200 orelse Code =:= 201 orelse Code =:= 204
->
{ok, Return};
{ok, {Reason, _Header, Body}} ->
{error, Reason, Body}
end.
api_path(Path) -> api_path(Path) ->
?HOST ++ filename:join([?BASE_PATH, ?API_VERSION, Path]). emqx_mgmt_api_test_util:api_path([Path]).
json(Data) -> json(Data) ->
{ok, Jsx} = emqx_json:safe_decode(Data, [return_maps]), {ok, Jsx} = emqx_json:safe_decode(Data, [return_maps]),

1
changes/v5.0.15/fix-9781.en.md Normal file
View File

@ -0,0 +1 @@
Trace files were left on a node when creating a zip file for download. They are now removed when the file is sent. Also, concurrent downloads will no longer interfere with each other.

1
changes/v5.0.15/fix-9781.zh.md Normal file
View File

@ -0,0 +1 @@
当下载 日志追踪 的日志时，一些中间文件将存留在处理节点上，现在这个问题得到了修复。同时，并发下载日志将不再相互干扰。