In 2017, we won a competition to develop the transactional core for Alfa-Bank's investment business and began work (at HighLoad++ 2018 with a presentation on the investment business core. Vladimir Dryinkin, head of the transactional core direction of Alfa-Bank's investment business). This system was intended to aggregate transaction data from various sources in different formats, unify the data, store it, and provide access.
During the development process, the system evolved and gained functionality, and at some point, we realized that we were crystallizing something much larger than just application software created to solve a strictly defined range of tasks: we ended up with a system for building distributed applications with persistent storage.. The experience we gained formed the basis for a new product — (TDG).
I want to talk about the architecture of TDG and the solutions we arrived at during the development process, introduce you to the main features, and show how our product can serve as the foundation for building complete solutions.
Architecturally, we divided the system into separate roles, each responsible for addressing a specific set of tasks. One running instance of the application implements one or several types of roles. There can be multiple roles of the same type in a cluster:
Connector
The connector is responsible for communication with the outside world; its task is to accept a request, parse it, and if successful, send the data for processing to the input processor. We support HTTP, SOAP, Kafka, and FIX formats. The architecture allows for easy addition of support for new formats, with IBM MQ support coming soon. If the request parsing fails, the connector will return an error; otherwise, it will respond that the request has been successfully processed, even if an error occurred during its further processing. This is done intentionally to work with systems that are unable to repeat requests—or, on the contrary, do so too insistently. To avoid data loss, a repair queue is used: the object first goes into it and is only removed after successful processing. The administrator can receive alerts about objects that remain in the repair queue and, after fixing a software or hardware failure, attempt to process them again.
Input processor
The input processor classifies the incoming data based on distinctive features and invokes the appropriate handlers. Handlers are Lua code executed in a sandbox, meaning they cannot affect the system's operation. At this stage, the data can be transformed as needed, as well as run an arbitrary number of tasks that can implement the necessary logic. For example, in the MDM (Master Data Management) product built on the Tarantool Data Grid, when adding a new user, we initiate a separate task for creating a golden record to avoid slowing down the request processing. The sandbox supports requests for reading, modifying, and adding data, allowing some function to be performed across all roles of type storage and aggregate the result (map/reduce).
Handlers can be described in files:
sum.lua
local x, y = unpack(...)
return x + yAnd then declared in the configuration:
functions:
sum: { __file: sum.lua }
Why Lua? Lua is a very simple language. Based on our experience, just a couple of hours after getting acquainted with it, people start writing code that solves their problems. And this includes not only professional developers but also analysts, for instance. Furthermore, due to the JIT compiler, Lua runs very quickly.
Storage
Storage retains persistent data. Before saving, data is validated against the data schema. We use an extended format to describe the schema. . Example:
{
"name": "User",
"type": "record",
"logicalType": "Aggregate",
"fields": [
{ "name": "id", "type": "string"},
{"name": "first_name", "type": "string"},
{"name": "last_name", "type": "string"}
],
"indexes": ["id"]
}Based on this description, DDL (Data Definition Language) for the Tarantool DBMS is automatically generated and a schema for data access.
Asynchronous data replication is supported (synchronous is planned to be added).
Output processor
Sometimes it is necessary to notify external consumers of the arrival of new data; this is the role of the Output processor. After saving data, it can be sent to the appropriate handler (e.g., to format it as required by the consumer) — and subsequently passed to the connector for sending. A repair queue is also used here: if the object is not accepted, the administrator can retry later.
Scaling
The roles of connector, input processor, and output processor are stateless, allowing us to scale the system horizontally by simply adding new application instances with the necessary role enabled. For horizontal scaling, the storage uses to cluster organization using virtual buckets. When a new server is added, some buckets from the old servers are moved to the new server in the background; this process is transparent to users and does not affect the entire system's performance.
Data properties
Objects can be very large and may contain other objects. We ensure the atomicity of adding and updating data by saving the object with all dependencies in one virtual bucket. This prevents the "spreading" of the object across multiple physical servers.
Versioning is supported: each update of an object creates a new version, and we can always make a temporal snapshot to see how the world looked at that time. For data that doesn't require a long history, we can limit the number of versions or even store just one—the last one—effectively disabling versioning for a specific type. It's also possible to limit history by time: for example, deleting all objects of a certain type older than 1 year. Archiving is supported too: we can export objects older than the specified time, freeing up space in the cluster.
Since I have already learned to "somewhat" port QEMU to JavaScript, this time it was decided to do it wisely and not repeat past mistakes.
Among the interesting features, it is worth noting the ability to run tasks on a schedule, on user request, or programmatically from the sandbox:
Here we see another role—runner. This role has no state, and if necessary, additional instances of the application with this role can be added to the cluster. The responsibility of the runner is to execute tasks. As mentioned, it is possible to generate new tasks from the sandbox; they are saved in the queue to storage and then executed on the runner. This type of task is called a Job. We also have a type of task called a Task—these are user-defined tasks that can be scheduled (using cron syntax) or triggered on demand. To start and track such tasks, we have a handy task scheduler. For this functionality to be available, the scheduler role needs to be enabled; this role has state, so it does not scale, which is not required; however, like all other roles, it can have a replica that starts working if the master fails.
Logger
Another role is called logger. It collects logs from all members of the cluster and provides an interface for exporting and viewing them through a web interface.
Services
It's worth mentioning that the system allows for easy creation of services. In the configuration file, you can specify which requests to route to a user-defined handler executed in the sandbox. In this handler, you can, for example, perform some analytical query and return the result.
A service is described in the configuration file:
services:
sum:
doc: "adds two numbers"
function: sum
return_type: int
args:
x: int
y: int
The GraphQL API is generated automatically and the service becomes available for calls:
query {
sum(x: 1, y: 2)
} This will result in the handler being called sum, which will return the result:
3
Query profiling and metrics
To understand how the system works and to profile requests, we have implemented support for the OpenTracing protocol. The system can send information on demand to tools that support this protocol, such as Zipkin, allowing insight into how a request was processed:
Naturally, the system provides internal metrics that can be collected using Prometheus and visualized with Grafana.
Deploy
Tarantool Data Grid can be deployed from RPM packages or archives, using the utility provided or Ansible, and there is also support for Kubernetes ().
The application implementing business logic (configuration, handlers) is uploaded to the deployed Tarantool Data Grid cluster as an archive via the UI or using a script through the API we provide.
Examples of applications
What applications can be created using Tarantool Data Grid? In fact, most business tasks are in one way or another related to processing streams of data, storage, and access to them. So, if you have large streams of data that need to be reliably stored and accessed, our product can save you a lot of development time and allow you to focus on your business logic.
For example, we want to gather information about the real estate market, in order to later, for example, have information on the most favorable offers. In this case, we will highlight the following tasks:
- Robots that collect information from open sources will be our data sources. You can address this task using ready-made solutions or by writing code in any language.
- Then Tarantool Data Grid will accept and save the data. If the data format from different sources differs, you can write Lua code that will convert it to a common format. At the preprocessing stage, you can also filter out duplicate offers or further update information in the database about agents operating in the market.
- You now have a scalable solution in the cluster that can be populated with data and used for data retrieval. Next, you can implement new functionality, for example, create a service that makes a request for data and provides the best offer for the day — this will require a few lines in the configuration file and some Lua code.
What's next?
Our priority is to enhance development convenience through . For instance, this IDE supports profiling and debugging of handlers running in a sandbox.
We also pay great attention to security issues. Right now, we are undergoing certification by the FSTEC of Russia to confirm a high level of security and comply with the requirements for the certification of software products used in personal data information systems and state information systems.
Source: habr.com
