Geo-Distributed Microservices and Their Database: Fighting the High Latency

Hey, comrades!

My development journey for the first version of the geo-distributed messenger has come to an end. So in this last article in the series, I want to talk about the multi-region database deployment options that were valid for Messenger.

The top-level architecture of the geo-messenger is shown below:

The application layer’s microservices instances are scattered around the world in the cloud regions of choice. The API layer, powered by Kong Gateway, lets microservices communicate with each other through simple REST endpoints.

The Global Load Balancer accepts user requests at the nearest POP (Point-of-Presence) and forwards the requests to the microservice instances that are geographically closest to the user.

Once the microservice instance receives a request from the load balancer, it is very likely that the service will need to read data from the database or write changes to it. And this step can become a painful bottleneck – a problem that you will need to solve if the data (database) is located far away from the microservice instance and the user.

In this article, I’ll take a look at some multi-region database deployment options and demonstrate how to keep read and write latency down for database queries regardless of the user’s location.

So, if you’re still with me on this journey, then, as pirates used to say, “Anchor and hoist the mizzen!” Which means, “Pull up the anchor and get this ship sailing!”

Multi-region database deployment option

There is no silver bullet for multi-region database deployment. It’s all about tradeoffs. Each option offers benefits, while others cost you something.

YugabyteDB, my distributed SQL database of choice, supports four multi-region database deployment options that are used by geo-distributed applications:

  1. Single Stretched Cluster: The database cluster is “extended” over several regions. Regions are usually located in relatively close proximity (eg, the US Midwest and East regions).
  2. Single Stretched Cluster with Reed Replicas: As in the previous option, clusters are deployed in cloud regions in a geographic location (eg, North America) in relatively close proximity (eg, the US Midwest and eastern regions). But with this option, you can add read replica nodes to distant geographic locations (e.g., Europe and Asia) to improve the performance of read workloads.
  3. Single Geo-divided Cluster: The database cluster is spread over many distant geographic locations (such as North America, Europe, and Asia). Each geographic location has its own set of nodes deployed in one or more local regions (eg, the US Midwest and East regions). The data is automatically pinned to a specific set of nodes based on the value of the geo-segmentation column. With this deployment option, you achieve low latency for both worldwide read and write workloads.
  4. Multiple Clusters with Async Replication: Multiple standalone clusters have been deployed in different regions. Regions may be located in relatively close proximity (eg, US Midwest and East regions) or distant locations (eg, US East and Asia South regions). Changes are replicated asynchronously between groups. You’ll get less latency for both worldwide read and write workloads, similar to the previous option, but you’ll be dealing with multiple standalone clusters that exchange changes asynchronously.

Okay, mate, let’s go ahead and review the first three deployment options for the use case of Geo-Messenger. I’ll leave out the fourth one because it doesn’t fit Messenger’s architecture which requires a database cluster.

Single Stretched Cluster in the United States

The first cluster spans three regions in the United States—US West, Central, and East.
Single multi-regional database cluster

The application/microservices instance and the API server (not shown in the picture) are running in the same locations as well as in the Europe West and Asia East regions.

Database read/write latency for US traffic

Suppose Ms. Blue is working this week from Iowa, USA. She opens Jio-messenger to send a note to a corporate channel. Its traffic will be processed by microservice instances deployed in the US Central region.

Before Ms. Blue can send messages, the US Central Microservices instance must load the channel’s message history. Which database node will be serving that read request?

In my case, the US Central region is configured as preferred for YugabyteDB, which means that the database node from that region will handle all read requests from the application layer. This 10-15 ms To load the channel’s message history from that US Central database node to the application instance from the same region. Here is the output of my Spring Boot log with the last line showing query execution time:

