Add user information for connection.closed event under the amq.rabbitmq.event exchange

[galah92]

Hi RabbitMQ.

I'm working with the MQTT plugin, and trying to get notifications about MQTT clients connecting & disconnecting. I enabled the Event Exchange Plugin and am able to consume messages from a queue named channel.*.

While connection.created events contained the user & user_who_performed_action under the message headers, the connection.closed does not, and so there's no way for me to correlate between these two events.
I also couldn't find any relevant doc or code reference regarding these two event payloads.

Since I do get the consumerTag & name field, I though about fetching the user based on these from the HTTP API, but couldn't find anything relevant.

My request is to add the same correlation fields to the connection.closed events.

Attached example events:

connection.created:

{
   "fields":{
      "consumerTag":"amq.ctag-J3ZBInPWRiX8AsvCw83KMQ",
      "deliveryTag":41,
      "redelivered":false,
      "exchange":"amq.rabbitmq.event",
      "routingKey":"connection.created"
   },
   "properties":{
      "contentType":"undefined",
      "contentEncoding":"undefined",
      "headers":{
         "ssl":false,
         "client_properties":[
            
         ],
         "frame_max":0,
         "channel_max":1,
         "channels":1,
         "variable_map":"#{<<\"client_id\">> => <<\"mqttjs_b99a1b36\">>}",
         "user_who_performed_action":"device",
         "node":"rabbit@bf5a60dffe48",
         "connected_at":1667573092655,
         "type":"direct",
         "vhost":"/",
         "user":"device",
         "peer_port":35884,
         "peer_host":"{0,0,0,0,0,65535,44063,1}",
         "name":"172.31.0.1:35884 -> 172.31.0.3:1883",
         "port":1883,
         "host":"{0,0,0,0,0,65535,44063,3}",
         "protocol":"`{'MQTT',\"3.1.1\"}`",
         "pid":"<[email protected]>",
         "timestamp_in_ms":1667573092655
      },
      "deliveryMode":2,
      "priority":"undefined",
      "correlationId":"undefined",
      "replyTo":"undefined",
      "expiration":"undefined",
      "messageId":"undefined",
      "timestamp":1667573092,
      "type":"undefined",
      "userId":"undefined",
      "appId":"undefined",
      "clusterId":"undefined"
   },
   "content":"<Buffer >"
}

connection.closed:

{
   "fields":{
      "consumerTag":"amq.ctag-J3ZBInPWRiX8AsvCw83KMQ",
      "deliveryTag":42,
      "redelivered":false,
      "exchange":"amq.rabbitmq.event",
      "routingKey":"connection.closed"
   },
   "properties":{
      "contentType":"undefined",
      "contentEncoding":"undefined",
      "headers":{
         "name":"172.31.0.1:35884 -> 172.31.0.3:1883",
         "node":"rabbit@bf5a60dffe48",
         "pid":"<[email protected]>",
         "timestamp_in_ms":1667573092665
      },
      "deliveryMode":2,
      "priority":"undefined",
      "correlationId":"undefined",
      "replyTo":"undefined",
      "expiration":"undefined",
      "messageId":"undefined",
      "timestamp":1667573092,
      "type":"undefined",
      "userId":"undefined",
      "appId":"undefined",
      "clusterId":"undefined"
   },
   "content":"<Buffer >"
}

Thanks

[lukebakken]

I vaguely remember this being requested before but the user information is not available when a connection is closed. @michaelklishin @ansd ?

[michaelklishin]

Correct, sometimes connections are closed before authenticating successfully so we don't have anything to populate that field with

[galah92]

@michaelklishin that's understandable, but on the "happy path", when disconnected connections did authenticated, is it still problematic?
Do I have another way to achieve what I want? Because currently the closed event is pretty unuseful.

[michaelklishin]

connection.close handler should have access to the username from connection state but the event may be emitted elsewhere, where it's not the case. There are several scenarios and code paths that deal with connection closure but only one that leads to a successfully authenticated connection.
This is the case for every protocol we support.

[galah92]

