Incoming message interceptors

[LoisSotoLopez]

Proposed Changes

This PR 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.

-callback intercept(Msg, MsgInterceptorCtx, Group, Config) -> Msg when
    Msg :: mc:state(),
    MsgInterceptorCtx :: msg_interceptor_ctx(),
    Group :: incoming_message_interceptors | outgoing_message_interceptors,
    Config :: #{atom() := term()}.

The intercept/4 callback receives a Msg; a context map provided by
the protocol-specific module that invoked the interception; a Group
specifying whether Msg is a incoming or a outgoing message; and a
configuration map specific to the interceptor.

Some of those protocol-specific modules that invoke the interception and
provide the context map are rabbit_channel, rabbit_amqp_session and
rabbit_mqtt_processor. The msg_interceptor_ctx() type makes
mandatory to provide certain information on the context map, such as the
protocol used to send/get the message, the virtual host where it
arrived/departed or the username publishing/getting the message. Other
pieces of information such as the name of the connection used to
publish/get the message may be added to the context map but are not
mandatory, so users providing custom interceptors should be aware of
this when implementing such modules.

For an interceptor to be invoked it needs to be properly configured
through the .conf file.

For just enabling an interceptor it would be enough to add a single line
to the .conf file. For an interceptor module named foo_mod.erl that
line would be:

message_interceptors.incoming.foo_mod.enabled=true

Notice this configuration line only enables the module as an interceptor
for incoming messages. To also make the same module intercept outgoing
messages two lines are needed.

message_interceptors.incoming.foo_mod.enabled=true
message_interceptors.outgoing.foo_mod.enabled=true

Thid last two-line configuration will generate an Erlang configuration
like this:

