Add incoming message interceptors

This commit enables users to provide custom message interceptor modules,
i.e. modules to process incoming and outgoing messages. The
`rabbit_message_interceptor` behaviour defines a `intercept/4` callback,
for those modules to implement.

Co-authored-by: Péter Gömöri <[email protected]>

Author: LoisSotoLopez

.github/workflows/test-make-target.yaml

@@ -57,7 +57,8 @@ jobs:
       uses: dsaltares/fetch-gh-release-asset@master
       if: inputs.mixed_clusters
       with:
-        version: 'tags/v4.0.5'
+        repo: 'rabbitmq/server-packages'
+        version: 'tags/alphas.1744021065493'
         regex: true
         file: "rabbitmq-server-generic-unix-\\d.+\\.tar\\.xz"
         target: ./

deps/amqp10_client/src/amqp10_client.erl

@@ -47,6 +47,7 @@
 
 -type terminus_durability() :: amqp10_client_session:terminus_durability().
 
+-type terminus_address() :: amqp10_client_session:terminus_address().
 -type target_def() :: amqp10_client_session:target_def().
 -type source_def() :: amqp10_client_session:source_def().
 
@@ -64,6 +65,7 @@
               snd_settle_mode/0,
               rcv_settle_mode/0,
               terminus_durability/0,
+              terminus_address/0,
               target_def/0,
               source_def/0,
               attach_role/0,
@@ -170,7 +172,7 @@ attach_sender_link_sync(Session, Name, Target) ->
 %% @doc Synchronously attach a link on 'Session'.
 %% This is a convenience function that awaits attached event
 %% for the link before returning.
--spec attach_sender_link_sync(pid(), binary(), binary(),
+-spec attach_sender_link_sync(pid(), binary(), terminus_address(),
                               snd_settle_mode()) ->
     {ok, link_ref()} | link_timeout.
 attach_sender_link_sync(Session, Name, Target, SettleMode) ->
@@ -179,7 +181,7 @@ attach_sender_link_sync(Session, Name, Target, SettleMode) ->
 %% @doc Synchronously attach a link on 'Session'.
 %% This is a convenience function that awaits attached event
 %% for the link before returning.
--spec attach_sender_link_sync(pid(), binary(), binary(),
+-spec attach_sender_link_sync(pid(), binary(), terminus_address(),
                               snd_settle_mode(), terminus_durability()) ->
     {ok, link_ref()} | link_timeout.
 attach_sender_link_sync(Session, Name, Target, SettleMode, Durability) ->
@@ -199,7 +201,7 @@ attach_sender_link_sync(Session, Name, Target, SettleMode, Durability) ->
 %% This is asynchronous and will notify completion of the attach request to the
 %% caller using an amqp10_event of the following format:
 %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}}
--spec attach_sender_link(pid(), binary(), binary()) -> {ok, link_ref()}.
+-spec attach_sender_link(pid(), binary(), terminus_address()) -> {ok, link_ref()}.
 attach_sender_link(Session, Name, Target) ->
     % mixed should work with any type of msg
     attach_sender_link(Session, Name, Target, mixed).
@@ -208,7 +210,7 @@ attach_sender_link(Session, Name, Target) ->
 %% This is asynchronous and will notify completion of the attach request to the
 %% caller using an amqp10_event of the following format:
 %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}}
--spec attach_sender_link(pid(), binary(), binary(),
+-spec attach_sender_link(pid(), binary(), terminus_address(),
                          snd_settle_mode()) ->
     {ok, link_ref()}.
 attach_sender_link(Session, Name, Target, SettleMode) ->
@@ -218,7 +220,7 @@ attach_sender_link(Session, Name, Target, SettleMode) ->
 %% This is asynchronous and will notify completion of the attach request to the
 %% caller using an amqp10_event of the following format:
 %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}}
