Saving high-frequency events to a connection-limit constrained database

Question

We've got a situation where I have to deal with a massive influx of events coming in to our server, at about 1000 events per second, on average (peak could be ~2000).

The problem

Our system is hosted on Heroku and uses a relatively expensive Heroku Postgres DB, that allows a maximum of 500 DB connections. We use connection pooling to connect from the server to the DB.

Events come in faster than the DB connection pool can handle

The problem we have is that events come faster than the connection pool can handle. By the time one connection has finished the network roundtrip from the server to the DB, so it can get released back to the pool, more than n additional events come in.

Eventually the events stack up, waiting to get saved and because there are no available connections in the pool, they time out and the whole system is rendered non-operational.

We've solved the emergency by emitting the offending high-frequency events at a slower pace from the clients, but we still want to know how to handle this scenarios in the event we need to handle that high-frequency events.

Constraints

Other clients might want to read events concurrently

Other clients continuously request to read all the events with a particular key, even if they are not saved in the DB yet.

A client can query GET api/v1/events?clientId=1 and get all the events sent by client 1, even if those events are not done saving in the DB just yet.

Are there any "classroom" examples on how to deal with this?

Possible solutions

Enqueue the events on our server

We could enqueue the events on the server (with the queue having a maximum concurrency of 400 so the connection pool doesn't run out).

This is bad idea because:

• It will eat up available server memory. The stacked-up enqueued events will consume massive amounts of RAM.
• Our servers restart once every 24 hours. This is a hard limit imposed by Heroku. The server can restart while events are enqueued causing us to lose the enqueued events.
• It introduces state on the server, thus hurting scalability. If we have a multi-server setup and a client wants to read all the enqueued + saved events, we won't know on which server the enqueued events live.

Use a separate message queue

I assume we could use a message queue, (like RabbitMQ?), where we pump the messages in it and on the other end there is another server that only deals with saving the events on the DB.

I'm not sure if message queues allow querying enqueued events (that weren't saved yet) so if another client wants to read the messages of another client, I can just get the saved messages from the DB and the pending messages from the queue and concatenate them together so I can send them back to the read-request client.

Use multiple databases, each saving a portion of the messages with a central DB-coordinator server to manage them

Another solution we've though is to use multiple databases, with a central "DB coordinator/load balancer". Upon receiving an event it this coordinator would choose one of the databases to write the message to. This should allow us to use multiple Heroku databases thus upping the connection limit to 500 x number of databases.

Upon a read query, this coordinator could issue SELECT queries to each database, merge all the results and send them back to the client that requested the read.

This is bad idea because:

• This idea sounds like ... ahem.. over-engineering? Would be a nightmare to manage as well (backups etc..). It's complicated to build and maintain and unless it's absolutely necessary it sounds like a KISS violation.
• It sacrifices Consistency. Doing transactions across multiple DB's is a no-go if we go with this idea.

Answer 1

Input stream

It is not clear if your 1000 events/second represent peaks or if it's a continuous load:

• if it's a peak, you could use a message queue as buffer to spread the load on the DB server over a longer time;
• if it's constant load, the message queue alone is not sufficient, because the DB server will never be able to catch up. Then you'd need to think about a distributed database.

Proposed solution

Intuitively, in both cases, I'd go for a Kafka based event-stream:

• All events are systematically published on a kafka topic
• A consumer would subscribe to the events and store them to the database.
• A query processor will handle the requests from the clients and query the DB.

This is highly scalable at all levels:

• If DB server is the bottleneck, just add several consumers. Each could subscribe to the topic, and write to a different DB server. However, if the distribution occurs randomly across the DB servers, the query processor will not be able to predict the DB server to take and have to query several DB servers. This could lead to a new bottleneck on the query side.
• The DB distribution scheme could therefore be anticipated by organising the event stream into several topics (for example, using groups of keys or properties, to partition the DB according to a predictable logic).
• If one message server is not sufficient to handle a growing flood of input events, you could add kafka partitions to distribute kafka topics across several physical servers.

Offering events not yet written in the DB to clients

You want your clients to be able to get access also to information still in the pipe and not yet written to the DB. This is a little more delicate.

