Question for DB archive

[mntnclmbr]

what would be the best approach to archive all the messages received in a DB ? would it be to tap into AlterPublish path for the plugin and pipe to DB or you would suggest to instead create a dumb subscriber whose sole purpose is to subscribe to all topics and keep writing to a DB ?

[halfgaar]

Well, as it currently stands, you don't have the payload available to you in the plugin. This is for performance reasons. However, I do have some ideas how I would make that possible. So, that functionality may come.

But aside from that, I can give some insight. First, how many messages per second are we talking? One thing you do have to realize, is that MQTT messages can be dropped, even in QoS. QoS is merely a way for the sending side to know if a message is delivered, not a guarantee that nothing is dropped. So, if you need to process 1 million messages per second, one subscriber isn't going to be able to handle that.

If you do do it in the plugin, beware that if you do a direct call to a a database back-end, that will be blocking and you'll be holding up the event loops. You'd need to do it in some kind of async way, or producer/consumer model with semaphore controlled queues. You'd have an extra thread per event loop that will be writing the messages to the DB.

Also, you should make use of multi-value insert statements.

Doing it in the plugin would also have the advantage of distributing the load over multiple threads (because you'd have a db writer thread per FlashMQ thread).

But, as stated, currently you can't do it in the plugin. I'd have to give an API change some thought.

If your message volume isn't super high, a subscriber on # is fine. With a persistent session, it won't even lose messages on restarts of the subscriber. You still want to do some kind of producer/consumer model, because any hold up in receiving messages will increase the chance of dropping them.

[mntnclmbr]

I was debating doing it with a subscriber on #, specifically in case of high volume messages and then having an an additional dependency of this new subscriber and managing it(cost, uptime, maintenance, etc) .

As a different approach, I was thinking about adding a hook to the RetainedMessagesDB::saveData() call as it seemed to be already backed by a circular buffer and setup in a way that multi threaded calls would be safe.

We are planning to build a simple IoT service where devices will push messages up, with publish being the crux and there may not really be a subscriber but just a DB where the data gets ingested and is accessed via DB's rest apis.

If we do not use any subscriber at all, i assume the messages get dropped but will still make its way to the the above function call if they are marked as "retain". Just replacing the persistencefile write calls with the DB calls in saveData() seemed like the simplest way to do it.

I think the messages should always get to this function and then based on the user selection in the config file, write to DB in addition to the actual persist handling. This seemed to be the easiest but probably not the cleanest I think.

Also, my approach is based on reading your code just in the last day, so I am sure I am making some dumb assumptions as well. :)

[halfgaar]

Remember that retained messages are set to a specific value; it's not a stream of messages. If you only dump the retained messages, you will miss transitions. And, the messages may be old.

Can you tell me what your message volume is expected to be?

[mntnclmbr]

we are not planning it to be huge for now but potentially in the range of 10s of thousands is what i would expect, when fully loaded

Actually, let me correct that, 10s of thousands of connections, the active message volume/sec should be a 10th of that

[halfgaar]

I think doing it with the plugin may still be your best choice. Forking the code has obvious downsides. I have some ideas how to make the payload available. I'll push a concept branch soon.

You may also want to test writing thousands of messages per second to a database. Multi-value statements is key. And, perhaps Timescaledb is an option, if it's time series based. If it's in a normal SQL database, actually removing old data is going to harder than inserting it. I have experienced this myself. That's one of many things a time series DB can help.

[slavslavov]

Hello @halfgaar, when you are thinking of enhancing the data available in the plugin could you also consider the following:

• Implement proxy protocol v2 per listener - https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt.
• Expose the client IP and port, and when proxy enabled the source and destination IP and port

The use case for this is to track/block IP addresses for some reasons when FlashMQ listens directly to the network or behind a proxy.

[halfgaar]

I created two issues. Feel free to give me pointers on HA proxy that would hasten my research :)

#25 is easy enough. I can probably have something ready soon.

On a side note, I already have on my list to support X-Forwarded-For for the websockets.

[slavslavov]

Sure, I can create a simple HA proxy configuration so you can use in your test. I will add some comments in the tickets based on previous experience.

[mntnclmbr]

makes sense, will wait for your concept branch :)

[halfgaar]

I pushed draft-change-payload. I'm still thinking over a few issues.

[slavslavov]

Is there any progress on this? Looks like I will need access to the payload from the plugin too.

[halfgaar]

Unfortunately no. I'll have to dust it off again.

Do you also need to change the payload?

[slavslavov]

In my case no, I don't need to change the payload but I don't know what @mntnclmbr's requirements are.

[mntnclmbr]

same for me as @slavslavov . I don't need to modify the payload

[halfgaar]

I don't know if you saw #33, but that's about done, to be released soon.

[slavslavov]

Hi Wiebe, I know you've been trying something more complex and probably you have a reason for not including the payload in aclCheck but I just drafted something in my fork that should work (unless I am not taking into account something related to the lifetime of MqttPacket). I have not tested it though - it just compiles. C++17 will have to be used as I used std::string_view which gives the possibility to get access to the payload without making any string copies. If this is acceptable, it can be part of your plugin V2 interface.
This PR is not intended to be merged here but can be a base for discussion - slavslavov#1

[halfgaar]

Hi Slav,

You seem to understand my code, and the concerns. It would need some plugin version detection and flashmq_plugin_alter_publish also needs the view, but I think this idea would work.

Where I got stuck myself is the ability to alter. It's incompatible with the design that it tries to limit unnecessary packet creations. I have some ideas still, but it would make it very complex.

The string_view is a nice solution that prevents unnecessary copies. However, C++17 would mean dropping support for Ubuntu 18, which is kind of a bummer (especially for myself because I'm lazy and haven't upgraded my PC...). But then again, I guess the time is about now because it's EOL soon anyway.

[slavslavov]

Great! If you decide to create a plugin V2 feature branch I may raise a PR to add one more function but I don't want to pollute this thread with the details.

[halfgaar]

We can continue discussion about that here: #33

Can you also start a discussion about your other idea? I also have another change: #34

[halfgaar]

I think everything from this discussion was resolved. Closing.