Add StatsD (updated)

Cargo.toml

@@ -34,6 +34,7 @@ rustls-pemfile = "1"
 hyper = { version = "0.14", features = ["full"] }
 phf = { version = "0.11.1", features = ["macros"] }
 exitcode = "1.1.2"
+cadence = "0.29"
 futures = "0.3"
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]

README.md

@@ -69,6 +69,32 @@ PGPASSWORD=postgres psql -h 127.0.0.1 -p 6432 -U postgres -c 'SELECT 1'
 | `query_parser_enabled`       | Enable the query parser which will inspect incoming queries and route them to a primary or replicas.                                       | `false`                          |
 | `primary_reads_enabled`      | Enable this to allow read queries on the primary; otherwise read queries are routed to the replicas.                                       | `true`                           |
 
+## Statsd Configuration
+Statsd is optional and can be configured in both UnixSocket and Udp modes
+
+
+UDP
+```toml
+[general.statsd]
+type = "Udp"
+[general.statsd.args]
+prefix = "prefix.pgcat"
+host = "statsd.host.ac"
+port = 8125
+```
+
+Unix Socket
+```toml
+[general.statsd]
+type = "UnixSocket"
+[general.statsd.args]
+prefix = "prefix.pgcat"
+path = "/var/run/statsd.socket"
+```
+
+**Note**: The statsd client in pgcat will require it's own thread to run so it doesn't interfere with the main server threads. Under default configs that would mean **5 threads** will be required to run pgcat.
+
+
 ## Local development
 
 1. Install Rust (latest stable will work great).

src/config.rs

@@ -20,6 +20,20 @@ pub const VERSION: &str = env!("CARGO_PKG_VERSION");
 /// Globally available configuration.
 static CONFIG: Lazy<ArcSwap<Config>> = Lazy::new(|| ArcSwap::from_pointee(Config::default()));
 
+#[derive(Clone, PartialEq, Serialize, Deserialize, Debug)]
+#[serde(tag = "type", content = "args")]
+pub enum StatsDMode {
+    UnixSocket {
+        prefix: String,
+        path: String,
+    },
+    Udp {
+        prefix: String,
+        host: String,
+        port: u16,
+    },
+}
+
 /// Server role: primary or replica.
 #[derive(Clone, PartialEq, Serialize, Deserialize, Hash, std::cmp::Eq, Debug, Copy)]
 pub enum Role {
@@ -188,6 +202,8 @@ pub struct General {
     pub tls_private_key: Option<String>,
     pub admin_username: String,
     pub admin_password: String,
+
+    pub statsd: Option<StatsDMode>,
 }
 
 impl General {
@@ -235,6 +251,7 @@ impl Default for General {
             port: Self::default_port(),
             enable_prometheus_exporter: Some(false),
             prometheus_exporter_port: 9930,
+            statsd: None,
             connect_timeout: General::default_connect_timeout(),
             idle_timeout: General::default_idle_timeout(),
             shutdown_timeout: Self::default_shutdown_timeout(),
@@ -577,6 +594,11 @@ impl Config {
             "Healthcheck timeout: {}ms",
             self.general.healthcheck_timeout
         );
+        if let Some(statsd_mode) = self.general.statsd.clone() {
+            info!("Statsd: {:?}", statsd_mode);
+        } else {
+            info!("Statsd: Not Enabled");
+        };
         info!("Connection timeout: {}ms", self.general.connect_timeout);
         info!("Idle timeout: {}ms", self.general.idle_timeout);
         info!(

src/messages.rs

@@ -380,7 +380,7 @@ pub fn row_description(columns: &Vec<(&str, DataType)>) -> BytesMut {
     let mut res = BytesMut::new();
     let mut row_desc = BytesMut::new();
 
-    // how many colums we are storing
+    // how many columns we are storing
     row_desc.put_i16(columns.len() as i16);
 
     for (name, data_type) in columns {

src/pool.rs

@@ -742,6 +742,7 @@ impl ManageConnection for ServerPool {
             self.address.name(),
             self.address.pool_name.clone(),
             self.address.username.clone(),
+            self.address.role,
         );
         self.stats.server_login(server_id);