In 2017, we won the competition for the development of the transactional core of Alfa-Bank's investment business and started working (at HighLoad++ 2018 with a report on the core of the investment business Vladimir Drynkin, head of the transactional core of the investment business of Alfa-Bank). This system was supposed to aggregate transaction data from different sources in various formats, bring the data to a unified form, save it and provide access to it.
During the development process, the system evolved and acquired functionality, and at some point we realized that we were crystallizing something much more than just application software created to solve a strictly defined range of tasks: we got system for building distributed applications with persistent storage. The experience gained by us formed the basis of a new product - (TDG).
I want to talk about the TDG architecture and the solutions that we came up with during the development process, introduce you to the main functionality and show how our product can become the basis for building complete solutions.
Architecturally, we divided the system into separate role of, each of which is responsible for solving a certain range of tasks. One running application instance implements one or more types of roles. A cluster can have multiple roles of the same type:
Connector
Connector is responsible for communication with the outside world; its task is to accept the request, parse it, and if it succeeds, then send the data for processing to the input processor. We support HTTP, SOAP, Kafka, FIX formats. The architecture makes it easy to add support for new formats, IBM MQ support coming soon. If parsing the request fails, connector will return an error; otherwise, it will reply that the request was processed successfully, even if an error occurred during its further processing. This is done on purpose in order to work with systems that do not know how to repeat requests - or vice versa, they do it too persistently. In order not to lose data, a repair queue is used: the object first enters it and only after successful processing is removed from it. The administrator can receive notifications about objects left in the repair queue, and after fixing a software error or hardware failure, retry.
input processor
The input processor classifies the received data by characteristic features and calls the appropriate handlers. Handlers are Lua code that runs in a sandbox, so they cannot affect the functioning of the system. At this stage, the data can be brought to the required form, and, if necessary, an arbitrary number of tasks can be launched that can implement the necessary logic. For example, in the MDM (Master Data Management) product built on Tarantool Data Grid, when adding a new user, in order not to slow down the processing of a request, we launch the creation of a golden record as a separate task. The sandbox supports requests for reading, changing and adding data, allows you to perform some function on all roles of the storage type and aggregation of 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, after a couple of hours after meeting him, people begin to write code that solves their problem. And these are not only professional developers, but, for example, analysts. In addition, thanks to the jit compiler, Lua is very fast.
Storage
Storage stores persistent data. Before saving, the data is validated against the data schema. To describe the circuit, we use the extended format . Example:
{
"name": "User",
"type": "record",
"logicalType": "Aggregate",
"fields": [
{ "name": "id", "type": "string"},
{"name": "first_name", "type": "string"},
{"name": "last_name", "type": "string"}
],
"indexes": ["id"]
}According to this description, DDL (Data Definition Language) is automatically generated for the Tarantula DBMS and schema for data access.
Asynchronous data replication is supported (in plans to add synchronous).
output processor
Sometimes you need to notify external consumers about the arrival of new data; for this, there is the Output processor role. After saving the data, they can be passed to the appropriate handler (for example, to bring them to the form that the consumer requires) - and then passed to the connector for sending. A repair queue is also used here: if no one has accepted the object, the administrator can try again later.
Scaling
The connector, input processor, and output processor roles are stateless, which allows us to scale the system horizontally by simply adding new application instances with the desired role type enabled. Storage is used for horizontal scaling to organizing a cluster using virtual buckets. After adding a new server, part of the buckets from the old servers in the background moves to the new server; this happens transparently to users and does not affect the operation of the entire system.
Data properties
Objects can be very large and contain other objects. We ensure the atomicity of adding and updating data by storing an object with all dependencies on one virtual bucket. This eliminates the "smearing" of the object across several physical servers.
Versioning is supported: each update of an object creates a new version, and we can always take a time slice and see how the world looked then. For data that doesn't need a long history, we can limit the number of versions or even store only one - the last one - that is, in fact, disable versioning for a particular type. You can also limit the history by time: for example, delete all objects of a certain type older than 1 year. Archiving is also supported: we can unload objects older than the specified time, freeing up space in the cluster.
Tasks
Of the interesting features, it is worth noting the ability to run tasks on a schedule, at the request of the user, or programmatically from the sandbox:
Here we see another role - runner. This role is stateless, and additional application instances with this role can be added to the cluster as needed. The runner's responsibility is to complete tasks. As mentioned, it is possible to generate new tasks from the sandbox; they are stored in the storage queue and then executed on the runner. This type of task is called Job. We also have a task type called Task - these are user-defined tasks that run on a schedule (cron syntax is used) or on demand. To launch and track such tasks, we have a handy task manager. In order for this functionality to be available, you must enable the scheduler role; this role has a state, so it does not scale, which, however, is not required; at the same time, she, like all other roles, can have a replica that starts working if the master suddenly fails.
Logger
Another role is called logger. It collects logs from all cluster members and provides an interface for uploading and viewing them through a web interface.
Services
It is worth mentioning that the system makes it easy to create services. In the configuration file, you can specify which requests to send to a user-written handler that runs in the sandbox. In this handler, you can, for example, execute some kind of analytical query and return the result.
The 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 automatically generated and the service becomes available to call:
query {
sum(x: 1, y: 2)
}
This will call the handler sumwhich will return the result:
3
Query profiling and metrics
To understand the operation of the system and profiling 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, which will allow you to figure out how the request was made:
Naturally, the system provides internal metrics that can be collected with Prometheus and visualized with Grafana.
Deploy
Tarantool Data Grid can be deployed from RPM packages or an archive, using a utility from the distribution or Ansible, there is also support for Kubernetes ().
The application that implements the business logic (configuration, handlers) is loaded into the deployed Tarantool Data Grid cluster as an archive via the UI or using a script, via the API provided by us.
Application examples
What applications can be created with Tarantool Data Grid? In fact, most business tasks are somehow related to the processing of data flow, storage and access to them. Therefore, if you have large data streams that need to be securely stored and accessed, then our product can save you a lot of development time and focus on its business logic.
For example, we want to collect information about the real estate market in order to later, for example, have information about the best offers. In this case, we will highlight the following tasks:
- Robots that collect information from open sources will be our data sources. You can solve this problem using ready-made solutions or by writing code in any language.
- Next, Tarantool Data Grid will accept and save the data. If the format of data from different sources is different, then you can write Lua code that will convert to a single format. During the pre-processing stage, you can also, for example, filter out recurring offers or additionally update the information about agents operating in the market in the database.
- Now you already have a scalable solution in the cluster that can be populated with data and make data selections. Then you can implement new functionality, for example, write a service that will query the data and give you 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 improve the convenience of development with the help of . For example, it is an IDE with support for profiling and debugging sandboxed handlers.
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 meet the requirements for certification of software products used in personal data information systems and government information systems.
Source: habr.com