Fixed linter errors

Revert "Do not unban replicas if a primary is available (#843)"

This reverts commit cdcfa99fb9.

.github/workflows/chart-lint-test.yaml

@@ -22,7 +22,7 @@ jobs:
       # Python is required because `ct lint` runs Yamale (https://github.com/23andMe/Yamale) and
       # yamllint (https://github.com/adrienverge/yamllint) which require Python
       - name: Set up Python
-        uses: actions/setup-python@v4.1.0
+        uses: actions/setup-python@v5.1.0
         with:
           python-version: 3.7
 
@@ -43,7 +43,7 @@ jobs:
         run: ct lint --config ct.yaml
 
       - name: Create kind cluster
-        uses: helm/kind-action@v1.7.0
+        uses: helm/kind-action@v1.10.0
         if: steps.list-changed.outputs.changed == 'true'
 
       - name: Run chart-testing (install)

.github/workflows/chart-release.yaml

@@ -32,7 +32,7 @@ jobs:
           version: v3.13.0
 
       - name: Run chart-releaser
-        uses: helm/chart-releaser-action@be16258da8010256c6e82849661221415f031968 # v1.5.0
+        uses: helm/chart-releaser-action@a917fd15b20e8b64b94d9158ad54cd6345335584 # v1.6.0
         with:
           charts_dir: charts
           config: cr.yaml

.github/workflows/publish-deb-package.yml

@@ -1,6 +1,9 @@
 name: pgcat package (deb)
 
 on:
+  push:
+    tags:
+      - v*
   workflow_dispatch:
     inputs:
       packageVersion:
@@ -16,6 +19,14 @@ jobs:
     runs-on: ${{ matrix.os }}
     steps:
     - uses: actions/checkout@v3
+    - name: Set package version
+      if: github.event_name == 'push'  # For push event
+      run: |
+        TAG=${{ github.ref_name }}
+        echo "packageVersion=${TAG#v}" >> "$GITHUB_ENV"
+    - name: Set package version (manual dispatch)
+      if: github.event_name == 'workflow_dispatch'  # For manual dispatch
+      run: echo "packageVersion=${{ github.event.inputs.packageVersion }}" >> "$GITHUB_ENV"
     - uses: actions-rs/toolchain@v1
       with:
         toolchain: stable
@@ -39,10 +50,10 @@ jobs:
           export ARCH=arm64
         fi
 
-        bash utilities/deb.sh ${{ inputs.packageVersion }}
+        bash utilities/deb.sh ${{ env.packageVersion }}
 
         deb-s3 upload \
           --lock \
           --bucket apt.postgresml.org \
-          pgcat-${{ inputs.packageVersion }}-ubuntu22.04-${ARCH}.deb \
+          pgcat-${{ env.packageVersion }}-ubuntu22.04-${ARCH}.deb \
           --codename $(lsb_release -cs)

.gitignore

@@ -12,3 +12,4 @@ dev/cache
 !dev/cache/.keepme
 .venv
 **/__pycache__
+.bundle

CONFIG.md

@@ -36,10 +36,11 @@ Port at which prometheus exporter listens on.
 ### connect_timeout
 ```
 path: general.connect_timeout
-default: 5000 # milliseconds
+default: 1000 # milliseconds
 ```
 
-How long to wait before aborting a server connection (ms).
+How long the client waits to obtain a server connection before aborting (ms).
+This is similar to PgBouncer's `query_wait_timeout`.
 
 ### idle_timeout
 ```
@@ -462,10 +463,18 @@ path: pools.<pool_name>.users.<user_index>.pool_size
 default: 9
 ```
 
-Maximum number of server connections that can be established for this user
+Maximum number of server connections that can be established for this user.
 The maximum number of connection from a single Pgcat process to any database in the cluster
 is the sum of pool_size across all users.
 
+### min_pool_size
+```
+path: pools.<pool_name>.users.<user_index>.min_pool_size
+default: 0
+```
+
+Minimum number of idle server connections to retain for this pool.
+
 ### statement_timeout
 ```
 path: pools.<pool_name>.users.<user_index>.statement_timeout
@@ -475,6 +484,16 @@ default: 0
 Maximum query duration. Dangerous, but protects against DBs that died in a non-obvious way.
 0 means it is disabled.
 
+### connect_timeout
+```
+path: pools.<pool_name>.users.<user_index>.connect_timeout
+default: <UNSET> # milliseconds
+```
+
+How long the client waits to obtain a server connection before aborting (ms).
+This is similar to PgBouncer's `query_wait_timeout`.
+If unset, uses the `connect_timeout` defined globally.
+
 ## `pools.<pool_name>.shards.<shard_index>` Section
 
 ### servers
@@ -502,4 +521,3 @@ default: "shard0"
 ```
 
 Database name (e.g. "postgres")
-

Cargo.lock

@@ -192,12 +192,11 @@ checksum = "604178f6c5c21f02dc555784810edfb88d34ac2c73b2eae109655649ee73ce3d"
 
 [[package]]
 name = "bb8"
