Message buffering and frequencies

[nateyazdani]

Given that plugins are stateless, the function of running them at certain frequencies has really become to reduce traffic to/from the database and to ensure that the respective models are updated at a known frequency. For example, statscache has plugins to count the volume of messages over given intervals (e.g., 1s, 5s) that are run at frequencies of the same intervals. So the volume per second plugin runs once per second. However, it cannot take advantage of this fact and simply record the number of messages received each time it runs, firstly because it is supposed to be stateless and secondly because some messages timestamped within the interval may arrive too late. Therefore, each plugin realistically needs to iterate over all messages received, check each timestamp, and then perform whatever processing needed. If I'm not missing anything, then plugins are not really affected or benefited by controlling their individual processing frequencies.

I would like to propose that we consider the following two options:

1. Real-time processing: Handle each message when it arrives, perhaps by implementing plugins as FedmsgConsumers directly. The obvious benefit to this is that stats models are always up-to-date, which is very cool, but the downside is potentially heavy traffic to the database when things get busy.
2. Standard update frequency: Use a single polling producer to run all plugins at an application-configurable interval (e.g., 1s). This isn't a whole lot different from what we're doing now. If the first option is too risky, this would be a good middle-ground, as database I/O will be more predictable.

Thoughts? If there are any alternative ideas or aspects that I'm missing, I'd be happy to hear about them.

[rtnpro]

👍 for raising this issue.

I agree with advantages and disadvantages of option 1. However, I am not totally comfortable with the current implementation, which is similar to option 2, as, the polling producers maintain state in memory. I will always support performing actions in realtime however db IO becomes a bottleneck. I'd like to propose an alternative solution, which takes the best of options 1 and 2 along with some more:

• We listen for events in realtime
• As soon as we get a new message, we enqueue it to a task queue
• some workers pick up the tasks from the task queue, perform some in memory operations (in case of volume plugins, etc.) and does a batch write to DB. So, the changes will not appear in realtime, but some delay is acceptable, until the time we go for real REALTIME model. The workers take care of atomicity of the entire operation. If something fails, rollback.

In the above option, we just need a single fedmsg listener which acts as a dispatcher and enqueues tasks. Now, whether some preprocessing happens before en queuing or after it is matter of discussion.

Now, if we want to go really realtime, we can use a faster data store as a caching layer to write processed data to, which, then gets propagated to a slower RDBMS datastore as batch writes. In our app for APIs, we just need to write some code, to collate data from the faster cache and the slower main DB to return results. And, we'll have an API server returning realtime data. Exposing a websocket API around statscache will allow people to build realtime reporting tools around statscache.

I'd like to emphasize on the fact that I want to minimize lossy in memory data crunching as much as possible.

Looking forward for your feedback and concerns :D

[ralphbean]

I'm fine with either option however the trade off, as I see it, is between robustness and complexity.

Spreading things out to a task queue (like redis) or a faster local data store (like redis) would work, but would add more components that future developers and admins would have to juggle. OTOH, it would let us scale out more workers processes since everything's not trapped in the fedmsg-hub daemon anymore.

Note, as things stand now I don't think we're actually lossy. If we crash and lose the "buckets", the initialize functions should know how to get the backlog from datagrepper at startup to figure out where they were, no? Startup is slow in that case, but it's not data-loss.

[nateyazdani]

I like the idea of making statscache distributed, but to avoid the potential for data loss, I think that it's necessary to ensure that messages are processed in order. Otherwise, we may end up playing a delicate game to prevent data loss, as with the task queue described by @rtnpro.

For example, say two messages are on such a task queue, and workers A and B grab the first and the second messages, respectively. If worker A goes offline before committing its change, worker B will continue along and commit its change anyway. The message that A was processing will have been lost, and that data loss will be difficult to determine, particularly since the database models will be "up-to-date" as of the timestamp on the second message processed by worker B. We could alleviate this issue by requiring a confirmation to mark the task as resolved (otherwise placing it back on the queue), but then if the Redis instance goes offline before synchronizing to disk, we'd risk the same issue.

To avoid that complexity, I would suggest that if statscache is to become distributed (which I agree is a good idea), it would best be done at the plugin level (i.e., one worker per plugin), where each worker processes messages in order. This way, we can rely on the latest timestamp in the database models to determine from what point we need to backprocess fedmsg traffic from datagrepper (as mentioned by @ralphbean). Going further with that, statscache could (potentially) be split into two parts: the web frontend and a framework of sorts for these plugin workers, built atop either FedmsgConsumer or a statscache message queue using Redis.

Speaking of backprocessing, that's actually what got me started on this. As things currently stand, each plugin is responsible for doing this in their initialize() method, even though the code is largely similar. I've been trying to centralize this process, but the separation of plugins amongst the different polling producers has been making things tricky.

Another possibility is to do real-time processing, but utilizing batch database updates like @rtnpro describes. Plugins would process messages as they arrive, in real-time, and accumulate the database change in memory, until a polling producer triggers a batch database update at regular intervals. This could be extended to support distributed processing later, be running each plugin in a separate worker process and using some sort of message bus/queue to dispatch incoming fedmsg traffic.