If that single scenario can populate the user field, and I can guarantee that only authenticated clients live in my system (which I'm doing anyway), won't that work?

[galah92]

By the way, I the user header is preset for the channel.created & channel.closed events, when in case of MQTT clients, are fired on the same time. The problem there is the for some reason, at least from my tests, I'm getting two channel.created & channel.closed events for a single MQTT client. Does that make sense to you in a way?

[ansd]

The problem there is the for some reason, at least from my tests, I'm getting two channel.created & channel.closed events for a single MQTT client. Does that make sense to you in a way?

In the current implementation, MQTT is proxied via AMQP 0.9.1. It uses something called an "AMQP direct connection" underneath. This creates 2 channels in the rabbit app (server side). Additionally channels are created in the client side (MQTT plugin). (So, it's nothing to worry about, just some implementation details.)

We are currently working on "native MQTT" in #5895 which does not use any channels anymore. In that case you won't see channel.created and channel.closed events at all.

From my point of view, it makes sense to add the user to the connection.closed event (whenever that user is available). It's currently not only missing from MQTT, but also from AMQP 0.9.1 for example.

Feel free to create a PR including a test.
For the test, you can check out

rabbitmq-server/deps/rabbitmq_mqtt/test/integration_SUITE.erl

Lines 157 to 208 in add7b74

	events(Config) ->
	ok = rabbit_ct_broker_helpers:add_code_path_to_all_nodes(Config, event_recorder),
	Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
	ok = gen_event:add_handler({rabbit_event, Server}, event_recorder, []),
	ClientId = atom_to_binary(?FUNCTION_NAME),
	{C, _} = connect(ClientId, Config),
	[E0, E1] = get_events(Server),
	assert_event_type(user_authentication_success, E0),
	assert_event_prop([{name, <<"guest">>},
	{connection_type, network}],
	E0),
	assert_event_type(connection_created, E1),
	assert_event_prop({protocol, {'MQTT', "3.1.1"}}, E1),
	{ok, _, _} = emqtt:subscribe(C, <<"TopicA">>, qos0),
	[E2, E3] = get_events(Server),
	assert_event_type(queue_created, E2),
	QueueNameBin = <<"mqtt-subscription-", ClientId/binary, "qos0">>,
	QueueName = {resource, <<"/">>, queue, QueueNameBin},
	assert_event_prop([{name, QueueName},
	{durable, true},
	{auto_delete, false},
	{exclusive, true},
	{type, rabbit_mqtt_qos0_queue},
	{arguments, []}],
	E2),
	assert_event_type(binding_created, E3),
	assert_event_prop([{source_name, <<"amq.topic">>},
	{source_kind, exchange},
	{destination_name, QueueNameBin},
	{destination_kind, queue},
	{routing_key, <<"TopicA">>},
	{arguments, []}],
	E3),
	{ok, _, _} = emqtt:unsubscribe(C, <<"TopicA">>),
	[E4] = get_events(Server),
	assert_event_type(binding_deleted, E4),
	ok = emqtt:disconnect(C),
	[E5, E6] = get_events(Server),
	assert_event_type(connection_closed, E5),
	assert_event_type(queue_deleted, E6),
	assert_event_prop({name, QueueName}, E6),
	ok = gen_event:delete_handler({rabbit_event, Server}, event_recorder, []).

for inspiration.

[michaelklishin]

If someone from the core team thinks that such "inherently optional" metrics make sense, we can add them. But it will mislead some, so we might have to explicitly set this username field to undefined in code paths when the connection is closed before successful authentication or access to connection state would be difficult.

I don't think refactoring a decade old code path just to access to this field would be worth the effort and risk.

[galah92]

Sounds like when #5895 will be done I won't be able to utilize the channel.created and channel.closed events, but also that there's no intention to add the user fields to the connection.created & connection.closed fields, meaning there's no reliable way to detect clients connecting/disconnecting. That mostly means I won't be able to use RabbitMQ as an MQTT broker for my application.

[lukebakken]

@ansd @ChunyiLyu @michaelklishin care to comment on the above? I would assume appropriate events with the refactored MQTT support in #5895

[ansd]

Correct, the user will be included in connection.closed event in Native MQTT.

[lukebakken]

@galah92 -

there's no intention to add the user fields to the connection.created & connection.closed fields, meaning there's no reliable way to detect clients connecting/disconnecting

That is not correct. The user will be present in those events if the user has been authenticated. Note that this applies only to the "Native MQTT" support being added in #5895, so keep an eye on that PR.

[galah92]

Great, I wasn't sure this is the case. I'll be sure to follow #5895, but is there any ETA for that?

[ansd]

ETA is some time next year. Either in RabbitMQ 3.12 or 3.13.
Native MQTT ships in 3.12.

[galah92]

Hi all, I'm evaluating rabbitmq:3.12-rc and was happy to see it works! The properties.headers.user is populated as expected.

However, I'm going to use a single user to represent all of my IoT devices, where each of them will be connected with a different client_id, which is substantial in the MQTT world.

I am getting the client_id in properties.headers.client_properties, but only for the connection.created event and not for the connection.closed event. Is it possible to add it as well?

Thank you, 3.12 looks great until now.

In general, I wonder if it makes sense to match all the fields under the created event to the ones under the closed events.

[lukebakken]

@ansd my reading of this conversation is that connection.closed, using native MQTT, should have user information (if available). Is that still correct?

@galah92 -

I am getting the client_id in properties.headers.client_properties, but only for the connection.created event and not for the connection.closed event. Is it possible to add it as well?

If you share the code you are using to test this, I could check using the latest RabbitMQ source code. A git repository that I can clone and run would be ideal. Thanks.

[galah92]

@lukebakken Sure - https://github.com/galah92/rabbitmq-mqtt-lifecycle
I'm open to use other lifecycle events as well, as mentioned.

@michaelklishin I understand, but note that I'm not asking for all fields because of course some of them might not be present. But in case they are, in the "happy path", why not include some ways to correlate between these events? In the end, I'm only asking for a way to detect device disconnections by their client_id.

[ansd]

connection.closed, using native MQTT, should have user information (if available). Is that still correct?

@lukebakken yes, that's still correct.

@galah92 thanks for testing Native MQTT in 3.12 and thanks for your feedback. Your request makes sense to me. #8098 will include all infos from the MQTT connection created event in the MQTT connection closed event. Please try it out using the following docker image: pivotalrabbitmq/rabbitmq:mqtt-connection-closed-event-otp-min-bazel and provide feedback whether it works for you. (This Docker image got built by a GitHub action that builds Docker images for every branch pushed to rabbitmq-server.)

[galah92]

@ansd works. I am getting the exact same data under properties.headers.client_properties for both connection.created & connection.closed. It's an array that looks like the following: [ '{client_id,longstr,<<"mqttjs_086f3382">>}' ]

[lukebakken]

Thanks @ansd! @galah92 thanks for providing the code you are using to test.