In this article, we will explain how and why we developed the – a mechanism that transfers information between client applications and 1C:Enterprise servers – from task assignment to architectural planning and implementation details.
The Interaction System (hereafter referred to as IS) is a distributed fault-tolerant messaging exchange system with guaranteed delivery. The IS is designed as a high-load service with significant scalability, available both as an online service (provided by 1C) and as a packaged product that can be deployed on your own server infrastructure.
The IS uses a distributed storage and a search system . We will also discuss Java and how we horizontally scale PostgreSQL.
Task Definition
To understand why we created the Interaction System, let me explain a bit about how business application development works at 1C.
First, a little about us for those who may not know what we do :) We create the technology platform "1C:Enterprise." The platform includes a development tool for business applications, as well as a runtime that allows business applications to operate in a cross-platform environment.
Client-server development paradigm
Business applications built on "1C:Enterprise" operate in a three-tiered architecture of "DBMS – application server – client." Application code written in , can run on the application server or on the client. All interactions with application objects (directories, documents, etc.), as well as reading and writing to the database, are performed solely on the server. The functionality of forms and command interfaces is also implemented on the server. On the client side, tasks include receiving, opening, and displaying forms, "communicating" with the user (alerts, questions…), executing small calculations on forms that require quick responses (for instance, multiplying price by quantity), handling local files, and interacting with hardware.
In the application code, in the headers of procedures and functions, it is necessary to explicitly specify where the code will execute — using the directives &НаКлиенте / &НаСервере (&AtClient / &AtServer in the English version of the language). 1C developers might correct me now, saying that directives actually , but this is not significant for us right now.
The client code can invoke server code, but server code cannot call client code. This is a fundamental limitation imposed for various reasons. In particular, server code must be written to execute uniformly, regardless of where it is invoked—from the client or the server. And if server code is called from another server code, the client is absent as such. Additionally, during the execution of server code, the invoking client may have closed or exited the application, leaving the server with no one to call.
The code handling button clicks: calling a server procedure from the client will work, but calling a client procedure from the server will not.
This means that if we want to send a message from the server to the client application, for example, indicating that the long-running report has finished generating and is available for viewing—there is no way for us to do this. We have to resort to making periodic requests to the server from the client code. However, this approach burdens the system with unnecessary calls and generally does not look very elegant.
There's also the need, for example, when a phone call comes in, - to notify the client application about it, so that it can find the caller in the database of counterparties using the caller’s number and present information about the calling counterparty to the user. Or, for instance, to inform the client application of the customer when an order arrives at the warehouse. In general, there are many cases where such a mechanism would be useful.
The actual task
Create a messaging exchange mechanism. Fast, reliable, with guaranteed delivery and flexible message searching capabilities. Based on this mechanism, implement a messenger (messages, video calls) that operates within 1C applications.
Design a horizontally scalable system. Increasing load should be handled by adding more nodes.
Implementation
We decided not to integrate the messaging service directly into the 1C:Enterprise platform, but to implement it as a separate product whose API can be called from the code of 1C application solutions. This was done for several reasons, the main one being the desire to enable messaging between different 1C applications (for example, between Inventory Management and Accounting). Different 1C applications may operate on different versions of the 1C:Enterprise platform, be hosted on different servers, etc. In such conditions, implementing the messaging service as a separate product located 'alongside' the 1C installations is the optimal solution.
Thus, we chose to create the messaging service as a separate product. For small companies, we recommend using the messaging server that we have set up in our cloud (wss://1cdialog.com) to avoid the overhead costs associated with local installation and server configuration. Larger clients may find it beneficial to install their own messaging server on their hardware. We used a similar approach in our cloud SaaS product. – it is released as a packaged product for installation at client sites and is also hosted in our cloud. .
Application
To distribute the load and ensure fault tolerance, we will deploy multiple Java applications and place a load balancer in front of them. If we need to pass a message from node to node, we will use publish/subscribe with Hazelcast.
Communication between the client and the server will be through websockets. This is well-suited for real-time systems.
Distributed cache
We considered Redis, Hazelcast, and Ehcache. It's 2015. Redis has just released a new cluster (too new, too risky), and there are limitations with Sentinel. Ehcache cannot cluster (this feature appeared later). We decided to try Hazelcast 3.4.
Hazelcast clusters out of the box. In a single-node mode, it is not very useful and can only serve as a cache – it cannot persist data to disk, and if the sole node is lost, data is lost. We deploy multiple Hazelcasts, backing up critical data between them. We do not back up the cache – it's not worth it.
For us, Hazelcast is:
- A storage for user sessions. Fetching each session from the database takes time, so we store all sessions in Hazelcast.
- Cache. If you're looking for a user profile, check the cache. If you've sent a new message, store it in the cache.
- Topics for application instance communication. A node generates an event and places it in the Hazelcast topic. Other application nodes subscribed to this topic receive and process the event.
- Cluster locks. For example, we create a discussion using a unique key (singleton discussion in the context of a 1C database):
conversationKeyChecker.check("GAS STATION");
doInClusterLock("GAS STATION", () -> {
conversationKeyChecker.check("GAS STATION");
createChannel("GAS STATION");
});We checked that the channel does not exist. We took a lock, checked again, and created it. If we do not check after taking the lock, there's a chance that another thread also checked at that moment and is now trying to create the same discussion, which already exists. Locking via synchronized or regular Java Lock is not allowed. Through the database is slow, and we don't want to burden the database; using Hazelcast is just what we need.
Choosing the DBMS
We have extensive and successful experience with PostgreSQL and collaborating with developers of this DBMS.
It's not easy with PostgreSQL clusters — there are , , , but generally speaking, it's not like noSQL that scales out of the box. We didn’t consider noSQL as the primary storage, sticking with Hazelcast, with which we hadn’t worked before.
If we need to scale a relational DB, it means . As you know, with sharding, we split the database into separate parts so that each can be moved to a separate server.
Our first sharding approach anticipated the possibility of distributing each of our application's tables across different servers in varying proportions. If there are many messages on server A, let's move part of that table to server B. This solution was a clear case of premature optimization, so we decided to limit ourselves to a multi-tenant approach.
You can read about multi-tenant on the site .
In the SV, there are the concepts of application and subscriber. An application is a specific installation of a business application, such as ERP or Accounting, with its own users and business data. A subscriber is an organization or individual on behalf of whom the application is registered on the SV server. A subscriber can have several registered applications, and these applications can exchange messages with each other. The subscriber has become a tenant in our system. Messages from multiple subscribers can reside in the same physical database; if we see that a subscriber is generating a lot of traffic, we move it to a separate physical database (or even to a separate DB server).
We have a main DB that stores a routing table with information about the location of all subscriber databases.
To ensure that the main DB doesn't become a bottleneck, we keep the routing table (and other frequently requested data) in cache.
If a subscriber's DB starts to slow down, we will partition it internally. In other projects, we use partitioning for large tables with .
Since losing user messages is detrimental, we maintain our DBs with replicas. The combination of synchronous and asynchronous replicas allows us to safeguard against the loss of the primary DB. Message loss will only occur in the event of a simultaneous failure of the primary DB and its synchronous replica.
If the synchronous replica fails, the asynchronous replica becomes synchronous.
If the primary DB fails, the synchronous replica becomes the primary DB, and the asynchronous replica becomes the synchronous replica.
Elasticsearch for search
Since, among other things, the SV is also a messenger, a fast, convenient, and flexible search is needed, accounting for morphology and fuzzy matching. We decided not to reinvent the wheel and use the open-source search engine Elasticsearch, based on the library . We also deploy Elasticsearch in a cluster (master – data – data) to avoid issues in case of node failures.
On GitHub, we found for Elasticsearch and we use it. In the Elasticsearch index, we store root words (defined by the plugin) and N-grams. As the user enters text for the search, we look for the typed text among the N-grams. When saving in the index, the word 'texts' will be broken down into the following N-grams:
[te, tek, teks, tekst, teksty, ek, eks, ekst, eksty, ks, kst, ksty, st, sty, ty],
The root of the word 'text' will also be saved. This approach allows for searching at the beginning, middle, and end of the word.
Overall Picture
Repeated image from the beginning of the article, but with explanations:
- The load balancer exposed to the internet; we use nginx, but it can be anything.
- Java application instances communicate with each other via Hazelcast.
- For working with web sockets, we use .
- The Java application is written in Java 8 and consists of bundles. . We plan to migrate to Java 10 and switch to modules.
Development and Testing
During the development and testing of the system, we encountered a number of interesting features of the products we used.
Load Testing and Memory Leaks
The release of each version of the system includes load testing. It is considered successful when:
- The test ran for several days without service failures.
- The response time for key operations did not exceed the comfortable threshold.
- The performance degradation compared to the previous version is no more than 10%.
We fill the test database with data – for this, we obtain information about the most active subscriber from the production server, multiply their figures by 5 (number of messages, discussions, users), and thus we test.
We conduct load testing of the interaction system in three configurations:
- Stress Test
- Only connections
- Subscriber Registration
During stress testing, we run several hundred threads that continuously load the system: they send messages, create discussions, and retrieve lists of messages. We simulate actions of regular users (fetching the list of my unread messages, sending messages) and software solutions (sending a package to another configuration, processing a notification).
For example, this is what a part of the stress test looks like:
- A user logs into the system
- Requests their unread discussions
- Has a 50% chance of reading messages
- Has a 50% chance of sending messages
- Next, the user:
- Creates a new discussion with a 20% probability
- Randomly selects any of their discussions
- Enters inside
- Requests messages, user profiles
- Creates five messages addressed to random users from this discussion
- Exits the discussion
- Repeats 20 times
- Logs out, returns back to the beginning of the scenario
- A chatbot logs into the system (emulating message exchanges from application code)
- Creates a new channel for data exchange (special discussion) with a 50% probability
- Writes a message in any of the existing channels with a 50% probability
The 'Only Connections' scenario did not arise by chance. There are situations where users connect the system but have not yet engaged. Every day at 09:00, each user starts their computer, connects to the server, and remains silent. These individuals are dangerous, and there are many of them – from the packets, they only send PING/PONG, but they maintain a connection to the server (they cannot disconnect – what if there’s a new message?). The test simulates a situation where a large number of such users are trying to authenticate in the system within half an hour. It resembles a stress test, but its focus is specifically on this initial login – to ensure that there are no failures (a person does not use the system, yet it already fails – it's hard to think of something worse).
The subscriber registration scenario begins with the first launch. We conducted a stress test and were confident that the system does not lag in correspondence. However, as users began to register, the registration started to drop out due to timeouts. During registration, we used , which is tied to the system's entropy. The server could not accumulate enough entropy and froze for tens of seconds when requesting new SecureRandom. There are many solutions to this situation, such as switching to a less secure /dev/urandom, installing a special board that generates entropy, or generating random numbers in advance and storing them in a pool. We temporarily addressed the issue with a pool, but since then, we run a separate test for the registration of new subscribers.
As a load generator, we use . It cannot work with websockets; a plugin is required. The first results in a search for 'jmeter websocket' are , which recommend .
. We decided to start with it.
Almost immediately after commencing serious testing, we discovered that memory leaks began occurring in JMeter.
The plugin is a whole different story, having 176 stars and 132 forks on GitHub. The author hasn't committed to it since 2015 (we acquired it in 2015, and at that time, there were no suspicions), with several GitHub issues regarding memory leaks and 7 open pull requests.
If you decide to conduct load testing using this plugin, please take note of the following discussions:
- In a multithreaded environment, a regular LinkedList was used, which ultimately resulted in at runtime. This can be resolved by either switching to ConcurrentLinkedDeque or using synchronized blocks. We chose the first option ().
- Memory leak, information about the connection is not deleted upon disconnection ().
- In streaming mode (when the websocket doesn't close at the end of the sample but continues to be used in the plan), Response patterns don't work ().
This is one of those on GitHub. What we did:
- We took (@elyrank) – which fixes issues 1 and 3.
- We resolved issue 2.
- We upgraded Jetty from 9.2.14 to 9.3.12.
- We wrapped SimpleDateFormat in ThreadLocal; SimpleDateFormat is not thread-safe, which caused NPE at runtime.
- We eliminated another memory leak (the connection wasn’t properly closed upon disconnection).
And yet it leaks!
Memory now runs out not in a day, but in two. With hardly any time left, we decided to run fewer threads, but on four agents. That should have been enough for at least a week.
Two days passed...
Now memory is running out for Hazelcast. The logs showed that after a couple of days of testing, Hazelcast starts complaining about memory shortages, and shortly after, the cluster collapses and nodes continue to die one by one. We connected JVisualVM to Hazelcast and saw a 'rising saw' – it regularly triggered GC but couldn't clear memory.
It turned out that in Hazelcast 3.4, when deleting a map / multiMap (map.destroy()), memory is not fully released:
The bug is now fixed in 3.5, but it was a problem back then. We were creating new multiMaps with dynamic names and deleting them according to our logic. The code looked something like this:
public void join(Authentication auth, String sub) {
MultiMap sessions = instance.getMultiMap(sub);
sessions.put(auth.getUserId(), auth);
}
public void leave(Authentication auth, String sub) {
MultiMap sessions = instance.getMultiMap(sub);
sessions.remove(auth.getUserId(), auth);
if (sessions.size() == 0) {
sessions.destroy();
}
}Call:
service.join(auth1, "NEW_MESSAGES_IN_DISCUSSION_UUID1");
service.join(auth2, "NEW_MESSAGES_IN_DISCUSSION_UUID1");The multiMap was created for each subscription and removed when it was no longer needed. We decided to create a Map, where the key would be the subscription name, and the values would be session identifiers (which could then be used to obtain user IDs if necessary).
public void join(Authentication auth, String sub) {
addValueToMap(sub, auth.getSessionId());
}
public void leave(Authentication auth, String sub) {
removeValueFromMap(sub, auth.getSessionId());
}The graphs have improved.
What else we learned about load testing
- JSR223 needs to be written in Groovy and enable compilation cache - this is significantly faster. .
- JMeter-Plugins graphs are easier to understand than the standard ones. .
About our experience with Hazelcast
Hazelcast was a new product for us; we started working with it from version 3.4.1 and currently, our production server runs version 3.9.2 (as of this writing, the latest version of Hazelcast is 3.10).
ID Generation
We started with integer identifiers. Let's imagine we need another Long for a new entity. A sequence in the database doesn't work since the tables are sharded - it would mean having a message ID=1 in DB1 and a message ID=1 in DB2. You can't store such an ID in Elasticsearch or Hazelcast, and the worst part is if you want to merge data from two databases into one (for instance, deciding that one database is sufficient for these subscribers). You can create several AtomicLongs in Hazelcast and maintain a counter there, thus the performance for retrieving a new ID becomes incrementAndGet plus the time for the request to Hazelcast. But Hazelcast has something more optimal - FlakeIdGenerator. Each client is given a range of IDs upon request; for example, the first one gets from 1 to 10,000, the second from 10,001 to 20,000, and so on. Now the client can issue new identifiers on its own until the assigned range runs out. It's fast, but when the application (and Hazelcast client) restarts, a new sequence begins - hence skips, etc. Additionally, developers don't quite understand why IDs are integers but are so disparate. We weighed it all and switched to UUIDs.
By the way, for those who want to be like Twitter, there is a library called Snowcast – it's an implementation of Snowflake on top of Hazelcast. You can take a look here:
But we haven’t gotten to that yet.
TransactionalMap.replace
Here's another surprise: TransactionalMap.replace does not work. Here's a test:
@Test
public void replaceInMap_putsAndGetsInsideTransaction() {
hazelcastInstance.executeTransaction(context -> {
HazelcastTransactionContextHolder.setContext(context);
try {
context.getMap("map").put("key", "oldValue");
context.getMap("map").replace("key", "oldValue", "newValue");
String value = (String) context.getMap("map").get("key");
assertEquals("newValue", value);
return null;
} finally {
HazelcastTransactionContextHolder.clearContext();
}
});
}
Expected : newValue
Actual : oldValueI had to write my own replace using getForUpdate:
protected boolean replaceInMap(String mapName, K key, V oldValue, V newValue) {
TransactionalTaskContext context = HazelcastTransactionContextHolder.getContext();
if (context != null) {
log.trace("[CACHE] Replacing value in a transactional map");
TransactionalMap map = context.getMap(mapName);
V value = map.getForUpdate(key);
if (oldValue.equals(value)) {
map.put(key, newValue);
return true;
}
return false;
}
log.trace("[CACHE] Replacing value in a not transactional map");
IMap map = hazelcastInstance.getMap(mapName);
return map.replace(key, oldValue, newValue);
}Test not only common data structures but also their transactional versions. Sometimes, IMap works, while TransactionalMap does not.
Upload a new JAR without downtime
Initially, we decided to store our class objects in Hazelcast. For example, we have a class Application, and we want to save and read it. Saving:
IMap map = hazelcastInstance.getMap("application");
map.set(id, application);Reading:
IMap map = hazelcastInstance.getMap("application");
return map.get(id);Everything works. Then we decided to build an index in Hazelcast to search through it:
map.addIndex("subscriberId", false);When we started writing a new entity, we began receiving ClassNotFoundException. Hazelcast was trying to augment the index, but it didn’t know about our class and requested the JAR containing that class. We did just that, everything worked, but a new problem arose: how to update the JAR without completely stopping the cluster? Hazelcast does not capture the new JAR during rolling updates. At that moment, we decided that we could live without searching by index. After all, if Hazelcast is used as a key-value store, won’t everything work? Not quite. There is again a different behavior between IMap and TransactionalMap. While IMap doesn’t care, TransactionalMap throws an error.
IMap. We are recording 5000 objects, reading them. Everything is as expected.
@Test
void get5000() {
IMap map = hazelcastInstance.getMap("application");
UUID subscriberId = UUID.randomUUID();
for (int i = 0; i < 5000; i++) {
UUID id = UUID.randomUUID();
String title = RandomStringUtils.random(5);
Application application = new Application(id, title, subscriberId);
map.set(id, application);
Application retrieved = map.get(id);
assertEquals(id, retrieved.getId());
}
}And it doesn't work in a transaction, we get ClassNotFoundException:
@Test
void get_transaction() {
IMap map = hazelcastInstance.getMap("application_t");
UUID subscriberId = UUID.randomUUID();
UUID id = UUID.randomUUID();
Application application = new Application(id, "qwer", subscriberId);
map.set(id, application);
Application retrievedOutside = map.get(id);
assertEquals(id, retrievedOutside.getId());
hazelcastInstance.executeTransaction(context -> {
HazelcastTransactionContextHolder.setContext(context);
try {
TransactionalMap transactionalMap = context.getMap("application_t");
Application retrievedInside = transactionalMap.get(id);
assertEquals(id, retrievedInside.getId());
return null;
} finally {
HazelcastTransactionContextHolder.clearContext();
}
});
}In version 3.8, a User Class Deployment mechanism was introduced. You can designate one main node and update the JAR file on it.
Now we have completely changed our approach: we serialize to JSON ourselves and save it in Hazelcast. Hazelcast doesn't need to know the structure of our classes, and we can update without downtime. Versioning of domain objects is managed by the application. Different versions of the application can run simultaneously, and there can be situations where a new application writes objects with new fields while an old one is not aware of these fields. At the same time, the new application reads objects written by the old application, which do not have the new fields. We handle such situations within the application, but for simplicity, we do not change or remove fields; we only extend classes by adding new fields.
How we ensure high performance
Four trips to Hazelcast – good, two to the database – bad
Accessing data from the cache is always better than from the database, but we also don't want to store unnecessary records. The decision on what to cache is postponed until the final stages of development. Once the new functionality is coded, we enable logging of all queries in PostgreSQL (set log_min_duration_statement to 0) and run load testing for about 20 minutes. The logs collected by tools like pgFouine and pgBadger can generate analytical reports. In these reports, we primarily look for slow and frequent queries. For slow queries, we create an execution plan (EXPLAIN) and evaluate whether it can be optimized. Frequent queries with the same input data cache well. We try to keep queries 'flat,' involving only one table per query.
Operation
The SV online service was launched in the spring of 2017, and the standalone SV product was released in November 2017 (at that time in beta status).
Throughout more than a year of operation, there have been no major issues with the SV online service. , we collect and deploy from .
The SV server distribution is supplied as native packages: RPM, DEB, MSI. Additionally, for Windows, we provide a single installer in the form of one EXE that installs the server, Hazelcast, and Elasticsearch on a single machine. Initially, we referred to this installation version as 'demonstration,' but it has now become clear that this is the most popular deployment option.
Source: habr.com