Hibernate: select message0_.country_code as country_1_1_, message0_.id as id2_1_, message0_.attachment as attachme3_1_, message0_.channel_id as channel_4_1_, message0_.message as message5_1_, message0_.sender_country_code as sender_c6_1_, message0_.sender_id as sender_i7_1_, message0_.sent_at as sent_at8_1_ from message message0_ where message0_.channel_id=? order by message0_.id asc

INFO 11744 --- [-nio-80-exec-10] i.StatisticalLoggingSessionEventListener : Session Metrics {
1337790 nanoseconds spent acquiring 1 JDBC connections;
0 nanoseconds spent releasing 0 JDBC connections;
1413081 nanoseconds spent preparing 1 JDBC statements;
14788369 nanoseconds spent executing 1 JDBC statements; (14 ms!)

Next, when Ms. Blue sends the message to the channel, the latency between the microservice and the database will be approx. 90 ms, It takes more time than previous operation because hibernate my . Generates multiple SQL queries for JpaRepository.save(message) method call (which can of course be optimized), and then YugabyteDB needs to store a copy of the message on all nodes running in US West, Central and East. Here’s what the output and latency look like:

Hibernate: select message0_.country_code as country_1_1_0_, message0_.id as id2_1_0_, message0_.attachment as attachme3_1_0_, message0_.channel_id as channel_4_1_0_, message0_.message as message5_1_0_, message0_.sender_country_code as sender_c6_1_0_, message0_.sender_id as sender_i7_1_0_, message0_.sent_at as sent_at8_1_0_ from message message0_ where message0_.country_code=? and message0_.id=?
Hibernate: select nextval ('message_id_seq')
Hibernate: insert into message (attachment, channel_id, message, sender_country_code, sender_id, sent_at, country_code, id) values (?, ?, ?, ?, ?, ?, ?, ?)

31908 nanoseconds spent acquiring 1 JDBC connections;
0 nanoseconds spent releasing 0 JDBC connections;
461058 nanoseconds spent preparing 3 JDBC statements;
91272173 nanoseconds spent executing 3 JDBC statements; (90 ms)

Database read/write latency for APAC traffic

Remember that there will be some tradeoffs with each multi-region database deployment. With current database deployments, read/write latency is low for US-based traffic, but high for traffic coming from other, more distant locations. Let’s look at an example.

Imagine that Mr. Red, a colleague of Ms. Blue’s, receives a push notification about Ms. Blue’s latest message. Since Mr. Red is on a business trip in Taiwan, his traffic will be handled by an instance of an app deployed on that island.

However, there are no database nodes deployed in or near Taiwan, so microservice instances have to query database nodes running in the US. that’s why it takes 165 ms To load the entire channel history on average before seeing Mr Red’s message from Miss Blue:

Hibernate: select message0_.country_code as country_1_1_, message0_.id as id2_1_, message0_.attachment as attachme3_1_, message0_.channel_id as channel_4_1_, message0_.message as message5_1_, message0_.sender_country_code as sender_c6_1_, message0_.sender_id as sender_i7_1_, message0_.sent_at as sent_at8_1_ from message message0_ where message0_.channel_id=? order by message0_.id asc

[p-nio-80-exec-8] i.StatisticalLoggingSessionEventListener : Session Metrics {
153152267 nanoseconds spent acquiring 1 JDBC connections;
0 nanoseconds spent releasing 0 JDBC connections;
153217915 nanoseconds spent preparing 1 JDBC statements;
165798894 nanoseconds spent executing 1 JDBC statements; (165 ms)

When Mr. Red replies to Ms. Blue in the same channel, it has about . takes time 450 ms For Hibernate to prepare, send and store messages in the database. Well, thanks to the laws of physics, the packet with the message has to travel through the Pacific Ocean from Taiwan, and then the message copy has to be stored in the US West, Central and East:

Hibernate: select message0_.country_code as country_1_1_0_, message0_.id as id2_1_0_, message0_.attachment as attachme3_1_0_, message0_.channel_id as channel_4_1_0_, message0_.message as message5_1_0_, message0_.sender_country_code as sender_c6_1_0_, message0_.sender_id as sender_i7_1_0_, message0_.sent_at as sent_at8_1_0_ from message message0_ where message0_.country_code=? and message0_.id=?
select nextval ('message_id_seq')
insert into message (attachment, channel_id, message, sender_country_code, sender_id, sent_at, country_code, id) values (?, ?, ?, ?, ?, ?, ?, ?)

 i.StatisticalLoggingSessionEventListener : Session Metrics 
 23488 nanoseconds spent acquiring 1 JDBC connections;
 0 nanoseconds spent releasing 0 JDBC connections;
 137247 nanoseconds spent preparing 3 JDBC statements;
 454281135 nanoseconds spent executing 3 JDBC statements (450 ms);

Now the high latency for APAC-based traffic may not be a big deal for applications that don’t target users in the area, but in my case it’s not. My geo-messenger has to work smoothly around the world. Let’s fight this high latency, friend! We’ll start with the reading!

Deploying Read Replicas in Distant Locations

The simplest way to improve latency for read workloads in YugabyteDB is to deploy read replica nodes in remote locations. This is a purely operational task that can be performed on a live cluster.

So, I “attached” a replication node to my live US-based database cluster, and that replica was placed near a microservice instance in the Asia East region that serves Mr.

Single multi-regional database cluster with read replication

Then I requested the application instance from Taiwan to use that replication node for database query. The latency time for preloading channel history for Mr. Red decreased from 165 ms to . It is done 10-15 ms, It’s just as fast for Ms. Blue, which is based in the United States.

Hibernate: select message0_.country_code as country_1_1_, message0_.id as id2_1_, message0_.attachment as attachme3_1_, message0_.channel_id as channel_4_1_, message0_.message as message5_1_, message0_.sender_country_code as sender_c6_1_, message0_.sender_id as sender_i7_1_, message0_.sent_at as sent_at8_1_ from message message0_ where message0_.channel_id=? order by message0_.id asc

 i.StatisticalLoggingSessionEventListener : Session Metrics 
 1210615 nanoseconds spent acquiring 1 JDBC connections;
 0 nanoseconds spent releasing 0 JDBC connections;
 1255989 nanoseconds spent preparing 1 JDBC statements;
 12772870 nanoseconds spent executing 1 JDBC statements; (12 mseconds)

As a result, with Read Replica, My Geo-Messenger can place read requests at low latency regardless of the user’s location!

But it is too early for the celebration. Writing is still very slow.

Imagine Mr. Red sends another message to the channel. The microservice instance from Taiwan will ask the replication node to execute the query. And replication will forward almost all requests generated by Hibernate to US-based nodes that store primary copies of records. Thus, the latency can still be as high as 640 ms For APAC traffic:

Hibernate: set transaction read write;
Hibernate: select message0_.country_code as country_1_1_0_, message0_.id as id2_1_0_, message0_.attachment as attachme3_1_0_, message0_.channel_id as channel_4_1_0_, message0_.message as message5_1_0_, message0_.sender_country_code as sender_c6_1_0_, message0_.sender_id as sender_i7_1_0_, message0_.sent_at as sent_at8_1_0_ from message message0_ where message0_.country_code=? and message0_.id=?
Hibernate: select nextval ('message_id_seq')
Hibernate: insert into message (attachment, channel_id, message, sender_country_code, sender_id, sent_at, country_code, id) values (?, ?, ?, ?, ?, ?, ?, ?)

 i.StatisticalLoggingSessionEventListener : Session Metrics 
 23215 nanoseconds spent acquiring 1 JDBC connections;
 0 nanoseconds spent releasing 0 JDBC connections;
 141888 nanoseconds spent preparing 4 JDBC statements;
 640199316 nanoseconds spent executing 4 JDBC statements; (640 ms)

Finally, friend, let’s settle this issue once and forever!

Switching to a Global Geo-Partitioned Cluster

A global geospatial cluster can provide fast read and write access to remote locations, but it requires you to introduce a special geospatial column to the database schema. Based on the value of the column, the database will automatically decide to which geography the record belongs to the database node.

database schema change

In short, my tables, such as the message one, define country_code column:

CREATE TABLE Message(
    id integer NOT NULL DEFAULT nextval('message_id_seq'),
    channel_id integer,
    sender_id integer NOT NULL,
    message text NOT NULL,
    attachment boolean NOT NULL DEFAULT FALSE,
    sent_at TIMESTAMP NOT NULL DEFAULT NOW(),
    country_code varchar(3) NOT NULL
) PARTITION BY LIST (country_code);

Depending on the value of that column, a record may be placed in one of the following database partitions:

CREATE TABLE Message_USA
    PARTITION OF Message
    (id, channel_id, sender_id, message, sent_at, country_code, sender_country_code,
    PRIMARY KEY(id, country_code))
    FOR VALUES IN ('USA') TABLESPACE us_central1_ts;

CREATE TABLE Message_EU
    PARTITION OF Message
    (id, channel_id, sender_id, message, sent_at, country_code, sender_country_code,
    PRIMARY KEY(id, country_code))
    FOR VALUES IN ('DEU') TABLESPACE europe_west3_ts;

CREATE TABLE Message_APAC
    PARTITION OF Message
    (id, channel_id, sender_id, message, sent_at, country_code, sender_country_code,
    PRIMARY KEY(id, country_code))
    FOR VALUES IN ('TWN') TABLESPACE asia_east1_ts;

Each partition is mapped to one of the tablespaces:

CREATE TABLESPACE us_central1_ts WITH (
  replica_placement="{"num_replicas": 1, "placement_blocks":
  [{"cloud":"gcp","region":"us-central1","zone":"us-central1-b","min_num_replicas":1}]}"
);

CREATE TABLESPACE europe_west3_ts WITH (
  replica_placement="{"num_replicas": 1, "placement_blocks":
  [{"cloud":"gcp","region":"europe-west3","zone":"europe-west3-b","min_num_replicas":1}]}"
);

