PgCat Query Mirroring #341

drdrsh · 2023-03-04T21:16:28Z

This is an implementation of Query mirroring in PgCat (outlined here #302)

In configs, we match mirror hosts with the servers handling the traffic. A mirror host will receive the same protocol messages as the main server it was matched with.

This is done by creating an async task for each mirror server, it communicates with the main server through two channels, one for the protocol messages and one for the exit signal. The mirror server sends the protocol packets to the underlying PostgreSQL server. We receive from the underlying PostgreSQL server as soon as the data is available and we immediately discard it. We use bb8 to manage the life cycle of the connection, not for pooling since each mirror server handler is more or less single-threaded.

We don't have any connection pooling in the mirrors. Matching each mirror connection to an actual server connection guarantees that we will not have more connections to any of the mirrors than the parent pool would allow.

The config setup looks something like this

# Shard 0
[pools.sharded_db.shards.0]
# [ host, port, role ]
servers = [
    [ "127.0.0.1", 5432, "primary"],
    [ "127.0.0.1", 5432, "replica"],
    [ "127.0.0.1", 5432, "replica"],
]
# Database name (e.g. "postgres")
database = "shard0"

mirrors = [
    [ "localhost", 5432, 2], # Mirrors traffic to the second replica
    [ "localhost", 5432, 1], # Mirrors traffic to the first replica
    [ "localhost", 5432, 0]  # Mirrors traffic to the primary
]

Preview

Here I created 3 mirrors all pointing at the same server so sending one query from PgCat yields a total of 4 queries made a gainst the database from 4 different connections.
psql

Postgres

drdrsh · 2023-03-04T21:18:19Z

src/mirrors.rs

+    }
+
+    pub fn send(self: &mut Self, bytes: &BytesMut) {
+        let cpy = bytes.clone();


This is happening on the client<->server task so I would like to avoid cloning all the bytes as much as possible. Not sure how to do it given server.send requires BytesMut

I figured it out. I will change send to use ByteMut.freeze which yields a frozen chunk of memory that is cheap to clone of the type Bytes

Well, changing server.send to take Bytes instead of BytesMut will require doing a ton of BytesMut.clone() in the server critical path which is not something I want to do.

Instead, I have the communication channel use Bytes and I make just one clone per server to create a frozen Bytes object, when I do fan-out I send that same frozen object across the channel and then create a new BytesMut in the mirror task

drdrsh · 2023-03-05T15:41:13Z

src/mirrors.rs

+        let mut delay = Duration::from_secs(0);
+        let min_backoff = Duration::from_millis(100);
+        let max_backoff = Duration::from_secs(5);
+        let mut retries = 0;


Avoid too aggressive retries.

drdrsh · 2023-03-05T21:10:00Z

src/server.rs

 }

 impl Drop for Server {
    /// Try to do a clean shut down. Best effort because
    /// the socket is in non-blocking mode, so it may not be ready
    /// for a write.
    fn drop(&mut self) {
+        self.mirror_disconnect();


Dropping the bytes channel sender should be enough to kill the mirror task but just in case the bytes channel is blocked, we send a signal to the exit channel which has the one purpose of transmitting exit signals

src/pool.rs

src/mirrors.rs

Co-authored-by: Nicholas Dujay <[email protected]>

…a_mirror3

levkk · 2023-03-07T10:02:23Z

src/mirrors.rs

+        );
+        stats.server_login(server_id);
+
+        match Server::startup(


Not using bb8? I guess it makes sense because MirrorManager is owned by Server which itself is managed by bb8.

Right. But to be honest, I hate the connection management part of this PR to the point I am okay to create a bb8 pool with one connection to not have to manage the lifetime of the connection.

I decided to go with the bb8 route even if the pool is tiny to avoid having to handle connection lifetime

levkk · 2023-03-07T10:08:13Z

src/server.rs

+
+    pub fn mirror_send(&mut self, bytes: &BytesMut) {
+        match self.mirror_manager.as_mut() {
+            Some(manager) => manager.send(bytes),


Mirroring should be best effort imo, this will block if the mirror channel buffer is full because the mirror can't absorb any more traffic, I think?

Yes, mirroring is best-effort. manager.send uses try_send under the hood. The reason I have a send method on manager is to handle the fanout because we could have one Server matched with more than one MirroredClient
https://github.com/levkk/pgcat/pull/341/files#diff-453e032b0a6294b617b502297ffd4ffcce47d4f81e06f16b602c0d6f89afebb9R215-R225

levkk · 2023-03-10T02:05:22Z

.circleci/config.yml

@@ -18,28 +18,28 @@ jobs:
          RUSTFLAGS: "-Zprofile -Ccodegen-units=1 -Copt-level=0 -Clink-dead-code -Coverflow-checks=off -Zpanic_abort_tests -Cpanic=abort -Cinstrument-coverage"
          RUSTDOCFLAGS: "-Cpanic=abort"
      - image: postgres:14
-        command: ["postgres", "-p", "5432", "-c", "shared_preload_libraries=pg_stat_statements"]
+        command: ["postgres", "-p", "5432", "-c", "shared_preload_libraries=pg_stat_statements", "-c", "pg_stat_statements.track=all", "-c", "pg_stat_statements.max=100000"]


Do we need to update the dev/docker-compose.yml as well?

levkk · 2023-03-10T02:06:15Z

src/mirrors.rs

+                    recv_result = server.recv() => {
+                        match recv_result {
+                            Ok(message) => trace!("Received from mirror: {} {:?}", String::from_utf8_lossy(&message[..]), address.clone()),
+                            Err(err) => error!("Failed to receive from mirror {:?} {:?}", err, address.clone())


Should you mark_bad here and make bb8 create you a new server?

I left that up to server.recv. It calls mark_bad in a handful of places. Similarly for server.send. Double marking bad should be fine.

Do you think we should mark_bad to be safe or just leave it up to the server logic to handle it?

Done. An extra mark_bad does not hurt. It documents the behavior here

levkk · 2023-03-10T02:21:13Z

src/mirrors.rs

+
+        Pool::builder()
+            .max_size(1)
+            .connection_timeout(std::time::Duration::from_millis(10_000))


You might want to use the config values here, so we don't get long timeouts unexpectedly.

levkk · 2023-03-10T02:21:43Z

src/mirrors.rs

+                let mut server = match pool.get().await {
+                    Ok(server) => server,
+                    Err(err) => {
+                        error!(


mark_bad ?

mark_bad works if we have a server connection. In this case we failed to checkout a connection from the pool so we have no server to mark_bad.

In the non-mirrored version, we ban the server but in the mirrored mode, it doesn't make sense to ban (we don't have banlists and banning in a mirrored setup is not very useful)

levkk · 2023-03-10T02:22:58Z

src/pool.rs

                    for (address_index, server) in shard.servers.iter().enumerate() {
+                        let mut mirror_addresses: Vec<Address> = vec![];


You don't need a type annotation typically, Rust should infer it.

levkk · 2023-03-10T02:24:08Z

src/pool.rs

+                                    host: mirror_settings.host.clone(),
+                                    port: mirror_settings.port,
+                                    role: server.role,
+                                    address_index: 0,


We send stats from the mirrors , so we should make sure these unique identifiers are unique.

I think the stats use the address_id, not the address_index. The address_id is unique.

Here is the address_index callsites
https://cs.github.com/levkk/pgcat?q=.address_index

They are both irrelevant to mirrors

I followed the same pattern we do for the server addresses. For the server addresses, we set the address_index to equal the index of the server in the configs array. We do the same for mirrors, we set the mirror address_index to be the index of the mirror in the mirror config array

https://github.com/levkk/pgcat/blob/main/src/pool.rs#L247-L254

drdrsh · 2023-03-10T03:49:32Z

.circleci/run_tests.sh

+#
+# Ruby integration tests
+# These tests create their own PgCat servers so we want to run them after starting toxiproxy
+# and before starting PgCat
+#
+cd tests/ruby
+sudo bundle install
+bundle exec rspec *_spec.rb --format documentation || exit 1
+cd ../..
+


I am wondering if the fact that we have an extra PgCat running in the background is making the specs flake out. So I moved the ruby tests up before we start any PgCats

drdrsh · 2023-03-10T04:07:20Z

I disabled another uber flaky test

levkk

Awesome!

mismaah · 2024-05-21T05:24:59Z

Can this be a replacement for replication (like repmgr)?

First cut Query Mirroring

393b48c

drdrsh commented Mar 4, 2023

View reviewed changes

drdrsh added 9 commits March 4, 2023 18:17

one clone

396ffc7

fmt

a6b9df6

Add connection retries

993d8ed

Better handling of disconnection and recv

4c9025c

Simpler event model

226c051

revert

23e75ed

revert

cd8eb9b

whitespace

8d4af57

comments

7392b3d

drdrsh changed the title ~~First cut Query Mirroring~~ PgCat Query Mirroring Mar 5, 2023

drdrsh requested a review from levkk March 5, 2023 15:33

drdrsh commented Mar 5, 2023

View reviewed changes

drdrsh added 5 commits March 5, 2023 09:42

refactor

53b8422

Add tests

a78c0c6

test channel overrun

d11fd8f

logs

786ba14

logs

31254eb

drdrsh marked this pull request as ready for review March 5, 2023 17:11

drdrsh added 2 commits March 5, 2023 11:23

more messages to cover recv call

7d123aa

add a test to detect failure to close mirror connections

4947e69

drdrsh commented Mar 5, 2023

View reviewed changes

dat2 reviewed Mar 6, 2023

View reviewed changes

src/pool.rs Outdated Show resolved Hide resolved

dat2 reviewed Mar 6, 2023

View reviewed changes

src/mirrors.rs Outdated Show resolved Hide resolved

drdrsh and others added 5 commits March 6, 2023 09:52

Update src/mirrors.rs

cbe934b

Co-authored-by: Nicholas Dujay <[email protected]>

Update src/pool.rs

0b3dd37

Co-authored-by: Nicholas Dujay <[email protected]>

Merge branch 'main' of github.com:drdrsh/pgcat into mostafa_mirror3

0604f9a

Merge branch 'mostafa_mirror3' of github.com:drdrsh/pgcat into mostaf…

d22343a

…a_mirror3

comments

672edc3

levkk reviewed Mar 7, 2023

View reviewed changes

drdrsh added 14 commits March 7, 2023 11:35

test for retries and recovery

035fb8d

make mirror specs less flaky

d291a1a

Simplify

8b25b67

remove redundent continue

92ff1bd

rename method

b24c187

maybe configs would fix flakiness?

fa509c2

one more to go

781843e

drop connections after each test run

d254c0f

some give for mirror tests

cfc347b

make address_index/address_id safer

9cde6e0

clean up

2ddd1c7

one address_id

795f13d

remove flaky expectation

476bd43

build

87bf33c

levkk reviewed Mar 10, 2023

View reviewed changes

drdrsh added 4 commits March 9, 2023 21:04

address comments

5e2e205

revert

bd437f3

mirror_idx

2dfb6ff

move tests around

69e0a0a

drdrsh commented Mar 10, 2023

View reviewed changes

restore

e7d4114

levkk approved these changes Mar 10, 2023

View reviewed changes

drdrsh merged commit aa89e35 into postgresml:main Mar 10, 2023

drdrsh mentioned this pull request Mar 10, 2023

Feature request: traffic mirroring for load testing #302

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

PgCat Query Mirroring #341

PgCat Query Mirroring #341

drdrsh commented Mar 4, 2023 •

edited

Loading

drdrsh Mar 4, 2023 •

edited

Loading

drdrsh Mar 4, 2023

drdrsh Mar 5, 2023

drdrsh Mar 5, 2023

drdrsh Mar 5, 2023 •

edited

Loading

levkk Mar 7, 2023 •

edited

Loading

drdrsh Mar 7, 2023

drdrsh Mar 8, 2023

levkk Mar 7, 2023

drdrsh Mar 7, 2023 •

edited

Loading

levkk Mar 10, 2023

drdrsh Mar 10, 2023

levkk Mar 10, 2023

drdrsh Mar 10, 2023

drdrsh Mar 10, 2023

drdrsh Mar 10, 2023

levkk Mar 10, 2023

drdrsh Mar 10, 2023

levkk Mar 10, 2023

drdrsh Mar 10, 2023 •

edited

Loading

levkk Mar 10, 2023

drdrsh Mar 10, 2023

levkk Mar 10, 2023

drdrsh Mar 10, 2023

drdrsh Mar 10, 2023

drdrsh Mar 10, 2023 •

edited

Loading

drdrsh Mar 10, 2023

drdrsh Mar 10, 2023

drdrsh commented Mar 10, 2023

levkk left a comment

mismaah commented May 21, 2024

		for (address_index, server) in shard.servers.iter().enumerate() {
		let mut mirror_addresses: Vec<Address> = vec![];

PgCat Query Mirroring #341

PgCat Query Mirroring #341

Conversation

drdrsh commented Mar 4, 2023 • edited Loading

Preview

drdrsh Mar 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drdrsh Mar 5, 2023 • edited Loading

Choose a reason for hiding this comment

levkk Mar 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drdrsh Mar 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drdrsh Mar 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drdrsh Mar 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drdrsh commented Mar 10, 2023

levkk left a comment

Choose a reason for hiding this comment

mismaah commented May 21, 2024

drdrsh commented Mar 4, 2023 •

edited

Loading

drdrsh Mar 4, 2023 •

edited

Loading

drdrsh Mar 5, 2023 •

edited

Loading

levkk Mar 7, 2023 •

edited

Loading

drdrsh Mar 7, 2023 •

edited

Loading

drdrsh Mar 10, 2023 •

edited

Loading

drdrsh Mar 10, 2023 •

edited

Loading