-version = "0.8.1"
+version = "0.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "98b4b0f25f18bcdc3ac72bdb486ed0acf7e185221fd4dc985bc15db5800b0ba2"
+checksum = "d89aabfae550a5c44b43ab941844ffcd2e993cb6900b342debf59e9ea74acdb8"
 dependencies = [
  "async-trait",
- "futures-channel",
  "futures-util",
  "parking_lot",
  "tokio",

Cargo.toml

@@ -8,7 +8,7 @@ edition = "2021"
 tokio = { version = "1", features = ["full"] }
 bytes = "1"
 md-5 = "0.10"
-bb8 = "0.8.1"
+bb8 = "=0.8.6"
 async-trait = "0.1"
 rand = "0.8"
 chrono = "0.4"

charts/pgcat/Chart.yaml

@@ -5,4 +5,4 @@ maintainers:
   - name: Wildcard
     email: support@w6d.io
 appVersion: "1.2.0"
-version: 0.2.0
+version: 0.2.4

charts/pgcat/templates/secret.yaml

@@ -15,6 +15,7 @@ stringData:
     connect_timeout = {{ .Values.configuration.general.connect_timeout }}
     idle_timeout = {{ .Values.configuration.general.idle_timeout | int }}
     server_lifetime = {{ .Values.configuration.general.server_lifetime | int }}
+    server_tls = {{ .Values.configuration.general.server_tls }}
     idle_client_in_transaction_timeout = {{ .Values.configuration.general.idle_client_in_transaction_timeout | int }}
     healthcheck_timeout = {{ .Values.configuration.general.healthcheck_timeout }}
     healthcheck_delay = {{ .Values.configuration.general.healthcheck_delay }}
@@ -58,11 +59,21 @@ stringData:
     ##
     [pools.{{ $pool.name | quote }}.users.{{ $index }}]
     username = {{ $user.username | quote }}
+    {{- if $user.password }}
     password = {{ $user.password | quote }}
+    {{- else if and $user.passwordSecret.name $user.passwordSecret.key }}
+    {{- $secret := (lookup "v1" "Secret" $.Release.Namespace $user.passwordSecret.name) }}
+    {{- if $secret }}
+    {{- $password := index $secret.data $user.passwordSecret.key | b64dec }}
+    password = {{ $password | quote }}
+    {{- end }}
+    {{- end }}
     pool_size = {{ $user.pool_size }}
-    statement_timeout = {{ $user.statement_timeout }}
-    min_pool_size = 3
-    server_lifetime = 60000
+    statement_timeout = {{ default 0 $user.statement_timeout }}
+    min_pool_size = {{ default 3 $user.min_pool_size }}
+    {{- if $user.server_lifetime }}
+    server_lifetime = {{ $user.server_lifetime }}
+    {{- end }}
     {{-     if and $user.server_username $user.server_password }}
     server_username = {{ $user.server_username | quote }}
     server_password = {{ $user.server_password | quote }}

charts/pgcat/values.yaml

@@ -175,6 +175,9 @@ configuration:
     # Max connection lifetime before it's closed, even if actively used.
     server_lifetime: 86400000  # 24 hours
 
+    # Whether to use TLS for server connections or not.
+    server_tls: false
+
     # How long a client is allowed to be idle while in a transaction (ms).
     idle_client_in_transaction_timeout: 0  # milliseconds
 
@@ -315,7 +318,9 @@ configuration:
     #  ## Credentials for users that may connect to this cluster
     #  ## @param users [array]
     #  ## @param users[0].username Name of the env var (required)
-    #  ## @param users[0].password Value for the env var (required)
+    #  ## @param users[0].password Value for the env var (required) leave empty to use existing secret see passwordSecret.name and passwordSecret.key
+    #  ## @param users[0].passwordSecret.name Name of the secret containing the password
+    #  ## @param users[0].passwordSecret.key Key in the secret containing the password
     #  ## @param users[0].pool_size Maximum number of server connections that can be established for this user
     #  ## @param users[0].statement_timeout Maximum query duration. Dangerous, but protects against DBs that died in a non-obvious way.
     #  users: []

cr.yaml

@@ -1 +1,2 @@
 sign: false
+pages_branch: main

pgcat.toml

@@ -179,7 +179,8 @@ primary_reads_enabled = true
 # `random`: picks a shard at random
 # `random_healthy`: picks a shard at random favoring shards with the least number of recent errors
 # `shard_<number>`: e.g. shard_0, shard_4, etc. picks a specific shard, everytime
-# no_shard_specified_behavior = "shard_0"
+# `fail`: fails to pick up shard. (require explicit shard setup)
+# default_shard = "shard_0"
 
 # So what if you wanted to implement a different hashing function,
 # or you've already built one and you want this pooler to use it?

postinst

@@ -7,3 +7,7 @@ systemctl enable pgcat
 if ! id pgcat 2> /dev/null; then
 	useradd -s /usr/bin/false pgcat
 fi
+
+if [ -f /etc/pgcat.toml ]; then
+	systemctl start pgcat
+fi

src/auth_passthrough.rs

@@ -1,3 +1,4 @@
+use crate::config::AuthType;
 use crate::errors::Error;
 use crate::pool::ConnectionPool;
 use crate::server::Server;
@@ -71,6 +72,7 @@ impl AuthPassthrough {
     pub async fn fetch_hash(&self, address: &crate::config::Address) -> Result<String, Error> {
         let auth_user = crate::config::User {
             username: self.user.clone(),
+            auth_type: AuthType::MD5,
             password: Some(self.password.clone()),
             server_username: None,
             server_password: None,

src/client.rs

@@ -14,7 +14,9 @@ use tokio::sync::mpsc::Sender;
 
 use crate::admin::{generate_server_parameters_for_admin, handle_admin};
 use crate::auth_passthrough::refetch_auth_hash;
-use crate::config::{get_config, get_idle_client_in_transaction_timeout, Address, PoolMode};
+use crate::config::{
+    get_config, get_idle_client_in_transaction_timeout, Address, AuthType, PoolMode,
+};
 use crate::constants::*;
 use crate::messages::*;
 use crate::plugins::PluginOutput;
@@ -463,8 +465,8 @@ where
             .count()
             == 1;
 
-        // Kick any client that's not admin while we're in admin-only mode.
         if !admin && admin_only {
+            // Kick any client that's not admin while we're in admin-only mode.
             debug!(
                 "Rejecting non-admin connection to {} when in admin only mode",
                 pool_name
@@ -481,72 +483,76 @@ where
         let process_id: i32 = rand::random();
         let secret_key: i32 = rand::random();
 
-        // Perform MD5 authentication.
-        // TODO: Add SASL support.
-        let salt = md5_challenge(&mut write).await?;
-
-        let code = match read.read_u8().await {
-            Ok(p) => p,
-            Err(_) => {
-                return Err(Error::ClientSocketError(
-                    "password code".into(),
-                    client_identifier,
-                ))
-            }
-        };
-
-        // PasswordMessage
-        if code as char != 'p' {
-            return Err(Error::ProtocolSyncError(format!(
-                "Expected p, got {}",
-                code as char
-            )));
-        }
-
-        let len = match read.read_i32().await {
-            Ok(len) => len,
-            Err(_) => {
-                return Err(Error::ClientSocketError(
-                    "password message length".into(),
-                    client_identifier,
-                ))
-            }
-        };
-
-        let mut password_response = vec![0u8; (len - 4) as usize];
-
-        match read.read_exact(&mut password_response).await {
-            Ok(_) => (),
-            Err(_) => {
-                return Err(Error::ClientSocketError(
-                    "password message".into(),
-                    client_identifier,
-                ))
-            }
-        };
-
         let mut prepared_statements_enabled = false;
 
         // Authenticate admin user.
         let (transaction_mode, mut server_parameters) = if admin {
             let config = get_config();
+            // TODO: Add SASL support.
+            // Perform MD5 authentication.
+            match config.general.admin_auth_type {
+                AuthType::Trust => (),
+                AuthType::MD5 => {
+                    let salt = md5_challenge(&mut write).await?;
 
-            // Compare server and client hashes.
-            let password_hash = md5_hash_password(
-                &config.general.admin_username,
-                &config.general.admin_password,
-                &salt,
-            );
+                    let code = match read.read_u8().await {
+                        Ok(p) => p,
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password code".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
 
-            if password_hash != password_response {
-                let error = Error::ClientGeneralError("Invalid password".into(), client_identifier);
+                    // PasswordMessage
+                    if code as char != 'p' {
+                        return Err(Error::ProtocolSyncError(format!(
+                            "Expected p, got {}",
+                            code as char
+                        )));
+                    }
 
-                warn!("{}", error);
-                wrong_password(&mut write, username).await?;
+                    let len = match read.read_i32().await {
+                        Ok(len) => len,
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password message length".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
 
-                return Err(error);
+                    let mut password_response = vec![0u8; (len - 4) as usize];
+
+                    match read.read_exact(&mut password_response).await {
+                        Ok(_) => (),
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password message".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
+
+                    // Compare server and client hashes.
+                    let password_hash = md5_hash_password(
+                        &config.general.admin_username,
+                        &config.general.admin_password,
+                        &salt,
+                    );
+
+                    if password_hash != password_response {
+                        let error =
+                            Error::ClientGeneralError("Invalid password".into(), client_identifier);
+
+                        warn!("{}", error);
+                        wrong_password(&mut write, username).await?;
+
+                        return Err(error);
+                    }
+                }
             }
-
             (false, generate_server_parameters_for_admin())
         }
         // Authenticate normal user.
@@ -573,92 +579,143 @@ where
             // Obtain the hash to compare, we give preference to that written in cleartext in config
             // if there is nothing set in cleartext and auth passthrough (auth_query) is configured, we use the hash obtained
             // when the pool was created. If there is no hash there, we try to fetch it one more time.
-            let password_hash = if let Some(password) = &pool.settings.user.password {
-                Some(md5_hash_password(username, password, &salt))
-            } else {
-                if !get_config().is_auth_query_configured() {
-                    wrong_password(&mut write, username).await?;
-                    return Err(Error::ClientAuthImpossible(username.into()));
-                }
+            match pool.settings.user.auth_type {
+                AuthType::Trust => (),
+                AuthType::MD5 => {
+                    // Perform MD5 authentication.
+                    // TODO: Add SASL support.
+                    let salt = md5_challenge(&mut write).await?;
 
-                let mut hash = (*pool.auth_hash.read()).clone();
+                    let code = match read.read_u8().await {
+                        Ok(p) => p,
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password code".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
 
-                if hash.is_none() {
-                    warn!(
-                        "Query auth configured \
-                          but no hash password found \
-                          for pool {}. Will try to refetch it.",
-                        pool_name
-                    );
+                    // PasswordMessage
+                    if code as char != 'p' {
+                        return Err(Error::ProtocolSyncError(format!(
+                            "Expected p, got {}",
+                            code as char
+                        )));
+                    }
 
-                    match refetch_auth_hash(&pool).await {
-                        Ok(fetched_hash) => {
-                            warn!("Password for {}, obtained. Updating.", client_identifier);
+                    let len = match read.read_i32().await {
+                        Ok(len) => len,
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password message length".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
+
+                    let mut password_response = vec![0u8; (len - 4) as usize];
+
+                    match read.read_exact(&mut password_response).await {
+                        Ok(_) => (),
+                        Err(_) => {
+                            return Err(Error::ClientSocketError(
+                                "password message".into(),
+                                client_identifier,
+                            ))
+                        }
+                    };
+
+                    let password_hash = if let Some(password) = &pool.settings.user.password {
+                        Some(md5_hash_password(username, password, &salt))
+                    } else {
+                        if !get_config().is_auth_query_configured() {
+                            wrong_password(&mut write, username).await?;
+                            return Err(Error::ClientAuthImpossible(username.into()));
+                        }
+
+                        let mut hash = (*pool.auth_hash.read()).clone();
+
+                        if hash.is_none() {
+                            warn!(
+                                "Query auth configured \
+                                  but no hash password found \
+                                  for pool {}. Will try to refetch it.",
+                                pool_name
+                            );
+
+                            match refetch_auth_hash(&pool).await {
+                                Ok(fetched_hash) => {
+                                    warn!(
+                                        "Password for {}, obtained. Updating.",
+                                        client_identifier
+                                    );
+
+                                    {
+                                        let mut pool_auth_hash = pool.auth_hash.write();
+                                        *pool_auth_hash = Some(fetched_hash.clone());
+                                    }
+
+                                    hash = Some(fetched_hash);
+                                }
+
+                                Err(err) => {
+                                    wrong_password(&mut write, username).await?;
+
+                                    return Err(Error::ClientAuthPassthroughError(
+                                        err.to_string(),
+                                        client_identifier,
+                                    ));
+                                }
+                            }
+                        };
+
+                        Some(md5_hash_second_pass(&hash.unwrap(), &salt))
+                    };
+
+                    // Once we have the resulting hash, we compare with what the client gave us.
+                    // If they do not match and auth query is set up, we try to refetch the hash one more time
+                    // to see if the password has changed since the pool was created.
+                    //
+                    // @TODO: we could end up fetching again the same password twice (see above).
+                    if password_hash.unwrap() != password_response {
+                        warn!(
+                            "Invalid password {}, will try to refetch it.",
+                            client_identifier
+                        );
+
+                        let fetched_hash = match refetch_auth_hash(&pool).await {
+                            Ok(fetched_hash) => fetched_hash,
+                            Err(err) => {
+                                wrong_password(&mut write, username).await?;
+
+                                return Err(err);
+                            }
+                        };
+
+                        let new_password_hash = md5_hash_second_pass(&fetched_hash, &salt);
+
+                        // Ok password changed in server an auth is possible.
+                        if new_password_hash == password_response {
+                            warn!(
+                                "Password for {}, changed in server. Updating.",
+                                client_identifier
+                            );
 
                             {
                                 let mut pool_auth_hash = pool.auth_hash.write();
-                                *pool_auth_hash = Some(fetched_hash.clone());
+                                *pool_auth_hash = Some(fetched_hash);
                             }
-
-                            hash = Some(fetched_hash);
-                        }
-
-                        Err(err) => {
+                        } else {
                             wrong_password(&mut write, username).await?;
-
-                            return Err(Error::ClientAuthPassthroughError(
-                                err.to_string(),
+                            return Err(Error::ClientGeneralError(
+                                "Invalid password".into(),
                                 client_identifier,
                             ));
                         }
                     }
-                };
-
-                Some(md5_hash_second_pass(&hash.unwrap(), &salt))
-            };
-
-            // Once we have the resulting hash, we compare with what the client gave us.
-            // If they do not match and auth query is set up, we try to refetch the hash one more time
-            // to see if the password has changed since the pool was created.
-            //
-            // @TODO: we could end up fetching again the same password twice (see above).
-            if password_hash.unwrap() != password_response {
-                warn!(
-                    "Invalid password {}, will try to refetch it.",
-                    client_identifier
-                );
-
-                let fetched_hash = match refetch_auth_hash(&pool).await {
-                    Ok(fetched_hash) => fetched_hash,
-                    Err(err) => {
-                        wrong_password(&mut write, username).await?;
-
-                        return Err(err);
-                    }
-                };
-
-                let new_password_hash = md5_hash_second_pass(&fetched_hash, &salt);
-
-                // Ok password changed in server an auth is possible.
-                if new_password_hash == password_response {
-                    warn!(
-                        "Password for {}, changed in server. Updating.",
-                        client_identifier
-                    );
-
-                    {
-                        let mut pool_auth_hash = pool.auth_hash.write();
-                        *pool_auth_hash = Some(fetched_hash);
-                    }
-                } else {
-                    wrong_password(&mut write, username).await?;
-                    return Err(Error::ClientGeneralError(
-                        "Invalid password".into(),
-                        client_identifier,
-                    ));
                 }
             }
-
             let transaction_mode = pool.settings.pool_mode == PoolMode::Transaction;
             prepared_statements_enabled =
                 transaction_mode && pool.prepared_statement_cache.is_some();
@@ -824,6 +881,7 @@ where
         };
 
         query_router.update_pool_settings(&pool.settings);
+        query_router.set_default_role();
 
         // Our custom protocol loop.
         // We expect the client to either start a transaction with regular queries

src/config.rs

@@ -208,6 +208,9 @@ impl Address {
 pub struct User {
     pub username: String,
     pub password: Option<String>,
+
+    #[serde(default = "User::default_auth_type")]
+    pub auth_type: AuthType,
     pub server_username: Option<String>,
     pub server_password: Option<String>,
     pub pool_size: u32,
@@ -225,6 +228,7 @@ impl Default for User {
         User {
             username: String::from("postgres"),
             password: None,
+            auth_type: AuthType::MD5,
             server_username: None,
             server_password: None,
             pool_size: 15,
@@ -239,6 +243,10 @@ impl Default for User {
 }
 
 impl User {
+    pub fn default_auth_type() -> AuthType {
+        AuthType::MD5
+    }
+
     fn validate(&self) -> Result<(), Error> {
         if let Some(min_pool_size) = self.min_pool_size {
             if min_pool_size > self.pool_size {
@@ -334,6 +342,9 @@ pub struct General {
     pub admin_username: String,
     pub admin_password: String,
 
+    #[serde(default = "General::default_admin_auth_type")]
+    pub admin_auth_type: AuthType,
+
     #[serde(default = "General::default_validate_config")]
     pub validate_config: bool,
 
@@ -348,6 +359,10 @@ impl General {
         "0.0.0.0".into()
     }
 
+    pub fn default_admin_auth_type() -> AuthType {
+        AuthType::MD5
+    }
+
     pub fn default_port() -> u16 {
         5432
     }
@@ -456,6 +471,7 @@ impl Default for General {
             verify_server_certificate: false,
             admin_username: String::from("admin"),
             admin_password: String::from("admin"),
+            admin_auth_type: AuthType::MD5,
             validate_config: true,
             auth_query: None,
             auth_query_user: None,
@@ -476,6 +492,15 @@ pub enum PoolMode {
     Session,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy, Hash)]
+pub enum AuthType {
+    #[serde(alias = "trust", alias = "Trust")]
+    Trust,
+
+    #[serde(alias = "md5", alias = "MD5")]
+    MD5,
+}
+
 impl std::fmt::Display for PoolMode {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
@@ -748,6 +773,7 @@ pub enum DefaultShard {
     Shard(usize),
     Random,
     RandomHealthy,
+    Fail,
 }
 impl Default for DefaultShard {
     fn default() -> Self {
@@ -762,6 +788,7 @@ impl serde::Serialize for DefaultShard {
             }
             DefaultShard::Random => serializer.serialize_str("random"),
             DefaultShard::RandomHealthy => serializer.serialize_str("random_healthy"),
+            DefaultShard::Fail => serializer.serialize_str("fail"),
         }
     }
 }
@@ -779,6 +806,7 @@ impl<'de> serde::Deserialize<'de> for DefaultShard {
         match s.as_str() {
             "random" => Ok(DefaultShard::Random),
             "random_healthy" => Ok(DefaultShard::RandomHealthy),
+            "fail" => Ok(DefaultShard::Fail),
             _ => Err(serde::de::Error::custom(
                 "invalid value for no_shard_specified_behavior",
             )),

src/errors.rs

@@ -30,6 +30,7 @@ pub enum Error {
     QueryRouterError(String),
     InvalidShardId(usize),
     PreparedStatementError,
+    NoShardSelected,
 }
 
 #[derive(Clone, PartialEq, Debug)]

src/pool.rs

@@ -720,6 +720,7 @@ impl ConnectionPool {
                             .unwrap()
                     });
                 }
+                DefaultShard::Fail => return Err(Error::NoShardSelected),
             },
         };
 

src/prometheus.rs

@@ -200,18 +200,17 @@ struct PrometheusMetric<Value: fmt::Display> {
 
 impl<Value: fmt::Display> fmt::Display for PrometheusMetric<Value> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let formatted_labels = self
-            .labels
+        let mut sorted_labels: Vec<_> = self.labels.iter().collect();
+        sorted_labels.sort_by_key(|&(key, _)| key);
+        let formatted_labels = sorted_labels
             .iter()
             .map(|(key, value)| format!("{}=\"{}\"", key, value))
             .collect::<Vec<_>>()
             .join(",");
         write!(
             f,
-            "# HELP {name} {help}\n# TYPE {name} {ty}\n{name}{{{formatted_labels}}} {value}\n",
+            "{name}{{{formatted_labels}}} {value}",
             name = format_args!("pgcat_{}", self.name),
-            help = self.help,
-            ty = self.ty,
             formatted_labels = formatted_labels,
             value = self.value
         )
@@ -247,7 +246,7 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
         labels.insert("pool", address.pool_name.clone());
         labels.insert("index", address.address_index.to_string());
         labels.insert("database", address.database.to_string());
-        labels.insert("user", address.username.clone());
+        labels.insert("username", address.username.clone());
 
         Self::from_name(&format!("databases_{}", name), value, labels)
     }
@@ -264,7 +263,8 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
         labels.insert("pool", address.pool_name.clone());
         labels.insert("index", address.address_index.to_string());
         labels.insert("database", address.database.to_string());
-        labels.insert("user", address.username.clone());
+        labels.insert("username", address.username.clone());
+
         Self::from_name(&format!("servers_{}", name), value, labels)
     }
 
@@ -276,7 +276,7 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
         labels.insert("role", address.role.to_string());
         labels.insert("index", address.address_index.to_string());
         labels.insert("database", address.database.to_string());
-        labels.insert("user", address.username.clone());
+        labels.insert("username", address.username.clone());
 
         Self::from_name(&format!("stats_{}", name), value, labels)
     }
@@ -288,6 +288,15 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
 
         Self::from_name(&format!("pools_{}", name), value, labels)
     }
+
+    fn get_header(&self) -> String {
+        format!(
+            "\n# HELP {name} {help}\n# TYPE {name} {ty}",
+            name = format_args!("pgcat_{}", self.name),
+            help = self.help,
+            ty = self.ty,
+        )
+    }
 }
 
 async fn prometheus_stats(
@@ -300,6 +309,7 @@ async fn prometheus_stats(
             push_pool_stats(&mut lines);
             push_server_stats(&mut lines);
             push_database_stats(&mut lines);
+            lines.push("".to_string()); // Ensure to end the stats with a line terminator as required by the specification.
 
             Response::builder()
                 .header("content-type", "text/plain; version=0.0.4")
@@ -313,6 +323,7 @@ async fn prometheus_stats(
 
 // Adds metrics shown in a SHOW STATS admin command.
 fn push_address_stats(lines: &mut Vec<String>) {
+    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u64>>> = HashMap::new();
     for (_, pool) in get_all_pools() {
         for shard in 0..pool.shards() {
             for server in 0..pool.servers(shard) {
@@ -322,7 +333,10 @@ fn push_address_stats(lines: &mut Vec<String>) {
                     if let Some(prometheus_metric) =
                         PrometheusMetric::<u64>::from_address(address, &key, value)
                     {
-                        lines.push(prometheus_metric.to_string());
+                        grouped_metrics
+                            .entry(key)
+                            .or_default()
+                            .push(prometheus_metric);
                     } else {
                         debug!("Metric {} not implemented for {}", key, address.name());
                     }
@@ -330,33 +344,53 @@ fn push_address_stats(lines: &mut Vec<String>) {
             }
         }
     }
+    for (_key, metrics) in grouped_metrics {
+        if !metrics.is_empty() {
+            lines.push(metrics[0].get_header());
+            for metric in metrics {
+                lines.push(metric.to_string());
+            }
+        }
+    }
 }
 
 // Adds relevant metrics shown in a SHOW POOLS admin command.
 fn push_pool_stats(lines: &mut Vec<String>) {
+    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u64>>> = HashMap::new();
     let pool_stats = PoolStats::construct_pool_lookup();
     for (pool_id, stats) in pool_stats.iter() {
         for (name, value) in stats.clone() {
             if let Some(prometheus_metric) =
                 PrometheusMetric::<u64>::from_pool(pool_id.clone(), &name, value)
             {
-                lines.push(prometheus_metric.to_string());
+                grouped_metrics
+                    .entry(name)
+                    .or_default()
+                    .push(prometheus_metric);
             } else {
                 debug!("Metric {} not implemented for ({})", name, *pool_id);
             }
         }
     }
+    for (_key, metrics) in grouped_metrics {
+        if !metrics.is_empty() {
+            lines.push(metrics[0].get_header());
+            for metric in metrics {
+                lines.push(metric.to_string());
+            }
+        }
+    }
 }
 
 // Adds relevant metrics shown in a SHOW DATABASES admin command.
 fn push_database_stats(lines: &mut Vec<String>) {
+    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u32>>> = HashMap::new();
     for (_, pool) in get_all_pools() {
         let pool_config = pool.settings.clone();
         for shard in 0..pool.shards() {
             for server in 0..pool.servers(shard) {
                 let address = pool.address(shard, server);
                 let pool_state = pool.pool_state(shard, server);
-
                 let metrics = vec![
                     ("pool_size", pool_config.user.pool_size),
                     ("current_connections", pool_state.connections),
@@ -365,7 +399,10 @@ fn push_database_stats(lines: &mut Vec<String>) {
                     if let Some(prometheus_metric) =
                         PrometheusMetric::<u32>::from_database_info(address, key, value)
                     {
-                        lines.push(prometheus_metric.to_string());
+                        grouped_metrics
+                            .entry(key.to_string())
+                            .or_default()
+                            .push(prometheus_metric);
                     } else {
                         debug!("Metric {} not implemented for {}", key, address.name());
                     }
@@ -373,6 +410,14 @@ fn push_database_stats(lines: &mut Vec<String>) {
             }
         }
     }
+    for (_key, metrics) in grouped_metrics {
+        if !metrics.is_empty() {
+            lines.push(metrics[0].get_header());
+            for metric in metrics {
+                lines.push(metric.to_string());
+            }
+        }
+    }
 }
 
 // Adds relevant metrics shown in a SHOW SERVERS admin command.
@@ -405,7 +450,7 @@ fn push_server_stats(lines: &mut Vec<String>) {
             crate::stats::ServerState::Idle => entry.idle_count += 1,
         }
     }
-
+    let mut grouped_metrics: HashMap<String, Vec<PrometheusMetric<u64>>> = HashMap::new();
     for (_, pool) in get_all_pools() {
         for shard in 0..pool.shards() {
             for server in 0..pool.servers(shard) {
@@ -428,7 +473,10 @@ fn push_server_stats(lines: &mut Vec<String>) {
                         if let Some(prometheus_metric) =
                             PrometheusMetric::<u64>::from_server_info(address, key, value)
                         {
-                            lines.push(prometheus_metric.to_string());
+                            grouped_metrics
+                                .entry(key.to_string())
+                                .or_default()
+                                .push(prometheus_metric);
                         } else {
                             debug!("Metric {} not implemented for {}", key, address.name());
                         }
@@ -437,6 +485,14 @@ fn push_server_stats(lines: &mut Vec<String>) {
             }
         }
     }
+    for (_key, metrics) in grouped_metrics {
+        if !metrics.is_empty() {
+            lines.push(metrics[0].get_header());
+            for metric in metrics {
+                lines.push(metric.to_string());
+            }
+        }
+    }
 }
 
 pub async fn start_metric_server(http_addr: SocketAddr) {

src/query_router.rs

@@ -386,6 +386,18 @@ impl QueryRouter {
         }
     }
 
+    /// Determines if a query is a mutation or not.
+    fn is_mutation_query(q: &sqlparser::ast::Query) -> bool {
+        use sqlparser::ast::*;
+
+        match q.body.as_ref() {
+            SetExpr::Insert(_) => true,
+            SetExpr::Update(_) => true,
+            SetExpr::Query(q) => Self::is_mutation_query(q),
+            _ => false,
+        }
+    }
+
     /// Try to infer which server to connect to based on the contents of the query.
     pub fn infer(&mut self, ast: &Vec<sqlparser::ast::Statement>) -> Result<(), Error> {
         if !self.pool_settings.query_parser_read_write_splitting {
@@ -428,8 +440,9 @@ impl QueryRouter {
                     };
 
                     let has_locks = !query.locks.is_empty();
+                    let has_mutation = Self::is_mutation_query(query);
 
-                    if has_locks {
+                    if has_locks || has_mutation {
                         self.active_role = Some(Role::Primary);
                     } else if !visited_write_statement {
                         // If we already visited a write statement, we should be going to the primary.
@@ -1048,6 +1061,11 @@ impl QueryRouter {
         self.active_shard
     }
 
+    /// Set active_role as the default_role specified in the pool.
+    pub fn set_default_role(&mut self) {
+        self.active_role = self.pool_settings.default_role;
+    }
+
     /// Get the current desired server role we should be talking to.
     pub fn role(&self) -> Option<Role> {
         self.active_role
@@ -1113,6 +1131,26 @@ mod test {
         assert_eq!(qr.role(), None);
     }
 
+    #[test]
+    fn test_split_cte_queries() {
+        QueryRouter::setup();
+        let mut qr = QueryRouter::new();
+        qr.pool_settings.query_parser_read_write_splitting = true;
+        qr.pool_settings.query_parser_enabled = true;
+
+        let query = simple_query(
+            "WITH t AS (
+                SELECT id FROM users WHERE name ILIKE '%ja%'
+            )
+            UPDATE user_languages
+               SET settings = '{}'
+              FROM t WHERE t.id = user_id;",
+        );
+        let ast = qr.parse(&query).unwrap();
+        assert!(qr.infer(&ast).is_ok());
+        assert_eq!(qr.role(), Some(Role::Primary));
+    }
+
     #[test]
     fn test_infer_replica() {
         QueryRouter::setup();

tests/python/test_auth.py

@@ -0,0 +1,71 @@
+import utils
+import signal
+
+class TestTrustAuth:
+    @classmethod
+    def setup_method(cls):
+        config= """
+        [general]
+        host = "0.0.0.0"
+        port = 6432
+        admin_username = "admin_user"
+        admin_password = ""
+        admin_auth_type = "trust"
+
+        [pools.sharded_db.users.0]
+        username = "sharding_user"
+        password = "sharding_user"
+        auth_type = "trust"
+        pool_size = 10
+        min_pool_size = 1
+        pool_mode = "transaction"
+
+        [pools.sharded_db.shards.0]
+        servers = [
+          [ "127.0.0.1", 5432, "primary" ],
+        ]
+        database = "shard0"
+        """
+        utils.pgcat_generic_start(config)
+
+    @classmethod
+    def teardown_method(self):
+        utils.pg_cat_send_signal(signal.SIGTERM)
+
+    def test_admin_trust_auth(self):
+        conn, cur = utils.connect_db_trust(admin=True)
+        cur.execute("SHOW POOLS")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)
+
+    def test_normal_trust_auth(self):
+        conn, cur = utils.connect_db_trust(autocommit=False)
+        cur.execute("SELECT 1")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)
+
+class TestMD5Auth:
+    @classmethod
+    def setup_method(cls):
+        utils.pgcat_start()
+
+    @classmethod
+    def teardown_method(self):
+        utils.pg_cat_send_signal(signal.SIGTERM)
+
+    def test_normal_db_access(self):
+        conn, cur = utils.connect_db(autocommit=False)
+        cur.execute("SELECT 1")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)
+
+    def test_admin_db_access(self):
+        conn, cur = utils.connect_db(admin=True)
+
+        cur.execute("SHOW POOLS")
+        res = cur.fetchall()
+        print(res)
+        utils.cleanup_conn(conn, cur)

tests/python/test_pgcat.py

@@ -1,30 +1,12 @@
-import os
+
 import signal
 import time
 
 import psycopg2
-
 import utils
 
 SHUTDOWN_TIMEOUT = 5
 
-def test_normal_db_access():
-    utils.pgcat_start()
-    conn, cur = utils.connect_db(autocommit=False)
-    cur.execute("SELECT 1")
-    res = cur.fetchall()
-    print(res)
-    utils.cleanup_conn(conn, cur)
-
-
-def test_admin_db_access():
-    conn, cur = utils.connect_db(admin=True)
-
-    cur.execute("SHOW POOLS")
-    res = cur.fetchall()
-    print(res)
-    utils.cleanup_conn(conn, cur)
-
 
 def test_shutdown_logic():
 
@@ -256,3 +238,5 @@ def test_shutdown_logic():
 
     utils.cleanup_conn(conn, cur)
     utils.pg_cat_send_signal(signal.SIGTERM)
+
+    # - - - - - - - - - - - - - - - - - -

tests/python/utils.py

@@ -1,20 +1,49 @@
-from typing import Tuple
 import os
-import psutil
 import signal
 import time
+from typing import Tuple
+import tempfile
 
+import psutil
 import psycopg2
 
 PGCAT_HOST = "127.0.0.1"
 PGCAT_PORT = "6432"
 
-def pgcat_start():
+
+def _pgcat_start(config_path: str):
     pg_cat_send_signal(signal.SIGTERM)
-    os.system("./target/debug/pgcat .circleci/pgcat.toml &")
+    os.system(f"./target/debug/pgcat {config_path} &")
     time.sleep(2)
 
 
+def pgcat_start():
+    _pgcat_start(config_path='.circleci/pgcat.toml')
+
+
+def pgcat_generic_start(config: str):
+    tmp = tempfile.NamedTemporaryFile()
+    with open(tmp.name, 'w') as f:
+        f.write(config)
+    _pgcat_start(config_path=tmp.name)
+
+
+def glauth_send_signal(signal: signal.Signals):
+    try:
+        for proc in psutil.process_iter(["pid", "name"]):
+            if proc.name() == "glauth":
+                os.kill(proc.pid, signal)
+    except Exception as e:
+        # The process can be gone when we send this signal
+        print(e)
+
+    if signal == signal.SIGTERM:
+        # Returns 0 if pgcat process exists
+        time.sleep(2)
+        if not os.system('pgrep glauth'):
+            raise Exception("glauth not closed after SIGTERM")
+
+
 def pg_cat_send_signal(signal: signal.Signals):
     try:
         for proc in psutil.process_iter(["pid", "name"]):
@@ -54,6 +83,27 @@ def connect_db(
 
     return (conn, cur)
 
+def connect_db_trust(
+    autocommit: bool = True,
+    admin: bool = False,
+) -> Tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]:
+
+    if admin:
+        user = "admin_user"
+        db = "pgcat"
+    else:
+        user = "sharding_user"
+        db = "sharded_db"
+
+    conn = psycopg2.connect(
+        f"postgres://{user}@{PGCAT_HOST}:{PGCAT_PORT}/{db}?application_name=testing_pgcat",
+        connect_timeout=2,
+    )
+    conn.autocommit = autocommit
+    cur = conn.cursor()
+
+    return (conn, cur)
+
 
 def cleanup_conn(conn: psycopg2.extensions.connection, cur: psycopg2.extensions.cursor):
     cur.close()

tests/ruby/Gemfile.lock

@@ -1,22 +1,33 @@
 GEM
   remote: https://rubygems.org/
   specs:
-    activemodel (7.0.4.1)
-      activesupport (= 7.0.4.1)
-    activerecord (7.0.4.1)
-      activemodel (= 7.0.4.1)
-      activesupport (= 7.0.4.1)
-    activesupport (7.0.4.1)
+    activemodel (7.1.4)
+      activesupport (= 7.1.4)
+    activerecord (7.1.4)
+      activemodel (= 7.1.4)
+      activesupport (= 7.1.4)
+      timeout (>= 0.4.0)
+    activesupport (7.1.4)
+      base64
+      bigdecimal
       concurrent-ruby (~> 1.0, >= 1.0.2)
+      connection_pool (>= 2.2.5)
+      drb
       i18n (>= 1.6, < 2)
       minitest (>= 5.1)
+      mutex_m
       tzinfo (~> 2.0)
     ast (2.4.2)
-    concurrent-ruby (1.1.10)
+    base64 (0.2.0)
+    bigdecimal (3.1.8)
+    concurrent-ruby (1.3.4)
+    connection_pool (2.4.1)
     diff-lcs (1.5.0)
-    i18n (1.12.0)
+    drb (2.2.1)
+    i18n (1.14.5)
       concurrent-ruby (~> 1.0)
-    minitest (5.17.0)
+    minitest (5.25.1)
+    mutex_m (0.2.0)
     parallel (1.22.1)
     parser (3.1.2.0)
       ast (~> 2.4.1)
@@ -24,7 +35,8 @@ GEM
     pg (1.3.2)
     rainbow (3.1.1)
     regexp_parser (2.3.1)
-    rexml (3.2.5)
+    rexml (3.3.6)
+      strscan
     rspec (3.11.0)
       rspec-core (~> 3.11.0)
       rspec-expectations (~> 3.11.0)
@@ -50,10 +62,12 @@ GEM
     rubocop-ast (1.17.0)
       parser (>= 3.1.1.0)
     ruby-progressbar (1.11.0)
+    strscan (3.1.0)
+    timeout (0.4.1)
     toml (0.3.0)
       parslet (>= 1.8.0, < 3.0.0)
     toxiproxy (2.0.1)
-    tzinfo (2.0.5)
+    tzinfo (2.0.6)
       concurrent-ruby (~> 1.0)
     unicode-display_width (2.1.0)
 

tests/ruby/load_balancing_spec.rb

@@ -56,6 +56,41 @@ describe "Random Load Balancing" do
       end
     end
   end
+
+  context "when all replicas are down " do
+    let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 5, "transaction", "random", "debug", {"default_role" => "replica"}) }
+
+    it "unbans them automatically to prevent false positives in health checks that could make all replicas unavailable" do
+      conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      failed_count = 0
+      number_of_replicas = processes[:replicas].length
+
+      # Take down all replicas
+      processes[:replicas].each(&:take_down)
+
+      (number_of_replicas + 1).times do |n|
+        conn.async_exec("SELECT 1 + 2")
+      rescue
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        failed_count += 1
+      end
+
+      expect(failed_count).to eq(number_of_replicas + 1)
+      failed_count = 0
+
+      # Ban_time is configured to 60 so this reset will only work
+      # if the replicas are unbanned automatically
+      processes[:replicas].each(&:reset)
+
+      number_of_replicas.times do
+        conn.async_exec("SELECT 1 + 2")
+      rescue
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        failed_count += 1
+      end
+      expect(failed_count).to eq(0)
+    end
+  end
 end
 
 describe "Least Outstanding Queries Load Balancing" do
@@ -161,4 +196,3 @@ describe "Least Outstanding Queries Load Balancing" do
     end
   end
 end
-