(3.13!) HTTP API: Gradually encode and send /queues response as JSON

NOTE this is currently branched off of `3.13.x` and should eventually
target `main` instead.

This change adds a helper to `rabbit_mgmt_util` which streams the reply
instead of relying on the standard `cowboy_rest` response tuple which
sets the response body in one go.

For a sizable broker with many many queues, querying the `/api/queues`
endpoint for all queues causes a significant memory spike on invocation.
Spiky workloads are generally hard to handle for any system, garbage
collectors included. The hope of this patch is to drop memory usage by
gradually converting queue metrics to JSON and sending them in small
chunks rather than converting the entire response to a large term
(albeit `iodata()`) at once.

TODO can we also run the augmentation gradually?

Author: the-mikedavis

deps/rabbitmq_management/src/rabbit_mgmt_util.erl

@@ -24,7 +24,7 @@
 -export([with_decode/4, not_found/3]).
 -export([with_channel/4, with_channel/5]).
 -export([props_to_method/2, props_to_method/4]).
--export([all_or_one_vhost/2, reply/3, responder_map/1,
+-export([all_or_one_vhost/2, reply/3, stream_reply/5, responder_map/1,
          filter_vhost/3]).
 -export([filter_conn_ch_list/3, filter_user/2, list_login_vhosts/2,
          list_login_vhosts_names/2]).
@@ -307,6 +307,81 @@ is_pagination_requested(undefined) ->
 is_pagination_requested(#pagination{}) ->
     true.
 
+-spec stream_reply(StatusCode, ChunkSize, Facts, ReqData, Context) -> ReqData1
+    when
+      StatusCode :: cowboy:http_status(),
+      ChunkSize :: pos_integer(),
+      Facts :: term(),
+      ReqData :: cowboy_req:req(),
+      Context :: term(),
+      ReqData1 :: cowboy_req:req().
+%% @doc Uses the Cowboy response streaming API to gradually encode the `Facts'
+%% into JSON and send on the connection.
+%%
+%% This helper can be used instead of `rabbit_mgmt_util:reply/3' when `Facts'
+%% could be large: a list with more than `ChunkSize' elements.
+%%
+%% If `Facts' is not a large list, this function delegates to `reply/3'.
+
+stream_reply(StatusCode, ChunkSize, Facts, ReqData, Context)
+  when is_integer(ChunkSize) ->
+    case maps:get(media_type, ReqData, undefined) of
+        %% Only application/json can be streamed currently.
+        {<<"application">>, <<"json">>, _} when is_list(Facts) andalso
+                                                length(Facts) > ChunkSize ->
+            stream_json_reply(StatusCode, ChunkSize, Facts, ReqData, Context);
+        _ ->
+            reply(Facts, ReqData, Context)
+    end.
+
+stream_json_reply(StatusCode, ChunkSize, Facts, ReqData, Context)
+  when is_list(Facts) ->
+    Facts1 = rabbit_mgmt_format:format_nulls(Facts),
+    ReqData1 = rabbit_mgmt_util:set_resp_header(<<"cache-control">>, "no-cache", ReqData),
+    try
+        ReqData2 = cowboy_req:stream_reply(StatusCode, ReqData1),
+        ok = cowboy_req:stream_body("[", nofin, ReqData2),
+        ok = stream_json_reply(Facts1, ChunkSize, ReqData2),
+        {stop, ReqData2, Context}
+    catch exit:{json_encode, E} ->
+            Error = iolist_to_binary(
+                      io_lib:format("JSON encode error: ~tp", [E])),
+            Reason = iolist_to_binary(
+                       io_lib:format("While encoding: ~n~tp", [Facts])),
+            internal_server_error(Error, Reason, ReqData1, Context)
+    end.
+
+stream_json_reply([], _ChunkSize, ReqData) ->
+    ReqData;
+stream_json_reply(Items, ChunkSize, ReqData) ->
+    {Chunk, Rest} = try
+                        lists:split(ChunkSize, Items)
+                    catch
+                        %% `lists:split/2' fails with `badarg' if the list is
+                        %% shorter than `ChunkSize'.
+                        error:badarg ->
+                            {Items, []}
+                    end,
+    EncodedChunk = json_encode_separated(Chunk),
+    case Rest of
+        [] ->
+            ok = cowboy_req:stream_body([EncodedChunk | "]"], fin, ReqData);
+        _ ->
+            ok = cowboy_req:stream_body([EncodedChunk | ","], nofin, ReqData),
+            stream_json_reply(Rest, ChunkSize, ReqData)
+    end.
+
+json_encode_separated(Items) ->
+    json_encode_separated(Items, []).
+
+json_encode_separated([] = _Items, IoData) ->
+    lists:reverse(IoData);
+json_encode_separated([Item | Rest], []) ->
+    json_encode_separated(Rest, [rabbit_json:encode(Item)]);
+json_encode_separated([Item | Rest], IoData) ->
+    IoData1 = [rabbit_json:encode(Item), "," | IoData],
+    json_encode_separated(Rest, IoData1).
+
 
 with_valid_pagination(ReqData, Context, Fun) ->
     try

deps/rabbitmq_management/src/rabbit_mgmt_wm_queues.erl

@@ -29,6 +29,10 @@
 
 -define(DEFAULT_SORT, ["vhost", "name"]).
 
+%% The number of queues to encode as JSON and send using Cowboy's response
+%% streaming API. TODO: tune this number according to performance tests.
+-define(RESPONSE_STREAM_CHUNK_SIZE, 1024).
+
 %%--------------------------------------------------------------------
 
 init(Req, State) ->
@@ -57,7 +61,8 @@ to_json(ReqData, {Mode, Context}) ->
         Data = rabbit_mgmt_util:augment_resources(Basic, ?DEFAULT_SORT,
                                                   ?BASIC_COLUMNS, ReqData,
                                                   Context, augment(Mode)),
-        rabbit_mgmt_util:reply(Data, ReqData, {Mode, Context})
+        rabbit_mgmt_util:stream_reply(
+          200, ?RESPONSE_STREAM_CHUNK_SIZE, Data, ReqData, {Mode, Context})
     catch
         {error, invalid_range_parameters, Reason} ->
             rabbit_mgmt_util:bad_request(iolist_to_binary(Reason), ReqData,