feat: add rows inserted, deleted and completed metrics

pgdog-stats/src/pool.rs

@@ -63,6 +63,12 @@ pub struct Counts {
     pub writes: usize,
     /// Password attempts.
     pub auth_attempts: usize,
+    /// Rows reported affected by INSERT command tags.
+    pub rows_inserted: usize,
+    /// Rows reported affected by UPDATE command tags.
+    pub rows_updated: usize,
+    /// Rows reported affected by DELETE command tags.
+    pub rows_deleted: usize,
 }
 
 impl Sub for Counts {
@@ -95,6 +101,9 @@ impl Sub for Counts {
             reads: self.reads.saturating_sub(rhs.reads),
             writes: self.writes.saturating_sub(rhs.writes),
             auth_attempts: self.auth_attempts.saturating_sub(rhs.auth_attempts),
+            rows_inserted: self.rows_inserted.saturating_sub(rhs.rows_inserted),
+            rows_updated: self.rows_updated.saturating_sub(rhs.rows_updated),
+            rows_deleted: self.rows_deleted.saturating_sub(rhs.rows_deleted),
         }
     }
 }
@@ -129,6 +138,9 @@ impl Add for Counts {
             reads: self.reads.saturating_add(rhs.reads),
             writes: self.writes.saturating_add(rhs.writes),
             auth_attempts: self.auth_attempts.saturating_add(rhs.auth_attempts),
+            rows_inserted: self.rows_inserted.saturating_add(rhs.rows_inserted),
+            rows_updated: self.rows_updated.saturating_add(rhs.rows_updated),
+            rows_deleted: self.rows_deleted.saturating_add(rhs.rows_deleted),
         }
     }
 }
@@ -167,6 +179,9 @@ impl Div<usize> for Counts {
             reads: self.reads.checked_div(rhs).unwrap_or(0),
             writes: self.writes.checked_div(rhs).unwrap_or(0),
             auth_attempts: self.auth_attempts.checked_div(rhs).unwrap_or(0),
+            rows_inserted: self.rows_inserted.checked_div(rhs).unwrap_or(0),
+            rows_updated: self.rows_updated.checked_div(rhs).unwrap_or(0),
+            rows_deleted: self.rows_deleted.checked_div(rhs).unwrap_or(0),
         }
     }
 }

pgdog-stats/src/server.rs

@@ -27,6 +27,9 @@ pub struct Counts {
     pub close: usize,
     pub cleaned: usize,
     pub prepared_sync: usize,
+    pub rows_inserted: usize,
+    pub rows_updated: usize,
+    pub rows_deleted: usize,
 }
 
 impl Add<Counts> for PoolCounts {
@@ -58,6 +61,9 @@ impl Add<Counts> for PoolCounts {
             writes: self.writes,
             reads: self.reads,
             auth_attempts: self.auth_attempts,
+            rows_inserted: self.rows_inserted + rhs.rows_inserted,
+            rows_updated: self.rows_updated + rhs.rows_updated,
+            rows_deleted: self.rows_deleted + rhs.rows_deleted,
         }
     }
 }
@@ -86,6 +92,9 @@ impl Add for Counts {
             close: self.close.saturating_add(rhs.close),
             cleaned: self.cleaned.saturating_add(rhs.cleaned),
             prepared_sync: self.prepared_sync.saturating_add(rhs.prepared_sync),
+            rows_inserted: self.rows_inserted.saturating_add(rhs.rows_inserted),
+            rows_updated: self.rows_updated.saturating_add(rhs.rows_updated),
+            rows_deleted: self.rows_deleted.saturating_add(rhs.rows_deleted),
         }
     }
 }

pgdog/src/admin/show_stats.rs

@@ -52,6 +52,9 @@ impl Command for ShowStats {
                         Field::numeric(&format!("{}_reads", prefix)),
                         Field::numeric(&format!("{}_writes", prefix)),
                         Field::numeric(&format!("{}_auth_attempts", prefix)),
+                        Field::numeric(&format!("{}_rows_inserted", prefix)),
+                        Field::numeric(&format!("{}_rows_updated", prefix)),
+                        Field::numeric(&format!("{}_rows_deleted", prefix)),
                     ]
                 })
                 .collect::<Vec<Field>>(),
@@ -101,7 +104,10 @@ impl Command for ShowStats {
                             .add(stat.connect_count)
                             .add(stat.reads)
                             .add(stat.writes)
-                            .add(stat.auth_attempts);
+                            .add(stat.auth_attempts)
+                            .add(stat.rows_inserted)
+                            .add(stat.rows_updated)
+                            .add(stat.rows_deleted);
                     }
 
                     messages.push(dr.message()?);

