What could cause such a large company as Lamoda, with a well-established process and dozens of interconnected services, to significantly change its approach? Motivations can vary greatly: from legislative changes to the inherent desire of all programmers to experiment.
However, this does not mean that one cannot expect additional benefits. Sergey Zaika will discuss what specific gains can be achieved by implementing an events-driven API on Kafka.). There will also be tales of bumps along the way and interesting discoveries — experiments cannot be done without them.
Disclaimer: This article is based on materials from the meetup that Sergey held in November 2018 at HighLoad++. The live experience of Lamoda working with Kafka attracted an audience just as much as other presentations in the schedule. We believe this is a great example of how finding like-minded people is always possible and necessary, and HighLoad++ organizers will continue to strive to create an environment conducive to this.
About the process
Lamoda is a large e-commerce platform with its own contact center, delivery service (as well as many partners), photo studio, and a huge warehouse, all working on its own software. There are dozens of payment methods, and B2B partners who can use some or all of these services want to know real-time information about their products. Moreover, Lamoda operates in three countries besides Russia, where things are slightly different. In total, there are probably over a hundred ways to configure a new order, each of which must be processed in its own way. All of this operates through dozens of services that communicate in sometimes non-obvious ways. There is also a central system, whose main responsibility is the order statuses. We call it BOB, and I work with it.
Refund Tool with events-driven API
The term events-driven is quite clichéd, and we will define later what it actually means. Let me start with the context in which we decided to test the events-driven API approach on Kafka.
In any store, besides orders that customers pay for, there are occasions when a refund is required because the customer is not satisfied with the item. This relatively short process involves confirming information, if necessary, and then processing the refund.
However, the return process has become complicated due to changes in legislation, and we had to implement a separate microservice for it.
Our motivation:
- Law FZ-54 — briefly, the law requires reporting to the tax authority about every monetary transaction, whether it's a return or receipt, within a quite short SLA of a few minutes. As an e-commerce company, we conduct a considerable number of operations. Technically, this means a new responsibility (and hence a new service) and adjustments in all involved systems.
- BOB split — an internal company project aimed at freeing BOB from numerous non-core responsibilities and reducing its overall complexity.
In this diagram, the main Lamoda systems are depicted. Currently, most of them represent more of a constellation of 5-10 microservices around a shrinking monolith.They are gradually growing, but we try to make them smaller, because deploying a dedicated fragment in the middle is risky — we cannot allow it to fail. All exchanges (the arrows) have to be reserved, preparing for the possibility that any of them may become unavailable.
There are also quite a few exchanges in BOB: payment systems, delivery, notifications, etc.
Technically, BOB consists of:
- ~150k lines of code + ~100k lines of tests;
- php7.2 + Zend 1 & Symfony Components 3;
- >100 APIs & ~50 outgoing integrations;
- 4 countries with their own business logic.
Deploying BOB is costly and painful; the amount of code and the tasks it resolves are such that no one can fully grasp it entirely. Overall, there are many reasons to simplify it.
Return process
Initially, the process involves two systems: BOB and Payment. Now, two more are introduced:
- Fiscalization Service, which will take on problems related to fiscalization and communication with external services.
- Refund Tool, which simply carries new exchanges so that BOB doesn't become bloated.
Now, the process looks like this:
- BOB receives a request to refund the money.
- BOB informs the Refund Tool about this.
- Refund Tool instructs Payment: 'Return the money.'
- Payment returns the money.
- Refund Tool and BOB synchronize their statuses with each other since both need this for now. We are not yet ready to completely switch to Refund Tool, as BOB has a UI, accounting reports, and generally much data that cannot be easily transferred. Thus, we’re stuck between two chairs.
- A request for fiscalization is sent.
As a result, we created an event bus on Kafka, where everything is connected. Hooray, now we have a single point of failure (sarcasm).
The pros and cons are quite obvious. We created the bus, which means that now all services depend on it. This simplifies design but introduces a single point of failure into the system. If Kafka goes down, the process will halt.
What is events-driven API
A good answer to this question can be found in Martin Fowler's report (GOTO 2017) .
In brief, here’s what we did:
- We wrapped all asynchronous exchanges through events storage. Instead of notifying each interested consumer over the network about status changes, we write an event of state change into a centralized storage, and interested consumers read everything that appears from there.
- An event in this case is a notification (notifications) that something has changed somewhere. For example, the status of an order has changed. A consumer who needs some accompanying data related to the status change that is not in the notification can retrieve their state by themselves.
- The maximum option is full-fledged event sourcing, state transfer, where the event contains all the information needed for processing: where it came from and which status it transitioned to, how the data changed, etc. The only question is the feasibility and the amount of information you can afford to store.
Within the launch of the Refund Tool, we used the third option. This simplified event processing, as there was no need to retrieve detailed information, plus it eliminated the scenario where each new event generates a spike of clarifying GET requests from consumers.
The Refund Tool service is not heavily loaded, so Kafka there is more of a trial than a necessity. I don't think that if the refund service became a high-load project, the business would be pleased.
Async exchange AS IS
For asynchronous exchanges, the PHP department usually uses RabbitMQ. They gathered data for the request, put it in the queue, and the consumer of the same service read it and sent it (or didn’t send it). For the API itself, Lamoda actively uses Swagger. We design the API, describe it in Swagger, generate client and server code. We also use a slightly extended JSON RPC 2.0.
ESB buses are used in some places, someone lives on ActiveMQ, but overall, RabbitMQ - standard.
Async exchange TO BE
When designing an exchange through events-bus, an analogy can be traced. We describe the future data exchange in a similar way through event structure descriptions. The YAML format, code generation had to be done manually, the generator creates DTOs according to the specification and teaches clients and servers to work with them. Generation is done in two languages - golang and php. This helps keep libraries consistent. The generator is written in golang, which earned it the name gogi.
Event sourcing on Kafka is a typical thing. There is a solution from the main enterprise version Kafka Confluent, there is also , a solution from our "brothers" in the domain area, Zalando. Our motivation to start with vanilla Kafka is to keep the solution free until we finally decide whether we will use it widely, as well as to leave ourselves space for maneuver and improvements: we want support for our JSON RPC 2.0, generators for two languages, and we'll see what else.
Ironically, even in such a fortunate case, when there is a roughly similar business like Zalando that has made a roughly similar solution, we cannot effectively use it.
Architecturally at launch, the pattern is as follows: we read directly from Kafka but write only through the events-bus. For reading from Kafka, there is a lot available: brokers, load balancers, and it is more or less ready for horizontal scaling, which we wanted to preserve. However, we wanted to wrap the writing through one Gateway aka Events-bus, and here's why.
Events-bus
Or event bus. This is simply a stateless HTTP gateway that takes on several important roles:
- Validation of producing - we check that the events comply with our specification.
- Master system for events, that is, this is the main and only system in the company that answers the question of what events with what structures are considered valid. Validation includes just data types and enums for strict specification of the content.
- Hash function for sharding - the structure of Kafka messages is key-value, and the hash from the key determines where to place it.
Why
We work in a large company with a well-established process. Why change anything? This is an experiment, and we expect to gain several benefits.
1:n+1 exchanges (one to many)
With Kafka, it is very easy to connect new consumers to the API.
Imagine you have a directory that needs to be kept up to date across several systems at once (including some new ones). Previously, we invented a bundle that implemented a set API, and the master system would report the addresses of consumers. Now the master system sends updates to a topic, and anyone interested can read them. A new system appeared—we subscribed it to the topic. Yes, it’s still a bundle, but simpler.
In the case of the refund tool, which is essentially a part of BOB, it's convenient for us to keep them synchronized through Kafka. Payment indicates that the money has been refunded: BOB and RT learn about it, change their statuses, and the Fiscalization Service is informed to issue a receipt.
We plan to create a unified Notifications Service that would inform the client about updates regarding their orders/returns. Currently, this responsibility is spread across systems. It will be enough for us to teach the Notifications Service to extract relevant information from Kafka and react to it (and disable these notifications in other systems). No new direct exchanges will be necessary.
Data-driven
Information between systems becomes transparent—no matter how complicated your 'bloody enterprise' is or how extensive your backlog might be. Lamoda has a Data Analytics department that gathers data from systems and presents it in a reusable format, both for business and for intelligent systems. Kafka allows them to quickly get a lot of data and keep this informational flow up to date.
Replication log
Messages do not disappear after being read, like in RabbitMQ. When an event contains sufficient information for processing, we have a history of the latest changes to the object, and, if desired, the possibility to apply these changes.
The retention period for the replication log depends on the intensity of writes to this topic. Kafka allows for flexible configuration of retention limits based on time and data volume. For intensive topics, it's crucial that all consumers manage to read the information before it disappears, even in the case of short-term failures. Typically, we manage to retain data for a few days, which is quite sufficient for support.
Next is a brief overview of the documentation for those unfamiliar with Kafka (the image is also from the documentation).
In AMQP, there are queues: we write messages to a queue for the consumer. Typically, a single queue is handled by one system with the same business logic. If you need to notify multiple systems, you can teach the application to write to multiple queues or configure an exchange with a fanout mechanism that clones them.
In Kafka, there is a similar abstraction topic, where you write messages, but they do not disappear after being read. By default, when connecting to Kafka, you receive all messages, and you also have the ability to save the position where you left off. That is, you read sequentially, can choose not to mark a message as read, but can save the ID where you will continue reading later. The ID where you left off is called the offset, and the mechanism is known as commit offset.
Accordingly, you can implement different logic. For example, we have BOB existing in 4 instances for different countries—Lamoda exists in Russia, Kazakhstan, Ukraine, and Belarus. Since they are deployed separately, they have slightly different configs and their own business logic. We specify in the message which country it relates to. Each BOB consumer in each country reads with different groupIds, and if a message does not pertain to it, they skip it, i.e., immediately commit offset +1. If the same topic is read by our Payment Service, it does this with a separate group, so their offsets do not overlap.
Requirements for events:
- Completeness of data. We would like the event to contain enough data to be processed.
- Integrity. We delegate the Events-bus to check that the event is consistent and that it can be processed.
- Order is important. In the case of a return, we have to work with history. With notifications, the order does not matter if they are homogeneous notifications; the email will be the same regardless of which order arrived first. In the case of a return, there is a clear process, and if the order is changed, exceptions will arise, refunds may not be created or processed, and we may end up in a different status.
- Consistency. We have a storage system, and now instead of using an API, we are creating events. We need a way to quickly and cost-effectively transmit information about new events and changes to existing ones to our services. This is achieved through a common specification in a separate git repository and code generators. Consequently, our clients and servers across different services are aligned.
Kafka at Lamoda
We have three Kafka installations:
- Logs;
- R&D;
- Events-bus.
Today we are only discussing the last item. Our events-bus doesn't have very large installations—3 brokers (servers) and only 27 topics. Generally, one topic corresponds to one process. However, this is a subtle point, and we will touch on it shortly.
Above is the rps graph. The refunds process is marked with a turquoise line (yes, the one along the X-axis), and the content update process is marked in pink.
The Lamoda catalog contains millions of products, with data being updated continuously. Some collections go out of style, while new ones are released, and new models constantly appear in the catalog. We try to predict what will be of interest to our customers tomorrow, so we continually buy new items, photograph them, and refresh the showcase.
The pink peaks represent product updates, meaning changes to products. You can see that the team was photographing and photographing, and then suddenly! — they uploaded a batch of events.
Lamoda Events use cases
The architecture we have built is used for operations such as:
- Tracking return statuses: call-to-action and tracking statuses from all involved systems. Payments, statuses, fiscalization, notifications. Here we tested an approach, created tools, gathered all the bugs, wrote documentation, and explained to colleagues how to use it.
- Updating product cards: configuration, metadata, characteristics. One system reads (which displays), while several write.
- Email, push, and SMS: an order has been collected, the order has arrived, the return has been accepted, etc., there are many of them.
- Stock, inventory updates — quantitative updates of items, just numbers: incoming stock, returns. It is necessary for all systems related to reserving products to operate with the most current data. Currently, the stock updating system is quite complex, and Kafka will allow us to simplify it.
- Data Analysis (R&D department), ML tools, analytics, statistics. We want the information to be transparent — Kafka is well-suited for this.
Now for the more interesting part about the bumps and discoveries that have occurred over the past six months.
Design Issues
Let's say we want to create something new — for instance, transfer the entire delivery process to Kafka. Currently, part of the process is implemented in Order Processing in BOB. The order delivery involves a status model for the transfer to the delivery service, moving to an interim warehouse, and so on. There is a whole monolith, even two, plus a bunch of APIs dedicated to delivery. They know much more about delivery.
It seems like these areas are similar, but the statuses differ for Order Processing in BOB and the delivery system. For example, some courier services do not send interim statuses, only final ones: 'delivered' or 'lost'. Others, on the contrary, report in great detail about the movement of goods. Everyone has their own validation rules: for some, a valid email means it will be processed; for others, it is not valid, but the order will still be processed because there is a phone contact, while some will say that such an order will not be processed at all.
Data Stream
In the case of Kafka, the question arises regarding the organization of the data stream. This task is linked to the choice of strategy on several points; let's go through them all.
In one topic or in different ones?
We have a specification for the event. In BOB, we write that a certain order needs to be delivered, specifying: order number, its contents, some SKUs, barcodes, etc. When the goods arrive at the warehouse, delivery will be able to receive statuses, timestamps, and everything else needed. But further, we want to receive updates on this data from BOB. Does this create a reverse process of obtaining data from delivery? Is it the same event? Or is it a separate exchange deserving a separate topic?
Most likely, they will be quite similar, and the temptation to create one topic is justified because having separate topics means separate consumers, separate configs, and separate generation of all this. But that's not a fact.
New field or new event?
However, if we use the same events, another problem arises. For example, not all delivery systems can generate a DTO that can create BOB. We send them an ID, but they do not store it because they do not need it. From the standpoint of starting the event-bus process, this field is mandatory.
If we establish a rule for the event-bus that this field is mandatory, we are forced to set additional validation rules either in BOB or in the handler of the starting event. Validation starts to spread across the service — this is not very convenient.
Another problem is the temptation for incremental development. We are told that we need to add something to the event and, perhaps, upon reflection, this should have been a separate event. But in our schema, a separate event is a separate topic. A separate topic encompasses the entire process I described above. The developer is tempted to simply add another field to the JSON schema and regenerate it.
In the case of refunds, we ended up with an event of events over six months. We had one meta-event called refund update, which included a type field describing what this update was about. From this, we had "wonderful" switches with validators that indicated how to validate this event with this type.
Event versioning
For message validation in Kafka, you can use , but it was necessary to plan for this from the start and use Confluent. In our case with versioning, we have to be cautious. It will not always be possible to reread messages from the replication log, because the model may have changed. Mainly, we try to build versions so that the model is backward compatible: for instance, making a field temporarily optional. If the differences are too significant, we start writing to a new topic, and we switch clients when they finish reading the old one.
Reading order guarantee for partitions
Topics within Kafka are divided into partitions. This is not very important while we are designing entities and exchanges, but it becomes crucial when deciding how to consume and scale them.
In a typical case, you write to a single topic in Kafka. By default, one partition is used, and all messages for that topic go into it. The consumer then reads these messages in order. Now, suppose you need to expand the system to allow two different consumers to read the messages. For example, if you send an SMS, you can instruct Kafka to create an additional partition, and Kafka will start dividing the messages into two parts — half to one and half to the other.
How does Kafka divide them? Each message has a body (where we store JSON) and a key. A hash function can be applied to this key to determine which partition the message will go to.
In our case with refunds, this is important; if we take two partitions, there is a chance that a parallel consumer will process the second event before the first, which would be problematic. The hash function guarantees that messages with the same key will end up in the same partition.
Events vs commands
This is another issue we faced. An event is a certain occurrence: we say that something happened (something_happened), for instance, an item was canceled or a refund occurred. If these events are being listened to, then upon "item canceled," an entity refund will be created, and "refund occurred" will be recorded somewhere in the setups.
However, usually when you design events, you don't want to write them in vain — you assume that someone will read them. There's a temptation to write not something_happened (item_canceled, refund_refunded), but rather something_should_be_done. For example, the item is ready for return.
On one hand, this hints at how the event will be used. On the other hand, it looks much less like a normal event title. Furthermore, it’s not far from the command do_something. But there is no guarantee that this event will be read by anyone; and if it is read, there is no assurance it was processed successfully; and if processed successfully, that something was done, and that something went well. At the moment an event becomes do_something, feedback becomes necessary, and that's a problem.
In asynchronous communication with RabbitMQ, once you read a message and made an HTTP call, you have a response — at least that the message was received. When you write to Kafka, you have a message stating that you wrote to Kafka, but you know nothing about how it was processed.
Therefore, in our case, it was necessary to introduce a response event and configure monitoring to ensure that if a certain number of events occurred, a corresponding number of response events should follow after a certain time. If this does not happen, it seems that something went wrong. For example, if we send the event 'item_ready_to_refund', we expect that a refund will be created, the money will be returned to the client, and we will receive the 'money_refunded' event. But this is not certain, hence the need for monitoring.
Details
There is a rather obvious problem: if you are reading from the topic sequentially, and you encounter a bad message, the consumer crashes, and you won't be able to proceed. You need to stop all consumers, commit the offset further to continue reading.
We were aware of this, we planned for it, and it still happened. This occurred because the event was valid from the perspective of the events bus, it was valid from the application validator's point of view, but it was not valid from the PostgreSQL perspective, since we have one system in MySQL with UNSIGNED INT, and the newly written system used PostgreSQL with just INT. Its size is slightly smaller, and the Id didn't fit. Symfony crashed with an exception. Of course, we caught the exception because we anticipated it, and we were planning to commit this offset, but before that, we wanted to increment the problem counter since the message was processed unsuccessfully. The counters for this project are also stored in the database, and Symfony had already closed communication with the database, and the second exception killed the entire process with no chance to commit the offset.
The service was down for a while—thankfully, with Kafka, this is not too alarming, because messages remain. When work resumes, they can be read again. This is convenient.
Kafka has the capability to set an arbitrary offset via tooling. However, to do this, you must stop all consumers—in our case, prepare a separate release that won't have any consumers for redeployments. Then, through tooling, you can shift the offset in Kafka, and the message will be processed.
Another detail— replication log vs rdkafka.so — is related to the specifics of our project. We use PHP, and in PHP, as a rule, all libraries communicate with Kafka through the rdkafka.so repository, and then there’s some kind of wrapper. Perhaps these are our personal difficulties, but it turned out that simply rereading a piece that has already been read is not so easy. In general, there were software issues.
Returning to the peculiarities of working with partitions, it is written directly in the documentation consumers >= topic partitions. But I learned about it much later than I would have liked. If you want to scale and have two consumers, you need at least two partitions. That is, if you had one partition that accumulated 20,000 messages, and you created a new one, the number of messages will not equal out anytime soon. Therefore, to have two parallel consumers, you need to understand partitions.
Monitoring
I think based on how we monitor it will be clearer what problems exist in the current approach.
For example, we count how many products in the database recently changed status, and, accordingly, events should have occurred based on these changes, and we send this number to our monitoring system. Then from Kafka, we receive a second number indicating how many events were actually recorded. Obviously, the difference between these two numbers should always be zero.
Furthermore, you need to monitor how things are going for the producer, whether the events-bus has received messages, and how things are going for the consumer. For example, in the graphs below, the Refund Tool is doing well, but BOB clearly has some issues (blue peaks).
I have already mentioned consumer-group lag. Roughly speaking, it is the number of unread messages. Overall, our consumers work quickly, so the lag is usually 0, but sometimes there can be a brief spike. Kafka can handle this out of the box, but you need to set a certain interval.
There is a project , which will give you more information about Kafka. It simply provides the status of a consumer group through the API, how that group is doing. In addition to OK and Failed, there is also a warning, and you will be able to find out that your consumers cannot keep up with the production pace—they can’t read what is being written in time. The system is quite smart and easy to use.
This is what the API response looks like. Here, the group is bob-live-fifa, partition refund.update.v1, status OK, lag 0—the last final offset is such.
Monitoring updated_at SLA (stuck) I have already mentioned. For example, the item has switched to the status that it is ready for return. We set up a Cron job that checks if, after 5 minutes, this object has not transitioned to refund (we process refunds through payment systems very quickly), then something has definitely gone wrong, and it's certainly a case for support. So we simply set up a Cron job that reads these cases, and if they are greater than 0, it sends an alert.
In summary, it's convenient to use events when:
- the information is needed by multiple systems;
- the result of the processing is not important;
- there are few events or the events are small.
It might seem that the article has a quite specific topic — asynchronous API on Kafka, but in connection with it, there is so much to recommend right away.
Firstly, the next event will take place in November, but the Saint Petersburg version will already be in April, and in June we will discuss high loads in Novosibirsk.
Secondly, the author of the report, Sergey Zaika, is part of the Program Committee of our new conference on knowledge management . The conference is one day long, it will be held on April 26, but the program is very rich.
And also in May there will be and (with DevOpsConf included) — you can still propose your topic there, share your experience, and complain about your own bumps and bruises.
Source: habr.com
