Optimization: stream HTTP responses from rabbit_prometheus_handler

deps/rabbitmq_prometheus/Makefile

@@ -9,7 +9,7 @@ endef
 PROJECT := rabbitmq_prometheus
 PROJECT_DESCRIPTION = Prometheus metrics for RabbitMQ
 PROJECT_MOD := rabbit_prometheus_app
-DEPS = accept cowboy rabbit rabbitmq_management_agent prometheus rabbitmq_web_dispatch
+DEPS = cowboy rabbit rabbitmq_management_agent prometheus rabbitmq_web_dispatch
 BUILD_DEPS = amqp_client rabbit_common rabbitmq_management
 TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers rabbitmq_stream
 

deps/rabbitmq_prometheus/src/rabbit_prometheus_handler.erl

@@ -59,7 +59,7 @@ setup_metrics(Registry) ->
                   {labels, ["registry", "content_type"]},
                   {registry, Registry}],
     ScrapeEncodedSize = [{name, ?SCRAPE_ENCODED_SIZE},
-                         {help, "Scrape size, encoded"},
+                         {help, "Scrape size, encoded (deprecated)"},
                          {labels, ["registry", "content_type", "encoding"]},
                          {registry, Registry}],
 
@@ -89,47 +89,34 @@ gen_response(_, Request) ->
     Request.
 
 gen_metrics_response(Registry, Request) ->
-    {Code, RespHeaders, Body} = reply(Registry, Request),
-
-    Headers = to_cowboy_headers(RespHeaders),
-    cowboy_req:reply(Code, maps:from_list(Headers), Body, Request).
-
-to_cowboy_headers(RespHeaders) ->
-    lists:map(fun to_cowboy_headers_/1, RespHeaders).
-
-to_cowboy_headers_({Name, Value}) ->
-    {to_cowboy_name(Name), Value}.
-
-to_cowboy_name(Name) ->
-    binary:replace(atom_to_binary(Name, utf8), <<"_">>, <<"-">>).
-
-reply(Registry, Request) ->
     case validate_registry(Registry, registry()) of
         {true, RealRegistry} ->
             format_metrics(Request, RealRegistry);
         {registry_conflict, _ReqR, _ConfR} ->
-            {409, [], <<>>}
+            cowboy_req:reply(409, #{}, <<>>, Request)
     end.
 
 format_metrics(Request, Registry) ->
-    AcceptEncoding = cowboy_req:header(<<"accept-encoding">>, Request, undefined),
+    %% Formatting registries produces large binaries. Fullsweep eagerly to
+    %% evict the large binaries faster and make GC cheaper.
+    process_flag(fullsweep_after, 0),
     ContentType = prometheus_text_format:content_type(),
-    Scrape = render_format(ContentType, Registry),
-    Encoding = accept_encoding_header:negotiate(AcceptEncoding, [<<"identity">>,
-                                                                 <<"gzip">>]),
-    encode_format(ContentType, binary_to_list(Encoding), Scrape, Registry).
-
-render_format(ContentType, Registry) ->
-    Scrape = prometheus_summary:observe_duration(
-               Registry,
-               ?SCRAPE_DURATION,
-               [Registry, ContentType],
-               fun () -> prometheus_text_format:format(Registry) end),
+    Req = cowboy_req:stream_reply(200, #{<<"content-type">> => ContentType}, Request),
+    Fmt = fun(Data, Size) ->
+        cowboy_req:stream_body(Data, nofin, Req),
+        Size + iolist_size(Data)
+    end,
+    Size = prometheus_summary:observe_duration(
+             Registry,
+             ?SCRAPE_DURATION,
+             [Registry, ContentType],
+             fun () -> prometheus_text_format:format_into(Registry, 0, Fmt) end),
+    cowboy_req:stream_body(<<>>, fin, Req),
     prometheus_summary:observe(Registry,
                                ?SCRAPE_SIZE,
                                [Registry, ContentType],
-                               iolist_size(Scrape)),
-    Scrape.
+                               Size),
+    Req.
 
 validate_registry(undefined, auto) ->
     {true, default};
@@ -146,20 +133,6 @@ telemetry_registry() ->
 registry() ->
     application:get_env(rabbitmq_prometheus, registry, auto).
 
-encode_format(ContentType, Encoding, Scrape, Registry) ->
-    Encoded = encode_format_(Encoding, Scrape),
-    prometheus_summary:observe(telemetry_registry(),
-                               ?SCRAPE_ENCODED_SIZE,
-                               [Registry, ContentType, Encoding],
-                               iolist_size(Encoded)),
-    {200, [{content_type, binary_to_list(ContentType)},
-           {content_encoding, Encoding}], Encoded}.
-
-encode_format_("gzip", Scrape) ->
-    zlib:gzip(Scrape);
-encode_format_("identity", Scrape) ->
-     Scrape.
-
 %% It's not easy to pass this information in a pure way (it'll require changing prometheus.erl)
 put_filtering_options_into_process_dictionary(Request) ->
     #{vhost := VHosts, family := Families} = cowboy_req:match_qs([{vhost, [], undefined}, {family, [], undefined}], Request),

deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl

@@ -373,7 +373,7 @@
 
 encoding_test(Config) ->
     {Headers, Body} = http_get(Config, [{"accept-encoding", "deflate"}], 200),
-    ?assertMatch("identity", proplists:get_value("content-encoding", Headers)),
+    ?assertMatch(undefined, proplists:get_value("content-encoding", Headers)),
     ?assertEqual(match, re:run(Body, "^# TYPE", [{capture, none}, multiline])).
 
 gzip_encoding_test(Config) ->
@@ -870,7 +870,7 @@
     ok.
 
 detailed_raft_metrics_test(Config) ->
     ComponentMetrics = #{#{module => "ra_log_wal", ra_system => "coordination"} => ["1.0"],
                          #{module => "ra_log_wal", ra_system => "quorum_queues"} => ["1.0"]},
     QQMetrics = #{#{queue => "a_quorum_queue", vhost => "/"} => ["1.0"]},
 

rabbitmq-components.mk

@@ -39,7 +39,6 @@ endif
 # We do that in this file, which is included by all projects, to ensure
 # all projects use the same versions. It avoids conflicts.
 
-dep_accept = hex 0.3.5
 dep_cowboy = hex 2.14.1
 dep_cowlib = hex 2.16.0
 dep_credentials_obfuscation = hex 3.5.0
@@ -50,7 +49,7 @@ dep_khepri = hex 0.17.2
 dep_khepri_mnesia_migration = hex 0.8.0
 dep_meck = hex 1.0.0
 dep_osiris = git https://github.com/rabbitmq/osiris v1.10.2
-dep_prometheus = hex 5.1.1
+dep_prometheus = git https://github.com/the-mikedavis/prometheus.erl ce32cfef7eee60f577c1f200304428509152a7ff
 dep_ra = hex 2.17.1
 dep_ranch = hex 2.2.0
 dep_recon = hex 2.5.6