pgdog/src/backend/pool/stats.rs

@@ -180,6 +180,9 @@ mod tests {
             reads: 25,
             writes: 50,
             auth_attempts: 30,
+            rows_inserted: 40,
+            rows_updated: 15,
+            rows_deleted: 5,
         }
         .into();
 
@@ -207,6 +210,9 @@ mod tests {
             reads: 10,
             writes: 20,
             auth_attempts: 20,
+            rows_inserted: 10,
+            rows_updated: 5,
+            rows_deleted: 2,
         }
         .into();
 
@@ -235,6 +241,9 @@ mod tests {
         assert_eq!(result.reads, 35);
         assert_eq!(result.writes, 70);
         assert_eq!(result.auth_attempts, 50);
+        assert_eq!(result.rows_inserted, 50);
+        assert_eq!(result.rows_updated, 20);
+        assert_eq!(result.rows_deleted, 7);
     }
 
     #[test]
@@ -263,6 +272,9 @@ mod tests {
             reads: 25,
             writes: 50,
             auth_attempts: 50,
+            rows_inserted: 40,
+            rows_updated: 20,
+            rows_deleted: 10,
         }
         .into();
 
@@ -290,6 +302,9 @@ mod tests {
             reads: 10,
             writes: 20,
             auth_attempts: 30,
+            rows_inserted: 15,
+            rows_updated: 8,
+            rows_deleted: 4,
         }
         .into();
 
@@ -318,6 +333,9 @@ mod tests {
         assert_eq!(result.reads, 15);
         assert_eq!(result.writes, 30);
         assert_eq!(result.auth_attempts, 20);
+        assert_eq!(result.rows_inserted, 25);
+        assert_eq!(result.rows_updated, 12);
+        assert_eq!(result.rows_deleted, 6);
     }
 
     #[test]
@@ -368,6 +386,9 @@ mod tests {
             reads: 10,
             writes: 20,
             auth_attempts: 10,
+            rows_inserted: 20,
+            rows_updated: 8,
+            rows_deleted: 4,
         }
         .into();
 
@@ -396,6 +417,9 @@ mod tests {
         assert_eq!(result.reads, 5);
         assert_eq!(result.writes, 10);
         assert_eq!(result.auth_attempts, 5);
+        assert_eq!(result.rows_inserted, 10);
+        assert_eq!(result.rows_updated, 4);
+        assert_eq!(result.rows_deleted, 2);
     }
 
     #[test]
@@ -439,6 +463,9 @@ mod tests {
             reads: 10,
             writes: 25,
             auth_attempts: 100,
+            rows_inserted: 50,
+            rows_updated: 30,
+            rows_deleted: 10,
         }
         .into();
 
@@ -460,6 +487,9 @@ mod tests {
             close: 2,
             cleaned: 4,
             prepared_sync: 5,
+            rows_inserted: 7,
+            rows_updated: 3,
+            rows_deleted: 1,
         };
 
         let result = pool_counts.inner + backend_counts;
@@ -487,6 +517,9 @@ mod tests {
         assert_eq!(result.reads, 10);
         assert_eq!(result.writes, 25);
         assert_eq!(result.auth_attempts, 100);
+        assert_eq!(result.rows_inserted, 57);
+        assert_eq!(result.rows_updated, 33);
+        assert_eq!(result.rows_deleted, 11);
     }
 
     #[test]

pgdog/src/backend/server.rs

@@ -657,6 +657,7 @@ impl Server {
                     "RESET" => self.client_params.clear(), // Someone reset params, we're gonna need to re-sync.
                     _ => (),
                 }
+                self.stats.rows_affected(&cmd);
                 self.statement_executed = true;
             }
             's' => self.statement_executed = true,

pgdog/src/backend/stats.rs

@@ -14,7 +14,7 @@ use crate::{
     config::Memory,
     net::{
         Parameters,
-        messages::{BackendPid, FrontendPid},
+        messages::{BackendPid, CommandComplete, FrontendPid},
     },
     state::State,
 };
@@ -194,6 +194,28 @@ impl Stats {
         self.local.last_checkout.bind += 1;
     }
 