Option 1: Using a cache to complement db queries

I have not analysed in depth, but the first idea that comes to my mind would be to make the query processor(s) a consumer(s) of the kafka topics, but in a different kafka consumer group. The request processor would then receive all the messages that the DB writer will receive, but independently. It could then keep them in a local cache. The queries would then run on DB + cache (+ elimination of duplicates).

The design would then look like:

The scalability of this query layer could be achieved by adding more query processors (each in its own consumer group).

Option 2: design a dual API

A better approach IMHO would be to offer a dual API (use the mechanism of the separate consumer group):

• a query API for accessing events in the DB and/or making analytics
• a streaming API that just forwards messages directly from the topic

The advantage, is that you let the client decide what is interesting. This could avoid that you systematically merge DB data with freshly cashed data, when the client is only interested in new incoming events. If the delicate merge between fresh and archived events is really needed, then the client would have to organise it.

Variants

I proposed kafka because it's designed for very high volumes with persistent messages so that you can restart the servers if needed.

You could build a similar architecture with RabbitMQ. However if you need persistent queues, it might decrease performance. Also, as far as I know, the only way to achieve the parallel consumption of the same messages by several readers (e.g. writer+cache) with RabbitMQ is to clone the queues. So a higher scalability might come at a higher price.

Answer 2

My guess is that you need to explore more carefully an approach that you have rejected

• Enqueue the events on our server

My suggestion would be to start reading through the various articles published about the LMAX architecture. They managed to make high volume batching work for their use case, and it may be possible to make your trade offs look more like theirs.

Also, you may want to see if you can get the reads out of the way - ideally you'd like to be able to scale them independently of the writes. That may mean looking into CQRS (command query responsibility segregation).

The server can restart while events are enqueued causing us to lose the enqueued events.

In a distributed system, I think you can be pretty confident that messages are going to get lost. You may be able to mitigate some of the impact of that by being judicious about your sequence barriers (for example - ensuring that the write to durable storage happens-before the event is shared outside of the system).

• Use multiple databases, each saving a portion of the messages with a central DB-coordinator server to manage them

Maybe -- I'd be more likely to look at your business boundaries to see if there are natural places to shard the data.

There are cases where losing data is an acceptable tradeoff?

Well, I suppose that there could be, but that's not where I was going. The point is that the design should have built into it the robustness required to progress in the face of message loss.

What this often looks like is a pull based model with notifications. Provider writes the messages into an ordered durable store. Consumer pulls the messages from the store, tracking its own high water mark. Push notifications are used as a latency reducing device -- but if the notification is lost, the message is still fetched (eventually) because the consumer is pulling on a regular schedule (the difference being that if the notification is received, the pull happens sooner).

See Reliable Messaging Without Distributed Transactions, by Udi Dahan (already referenced by Andy) and Polyglot Data by Greg Young.

Answer 3

If I understand correctly the current flow is:

1. Receive and event (I assume through HTTP?)
2. Request a connection from the pool.
3. Insert the event to the DB
4. Release the connection to the pool.

If so I think the first change to the design would be to stop having your even handling code return connections to the pool on every event. Instead create a pool of insertion threads/processes that is 1-to-1 with the number of DB connections. These will each hold a dedicated DB connection.

Using some sort of concurrent queue, you then have these threads pull messages from the concurrent queue and insert them. In theory they never need to return the connection to the pool or request a new one but you may need to build in handling in case the connection goes bad. It might be easiest to kill the thread/process and start a new one.

This should effectively eliminate the connection pool overhead. You will, of course need to be able to do push at least 1000/connections events per second on each connection. You may want to try different numbers of connections since having 500 connections working on the same tables could create come contention on the DB but that's whole different question. Another thing to consider is the use of batch inserts i.e. each thread pulls a number of messages and pushes them in all at once. Also, avoid having multiple connections attempting to update the same rows.

Answer 4

Assumptions

I'm going to assume that the load you describe is constant, as that is the more difficult scenario to solve for.

I'm also going to assume you have some way of running triggered, long-running workloads outside of your web application process.

Solution