--spec attach_sender_link(pid(), binary(), binary(),
+-spec attach_sender_link(pid(), binary(), terminus_address(),
                          snd_settle_mode(), terminus_durability()) ->
     {ok, link_ref()}.
 attach_sender_link(Session, Name, Target, SettleMode, Durability) ->

deps/amqp10_client/src/amqp10_client_session.erl

@@ -65,19 +65,22 @@
 -define(INITIAL_DELIVERY_COUNT, ?UINT_MAX - 2).
 
 -type link_name() :: binary().
--type link_address() :: binary().
+%% https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-address-string
+%% or
+%% https://docs.oasis-open.org/amqp/anonterm/v1.0/anonterm-v1.0.html
+-type terminus_address() :: binary() | null.
 -type link_role() :: sender | receiver.
--type link_target() :: {pid, pid()} | binary() | undefined.
+-type link_target() :: {pid, pid()} | terminus_address() | undefined.
 %% "The locally chosen handle is referred to as the output handle." [2.6.2]
 -type output_handle() :: link_handle().
 %% "The remotely chosen handle is referred to as the input handle." [2.6.2]
 -type input_handle() :: link_handle().
 
 -type terminus_durability() :: none | configuration | unsettled_state.
 
--type target_def() :: #{address => link_address(),
+-type target_def() :: #{address => terminus_address(),
                         durable => terminus_durability()}.
--type source_def() :: #{address => link_address(),
+-type source_def() :: #{address => terminus_address(),
                         durable => terminus_durability()}.
 
 -type attach_role() :: {sender, target_def()} | {receiver, source_def(), pid()}.
@@ -112,6 +115,7 @@
               terminus_durability/0,
               attach_args/0,
               attach_role/0,
+              terminus_address/0,
               target_def/0,
               source_def/0,
               filter/0,

deps/rabbit/priv/schema/rabbit.schema

@@ -2658,23 +2658,102 @@ end}.
 {mapping, "message_interceptors.incoming.$interceptor.overwrite", "rabbit.incoming_message_interceptors", [
     {datatype, {enum, [true, false]}}]}.
 