CREATE TABLESPACE asia_east1_ts WITH (
  replica_placement="{"num_replicas": 1, "placement_blocks":
  [{"cloud":"gcp","region":"asia-east1","zone":"asia-east1-b","min_num_replicas":1}]}"
);

And each tablespace belongs to a group of database nodes from a particular geography. For example, all records containing the country code of Taiwan (country_code=TWN) will be stored on cluster nodes from the Asia East cloud region as those nodes hold partitions and tablespaces for APAC data. If you want to know the details of geo-division then refer the following article.

Low read/write latency across continents

Therefore, I deployed a three-node geo-segmented cluster in the US Central, Europe West and Asia East regions.

single geo-partitioned database cluster

Now, let’s ensure that read latency is maintained for Mr. Red’s requests. With read replication deployment, it still takes 5-15 ms To load a channel’s message history (for channels and messages belonging to the APAC region):

Hibernate: select message0_.country_code as country_1_1_, message0_.id as id2_1_, message0_.attachment as attachme3_1_, message0_.channel_id as channel_4_1_, message0_.message as message5_1_, message0_.sender_country_code as sender_c6_1_, message0_.sender_id as sender_i7_1_, message0_.sent_at as sent_at8_1_ from message message0_ where message0_.channel_id=? and message0_.country_code=? order by message0_.id asc