Assuming that you have correctly identified your bottleneck - latency between your process and the Postgres database - that is the primary problem to solve for. The solution needs to account for your consistency constraint with other clients wanting to read the events as soon as practicable after they are received.

To solve the latency issue, you need to work in a way that minimizes the amount of latency incurred per event to be stored. This is the key thing you need to achieve if you aren't willing or able to change hardware. Given you are on PaaS services and have no control over hardware or network, the only way to reduce latency per event will be with some sort of batched write of events.

You will need to store a queue of events locally that gets flushed and written periodically to your db, either once it reaches a given size, or after an elapsed amount of time. A process will need to monitor this queue to trigger the flush to the store. There should be plenty of examples around on how to manage a concurrent queue that gets flushed periodically in your language of choice - Here is an example in C#, from the popular Serilog logging library's periodic batching sink.

This SO answer describes the fastest way to flush data in Postgres - although it would require your batching store the queue on disk, and there is likely a problem to be solved there when your disk disappears upon reboot in Heroku.

Constraint

Another answer has already mentioned CQRS, and that is the correct approach to solve for the constraint. You want to hydrate read models as each event is processed - a Mediator pattern can help encapsulate an event and distribute it to multiple handlers in-process. So one handler may add the event to your read model that is in-memory that clients can query, and another handler can be responsible for queuing the event for its eventual batched write.

The key benefit of CQRS is you decouple your conceptual read and write models - which is a fancy way of saying you write into one model, and you read from another totally different model. To get scalability benefits from CQRS you generally then want to ensure each model is stored separately in a way that is optimal for its usage patterns. In this case we can use an aggregate read model - for example, a Redis cache, or simply in-memory - to ensure our reads are fast and consistent, whilst we still use our transactional database to write our data to.

Answer 5

Events come in faster than the DB connection pool can handle

This is a problem if each process needed one database connection. The system should be designed so you have a pool of workers where each worker only need one database connection and each worker can process multiple events.

Message queue can be used with that design, you need message producer(s) that pushes events to the message queue and the workers (consumers) process the messages from the queue.

Other clients might want to read events concurrently

This constraint is only possible if the events stored in database without any processing (raw events). If events are getting processed before stored in database, then the only way to get the events are from database.

If the clients just want to query raw events then I would suggest using search engine like Elastic Search. You will even get the query/search API for free.

Given that it seems querying events before they are saved in database is important to you, a simple solution like Elastic Search should work. You basically just store all events in it and don't duplicate the same data by copying them into database.

Scaling Elastic Search is easy, but even with basic configuration it is quite high performant.

When you need processing, your process can get the events from ES, process and store them in database. I don't know what the performance level you need from this processing, but it would be completely separate from querying the events from ES. You shouldn't have connection issue anyway, as you can have a fixed number of worker and each with one database connection.

Answer 6

1k or 2k events (5KB) per second isn't that much for a database if it has an appropriate schema and storage engine. As suggested by @eddyce a master with one or more slaves can separate the read queries from committing writes. Using fewer DB connections will give you better overall throughput.

Other clients might want to read events concurrently

For these requests, they would need to also read from the master db as there would be replication lag to the read slaves.

I have used (Percona) MySQL with TokuDB engine for very high volume writes. There's also MyRocks engine based on LSMtrees that's good for write loads. For both these engines and likely also PostgreSQL there are settings for transaction isolation as well as commit sync behavior which can dramatically increase write capacity. In the past we accepted up to 1s lost data which was reported to the db client as committed. In other cases there were battery-backed SSDs to avoid loss.

Amazon RDS Aurora in the MySQL flavor is claimed to have 6x higher write throughput with zero-cost replication (akin to slaves sharing a filesystem with master). The Aurora PostgreSQL flavor also has a different advanced replication mechanism.

Answer 7

I’d drop heroku all together, that is to say, i’d drop a centralised approach: multiple writes that peak the maximum pool connection is one of the main reasons why db clusters where invented, mainly cause you don’t load the writing db(s) with read requests that can be performed by other db’s in the cluster, i’d try with a master-slave topology, moreover - as somebody else already mentioned, having your own db installations would make it possible to tune the whole system to make sure query propagation time would be correctly handled.

Good luck