+% Pseudo-key to include the interceptor in the list of interceptors.
+%  - If any other configuration is provided for the interceptor this
+%  configuration is not required.
+%  - If no other configuration is provided, this one is required so that the
+%  interceptor gets invoked.
+{mapping, "message_interceptors.incoming.$interceptor.enabled", "rabbit.incoming_message_interceptors", [
+    {datatype, {enum, [true]}}]}.
+
+{mapping, "message_interceptors.outgoing.$interceptor.enabled", "rabbit.outgoing_message_interceptors", [
+    {datatype, {enum, [true]}}]}.
+
+{mapping,
+    "message_interceptors.incoming.set_header_timestamp.overwrite",
+    "rabbit.incoming_message_interceptors",
+    [{datatype, {enum, [true, false]}}]}.
+{mapping,
+    "message_interceptors.incoming.rabbit_message_interceptor_routing_node.overwrite",
+    "rabbit.incoming_message_interceptors",
+    [{datatype, {enum, [true, false]}}]}.
+
+{mapping,
+    "message_interceptors.incoming.set_header_routing_node.overwrite",
+    "rabbit.incoming_message_interceptors",
+    [{datatype, {enum, [true, false]}}]}.
+{mapping,
+    "message_interceptors.incoming.rabbit_message_interceptor_timestamp.overwrite",
+    "rabbit.incoming_message_interceptors",
+    [{datatype, {enum, [true, false]}}]}.
+
 {translation, "rabbit.incoming_message_interceptors",
     fun(Conf) ->
-         case cuttlefish_variable:filter_by_prefix("message_interceptors", Conf) of
-             [] ->
-                 cuttlefish:unset();
-             L ->
-                 [begin
-                      Interceptor = list_to_atom(Interceptor0),
-                      case lists:member(Interceptor, [set_header_timestamp,
-                                                      set_header_routing_node]) of
-                          true ->
-                              {Interceptor, Overwrite};
-                          false ->
-                              cuttlefish:invalid(io_lib:format("~p is invalid", [Interceptor]))
-                      end
-                  end || {["message_interceptors", "incoming", Interceptor0, "overwrite"], Overwrite} <- L]
-         end
+        case cuttlefish_variable:filter_by_prefix("message_interceptors.incoming", Conf) of
+            [] ->
+                cuttlefish:unset();
+            L ->
+                InterceptorsConfig = [
+                    {Module0, Config, Value}
+                    || {["message_interceptors", "incoming", Module0, Config], Value} <- L
+                ],
+                {Result, Order0} = lists:foldl(
+                    fun({Interceptor0, Key0, Value}, {Acc, Order}) ->
+                        Interceptor = list_to_atom(Interceptor0),
+                        Key = list_to_atom(Key0),
+                        MapPutFun = fun(Old) -> maps:put(Key, Value, Old) end,
+                        % This Interceptor -> Module alias exists for
+                        % compatibility reasons
+                        Module = case Interceptor  of
+                                     set_header_timestamp ->
+                                         rabbit_message_interceptor_timestamp;
+                                     set_header_routing_node ->
+                                         rabbit_message_interceptor_routing_node;
+                                     _ ->
+                                         Interceptor
+                                 end,
+                        NewAcc = maps:update_with(Module,
+                                                  MapPutFun,
+                                                  #{Key => Value},
+                                                  Acc),
+                        {NewAcc, [Module| Order]}
+                    end,
+                    {#{}, []},
+                    InterceptorsConfig
+                ),
+                Order = lists:uniq(Order0),
+                [{O, maps:without([enabled], maps:get(O, Result))} || O <- Order]
+        end
+    end
+}.
+
+{translation, "rabbit.outgoing_message_interceptors",
+    fun(Conf) ->
+        case cuttlefish_variable:filter_by_prefix("message_interceptors.outgoing", Conf) of
+            [] ->
+                cuttlefish:unset();
+            L ->
+                InterceptorsConfig = [
+                    {Module0, Config, Value}
+                    || {["message_interceptors", "outgoing", Module0, Config], Value} <- L
+                ],
+                {Result, Order0} = lists:foldl(
+                    fun({Interceptor0, Key0, Value}, {Acc, Order}) ->
+                        Module = list_to_atom(Interceptor0),
+                        Key = list_to_atom(Key0),
+                        MapPutFun = fun(Old) -> maps:put(Key, Value, Old) end,
+                        NewAcc = maps:update_with(Module,
+                                                  MapPutFun,
+                                                  #{Key => Value},
+                                                  Acc),
+                        {NewAcc, [Module| Order]}
+                    end,
+                    {#{}, []},
+                    InterceptorsConfig
+                ),
+                Order = lists:uniq(Order0),
+                [{O, maps:without([enabled], maps:get(O, Result))} || O <- Order]
+        end
     end
 }.
 

deps/rabbit/src/rabbit.erl

@@ -1656,7 +1656,8 @@ persist_static_configuration() ->
       [classic_queue_index_v2_segment_entry_count,
        classic_queue_store_v2_max_cache_size,
        classic_queue_store_v2_check_crc32,
-       incoming_message_interceptors
+       incoming_message_interceptors,
+       outgoing_message_interceptors
       ]),
 
     %% Disallow the following two cases:

deps/rabbit/src/rabbit_amqp_management.erl

@@ -444,55 +444,40 @@ encode_queue(Q, NumMsgs, NumConsumers) ->
                            ShortName ->
                                ShortName
                        end}},
-               {{utf8, <<"arguments">>}, QArgs}
+               {{utf8, <<"arguments">>}, QArgs},
+               {{utf8, <<"replicas">>},
+                {array, utf8, [{utf8, atom_to_binary(R)} || R <- Replicas]}
+               }
               ],
-    KVList1 = if is_list(Replicas) ->
-                     [{{utf8, <<"replicas">>},
-                       {array, utf8, [{utf8, atom_to_binary(R)} || R <- Replicas]}
-                      } | KVList0];
-                 Replicas =:= undefined ->
-                     KVList0
-              end,
     KVList = case Leader of
-                 undefined ->
-                     KVList1;
+                 none ->
+                     KVList0;
                  _ ->
                      [{{utf8, <<"leader">>},
                        {utf8, atom_to_binary(Leader)}
-                      } | KVList1]
+                      } | KVList0]
              end,
     {map, KVList}.
 
 %% The returned Replicas contain both online and offline replicas.
 -spec queue_topology(amqqueue:amqqueue()) ->
