Support interceptors for native protocols

[ansd]

Is your feature request related to a problem? Please describe.

As described in https://github.com/rabbitmq/internals/blob/master/interceptors.md, rabbit_channel_interceptor provides the ability to create 3rd party extensions (e.g. Sharding Plugin) that intercept or modify AMQP 0.9.1 methods.
Such an extension mechanism doesn't exist for native MQTT, native AMQP, or native STOMP.

Describe the solution you'd like

The solution hasn't been fully designed yet, but here are some suggestions:

A 3rd party MQTT connection interceptor implementation will get as input the received MQTT packet and the MQTT processor state and outputs a modified MQTT packet (or forbids that packet).

The MQTT connection interceptor hook would be called earlier than the message container hook (see #10050), and for every MQTT packet type (not only for PUBLISH packets).

For AMQP 1.0, the AMQP 0.9.1 channel interceptor equivalent would be an AMQP 1.0 session interceptor.
We should also come up with a design that allows existing plugins currently targeting AMQP 0.9.1 (for example the Sharding Plugin) to be used with AMQP 1.0. Hence, it might make sense to allow hooks in rabbit_amqp_management before/after creating/updating/deleting exchanges, queues, and bindings.

So sum up, the solution should be a holistic in the sense that it:

• provides similar functionalities for Native AMQP 1.0, Native MQTT, AMQP 0.9.1
• works with existing plugins (e.g. the Sharding plugin)
• works with future extensions, potentially with message container hook Support custom message container hook #10050

Additional context

An example use case is described in https://groups.google.com/g/rabbitmq-users/c/gPJxxG2PGXM/m/EF2BDwpIAQAJ
For this specific use case, the PUBLISH packet could be added User Property containing the MQTT client ID. Upon translation to AMQP 0.9.1 later on, these will be converted automatically into AMQP 0.9.1 headers. (This will only work for MQTT 5.0 though as lower MQTT versions don't have a User Property.)

[getlarge]

After digging into the codebase to provide this feature, it looks like external rabbit_message_interceptor is what we need.
I'm still learning about Erlang and the codebase, so my question might be naive: Why was the rabbit_channel_interceptor designed for plugin use but not the rabbit_message_interceptor?
@ansd, has this issue progressed? Can I help or contribute in any way?

[michaelklishin]

rabbit_message_interceptor has a dedicated comment to what its purpose is: replace two small 3rd party plugins that were AMQP 0-9-1-specific and used channel interceptors.

In the current implementations of AMQP 1.0, MQTT and hopefully STOMP by 4.1.0, there are no channels because they do not build on AMQP 0-9-1. Therefore message interceptors were introduced to provide an alternative, in part for the above plugins.

It does what it was designed for. There are three places where it is used:

1. AMQP 0-9-1 channels

   rabbitmq-server/deps/rabbit/src/rabbit_channel.erl

   Line 1216 in 658d5b8

   Message = rabbit_message_interceptor:intercept(Message0),

2. AMQP 1.0 sessions

   rabbitmq-server/deps/rabbit/src/rabbit_amqp_session.erl

   Line 2407 in 658d5b8

   Mc2 = rabbit_message_interceptor:intercept(Mc1),

3. MQTT

   rabbitmq-server/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

   Line 1561 in 658d5b8

   Msg = rabbit_message_interceptor:intercept(Msg0),

I haven't seen much interest in "external interceptors" and there is a very real cost and risks to making such a feature extensible. This 3rd party code will be executed on the hottest code path there is, which means it will have a performance hit and a poorly written implementation can render the entire node unavailable for publishers. And, since nodes in most clusters run with the same set of plugins, the entire cluster.

There were very few channel interceptor examples that were actually used, so before 4.0 we have ported them to the core and that was that.

[michaelklishin]

An message interceptor is a function of M (a message container) to M. It can be a single function or a chain of them.

In fact, the only public function in that interface already folds over a list of functions stored in a persistent term named incoming_message_interceptors.

If there is/was a way to populate the list one way or another on node boot with a list of advanced.config (or maybe even rabbitmq.conf)-originating values, it would not be all that different from channel interceptors, that were registered using a dedicated function
on plugin boot.

[michaelklishin]

@getlarge a closer look suggests that in order to make the list of message interceptors extensible, rabbit_message_interceptor needs to change a bit and instead of storing interceptor names, it would store function references. The existing intercept/2 function with multiple heads will have to change to separate functions, rabbit_message_interceptor:set_header_routing_node/2 and rabbit_message_interceptor:set_header_timestamp/2, and whatever may be populating incoming_message_interceptors (a rabbit.schema configuration translation) will have to be adapted accordingly to populate it with function references such as fun rabbit_message_interceptor:set_header_routing_node/2. Or at least FMA ({Module, Function, Arity}) triplets but chances are that won't be necessary.

Assuming those changes are done, a plugin would simply need to add a tuple to a persistent term named incoming_message_interceptors on boot using persistent_term:put/2.

The tuple format will have to change from {InterceptorName, Overwrite} to {InterceptorFunction, Overwrite}, where InterceptorFunction is a function, such as fun rabbit_message_interceptor:set_header_routing_node/2 or any function a plugin might provide.

Once updated, the value will become available to all processes (connections, sessions, channels) and the core will fold (reduce) over them all to return a final message.
The functions that modify the message will have to make the updates using the mc module and be really careful about efficiency and error/exception handling.

[getlarge]

@michaelklishin, thank you for sharing so many details from the past/current design and the path for potential implementation.

I haven't seen much interest in "external interceptors" and there is a very real cost and risks to making such a feature extensible.

I know that such a "hook" would be called very often and could slow down message processing.
Is this warning an attempt to discourage anyone from tackling this challenge?
It counterbalances with your very encouraging advice and suggestions.

[michaelklishin]

I did not mean to discourage an attempt to make these changes.

The risks of a problematic interceptor implementation are real and very far reaching. Every publisher will be affected, regardless of the protocol or specific features used.

If someone contributes such a change it will be considered.

I cannot guarantee that end result would be accepted by our team but with persistent terms which are heavily optimized for frequent reads and infrequent writes, it might work out OK.

[gomoripeti]

We would like to implement MQTT packet interceptors feature

Any feedback is appreciated, what should or should not be as described bellow.

Implementation proposal

The implementation would take the best from AMQP 0.9.1 channel
interceptors and message interceptors.

I propose to make outgoing packets interceptible too (in addition to
the incoming ones). Can be useful when implementing custom counters of
outgoing events. (For example per-client_id publish-to-client count).

An interceptor behaviour could look like

-type interceptor_options() :: map().
-type interceptor_config() :: term().
-type interception_point() :: incoming_packet | outgoing_packet.

-callback init(
    rabbit_mqtt_processor:state(),
    interceptor_options()) -> interceptor_config().

-callback intercept(
    interception_point(),
    mqtt_packet(),
    interceptor_config()) -> mqtt_packet()  | {error, Reason}.

Init

When an MQTT connection is established, a CONNECT packet is received
and the connection/processor state is assembled, the interceptor
init function is called with the processor state and the interceptor
specific options (from rabbitmq.config). At this point all interceptor
specific initialization can be done and some data can be stored in the
interceptor config, which remains unchanged throughout the life of the
connection.

Note that in this case it is not possible to intercept and modify the
CONNECT packet itself. (This is similar to the channel interceptor
that cannot intercept connection methods.) The benefit is that
processor state calculation already happened so the interceptor does
not have to calculate eg user name or vhost replicating the logic
which is there in the mqtt_processor.

Should outgoing CONNACK packets be intercepted?

The interception point is introduced because in the MQTT protocol some
packets like PUBLISH can appear both client->server and server->
client.

Configuration

An ordered list of callback module and options pairs. (I suggest that
the order matters so that in the very unlikely case of conflicting
interceptors the operator can decide the order, and it is explicit. As
opposed to the channel interceptor where the registration order is
arbitrary and only one callback can intercept one method) In advanced
config format:

{rabbitmq_mqtt,
[{mqtt_interceptors,
[{Interceptor :: module(), Options :: map()}]},

The ini-style config format may look like the bellow in case there are no options

mqtt.interceptors.$number.$interceptor.enabled=true

Questions:

1. Can interceptor state change?
   a. config: initialised at connection init
   (rabbit_mqtt_processor:process_connect/5) and remains unchanged
   throughout the lifetime of the connections
   b. state: can change later, the intercept callback functions return
   an updated state for each intercepted packet

2. Can interceptors be enabled/disabled on-the-fly for existing connections?
   a. without broker restart but only for new connections
   b. apply change also for existing connections

   • mqtt_reader processes would receive a msg to refresh interceptors
     -> this would wake them all up from hibernation, not good
   • new interceptors would be initialised from latest mqtt_processor state

3. Should there be an init callback at all?
   a. yes
   b. no, intercept receiving processor state is enough. Interceptor
   can still store some state in the process dictionary but it is
   not very nice.

[ansd]

@gomoripeti thanks for offering to implement a solution.

What are your use cases of an "MQTT packet interceptor"?

We just internally discussed this issue as well as #10050. @kjnilsson made some very good suggestions.

Our latest thinking is as follows:

• Make rabbit_message_interceptor pluggable. (Plugins could possibly use rabbit_registry to register their modules.)
• Message interceptors keep getting called, like today, in the AMQP 0.9.1 channel, AMQP 1.0 session, and MQTT connection process before the message is routed.
• The message interceptors API changes from the current intercept(mc:state()) -> mc:state(). to something like intercept(mc:state(), OpaqueState :: term()) -> mc:annotations(). where OpaqueState is the state of the AMQP 0.9.1 channel, AMQP 1.0 session, or MQTT connection process. The 3rd party plugin should not know the exact state internals. Instead, there should be helper methods ("getters") to query this opaque state for certain fields. The goal here is to make this API stable, i.e. to avoid breaking plugins whenever the internal state of the AMQP 0.9.1 channel, AMQP 1.0 session, or MQTT connection process changes. The returned mc:annotations() will be merged with the message container's annotations.
• Introduce no channel interceptor equivalent for AMQP 1.0 or MQTT. (The channel interceptor API seems to be used only by the Sharding plugin.)
• No outgoing message interceptors for now.

Does this proposal fit your use cases?

[gomoripeti]

thanks David and team for the feedback

• "getters" I also originally had this idea (It is definitely nicer) but I see one benefit of an interceptor that it can be implemented independent of RabbitMQ. If I would need one field which does not have a getter yet, I have to wait for a new RabbitMQ release (patch or minor? a minor could be half a year away) and additionally I can't apply that interceptor for older RabbitMQ versions. (By older I mean that has the new interceptor mechanism but does not have the specific getter I need). The argument for non-opaque access is that usually community plugins depend already on a specific minor version of RabbitMQ. Similar to behaviour changes plugins need to adopt. But I understand that exposing the record internals would make it a semi-external API and would put burden on the Core Team that cannot be changed just any time. If we add getters for most of the fields which make sense I am happy with it.
• the "registration" is convenient for the end users but it is a bit magical and not explicit. I don't think the startup order of plugins is defined. What would be the order in which the interceptors are applied? Would it be too much from the user to ask to configure a sharding message interceptor in the config in addition to enabling the plugin? I would prefer a more static setup/configuration similar to eg the auth backends.
• "outgoing interceptor": we would like to have the option to implement custom metrics which might be too specific to introduce into RabbitMQ itself (maybe we only want to enable it on a few clusters) but still count outgoing messages ie deliveries to clients per some exotic grouping like client_id.
• what would happen with the sharding plugin and channel interceptor? (I might misunderstood the intention to rewrite it to message intercepting)

Another example is a get_empty rate-limiter interceptor (#12255) I'm bringing this up not because we definitely want to implement it but if such a feature can be implemented as an interceptor then it shows interceptors are powerful enough. This would need an outgoing interceptor that intercepts AMQP 0.9.1 packets (get_emtpy) (if I understand correctly) and the interceptor can return a delay result or noreply. That's already too much in 3 points for the proposed extended message interceptor (capture outgoing, capture packet instead of message, different outcome than just extra annotations).

RabbitMQ already provides many extension points which is great. I see interceptors as another hook-in point in RabbitMQ where it can be extended with exotic or too specific features without "bastardizing" RabbitMQ and without the burden for the Core Team of having to maintain it (and without forking). So from this point of view the more powerful and generic interceptors are the better. But I see the point of the Core Team to keep things simple and avoid introducing something that is not used.

To sum up a powerful interceptor would be useful for us in the future. Specifically intercepting "outgoing messages" would help us in our current need. I don't have a very strong feeling about the other points . And even without "outgoing" just the feature to enrich an incoming message with client_id is something some of our users would benefit from.

[kjnilsson]

• "getters" I also originally had this idea (It is definitely nicer) but I see one benefit of an interceptor that it can be implemented independent of RabbitMQ. If I would need one field which does not have a getter yet, I have to wait for a new RabbitMQ release (patch or minor? a minor could be half a year away) and additionally I can't apply that interceptor for older RabbitMQ versions. (By older I mean that has the new interceptor mechanism but does not have the specific getter I need). The argument for non-opaque access is that usually community plugins depend already on a specific minor version of RabbitMQ. Similar to behaviour changes plugins need to adopt. But I understand that exposing the record internals would make it a semi-external API and would put burden on the Core Team that cannot be changed just any time. If we add getters for most of the fields which make sense I am happy with it.

The API for retrieving context dependent details would have to be limited to data we can reasonably commit to maintain going forward. Generally this would be protocol specific fields, like MQTT client id, user, vhost, the protocol itself and version.

[ansd]

the "registration" is convenient for the end users but it is a bit magical and not explicit. I don't think the startup order of plugins is defined. What would be the order in which the interceptors are applied? Would it be too much from the user to ask to configure a sharding message interceptor in the config in addition to enabling the plugin? I would prefer a more static setup/configuration similar to eg the auth backends.

The module name of the plugin need to be set somewhere, right? I guess you're right, one could configure the plugin module name directly in the config and put it into persistent term, this is probably easier. If the module isn't loaded yet (i.e. plugin not enabled yet) or module isn't loaded anymore the exception could be caught by rabbit_message_interceptor and nothing happens.

[gomoripeti]

yes I think the message interceptor configuration could look very similar to how it looks today. Let's say set_header_timestamp would be implemented by rabbit_header_timestamp module and set_header_routing_node as rabbit_header_routing_node module respectively.
Then the advanced config could look like:

{rabbit, [
  {message_interceptors, [
    {rabbit_header_timestamp, #{overwrite => true}},
    {rabbit_header_routing_node, #{overwrite => false}},
    {my_plugin_interceptor, #{}}],

and the value of message_interceptors could be stored as is (or with minimal changes) in persistent_term.

The channel interceptor also handles cases when the callback module is not available https://github.com/rabbitmq/rabbitmq-server/blob/v4.0.5/deps/rabbit/src/rabbit_channel_interceptor.erl#L71