Merge remote-tracking branch 'origin/master' into stream-timestamp-offsetstream-timestamp-offset

12 files changed, 786 insertions, 0 deletions

diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_app.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_app.erl
new file mode 100644
index 0000000000..394f44a9d0
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_app.erl
@@ -0,0 +1,17 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_app).
+
+-behaviour(application).
+-export([start/2, stop/1]).
+
+start(_Type, _StartArgs) ->
+    rabbit_tracing_sup:start_link().
+
+stop(_State) ->
+    ok.
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_consumer.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_consumer.erl
new file mode 100644
index 0000000000..c610790086
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_consumer.erl
@@ -0,0 +1,247 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_consumer).
+
+-behaviour(gen_server).
+
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+-import(rabbit_misc, [pget/2, pget/3, table_lookup/2]).
+
+-record(state, {conn, ch, vhost, queue, file, filename, format, buf, buf_cnt,
+                max_payload}).
+-record(log_record, {timestamp, type, exchange, queue, node, connection,
+                     vhost, username, channel, routing_keys, routed_queues,
+                     properties, payload}).
+
+-define(DEFAULT_USERNAME, <<"guest">>).
+-define(DEFAULT_PASSWORD, <<"guest">>).
+
+-define(X, <<"amq.rabbitmq.trace">>).
+-define(MAX_BUF, 100).
+
+-export([start_link/1, info_all/1]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+         code_change/3]).
+
+start_link(Args) ->
+    gen_server:start_link(?MODULE, Args, []).
+
+info_all(Pid) ->
+    gen_server:call(Pid, info_all, infinity).
+
+%%----------------------------------------------------------------------------
+
+init(Args0) ->
+    process_flag(trap_exit, true),
+    Args = filter_optional_user_pass(Args0),
+    Name = pget(name, Args),
+    VHost = pget(vhost, Args),
+    Username = pget(tracer_connection_username, Args,
+                    rabbit_misc:get_env(rabbitmq_tracing, username, ?DEFAULT_USERNAME)),
+    Password = pget(tracer_connection_password, Args,
+                    rabbit_misc:get_env(rabbitmq_tracing, password, ?DEFAULT_PASSWORD)),
+    Username = rabbit_tracing_util:coerce_env_value(username, Username),
+    Password = rabbit_tracing_util:coerce_env_value(password, Password),
+    MaxPayload = pget(max_payload_bytes, Args, unlimited),
+    {ok, Conn} = amqp_connection:start(
+                   #amqp_params_direct{virtual_host = VHost,
+                                       username = Username,
+                                       password = Password}),
+    link(Conn),
+    {ok, Ch} = amqp_connection:open_channel(Conn),
+    link(Ch),
+    #'queue.declare_ok'{queue = Q} =
+        amqp_channel:call(Ch, #'queue.declare'{durable   = false,
+                                               exclusive = true}),
+    #'queue.bind_ok'{} =
+    amqp_channel:call(
+      Ch, #'queue.bind'{exchange = ?X, queue = Q,
+                        routing_key = pget(pattern, Args)}),
+    amqp_channel:enable_delivery_flow_control(Ch),
+    #'basic.consume_ok'{} =
+        amqp_channel:subscribe(Ch, #'basic.consume'{queue  = Q,
+                                                    no_ack = true}, self()),
+    {ok, Dir} = application:get_env(directory),
+    Filename = Dir ++ "/" ++ binary_to_list(Name) ++ ".log",
+    case filelib:ensure_dir(Filename) of
+        ok ->
+            case prim_file:open(Filename, [append]) of
+                {ok, F} ->
+                    rabbit_tracing_traces:announce(VHost, Name, self()),
+                    Format = list_to_atom(binary_to_list(pget(format, Args))),
+                    rabbit_log:info("Tracer opened log file ~p with "
+                                    "format ~p~n", [Filename, Format]),
+                    {ok, #state{conn = Conn, ch = Ch, vhost = VHost, queue = Q,
+                                file = F, filename = Filename,
+                                format = Format, buf = [], buf_cnt = 0,
+                                max_payload = MaxPayload}};
+                {error, E} ->
+                    {stop, {could_not_open, Filename, E}}
+            end;
+        {error, E} ->
+            {stop, {could_not_create_dir, Dir, E}}
+    end.
+
+handle_call(info_all, _From, State = #state{vhost = V, queue = Q}) ->
+    [QInfo] = rabbit_mgmt_db:augment_queues(
+                [rabbit_mgmt_wm_queue:queue(V, Q)],
+                rabbit_mgmt_util:no_range(), basic),
+    {reply, [{queue, rabbit_mgmt_format:strip_pids(QInfo)}], State};
+
+handle_call(_Req, _From, State) ->
+    {reply, unknown_request, State}.
+
+handle_cast(_C, State) ->
+    {noreply, State}.
+
+handle_info({BasicDeliver, Msg, DeliveryCtx},
+            State = #state{format = Format}) ->
+    amqp_channel:notify_received(DeliveryCtx),
+    {noreply, log(Format, delivery_to_log_record({BasicDeliver, Msg}, State),
+                  State),
+     0};
+
+handle_info(timeout, State) ->
+    {noreply, flush(State)};
+
+handle_info(_I, State) ->
+    {noreply, State}.
+
+terminate(shutdown, State = #state{conn = Conn, ch = Ch,
+                                   file = F, filename = Filename}) ->
+    flush(State),
+    catch amqp_channel:close(Ch),
+    catch amqp_connection:close(Conn),
+    catch prim_file:close(F),
+    rabbit_log:info("Tracer closed log file ~p~n", [Filename]),
+    ok;
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_, State, _) -> {ok, State}.
+
+%%----------------------------------------------------------------------------
+
+delivery_to_log_record({#'basic.deliver'{routing_key = Key},
+                        #amqp_msg{props   = #'P_basic'{headers = H},
+                                  payload = Payload}}, State) ->
+    {Type, Q, RQs} = case Key of
+                         <<"publish.", _Rest/binary>> ->
+                             {array, Qs} = table_lookup(H, <<"routed_queues">>),
+                             {published, none, [Q || {_, Q} <- Qs]};
+                         <<"deliver.", Rest/binary>> ->
+                             {received,  Rest, none}
+                     end,
+    {longstr, Node}   = table_lookup(H, <<"node">>),
+    {longstr, X}      = table_lookup(H, <<"exchange_name">>),
+    {array, Keys}     = table_lookup(H, <<"routing_keys">>),
+    {table, Props}    = table_lookup(H, <<"properties">>),
+    {longstr, Conn}   = table_lookup(H, <<"connection">>),
+    {longstr, VHost}  = table_lookup(H, <<"vhost">>),
+    {longstr, User}   = table_lookup(H, <<"user">>),
+    {signedint, Chan} = table_lookup(H, <<"channel">>),
+    #log_record{timestamp    = rabbit_mgmt_format:now_to_str_ms(
+                                 os:system_time(milli_seconds)),
+                type         = Type,
+                exchange     = X,
+                queue        = Q,
+                node         = Node,
+                connection   = Conn,
+                vhost        = VHost,
+                username     = User,
+                channel      = Chan,
+                routing_keys = [K || {_, K} <- Keys],
+                routed_queues= RQs,
+                properties   = Props,
+                payload      = truncate(Payload, State)}.
+
+log(text, Record, State) ->
+    Fmt = "~n========================================"
+        "========================================~n~s: Message ~s~n~n"
+        "Node:         ~s~nConnection:   ~s~n"
+        "Virtual host: ~s~nUser:         ~s~n"
+        "Channel:      ~p~nExchange:     ~s~n"
+        "Routing keys: ~p~n" ++
+        case Record#log_record.queue of
+            none -> "";
+            _    -> "Queue:        ~s~n"
+        end ++
+        case Record#log_record.routed_queues of
+            none -> "";
+            _    -> "Routed queues: ~p~n"
+        end ++
+        "Properties:   ~p~nPayload: ~n~s~n",
+    Args =
+        [Record#log_record.timestamp,
+         Record#log_record.type,
+         Record#log_record.node,    Record#log_record.connection,
+         Record#log_record.vhost,   Record#log_record.username,
+         Record#log_record.channel, Record#log_record.exchange,
+         Record#log_record.routing_keys] ++
+        case Record#log_record.queue of
+            none -> [];
+            Q    -> [Q]
+        end ++
+        case Record#log_record.routed_queues of
+            none -> [];
+            RQs  -> [RQs]
+        end ++
+        [Record#log_record.properties, Record#log_record.payload],
+    print_log(io_lib:format(Fmt, Args), State);
+
+log(json, Record, State) ->
+    print_log([rabbit_json:encode(
+                #{timestamp     => Record#log_record.timestamp,
+                  type          => Record#log_record.type,
+                  node          => Record#log_record.node,
+                  connection    => Record#log_record.connection,
+                  vhost         => Record#log_record.vhost,
+                  user          => Record#log_record.username,
+                  channel       => Record#log_record.channel,
+                  exchange      => Record#log_record.exchange,
+                  queue         => Record#log_record.queue,
+                  routed_queues => Record#log_record.routed_queues,
+                  routing_keys  => Record#log_record.routing_keys,
+                  properties    => rabbit_mgmt_format:amqp_table(
+                                     Record#log_record.properties),
+                  payload       => base64:encode(Record#log_record.payload)}),
+              "\n"],
+              State).
+
+print_log(LogMsg, State = #state{buf = Buf, buf_cnt = BufCnt}) ->
+    maybe_flush(State#state{buf = [LogMsg | Buf], buf_cnt = BufCnt + 1}).
+
+maybe_flush(State = #state{buf_cnt = ?MAX_BUF}) ->
+    flush(State);
+maybe_flush(State) ->
+    State.
+
+flush(State = #state{file = F, buf = Buf}) ->
+    prim_file:write(F, lists:reverse(Buf)),
+    State#state{buf = [], buf_cnt = 0}.
+
+truncate(Payload, #state{max_payload = Max}) ->
+    case Max =:= unlimited orelse size(Payload) =< Max of
+        true  -> Payload;
+        false -> <<Trunc:Max/binary, _/binary>> = Payload,
+                 Trunc
+    end.
+
+filter_optional_user_pass(Args) ->
+    case lists:member({tracer_connection_username,<<>>}, Args) of
+        true ->
+            [{K, V} || {K, V} <- Args,
+                       not lists:member(K, [tracer_connection_username,
+                                            tracer_connection_password,
+                                            <<"tracer_connection_username">>,
+                                            <<"tracer_connection_password">>])];
+        _ ->
+            Args
+    end.
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_consumer_sup.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_consumer_sup.erl
new file mode 100644
index 0000000000..b651a75a99
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_consumer_sup.erl
@@ -0,0 +1,25 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_consumer_sup).
+
+-behaviour(supervisor).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([start_link/1]).
+-export([init/1]).
+
+start_link(Args) -> supervisor2:start_link(?MODULE, Args).
+
+%%----------------------------------------------------------------------------
+
+init(Args) ->
+    {ok, {{one_for_one, 3, 10},
+          [{consumer, {rabbit_tracing_consumer, start_link, [Args]},
+            transient, ?WORKER_WAIT, worker,
+            [rabbit_tracing_consumer]}]}}.
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_files.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_files.erl
new file mode 100644
index 0000000000..dc3a70b7cc
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_files.erl
@@ -0,0 +1,48 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_files).
+
+-include_lib("kernel/include/file.hrl").
+
+-export([list/0, exists/1, delete/1, full_path/1]).
+
+%%--------------------------------------------------------------------
+
+list() ->
+    {ok, Dir} = application:get_env(rabbitmq_tracing, directory),
+    ok = filelib:ensure_dir(Dir ++ "/a"),
+    {ok, Names} = file:list_dir(Dir),
+    [file_info(Name) || Name <- Names].
+
+exists(Name) ->
+    filelib:is_regular(full_path(Name)).
+
+delete(Name) ->
+    ok = file:delete(full_path(Name)).
+
+full_path(Name0) when is_binary(Name0) ->
+    full_path(binary_to_list(Name0));
+full_path(Name0) ->
+    {ok, Dir} = application:get_env(rabbitmq_tracing, directory),
+    case rabbit_http_util:safe_relative_path(Name0) of
+        undefined -> exit(how_rude);
+        Name      -> Dir ++ "/" ++ Name
+    end.
+
+%%--------------------------------------------------------------------
+
+file_info(Name) ->
+    Size = case file:read_file_info(full_path(Name)) of
+               {ok, Info} ->
+                   Info#file_info.size;
+               {error, Error} ->
+                   rabbit_log:warning("error getting file info for ~s: ~p",
+                                      [Name, Error]),
+                   0
+           end,
+    [{name, list_to_binary(Name)}, {size, Size}].
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_mgmt.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_mgmt.erl
new file mode 100644
index 0000000000..c408150939
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_mgmt.erl
@@ -0,0 +1,25 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_mgmt).
+
+-behaviour(rabbit_mgmt_extension).
+
+-export([dispatcher/0, web_ui/0]).
+
+dispatcher() -> [{"/traces",                         rabbit_tracing_wm_traces, []},
+                 {"/traces/node/:node",              rabbit_tracing_wm_traces, []},
+                 {"/traces/:vhost",                  rabbit_tracing_wm_traces, []},
+                 {"/traces/node/:node/:vhost",       rabbit_tracing_wm_traces, []},
+                 {"/traces/:vhost/:name",            rabbit_tracing_wm_trace,  []},
+                 {"/traces/node/:node/:vhost/:name", rabbit_tracing_wm_trace, []},
+                 {"/trace-files",                    rabbit_tracing_wm_files,  []},
+                 {"/trace-files/node/:node",         rabbit_tracing_wm_files,  []},
+                 {"/trace-files/:name",              rabbit_tracing_wm_file,   []},
+                 {"/trace-files/node/:node/:name",   rabbit_tracing_wm_file,   []}].
+
+web_ui()     -> [{javascript, <<"tracing.js">>}].
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_sup.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_sup.erl
new file mode 100644
index 0000000000..d8fdd94d81
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_sup.erl
@@ -0,0 +1,50 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at https://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_sup).
+
+-behaviour(supervisor).
+
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-define(SUPERVISOR, ?MODULE).
+
+-export([start_link/0, start_child/2, stop_child/1]).
+-export([init/1]).
+
+%%----------------------------------------------------------------------------
+
+start_link() ->
+    supervisor:start_link({local, ?SUPERVISOR}, ?MODULE, []).
+
+start_child(Id, Args) ->
+    supervisor:start_child(
+      ?SUPERVISOR,
+      {Id, {rabbit_tracing_consumer_sup, start_link, [Args]},
+       temporary, ?SUPERVISOR_WAIT, supervisor,
+       [rabbit_tracing_consumer_sup]}).
+
+stop_child(Id) ->
+    supervisor:terminate_child(?SUPERVISOR, Id),
+    supervisor:delete_child(?SUPERVISOR, Id),
+    ok.
+
+%%----------------------------------------------------------------------------
+
+init([]) -> {ok, {{one_for_one, 3, 10},
+                  [{traces, {rabbit_tracing_traces, start_link, []},
+                    transient, ?WORKER_WAIT, worker,
+                    [rabbit_tracing_traces]}]}}.
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_traces.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_traces.erl
new file mode 100644
index 0000000000..fc2d2b1520
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_traces.erl
@@ -0,0 +1,116 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_traces).
+
+-behaviour(gen_server).
+
+-import(rabbit_misc, [pget/2]).
+
+-export([list/0, lookup/2, create/3, stop/2, announce/3]).
+
+-export([start_link/0]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+         code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-record(state, { table }).
+
+%%--------------------------------------------------------------------
+
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+list() ->
+    gen_server:call(?MODULE, list, infinity).
+
+lookup(VHost, Name) ->
+    gen_server:call(?MODULE, {lookup, VHost, Name}, infinity).
+
+create(VHost, Name, Trace) ->
+    gen_server:call(?MODULE, {create, VHost, Name, Trace}, infinity).
+
+stop(VHost, Name) ->
+    gen_server:call(?MODULE, {stop, VHost, Name}, infinity).
+
+announce(VHost, Name, Pid) ->
+    gen_server:cast(?MODULE, {announce, {VHost, Name}, Pid}).
+
+%%--------------------------------------------------------------------
+
+init([]) ->
+    {ok, #state{table = ets:new(anon, [private])}}.
+
+handle_call(list, _From, State = #state{table = Table}) ->
+    {reply, [augment(Trace) || {_K, Trace} <- ets:tab2list(Table)], State};
+
+handle_call({lookup, VHost, Name}, _From, State = #state{table = Table}) ->
+    {reply, case ets:lookup(Table, {VHost, Name}) of
+                []            -> not_found;
+                [{_K, Trace}] -> augment(Trace)
+            end, State};
+
+handle_call({create, VHost, Name, Trace0}, _From,
+            State = #state{table = Table}) ->
+    Already = vhost_tracing(VHost, Table),
+    Trace = pset(vhost, VHost, pset(name, Name, Trace0)),
+    true = ets:insert(Table, {{VHost, Name}, Trace}),
+    case Already of
+        true  -> ok;
+        false -> rabbit_trace:start(VHost)
+    end,
+    {reply, rabbit_tracing_sup:start_child({VHost, Name}, Trace), State};
+
+handle_call({stop, VHost, Name}, _From, State = #state{table = Table}) ->
+    true = ets:delete(Table, {VHost, Name}),
+    case vhost_tracing(VHost, Table) of
+        true  -> ok;
+        false -> rabbit_trace:stop(VHost)
+    end,
+    rabbit_tracing_sup:stop_child({VHost, Name}),
+    {reply, ok, State};
+
+handle_call(_Req, _From, State) ->
+    {reply, unknown_request, State}.
+
+handle_cast({announce, Key, Pid}, State = #state{table = Table}) ->
+    case ets:lookup(Table, Key) of
+        []           -> ok;
+        [{_, Trace}] -> ets:insert(Table, {Key, pset(pid, Pid, Trace)})
+    end,
+    {noreply, State};
+
+handle_cast(_C, State) ->
+    {noreply, State}.
+
+handle_info(_I, State) ->
+    {noreply, State}.
+
+terminate(_, _) -> ok.
+
+code_change(_, State, _) -> {ok, State}.
+
+%%--------------------------------------------------------------------
+
+pset(Key, Value, List) -> [{Key, Value} | proplists:delete(Key, List)].
+
+vhost_tracing(VHost, Table) ->
+    case [true || {{V, _}, _} <- ets:tab2list(Table), V =:= VHost] of
+        [] -> false;
+        _  -> true
+    end.
+
+augment(Trace) ->
+    Pid = pget(pid, Trace),
+    Trace1 = lists:keydelete(tracer_connection_password, 1,
+                             lists:keydelete(<<"tracer_connection_password">>, 1,
+                                             lists:keydelete(pid, 1, Trace))),
+    case Pid of
+        undefined -> Trace1;
+        _         -> rabbit_tracing_consumer:info_all(Pid) ++ Trace1
+    end.
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_util.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_util.erl
new file mode 100644
index 0000000000..93cb1dcb20
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_util.erl
@@ -0,0 +1,32 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_util).
+
+-export([coerce_env_value/2]).
+-export([apply_on_node/5]).
+
+coerce_env_value(username, Val) -> rabbit_data_coercion:to_binary(Val);
+coerce_env_value(password, Val) -> rabbit_data_coercion:to_binary(Val);
+coerce_env_value(_,        Val) -> Val.
+
+apply_on_node(ReqData, Context, Mod, Fun, Args) ->
+    case rabbit_mgmt_util:id(node, ReqData) of
+        none ->
+            apply(Mod, Fun, Args);
+        Node0 ->
+            Node = binary_to_atom(Node0, utf8),
+            case rpc:call(Node, Mod, Fun, Args) of
+                {badrpc, _} = Error ->
+                    Msg = io_lib:format("Node ~p could not be contacted: ~p",
+                                        [Node, Error]),
+                    rabbit_log:warning(Msg, []),
+                    rabbit_mgmt_util:bad_request(list_to_binary(Msg), ReqData, Context);
+                Any ->
+                    Any
+            end
+    end.
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_wm_file.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_file.erl
new file mode 100644
index 0000000000..17c225cba6
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_file.erl
@@ -0,0 +1,52 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+-module(rabbit_tracing_wm_file).
+
+-export([init/2, resource_exists/2, serve/2, content_types_provided/2,
+         is_authorized/2, allowed_methods/2, delete_resource/2]).
+-export([serve/1]).
+
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+
+%%--------------------------------------------------------------------
+init(Req, _State) ->
+    {cowboy_rest, rabbit_mgmt_cors:set_headers(Req, ?MODULE), #context{}}.
+
+
+content_types_provided(ReqData, Context) ->
+   {[{<<"text/plain">>, serve}], ReqData, Context}.
+
+allowed_methods(ReqData, Context) ->
+    {[<<"HEAD">>, <<"GET">>, <<"DELETE">>], ReqData, Context}.
+
+resource_exists(ReqData, Context) ->
+    Name = rabbit_mgmt_util:id(name, ReqData),
+    Exists = rabbit_tracing_util:apply_on_node(ReqData, Context, rabbit_tracing_files,
+                                               exists, [Name]),
+    {Exists, ReqData, Context}.
+
+serve(ReqData, Context) ->
+    Name = rabbit_mgmt_util:id(name, ReqData),
+    Content = rabbit_tracing_util:apply_on_node(ReqData, Context,
+                                                rabbit_tracing_wm_file,
+                                                serve, [Name]),
+    {Content, ReqData, Context}.
+
+serve(Name) ->
+    Path = rabbit_tracing_files:full_path(Name),
+    {ok, Content} = file:read_file(Path),
+    Content.
+
+delete_resource(ReqData, Context) ->
+    Name = rabbit_mgmt_util:id(name, ReqData),
+    ok = rabbit_tracing_util:apply_on_node(ReqData, Context, rabbit_tracing_files,
+                                           delete, [Name]),
+    {true, ReqData, Context}.
+
+is_authorized(ReqData, Context) ->
+    rabbit_mgmt_util:is_authorized_admin(ReqData, Context).
+
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_wm_files.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_files.erl
new file mode 100644
index 0000000000..74dc527524
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_files.erl
@@ -0,0 +1,30 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_wm_files).
+
+-export([init/2, to_json/2, content_types_provided/2, is_authorized/2]).
+
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+
+%%--------------------------------------------------------------------
+init(Req, _State) ->
+    {cowboy_rest, rabbit_mgmt_cors:set_headers(Req, ?MODULE), #context{}}.
+
+
+content_types_provided(ReqData, Context) ->
+   {[{<<"application/json">>, to_json}], ReqData, Context}.
+
+to_json(ReqData, Context) ->
+    List = rabbit_tracing_util:apply_on_node(ReqData, Context,
+                                             rabbit_tracing_files, list, []),
+    rabbit_mgmt_util:reply(List, ReqData, Context).
+
+is_authorized(ReqData, Context) ->
+    rabbit_mgmt_util:is_authorized_admin(ReqData, Context).
+
+%%--------------------------------------------------------------------
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_wm_trace.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_trace.erl
new file mode 100644
index 0000000000..92474be77e
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_trace.erl
@@ -0,0 +1,105 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+-module(rabbit_tracing_wm_trace).
+
+-export([init/2, resource_exists/2, to_json/2,
+         content_types_provided/2, content_types_accepted/2,
+         is_authorized/2, allowed_methods/2, accept_content/2,
+         delete_resource/2]).
+
+-define(ERR, <<"Something went wrong trying to start the trace - check the "
+               "logs.">>).
+
+-import(rabbit_misc, [pget/2, pget/3]).
+
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+
+%%--------------------------------------------------------------------
+init(Req, _State) ->
+    {cowboy_rest, rabbit_mgmt_cors:set_headers(Req, ?MODULE), #context{}}.
+
+
+content_types_provided(ReqData, Context) ->
+   {[{<<"application/json">>, to_json}], ReqData, Context}.
+
+content_types_accepted(ReqData, Context) ->
+   {[{<<"application/json">>, accept_content}], ReqData, Context}.
+
+allowed_methods(ReqData, Context) ->
+    {[<<"HEAD">>, <<"GET">>, <<"PUT">>, <<"DELETE">>], ReqData, Context}.
+
+resource_exists(ReqData, Context) ->
+    {case trace(ReqData, Context) of
+         not_found -> false;
+         _         -> true
+     end, ReqData, Context}.
+
+to_json(ReqData, Context) ->
+    rabbit_mgmt_util:reply(trace(ReqData, Context), ReqData, Context).
+
+accept_content(ReqData0, Ctx) ->
+    case rabbit_mgmt_util:vhost(ReqData0) of
+        not_found ->
+            not_found;
+        VHost ->
+            Name = rabbit_mgmt_util:id(name, ReqData0),
+            rabbit_mgmt_util:with_decode(
+              [format, pattern], ReqData0, Ctx,
+              fun([_, _], Trace, ReqData) ->
+                      Fs = [fun val_payload_bytes/5, fun val_format/5,
+                            fun val_create/5],
+                      case lists:foldl(fun (F,  ok)  -> F(ReqData, Ctx, VHost,
+                                                          Name, Trace);
+                                           (_F, Err) -> Err
+                                       end, ok, Fs) of
+                          ok  -> {true, ReqData, Ctx};
+                          Err -> rabbit_mgmt_util:bad_request(Err,
+                                                              ReqData,
+                                                              Ctx)
+                      end
+              end)
+    end.
+
+delete_resource(ReqData, Context) ->
+    VHost = rabbit_mgmt_util:vhost(ReqData),
+    rabbit_tracing_util:apply_on_node(ReqData, Context, rabbit_tracing_traces, stop,
+                                      [VHost, rabbit_mgmt_util:id(name, ReqData)]),
+    {true, ReqData, Context}.
+
+is_authorized(ReqData, Context) ->
+    rabbit_mgmt_util:is_authorized_admin(ReqData, Context).
+
+%%--------------------------------------------------------------------
+
+trace(ReqData, Context) ->
+    case rabbit_mgmt_util:vhost(ReqData) of
+        not_found -> not_found;
+        VHost     ->
+            Name = rabbit_mgmt_util:id(name, ReqData),
+            rabbit_tracing_util:apply_on_node(ReqData, Context, rabbit_tracing_traces,
+                                              lookup, [VHost, Name])
+    end.
+
+val_payload_bytes(_ReqData, _Context, _VHost, _Name, Trace) ->
+    case is_integer(maps:get(max_payload_bytes, Trace, 0)) of
+        false -> <<"max_payload_bytes not integer">>;
+        true  -> ok
+    end.
+
+val_format(_ReqData, _Context, _VHost, _Name, Trace) ->
+    case lists:member(maps:get(format, Trace), [<<"json">>, <<"text">>]) of
+        false -> <<"format not json or text">>;
+        true  -> ok
+    end.
+
+val_create(ReqData, Context, VHost, Name, Trace) ->
+    case rabbit_tracing_util:apply_on_node(
+           ReqData, Context, rabbit_tracing_traces, create,
+           [VHost, Name, maps:to_list(Trace)]) of
+        {ok, _} -> ok;
+        _       -> ?ERR
+    end.
diff --git a/deps/rabbitmq_tracing/src/rabbit_tracing_wm_traces.erl b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_traces.erl
new file mode 100644
index 0000000000..f41dc76336
--- /dev/null
+++ b/deps/rabbitmq_tracing/src/rabbit_tracing_wm_traces.erl
@@ -0,0 +1,39 @@
+%%  The contents of this file are subject to the Mozilla Public License
+%%  Version 1.1 (the "License"); you may not use this file except in
+%%  compliance with the License. You may obtain a copy of the License
+%%  at https://www.mozilla.org/MPL/
+%%
+%%  Software distributed under the License is distributed on an "AS IS"
+%%  basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%%  the License for the specific language governing rights and
+%%  limitations under the License.
+%%
+%%  The Original Code is RabbitMQ.
+%%
+%%  The Initial Developer of the Original Code is GoPivotal, Inc.
+%%  Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(rabbit_tracing_wm_traces).
+
+-export([init/2, to_json/2, content_types_provided/2, is_authorized/2]).
+
+-include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
+
+%%--------------------------------------------------------------------
+init(Req, _State) ->
+    {cowboy_rest, rabbit_mgmt_cors:set_headers(Req, ?MODULE), #context{}}.
+
+
+content_types_provided(ReqData, Context) ->
+   {[{<<"application/json">>, to_json}], ReqData, Context}.
+
+to_json(ReqData, Context) ->
+    List = rabbit_tracing_util:apply_on_node(ReqData, Context,
+                                             rabbit_tracing_traces, list, []),
+    rabbit_mgmt_util:reply(List, ReqData, Context).
+
+is_authorized(ReqData, Context) ->
+    rabbit_mgmt_util:is_authorized_admin(ReqData, Context).
+
+%%--------------------------------------------------------------------