[Suggestion] Stream HTTP responses from rabbit_prometheus_handler

[the-mikedavis]

RabbitMQ series

4.1.x

Operating system (distribution) used

macOS

How is RabbitMQ deployed?

Other

What would you like to suggest for a future version of RabbitMQ?

When formatting the reply to requests of /metrics/[:registry], rabbit_prometheus_handler uses prometheus_text_format:format/1 to build the response in Prometheus text format. Under the hood that function is building the response by writing into a ram_file. ram_file is a port (driver) wrapping a growable vector of bytes.

Buffering the whole response is necessary when the content-encoding is gzip, but for identity we could instead stream the response with cowboy_req:stream_reply/3 and cowboy_req:stream_body/3 and avoid buffering the response at all. I expect that this would be a nice improvement to the endpoints' memory use, which is valuable because some registries like per-object can send large responses on clusters with many objects.

To do this we would need to add a new formatting function to prometheus_text_format, say format_into/3 which would accept a function to pass data into, which would be file:write/2 for the default implementation.

format_into/3 function...

-spec format(Registry :: prometheus_registry:registry()) -> binary().
format(Registry) ->
    {ok, Fd0} = ram_file:open("", [write, read, binary]),
    FormatCallback = fun
        (Fd, Data) when is_list(Data) orelse is_binary(Data) ->
            file:write(Fd, Data);
        (Fd, eof) ->
            {ok, Size} = ram_file:get_size(Fd),
            {ok, Str} = file:pread(Fd, 0, Size),
            ok = file:close(Fd),
            Str
    end,
    format_into(Registry, FormatCallback, Fd0).

-spec format_into(Registry :: prometheus_registry:registry(), fun(), term()) -> term().
format_into(Registry, FormatCallback, State) ->
    Callback = fun(_, Collector) ->
        registry_collect_callback(FormatCallback, State, Registry, Collector)
    end,
    prometheus_registry:collect(Registry, Callback),
    FormatCallback(State, "\n"),
    FormatCallback(State, eof).

---

This change might be hard to measure precisely since memory allocated by a port driver is not tracked by the VM. Instead we would probably want to measure available memory while spamming /metrics/per-object on a cluster with many queues for example (like the 100k-classic-queues.json sample-config).

[michaelklishin]

Sounds reasonable, thank you for digging in.