[{rabbit, [{message_interceptors,
            #{incoming => [{foo_mod, #{}}],
              outgoing => [{foo_mod, #{}}]}},
           ...]},
  ...].

The empty map after foo_mod is the configuration map that will be
received by the invoke/4 function as fourth argument. To populate this
map users can add more lines to the .config file. For example:

message_interceptors.incoming.foo_mod.enabled=true
message_interceptors.incoming.foo_mod.delay=100
message_interceptors.incoming.foo_mod.period=3000
message_interceptors.outgoing.foo_mod.delay=200
message_interceptors.outgoing.foo_mod.mode=idle

would result on a config like the following.

[{rabbit, [{message_interceptors,
            #{incoming => [{foo_mod, #{delay => 100, period => 3000}}],
              outgoing => [{foo_mod, #{delay => 200, mode => idle}}]}},
           ...]},
  ...].

The configuration map can be different for the same interceptor used as
incoming or outgoing interceptor. Also, the enabled key never gets
added to the resulting configuration map. In fact, if other keys are
used for the same group (incoming or outgoing) you don't need to specify
the enabled key. Therefore the configuration above could be simplified
to:

message_interceptors.incoming.foo_mod.delay=100
message_interceptors.incoming.foo_mod.period=3000
message_interceptors.outgoing.foo_mod.delay=200
message_interceptors.outgoing.foo_mod.mode=idle

Users providing custom interceptors should also provide the Cuttlefish
schemas providing mappings for the configuration parameters of their
interceptors. This commit already provides a Cuttlefish translation
for those parameters. Therefore, those user-provided mappings should
keep the expected format:

{mapping,
 "message_interceptor.<incoming|outgoing>.<interceptor_module_name>.<config_key>",
 "rabbit.<incoming|outgoing>_message_interceptors",
 <mapping_options>}.

For compatibility reasons, an alias can be used in
<interceptor_module_name> instead of the actual module name, but only
for the interceptors that already existed in RabbitMQ before this
commit. The set_header_timestamp and set_header_routing_node aliases
can be used as <interceptor_module_name>. However, using the actual
module names, wich are rabbit_message_interceptor_timestamp and
rabbit_message_interceptor_routing_node respectively is recommended.

The order in which the interceptors are invoked is the same as the order
in which they appear in the .conf file. For instance, the following
config lines

message_interceptors.outgoing.mod1.enabled=true
message_interceptors.incoming.mod3.config1=1
message_interceptors.incoming.mod1.config1=1
message_interceptors.incoming.mod2.config2=2

would result in the following Erlang configuration

[{rabbit, [{message_interceptors,
            #{incoming => [{mod3, _Cfg1}, {mod1, _Cfg1}, {mod2, _Cfg2}],
              outgoing => [{mod1, _Cfg4}]}}
           ...]},
  ...].

and the following order of invocation for incoming messages:
mod3 (first) -> mod1 -> mod2 (last).

Types of Changes

• Bug fix (non-breaking change which fixes issue #NNNN)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause an observable behavior change in existing systems)
• Documentation improvements (corrections, new content, etc)
• Cosmetic change (whitespace, formatting, etc)
• Build system and/or CI

Checklist

• I have read the CONTRIBUTING.md document
• I have signed the CA (see https://cla.pivotal.io/sign/rabbitmq)
• I have added tests that prove my fix is effective or that my feature works
• All tests pass locally with my changes.
• If relevant, I have added necessary documentation to https://github.com/rabbitmq/rabbitmq-website
• If relevant, I have added this change to the first version(s) in release-notes that I expect to introduce it

Further Comments

[kjnilsson]

I think the behaviour can be reduced to a single behaviour for both incoming and outgoing interceptors and the protocol behaviour can be replaced by a context map.

The formatting is all over the place. Please use a standard editor such as vim or emacs, most come with a reasonable indentation implementation. case | fun keywords should line up with their respective end markers. Try to keep to the 80 char per line limit also.

[kjnilsson]

I think this can be replaced with a map that the intercept function takes.

e.g.

#{protocol => mqtt,
    vhost => <<"/">>,
    client_id =><<"test">>}

[kjnilsson]

i don't think we need to distinguish incoming and outgoing interceptors are the module / behaviour level.
We should have a single behaviour rabbit_message_interceptor with a single intercept function that takes the message container, the protocol context map (see another comment) and the config.

Instead they would be distinguined by which configuration they are added to. It is fine if this PR only implements incoming interceptors.

[LoisSotoLopez]

I think the behaviour can be reduced to a single behaviour for both incoming and outgoing interceptors and the protocol behaviour can be replaced by a context map.

Sounds good to me. The context map could be set on initialization by each protocol accessor (AMQP 0.9.1 channel, AMQP 1.0 session, or MQTT connection process), and then passed to rabbit_incoming_message_interceptor (or rabbit_message_interceptor). If we go for the rabbit_message_interceptor (both incoming and outgoing messages) a keyword incoming | outgoing should be passed along the context map so that the right list of interceptors gets invoked.

Any additional comments on not using the getters approach and going for both a incoming+outgoing approach, @ansd ?

[kjnilsson]

we're getting closer I think a few suggested changes and some formatting advise.

[kjnilsson]

I think it would be better to use a tuple key for persistent term than doing this. e.g. {message_interceptors, incoming | outgoing}

[LoisSotoLopez]

This needs for ad hoc code to be put in rabbit:persist_static_configuration

    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,
       outgoing_message_interceptors
      ]),

needs to become something like

    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, {message_interceptors, incoming}
       outgoing_message_interceptors, {message_interceptors, outgoing}
      ]),

and

persist_static_configuration(Params) ->
    lists:foreach(
      fun(Param) ->
              case application:get_env(?MODULE, Param) of
                  {ok, Value} ->
                      ok = persistent_term:put(Param, Value);
                  undefined ->
                      ok
              end
      end, Params).

needs to become

persist_static_configuration(Params) ->
    lists:foreach(
        fun({Param, Alias}) ->
            case application:get_env(?MODULE, Param) of
                {ok, Value} ->
                    ok = persistent_term:put(Alias, Value);
                undefined ->
                    ok
            end;
           (Param) ->
            case application:get_env(?MODULE, Param) of
                {ok, Value} ->
                    ok = persistent_term:put(Param, Value);
                undefined ->
                    ok
            end
        end, Params).

Is that fine? What about this other option:

The cuttlefish translations could generate a rabbit.message_interceptors.<incoming|outgoing> config instead of rabbit.<incoming_message_interceptors|outgoing_message_interceptors> and the rabbit:persist_static_configuration would only receive message_interceptors instead of both <incoming|outgoing>_message_interceptors. The interceptor module would read app conf. with message_interceptors and then then get either the incoming or the outgoing ones from the read term.

The app conf would look like

[{rabbit, [{message_interceptors, #{incoming => [...], outgoing => [...] ...

instead of

[{rabbit, [...
    {incoming_message_interceptors, ...},
    {outgoing_message_interceptors,...},
...

[kjnilsson]

I see. that seems complicated. Perhaps just resolve incoming to incoming_message_interceptors etc in a function rather than doing atom to list conversions.

resolve(incoming) ->
   incoming_message_interceptors;
resolve(outgoing) ->
   outgoing_message_interceptors.

[ansd]

Yes, calling atom_to_list for each message that RabbitMQ receives creates lots of garbage, I wouldn't be surprised if performance drops by a few percent.

What Karl suggested is great.

Alternatively, it might be simpler to omit these translations entirely. For example instead of

-spec intercept(Msg, MsgInterceptorCtx, MsgDirection) -> Resp when
    Msg :: mc:state(),
    MsgInterceptorCtx :: map(),
    MsgDirection:: incoming | outgoing,
    Resp :: mc:state().

we could have

-spec intercept(Msg, Context, Group) -> Msg when
    Msg :: mc:state(),
    Context :: map(),
    Group :: incoming_message_interceptors | outgoing_message_interceptors.

[kjnilsson]

I like the group option actually. I think it should be it's own argument to intercept/4 though so we don't have to modify the context map for each call.

[kjnilsson]

I feel the direction should be passed to the interceptor, you may want to just have a single interceptor for both incoming and outgoing.

[LoisSotoLopez]

I assumed an interceptor would do the same independently of the direction of the message. It is the user the one that decides (if the interceptor allows it by providing the appropriate message_interceptors.<direction>. cuttlefish mapping) in which direction should the interceptor be applied, or if it should be applied in both.

[LoisSotoLopez]

As suggested by @gomoripeti , the rabbit_message_interceptor module could also add the message direction in the MsgInterceptorCtx map. How does that sound?

[kjnilsson]

that sounds good.

[kjnilsson]

Can we define some context properties that we'd always pass, e.g. vhost, username and the current protocol itself?

e.g.

#{protocol := amqp091 | amqp | mqttv3 | mqttv5, %% tags to be defined
   vhost := binary(),
   username := binary(),
   conn_name => binary() % optional
   atom() -> term()}

[LoisSotoLopez]

@kjnilsson Thanks on the formatting advices :) . I'm using neovim and I've tried to find out a proper solution on this, but I haven't found yet a solution that would allow me to format my contributions w/out re-formatting previous code. Will try to figure this out.

[kjnilsson]

@kjnilsson Thanks on the formatting advices :) . I'm using neovim and I've tried to find out a proper solution on this, but I haven't found yet a solution that would allow me to format my contributions w/out re-formatting previous code. Will try to figure this out.

It is an art rather than a science really. I use nvim too and it is fine for indentation - just hit '=' for your selection and it mostly does the right thing.

[ansd]

Thank you @LoisSotoLopez for this contribution!

Any additional comments on not using the getters approach and going for both a incoming+outgoing approach, @ansd ?

The only downside of the context map I see is that we duplicate some state information, for example each MQTT connection has higher memory usage after this PR. However, this context map is cleaner and simpler, so I prefer the context map.

Feel free to go with outgoing interceptors as well.

[kjnilsson]

he only downside of the context map I see is that we duplicate some state information, for example each MQTT connection has higher memory usage after this PR. However, this context map is cleaner and simpler, so I prefer the context map.

MQTT could add the context map to a persistent term to avoid this if necessary

[kjnilsson]

he only downside of the context map I see is that we duplicate some state information, for example each MQTT connection has higher memory usage after this PR. However, this context map is cleaner and simpler, so I prefer the context map.

MQTT could add the context map to a persistent term to avoid this if necessary

no scrap that, the context map is per vhost / user.

[ansd]

This should probably be x-opt-client-id or x-opt-mqtt-client-id or x-opt-mqtt-publisher-client-id such that this message annotation will be received by an AMQP 1.0 client in the message-annotation section.
The current annotation client_id also gets lost when the MQTT client publishes into a stream. Maybe this key should be a configuration option by the plugin.

[kjnilsson]

x-opt-mqtt-client-id gets my vote.

[ansd]

yes, I suppose it's best to include mqtt so that it's clear that this is an MQTT client ID.

"Client ID" is otherwise a broad term which could for example refer to a JMS client ID (https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#client-identifier) whose purpose is similar though:

The purpose of client identifier is to associate a connection and its objects with a state maintained on behalf of the client by a provider. By definition, the client state identified by a client identifier can be ‘in use’ by only one client at a time.

[kjnilsson]

The only downside of the context map I see is that we duplicate some state information, for example each MQTT connection has higher memory usage after this PR. However, this context map is cleaner and simpler, so I prefer the context map.

MQTT could consider building it on the fly to keep memory overhead low

[ansd]

Can you add unit tests?
deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets is the file to add these tests.

[kjnilsson]

Nearly there! couple of tweaks

[kjnilsson]

this is great and what we discussed as we think lower memory overhead is preferrable. It would be good if it at all possible you could run some throughput tests so we can get a view on how much this might affect performance. https://www.rabbitmq.com/blog/2023/03/21/native-mqtt#latency-and-throughput

[LoisSotoLopez]

Run the benchmarks against the last commit (build on publish) and with the last one commit building the context map on init (build on init).

On an Arch Linux w/ 10 cores and 24 GB of RAM.

QOS1 build on publish

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           63.151
Average Runtime (sec):         62.670
Pub time min (ms):             0.134
Pub time max (ms):             39.215
Pub time mean mean (ms):       6.258
Pub time mean std (ms):        0.044
Average Bandwidth (msg/sec):   159.574
Total Bandwidth (msg/sec):     15957.402

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      100.000% (1000000/1000000)
Forward latency min (ms):         0.126
Forward latency max (ms):         36.001
Forward latency mean std (ms):    0.020
Total Mean forward latency (ms):  3.128


QOS0 build on publish

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           1.352
Average Runtime (sec):         0.966
Pub time min (ms):             0.002
Pub time max (ms):             52.091
Pub time mean mean (ms):       0.036
Pub time mean std (ms):        0.014
Average Bandwidth (msg/sec):   14429.931
Total Bandwidth (msg/sec):     1442993.119

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      75.687% (756868/1000000)
Forward latency min (ms):         7.968
Forward latency max (ms):         4167.565
Forward latency mean std (ms):    345.354
Total Mean forward latency (ms):  1803.073


QOS1 build on init

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           63.087
Average Runtime (sec):         62.568
Pub time min (ms):             0.142
Pub time max (ms):             31.975
Pub time mean mean (ms):       6.248
Pub time mean std (ms):        0.054
Average Bandwidth (msg/sec):   159.839
Total Bandwidth (msg/sec):     15983.907

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      100.000% (1000000/1000000)
Forward latency min (ms):         0.138
Forward latency max (ms):         19.824
Forward latency mean std (ms):    0.025
Total Mean forward latency (ms):  3.112


QOS0 build on init

================= TOTAL PUBLISHER (100) =================
Total Publish Success Ratio:   100.000% (1000000/1000000)
Total Runtime (sec):           1.423
Average Runtime (sec):         0.961
Pub time min (ms):             0.002
Pub time max (ms):             38.604
Pub time mean mean (ms):       0.030
Pub time mean std (ms):        0.008
Average Bandwidth (msg/sec):   12087.794
Total Bandwidth (msg/sec):     1208779.411

================= TOTAL SUBSCRIBER (100) =================
Total Forward Success Ratio:      75.167% (751674/1000000)
Forward latency min (ms):         1.671
Forward latency max (ms):         4043.116
Forward latency mean std (ms):    207.787
Total Mean forward latency (ms):  1765.429

I get that the under-100% Total Forward Success Ratio on subscribers for QOS 0 comes from me using not a boxed environment + reduced RAM compared to the one used on the doc page you linked.

[kjnilsson]

So unexpectedly the qos0 test is faster creating the map dynamically?

[LoisSotoLopez]

It seems it is, but does not really make too much sense to me. I'll repeat this benchmark on a dedicated machine just in case after addressing the squashed commit writing.

[kjnilsson]

This looks good to me, a couple of lines to indent with =.

Then just squash all commits and write a nice single commit for the feature and update PR description and we can merge I think.

[LoisSotoLopez]

It might be better to leave full support for outgoing interceptors out of this PR. It lays the groudwork for those interceptors, but triggering them is not included in this PR to not grow the scope. Wdyt?

[ansd]

It might be better to leave full support for outgoing interceptors out of this PR. It lays the groudwork for those interceptors, but triggering them is not included in this PR to not grow the scope. Wdyt?

That's okay, we can add them at a later point.

[ansd]

1. Why is this called message_interceptor instead of message_interceptors as in file deps/rabbit/priv/schema/rabbit.schema?
2. I think it's not great that this Cuttleflish mapping defined in the MQTT plugin sets the environment in app rabbit. Do you see this approach done for any other configuration in any other plugin?
3. I tried it out by setting the following in rabbitmq.conf:

message_interceptor.incoming.rabbit_mqtt_message_interceptor_client_id.annotation_key = my-key

and this setting won't be set in the application environment of app rabbit.

Did you test this once that this Cuttlefish translation actually works?

[gomoripeti]

just about 2.

when I originally proposed the config format for message interceptors I only had the advanced config format in mind:

{rabbit,
  [{incoming_message_interceptors,
     [{module, #{config_key => config_value},
      ...]},

This way the config of the modules is handled uniformly and easy to put them to persistent_term in one go. That's the motivation why all interceptor's config goes under rabbit app.

I did not consider that in cuttlefish format we also need to define the mapping/type of each config_key+config_value and this config schema logically belongs to the plugin which defines the module as well.
The logic of the current schema is that plugin schemas can extend the generic schema that is defined by the transformation in rabbit.schema. And it should give some uniformity to the schema names (after point 1. is fixed :) )

Alternatively callback modules can handle their own configs and not receive as an argument (and then config can live in the plugin's app env and cached under different persistent_term keys)

[ansd]

Thanks for the clarification. Ok, after fixing 1. I now see my above setting taking effect successfully.

However, the following should work as well if I write my own third party plugin, correct?

message_interceptors.incoming.my_third_party_module.my_custom_key = some-value

With this PR, I get the following error:

2025-04-15 13:57:39.511692+00:00 [error] <0.162.0> You've tried to set message_interceptors.incoming.my_third_party_module.my_custom_key, but there is no setting with that name.
2025-04-15 13:57:39.511723+00:00 [error] <0.162.0>   Did you mean one of these?
2025-04-15 13:57:39.731574+00:00 [error] <0.162.0>     message_interceptors.incoming.$interceptor.overwrite
2025-04-15 13:57:39.731619+00:00 [error] <0.162.0>     message_interceptors.incoming.$interceptor.enabled
2025-04-15 13:57:39.731645+00:00 [error] <0.162.0>     message_interceptors.incoming.set_header_routing_node.overwrite
2025-04-15 13:57:39.731662+00:00 [error] <0.162.0> Error preparing configuration in phase transform_datatypes:
2025-04-15 13:57:39.731676+00:00 [error] <0.162.0>   - Conf file attempted to set unknown variable: message_interceptors.incoming.my_third_party_module.my_custom_key

[ansd]

Another thing to consider is: If I disable a plugin, I would like the plugin's interceptors to be removed as well. rabbit_registry is an advantage here.
What do you think about the following?
Each plugin that defines interceptors (e.g. MQTT) has its own Cuttlefish translation not interfering with app rabbit. Then upon plugin activation and deactivation the plugin itself ensures that it adds and removes its interceptors to/from persistent term.

[gomoripeti]

No, I don't think my_custom_key should just work. You need to define a type (mapping) for it in the schema. annotation_key is an example of that.

[LoisSotoLopez]

However, the following should work as well if I write my own third party plugin, correct?

If you want to use an interceptor you can just do message_interceptors.incoming.my_third_party_module.enable=true or if you want the plugin to have a custom configuration key such as my_custom_key you would need to provide a .schema file with a Cuttlefish mapping for that key.

[gomoripeti]

I imagined the list of interceptors to be quite static and set up by the operators. (Also their order of execution is explicitly defined so in the unlikely event of conflict it is defined what will happen) (I personally don't really like how channel interceptors work, they register dynamically, but as plugin startup order is not defined the channel interceptor order is also not known)

I understand from a user's perspective that it would be expected that if a plugin is disabled then its interceptors are also deactivated, but this is quite hard to implement without the dynamic registration similar to channel interceptors. Currently (maybe accidentally) RabbitMQ can still call code from disabled plugins, this has its pros and cons. But if the persistent_term registration is bound to plugin startup and stop, but the list of message interceptors is hardcoded in config, this detaches the two and an interceptor might be called without a config. That's why I think its simpler to have the list of interceptor callbacks and their config in one place, all handled by rabbit app. (With the caveat that a disabled plugin's interceptor will still work if the plugin is disabled, it will only not be invoked if the plugin is deleted or not available)

[gomoripeti]

There is existing example of callbacks and config being separated: the list of auth_backends are set in rabbit app env but their config is in the plugin apps' env and the plugin also must be started. In case of auth_backend_http if it is configured as an auth_backend but the rabbitmq_auth_backend_http plugin is not enabled, it wont start inet and the backend wont work. In this case disabling the plugin won't automagically disable the auth_backend. But the order of auth_backends is well defined. So it has pros and cons.

[gomoripeti]

I found some existing examples where a plugin's schema maps to rabbit app's env eg:
https://github.com/rabbitmq/rabbitmq-server/blob/main/deps/rabbitmq_aws/priv/schema/rabbitmq_aws.schema#L18

and all the peer discovery settings live under rabbit.cluster_formation..., which is similar to what we try to achieve here, eg:
https://github.com/rabbitmq/rabbitmq-server/blob/main/deps/rabbitmq_peer_discovery_aws/priv/schema/rabbitmq_peer_discovery_aws.schema#L10

[ansd]

Thanks for the context.

Sum up up, I see the following issues with this PR:

1. Setting the right configuration was too user unfriendly, e.g. the user has to set

   message_interceptor.incoming.rabbit_mqtt_message_interceptor_client_id.annotation_key = x-opt-mqtt-client-id
   

   just to enable the MQTT message interceptor.
2. The code that parses was too difficult to understand
3. MQTT plugin was setting the env for app rabbit, which is an anti-pattern
4. disabling a plugin (e.g. MQTT), left its message interceptors still in place

This is now all fixed in #13760

[gomoripeti]

this mapping should be removed, it is superseded by the explicit message_interceptors.incoming.set_header_timestamp.overwrite and message_interceptors.incoming.rabbit_message_interceptor_routing_node.overwrite

[ansd]

Thank you for your great work and great collaboration @LoisSotoLopez and @gomoripeti! I think we all settled with a good solution 🙂

Since I can't push to your cloudamqp fork, I created a new branch/PR (#13760) with your original commit and take the liberty to close this PR. From my point of view, we can merge the new PR. Let me know what you think.

[LoisSotoLopez]

Changes you proposed on the new PR seem reasonable to me. I'll be running the benchmarks again today so you can expect to get those results before 4 pm CET.

[LoisSotoLopez]

Ah, I had the brilliant idea of running the benchmarks on a burstable instance. Also less performant than my laptop 👏 . Will try to run those benchmarks during the holidays.