From Database to Memory

We were able to solve both problems: A change in our data supply, away from a pull to a push principle, now makes it possible to distribute data to all relevant services almost simultaneously. The distribution is done by means of a messaging system. Each instance of each service is registered as a recipient and holds the processed data directly in memory. Requests can thus be answered quickly without a database query.

In the following, I will go into more detail about the two problems mentioned and then outline our solution.

Problems of the data supply

If a certain service experiences a higher load, i.e. a higher number of calls, this can be intercepted in a targeted manner by horizontal scaling, i.e. the use of additional instances of this service. However, since the instances share a database, the possible load is limited by this database. Here, too, horizontal scaling can be achieved by increasing the number of nodes in the database cluster. So much for the theory.

We use MongoDB as our database and have had rather bad experiences with this type of scaling. The increased write accesses of the data import often lead to global locks, which then temporarily block read accesses and thus lead to high latencies for the customer.

For scalability, it is also important not only to be able to increase the load, but also the amount of data that can be distributed to services in a timely manner.

With our architecture, it also became increasingly difficult to keep the database synchronized across all services. Due to the deviating runtimes of the import processes, e.g. due to different system loads or different processing steps, there are conceptual temporary deviations between the services. With the ever-growing number of new products and product data changes, these inconsistencies are increasingly amplified.

Alternative data supply via messaging system

Another factor influencing the scalability and robustness of distributed systems is the type of communication. The many services often do not communicate directly with each other, but instead asynchronously by sending messages over an event bus. According to the publish-subscribe principle, a message can have more than one recipient. The recipient(s) do not even have to be known.

This form of message distribution can of course also be used to supply data. We use it now, since it offers the following advantage in our case: There is only one central regular process that fetches modified product data and then distributes it to all recipients via the event bus (cf. Fig. 4). This happens with low latency and thus we manage to keep the services relatively synchronous despite large amounts of data.

Event sourcing

However, this type of data sourcing alone does not replace a database because it is not persistent. A service would only know the data it receives from the time it registers on the event bus. Thus, each restart would result in data loss and require reprocessing of all messages to get back to the current state.

This corresponds to an essential part of how event sourcing works. Instead of storing the current state as in a database, all messages that led to that state are stored in sequence in a message log. The final state can then be restored by sequentially re-reading and processing all messages from the message log. At the same time, the message log is also the event bus through which new messages are distributed to all registered recipients.

By using event sourcing, we can run our services without a database. All data that we read from the message log at startup, as well as that we receive thereafter, is stored directly in the memory of the respective instance (see Fig. 5).

In order to save storage space, only the data of a product that is relevant for the service is stored, of course. Likewise, we do not use the Java heap but an off-heap memory. There are two reasons for this. On the one hand, it almost halves the required memory, since strings with ASCII characters are represented with only one byte (UTF-8) instead of two bytes (UTF-16) as in Java. Secondly (and much more importantly), it saves the Java garbage collector from having to manage several gigabytes of data, thus preventing unnecessary pausing of the virtual machine.

Eliminating database access improves service response times while allowing cost-efficient and true linear scaling. Linear means that twice as many instances can now handle exactly twice the load; 100 instances, a hundred times the load, etc. - as long as you neglect the limited network bandwidth. The database cluster is no longer a limiting factor, and it is cheaper to provide instances with lots of memory than to run additional nodes for a database cluster.

There are even more advantages. Since messages in a message log are immutable and append-only, not only the current state but any state can be restored for a point in time. One receives thereby quasi automatically a versioning. For debugging, each change to the system can be individually traced and repeated step by step.

Where there is light, there is also shadow

Apart from the fact that event sourcing is rather unknown and not so widespread, there are no frameworks and the complexity is quite high, it violates an important characteristic of micro services: Short startup times (see Twelve-Factor Apps).

This is countered by the time-consuming reading of the entire event log to recover the dataset. Since only the current state is usually of interest to a service, there are various approaches to achieving this as quickly as possible:

• Variant a) The event log can be read backwards. However, it is extremely difficult to determine how far backward to read to get all the messages needed to get the correct state.
• Variant b) The current state and the time of the last message is persisted locally in a database. After a restart, the event log does not have to be read from the beginning, but only from the stored time.
• Variant c) A compact snapshot of the event log is created regularly (e.g. every two hours), which only contains the last version of a message (see Fig. 6). After a restart, only a reduced event log in the form of a snapshot and the messages from the time of snapshot creation need to be read.