i.StatisticalLoggingSessionEventListener : Session Metrics 
1516450 nanoseconds spent acquiring 1 JDBC connections;
0 nanoseconds spent releasing 0 JDBC connections;
1640860 nanoseconds spent preparing 1 JDBC statements;
7495719 nanoseconds spent executing 1 JDBC statements; (7 ms)

and…drumroll please….when Mr.Red posts a message to an APAC channel, the write latency drops from 400-650 ms 6 ms on average!

Hibernate: select message0_.country_code as country_1_1_0_, message0_.id as id2_1_0_, message0_.attachment as attachme3_1_0_, message0_.channel_id as channel_4_1_0_, message0_.message as message5_1_0_, message0_.sender_country_code as sender_c6_1_0_, message0_.sender_id as sender_i7_1_0_, message0_.sent_at as sent_at8_1_0_ from message message0_ where message0_.country_code=? and message0_.id=?
Hibernate: select nextval ('message_id_seq')
Hibernate: insert into message (attachment, channel_id, message, sender_country_code, sender_id, sent_at, country_code, id) values (?, ?, ?, ?, ?, ?, ?, ?)

1123280 nanoseconds spent acquiring 1 JDBC connections;
0 nanoseconds spent releasing 0 JDBC connections;
123249 nanoseconds spent preparing 3 JDBC statements;
6597471 nanoseconds spent executing 3 JDBC statements; (6 ms)