+    /// Record rows affected from a Postgres CommandComplete message.
+    pub fn rows_affected(&mut self, cmd: &CommandComplete) {
+        let Ok(Some(rows)) = cmd.rows() else {
+            return;
+        };
+        match cmd.tag() {
+            "INSERT" => {
+                self.local.total.rows_inserted += rows;
+                self.local.last_checkout.rows_inserted += rows;
+            }
+            "UPDATE" => {
+                self.local.total.rows_updated += rows;
+                self.local.last_checkout.rows_updated += rows;
+            }
+            "DELETE" => {
+                self.local.total.rows_deleted += rows;
+                self.local.last_checkout.rows_deleted += rows;
+            }
+            _ => {}
+        }
+    }
+
     /// A transaction has been completed.
     pub fn transaction(&mut self, now: Instant) {
         self.transaction_state(now, State::Idle);

pgdog/src/net/messages/command_complete.rs

@@ -113,3 +113,34 @@ impl Protocol for CommandComplete {
         'C'
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn rows_and_tag_for_dml() {
+        let insert = CommandComplete::from_str("INSERT 0 5");
+        assert_eq!(insert.tag(), "INSERT");
+        assert_eq!(insert.rows().unwrap(), Some(5));
+
+        let update = CommandComplete::from_str("UPDATE 3");
+        assert_eq!(update.tag(), "UPDATE");
+        assert_eq!(update.rows().unwrap(), Some(3));
+
+        let delete = CommandComplete::from_str("DELETE 2");
+        assert_eq!(delete.tag(), "DELETE");
+        assert_eq!(delete.rows().unwrap(), Some(2));
+    }
+
+    #[test]
+    fn rows_none_for_non_dml() {
+        let begin = CommandComplete::from_str("BEGIN");
+        assert_eq!(begin.tag(), "BEGIN");
+        assert_eq!(begin.rows().unwrap(), None);
+
+        let select = CommandComplete::from_str("SELECT 10");
+        assert_eq!(select.tag(), "SELECT");
+        assert_eq!(select.rows().unwrap(), Some(10));
+    }
+}

pgdog/src/stats/pools.rs

@@ -87,6 +87,12 @@ impl Pools {
         let mut total_sv_xact_idle = vec![];
         let mut total_auth_attempts = vec![];
         let mut avg_auth_attempts = vec![];
+        let mut total_rows_inserted = vec![];
+        let mut avg_rows_inserted = vec![];
+        let mut total_rows_updated = vec![];
+        let mut avg_rows_updated = vec![];
+        let mut total_rows_deleted = vec![];
+        let mut avg_rows_deleted = vec![];
 
         let general = &crate::config::config().config.general;
 
@@ -321,6 +327,36 @@ impl Pools {
                         labels: labels.clone(),
                         measurement: averages.auth_attempts.into(),
                     });
+
+                    total_rows_inserted.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: totals.rows_inserted.into(),
+                    });
+
+                    avg_rows_inserted.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: averages.rows_inserted.into(),
+                    });
+
+                    total_rows_updated.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: totals.rows_updated.into(),
+                    });
+
+                    avg_rows_updated.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: averages.rows_updated.into(),
+                    });
+
+                    total_rows_deleted.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: totals.rows_deleted.into(),
+                    });
+
+                    avg_rows_deleted.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: averages.rows_deleted.into(),
+                    });
                 }
             }
         }
@@ -700,6 +736,54 @@ impl Pools {
             metric_type: None,
         }));
 
+        metrics.push(Metric::new(PoolMetric {
+            name: "total_rows_inserted".into(),
+            measurements: total_rows_inserted,
+            help: "Total rows reported affected by INSERT command tags.".into(),
+            unit: None,
+            metric_type: Some("counter".into()),
+        }));
+
+        metrics.push(Metric::new(PoolMetric {
+            name: "avg_rows_inserted".into(),
+            measurements: avg_rows_inserted,
+            help: "Average rows inserted per statistics period.".into(),
+            unit: None,
+            metric_type: None,
+        }));
+
+        metrics.push(Metric::new(PoolMetric {
+            name: "total_rows_updated".into(),
+            measurements: total_rows_updated,
+            help: "Total rows reported affected by UPDATE command tags.".into(),
+            unit: None,
+            metric_type: Some("counter".into()),
+        }));
+
+        metrics.push(Metric::new(PoolMetric {
+            name: "avg_rows_updated".into(),
+            measurements: avg_rows_updated,
+            help: "Average rows updated per statistics period.".into(),
+            unit: None,
+            metric_type: None,
+        }));
+
+        metrics.push(Metric::new(PoolMetric {
+            name: "total_rows_deleted".into(),
+            measurements: total_rows_deleted,
+            help: "Total rows reported affected by DELETE command tags.".into(),
+            unit: None,
+            metric_type: Some("counter".into()),
+        }));
+
+        metrics.push(Metric::new(PoolMetric {
+            name: "avg_rows_deleted".into(),
+            measurements: avg_rows_deleted,
+            help: "Average rows deleted per statistics period.".into(),
+            unit: None,
+            metric_type: None,
+        }));
+
         Pools { metrics }
     }