Closed jeremycline closed 6 years ago
pypingou
commented
7 years ago I'm all for switching if we have something more reliable.
We may want to get @puiterwijk 's attention to this as well since I believe he uses pika for basset (though w/o twisted I think)
jeremycline
commented
7 years ago I spent a lot of time thinking about this this afternoon and wrote up our various design options for the consumer pieces of FMN:
I think the general front-end --> processor/worker --> backend design is a solid overall design. Assuming we continue with that design, we have four major pieces to consider:
Message front-end - collects raw fedmsgs for processing. This is mostly IO-bound.
Message processors - processes the fedmsgs to determine the recipients and medium (email, irc, etc). This is a CPU-bound task.
Message back-ends - sends the fedmsgs as human-readable text to the recipients. This is an IO-bound task.
Inter-process communication to tie the front-end(s) to the processor(s) and the processor(s) to the back-end process(es) (unless the back-ends are invoked from the processor processes)
These pieces should make the features we need from FMN easy to implement.
Dispatching messages individually or batched (based on time/size constraints).
Scale-able: Computing who should receive a notification is expensive since a
set of filters must be applied to a message for each back-end that a user has
configured, and for each user. That is, the number of filters run on a message
is users x back-ends x filters. As the number of users goes up, the amount
of compute time goes up. As the number of fedmsgs emitted per minute increases,
so to does the computation requirements.
Simple debugging tools (records of what got sent where and for what reason)
The selection of this piece will drive decisions in the other components, so it is covered first. We have several options for inter-process communication. Currently, FMN uses RabbitMQ (except when it doesn't and instead stashes messages in the database).
RabbitMQ supports message durability so if it goes down, messages are not lost. Messages are placed in queues and once they are received and acknowledged, they are removed from the queue.
We already use ZeroMQ for fedmsg, so we could use ZeroMQ here to pass messages between the processes. It's essentially fancy sockets, and there is no durability.
With this approach, we place incoming messages into a database table. Message processors select a message and mark it as "in-progress". Once it is processed, it is marked as "done" and the message with its recipients is placed in the database. The same process is applied by the back-end process(es). This approach requires a recovery mechanism to handle processors that die while processing a message and generally involves re-inventing several wheels. However, it's worth noting that it's possible.
The FMN message front-end is the component that gets the fedmsgs the other services have emitted. We can do this in one of two ways:
This is the current approach. In this approach, we have a fedmsg consumer that subscribes to all topics on the fedmsg "bus".
In this approach, we would drop our fedmsg consumer that subscribes to all fedmsg topics. Instead, we would rely on the datanommer-backed datagrepper service to collect fedmsgs.
Message processors (workers) are processes that accept a fedmsg from the front-end
as input and outputs a map of back-ends to recipients for the message. They do this
by using user-defined filters composed of FMN rules that evaluate to either true or
false. Each user can define one or more filters for each back-end type. All the
filters for a back-end must evaluate to True for the message to be sent.
We need multiple workers because it is computationally expensive to compare filters to messages. Although there can be optimizations in this area, it still takes a lot of CPU time.
These can be implemented in several ways:
Celery is a task system implemented in Python. It supports RabbitMQ, Redis, and Amazon SQS as the message broker. Tasks are assigned to workers (multiprocessing, eventles, gevents or single threaded) and the results are placed in a results store (an AMQP queue, Redis, memcached, a database via SQLAlchemy, etc).
In addition to basic task dispatching and results aggregation, Celery offers task monitoring, time and rate limits on tasks, optional worker reaping to avoid resource leaks, and scheduling.
The current implementation is a Twisted application which you can run multiple instances of. This is sub-optimal since it is a CPU-bound process with little or no IO (depending on how caching is implemented) and thus it gets little benefit from using Twisted (but it is much more confusing because of it).
We could keep this approach and simply drop the use of Twisted.
Continue handling the worker lifecycle ourselves, but switch to using ZeroMQ or PostgreSQL for IPC.
FMN defines an API for back-ends via a class. This API needs to be carefully evaluated and any gaps need to be addressed with additional APIs.
When it comes to how these back-ends are used, we have some choices:
Back-ends are currently implemented using Twisted. There are some advantages to this approach:
Back-ends are IO-bound and can benefit greatly by the reactor pattern Twisted uses.
Twisted provides a set of asynchronous APIs for email, IRC, and more. These are very helpful.
It makes everything easier if we just run one process for the back-end since we can rate-limit. However, we might be able to implement this with scale-able back-ends so that should be investigated.
There are also some cons to this approach. Namely, Twisted is not particularly easy to get into so it makes it less approachable. However, this can be mitigated with careful documentation.
If we can make it so each message processor can have its own set of back-ends and connections to those back-ends (e.g. they can all log into IRC), then we could attach the dispatch of the message to the message processor. This is not likely to be easy, and if RabbitMQ + Celery is used, probably not the right design since there's going to be a results table.
I tried to cover every reasonable (and not so reasonable) option without too much bias, but I clearly have opinions. This is what I think we should do:
Inter-process communication - RabbitMQ.
Message front-end - Continue to use a fedmsg consumer since integrating with datagrepper means polling. Switch from Pika to Kombu for the AMQP interactions.
Message processors - Celery tasks with RabbitMQ as the message broker and an SQL database via SQLAlchemy as the results store.
Message back-ends - Continue using Twisted to drive the back-ends since it involves a lot of IO. If we could implement this with Python 3.5 it would be even better since Twisted integrates with Python's async keywords.
jeremycline
commented
7 years ago Followup with some additional thoughts about the back-end:
If we opt to use a database as the results back-end for Celery, we have to poll the database for messages to send. This isn't very nice. However, if we use RabbitMQ as the results back-end (to avoid polling), we still have to handle batch messages. Batch messages add a lot of the complexity in the code today, I think. Where we choose to place the processed messages influences the complexity of implementing batch messages.
Batching is definitely one of the more complicated parts of FMN.
All implementations would be much easier if we simplified the batch configuration. At the moment, you can configure batches to be sent at regular intervals or when a certain number of messages have been queued up for you.
Restricting the configuration to be at regular intervals only would be nice because otherwise (if we want to use RabbitMQ for batching messages) we have to poll the queue. We should investigate what people's batch settings are. Maybe this is not a feature used by many/any people.
This approach is appealing since it means the place where results are stored is consistent. Unbatched messages can be sent without polling (using AMQP consumers). Batched messages are more difficult. One approach would be to route messages so that each user has a queue for batched messages. Then, we use Twisted to schedule a poll for each user's queue at the time interval they've configured.
Open questions:
Does it let us cancel the call or alter the interval? We need this since people will change their batching configuration. (yes)
Can RabbitMQ handle (potentially) thousands of queues? I think the answer is yes, but it's worth confirming. (yes, but the Erlang VM might need tweaking when the numbers get really big)
Can we drop the batch setting to send the messages when a certain number of queued messages is reached?
In this approach we would put unbatched messages in a RabbitMQ queue (so we don't have to poll), but we would insert messages we want to batch into the database. We would then poll the database on a regular interval. The downside to this approach is there are two totally separate places where messages are "queued" for dispatch.
In this approach we'd put everything in the database and poll it regularly to send unbatched messages. We would also need to poll it to send batched messages.
pypingou
commented
7 years ago I like the approach you're taking here and the careful description of your train of thoughts. Many thanks for being transparent and taking the time to do this.
When reading your last two comments, I ended up asking myself: what is it that we are trying to fix again? So I had to come back to this page to read the original comment. Not having followed really closely the state of things recently, do you have any numbers or trend explaining how broken FMN currently is? I am trying to asses (just for me) what would the priority of such rewrite be. Is it: the sky is falling, it breaks every two hours? Or more: it works but the way it does so is based on a pile of technical debt that we should pay back, preferably sooner rather than later?
jeremycline
commented
7 years ago I don't think it's critical at the moment, and it seems to be working okay for the moment (although I recall it getting "stuck" several times in the last few weeks). I wasn't prompted to write this due to it being on fire, I just started thinking about the design and wanted to capture my thoughts.
pypingou
commented
7 years ago Ok cool, 👍 on keeping notes of your thoughts, I was just curious if you were preparing to rewrite FMN based on them soon or not :)
jeremycline
commented
6 years ago For the record, I've done a lot of this:
The consumer remains about the same, but now it dispatches celery tasks (via RabbitMQ) rather than queuing messages directly.
Celery workers are used to process messages. This includes discovering recipients and formatting the message for delivery. It then places them in a queue based on the backend: fmn.backends.<backend name>
There's now a real Twisted application called the "delivery service". It subscribes to all the fmn.backends.<backend name> queues via kombu and sends the messages using the simplified backends. The backends no longer format messages (except for the IRC backend at the moment and only for confirmation messages).
Batched messages are currently handled using a celery beat task that queries the database. There's likely a neater way to do this with AMQP queues (one queue per user).
I've spent a good chunk of this week poking through FMN. I thought it would be helpful to have a sort of "State of the Union" and get some consensus on the direction we want to head with FMN consumers.
The basic FMN setup is: fedmsg consumer -> RabbitMQ -> fmn worker -> RabbitMQ -> fmn backend -> Email/IRC/etc.
We currently use the pika library to communicate with RabbitMQ. However, upstream appears to be dead and there are several issues that affect us. The most painful of these is that the Twisted integration pika offers doesn't work particularly well. The TwistedProtocolConnection is not API compatible with Twisted endpoints, for example, so we don't get easy reconnect support from Twisted, but because Twisted manages the TCP connection, we can't get particularly easy reconnect support from Pika either. Even if that PR got accepted, Pika isn't properly restarting the connection with the new TCP connection Twisted provides (this means that if the FMN <-> RabbitMQ connection is interrupted FMN crashes). The long and short of it is this is painfully complicated and we get very little out of it since we mostly use blocking calls within Twisted anyway.
What I would like to propose is that we replace pika with kombu. It provides a much simpler, higher-level interface, is active, and has thorough documentation. We can do this piece by piece, starting with the fedmsg consumer. With the workers, if we find that it's straightforward to integrate with the current twisted workers we can do that, or we can replace the workers with celery workers. Since we use so many blocking calls within the workers, it would probably be about the same anyway.
What do people think? Should we stick with Pika/Twisted and try to breathe some life into upstream? Should we switch to Kombu? Some other option? What about Twisted?