Mission accomplished, friend! Now my Geo-Messenger’s database can serve read and write across countries and continents with low latency. I just need to tell my messenger where to deploy microservice instances and database nodes.

The case of cross-continent queries

Now for a quick comment as to why I omitted the database deployment option with multiple standalone YugabyteDB clusters.

It was important for me to have a single database for Messenger so that:

  1. Users can join any geography related discussion channels.
  2. Microservice instances from any location can access data at any location.

For example, if Mr. Red joins a discussion channel related to US nodes (country_code=’USA') and sends a message there, the microservice instance from Taiwan will send the request to the Taiwan-based database node and that node will forward the request to the US-based counterpart. The latency for this operation involves three SQL requests and will be approx. 165 ms,

Hibernate: select message0_.country_code as country_1_1_0_, message0_.id as id2_1_0_, message0_.attachment as attachme3_1_0_, message0_.channel_id as channel_4_1_0_, message0_.message as message5_1_0_, message0_.sender_country_code as sender_c6_1_0_, message0_.sender_id as sender_i7_1_0_, message0_.sent_at as sent_at8_1_0_ from message message0_ where message0_.country_code=? and message0_.id=?
Hibernate: select nextval ('message_id_seq')
Hibernate: insert into message (attachment, channel_id, message, sender_country_code, sender_id, sent_at, country_code, id) values (?, ?, ?, ?, ?, ?, ?, ?)

i.StatisticalLoggingSessionEventListener : Session Metrics 
1310550 nanoseconds spent acquiring 1 JDBC connections;
0 nanoseconds spent releasing 0 JDBC connections;
159080 nanoseconds spent preparing 3 JDBC statements;
164660288 nanoseconds spent executing 3 JDBC statements; (164 ms)

165 ms is undoubtedly more than 6 ms (the latency when Mr. Red posts a message to a local APAC-based channel), but what’s important here is the ability to make cross-continent requests via a single database connection when necessary. . Also, as the execution plan shows, there is a lot of room for optimization at the Hibernate level. Currently, Hibernate my . translates to JpaRepository.save(message) 3 Call in JDBC statement. This can be further optimized to bring the latency down to a value much lower than 165 ms for cross-continent requests.

wrapping up

ok man!

As you can see, distributed databases can work seamlessly across all geographies. You have to choose a deployment option that works best for your application. In my case, the geo-partitioned YugabyteDB Cluster is best suited for Messenger’s needs.

Well, this article concludes the series about the growth journey of my Geo-Messenger. The application source code is available on GitHub so you can explore the logic and run the app in your environment.

After that, I’ll take a break and then start preparing for my upcoming SpringOne session featuring the version of Geo-Messenger running on Spring Cloud components. So, if it’s relevant to you, Follow me To inform about future articles related to Spring Cloud and geo-distributed apps.

Leave a Comment