-    {Leader :: undefined | node(), Replicas :: undefined | [node(),...]}.
+    {Leader :: node() | none, Replicas :: [node(),...]}.
 queue_topology(Q) ->
-    case amqqueue:get_type(Q) of
-        rabbit_quorum_queue ->
-            [{leader, Leader0},
-             {members, Members}] = rabbit_queue_type:info(Q, [leader, members]),
-            Leader = case Leader0 of
-                         '' -> undefined;
-                         _ -> Leader0
-                     end,
-            {Leader, Members};
-        rabbit_stream_queue ->
-            #{name := StreamId} = amqqueue:get_type_state(Q),
-            case rabbit_stream_coordinator:members(StreamId) of
-                {ok, Members} ->
-                    maps:fold(fun(Node, {_Pid, writer}, {_, Replicas}) ->
-                                      {Node, [Node | Replicas]};
-                                 (Node, {_Pid, replica}, {Writer, Replicas}) ->
-                                      {Writer, [Node | Replicas]}
-                              end, {undefined, []}, Members);
-                {error, _} ->
-                    {undefined, undefined}
-            end;
-        _ ->
-            Pid = amqqueue:get_pid(Q),
-            Node = node(Pid),
-            {Node, [Node]}
-    end.
+    Leader = case amqqueue:get_pid(Q) of
+                 {_RaName, Node} ->
+                     Node;
+                 none ->
+                     none;
+                 Pid ->
+                     node(Pid)
+             end,
+    Replicas = case amqqueue:get_type_state(Q) of
+                   #{nodes := Nodes} ->
+                       Nodes;
+                   _ ->
+                       [Leader]
+               end,
+    {Leader, Replicas}.
 
 decode_exchange({map, KVList}) ->
     M = lists:foldl(

deps/rabbit/src/rabbit_amqp_session.erl

@@ -283,7 +283,8 @@
           max_handle :: link_handle(),
           max_incoming_window :: pos_integer(),
           max_link_credit :: pos_integer(),
-          max_queue_credit :: pos_integer()
+          max_queue_credit :: pos_integer(),
+          msg_interceptor_ctx :: map()
          }).
 
 -record(state, {
@@ -474,7 +475,11 @@ init({ReaderPid, WriterPid, ChannelNum, MaxFrameSize, User, Vhost, ContainerId,
                            max_handle = EffectiveHandleMax,
                            max_incoming_window = MaxIncomingWindow,
                            max_link_credit = MaxLinkCredit,
-                           max_queue_credit = MaxQueueCredit
+                           max_queue_credit = MaxQueueCredit,
+                           msg_interceptor_ctx = #{protocol => ?PROTOCOL,
+                                                   username => User#user.username,
+                                                   vhost => Vhost,
+                                                   conn_name => ConnName}
                           }}}.
 
 terminate(_Reason, #state{incoming_links = IncomingLinks,
@@ -2411,7 +2416,8 @@ incoming_link_transfer(
                              trace_state = Trace,
                              conn_name = ConnName,
                              channel_num = ChannelNum,
-                             max_link_credit = MaxLinkCredit}}) ->
+                             max_link_credit = MaxLinkCredit,
+                             msg_interceptor_ctx = MsgInterceptorCtx}}) ->
 
     {PayloadBin, DeliveryId, Settled} =
     case MultiTransfer of
@@ -2436,7 +2442,9 @@ incoming_link_transfer(
     Mc0 = mc:init(mc_amqp, PayloadBin, #{}),
     case lookup_target(LinkExchange, LinkRKey, Mc0, Vhost, User, PermCache0) of
         {ok, X, RoutingKeys, Mc1, PermCache} ->
-            Mc2 = rabbit_message_interceptor:intercept(Mc1),
+            Mc2 = rabbit_message_interceptor:intercept(Mc1,
+                                                       MsgInterceptorCtx,
+                                                       incoming_message_interceptors),
             check_user_id(Mc2, User),
             TopicPermCache = check_write_permitted_on_topics(
                                X, User, RoutingKeys, TopicPermCache0),