From 80d13ea79cb79ada1642cac09a6b14f98061443a Mon Sep 17 00:00:00 2001 From: Jan-Erik Rediger Date: Wed, 29 Jul 2026 15:18:46 +0200 Subject: [PATCH] sqlite: Use an in-memory database when delay_ping_lifetime_io=true This restores `delay_ping_lifetime_io` behavior (not touching disk until `persist_ping_lifetime_data` is called, or in auto-mode when the counter/time is hitting the limit) The local benchmark shows it's twice as fast compared to current `main` (so compared to always hitting disk). --- .../rlb-tests/src/bin/delayed-ping-data.rs | 24 +- glean-core/rlb-tests/tests/it.rs | 4 + glean-core/src/database/sqlite.rs | 367 +++++++++++++++--- glean-core/src/database/sqlite/schema.rs | 29 +- 4 files changed, 357 insertions(+), 67 deletions(-) diff --git a/glean-core/rlb-tests/src/bin/delayed-ping-data.rs b/glean-core/rlb-tests/src/bin/delayed-ping-data.rs index cfa5fcc94a..07f1e642f4 100644 --- a/glean-core/rlb-tests/src/bin/delayed-ping-data.rs +++ b/glean-core/rlb-tests/src/bin/delayed-ping-data.rs @@ -13,7 +13,7 @@ use flate2::read::GzDecoder; use glean::{ClientInfoMetrics, ConfigurationBuilder, net, private::PingType}; pub mod glean_metrics { - use glean::{CommonMetricData, Lifetime, private::CounterMetric}; + use glean::{CommonMetricData, Lifetime, private::*}; #[allow(non_upper_case_globals)] pub static sample_counter: once_cell::sync::Lazy = @@ -27,6 +27,24 @@ pub mod glean_metrics { ..Default::default() }) }); + + #[allow(non_upper_case_globals)] + pub static labeled_counter: once_cell::sync::Lazy> = + once_cell::sync::Lazy::new(|| { + LabeledMetric::new( + glean::LabeledMetricData::Common { + cmd: CommonMetricData { + name: "labeled_counter".into(), + category: "test.metrics".into(), + send_in_pings: vec!["prototype".into()], + disabled: false, + lifetime: Lifetime::Ping, + ..Default::default() + }, + }, + None, + ) + }); } #[derive(Debug)] @@ -123,11 +141,13 @@ fn main() { match &*state { "accumulate_one_and_pretend_crash" => { log::debug!("incrementing by 1. exiting without shutdown."); - glean_metrics::sample_counter.add(1) + glean_metrics::sample_counter.add(1); + glean_metrics::labeled_counter.get("label").add(1); } "accumulate_ten_and_orderly_shutdown" => { log::debug!("incrementing by 10, waiting, shutdown. should trigger a flush."); glean_metrics::sample_counter.add(10); + glean_metrics::labeled_counter.get("label").add(10); glean::shutdown(); } "submit_ping" => { diff --git a/glean-core/rlb-tests/tests/it.rs b/glean-core/rlb-tests/tests/it.rs index 71bc65a68a..7df2bff308 100644 --- a/glean-core/rlb-tests/tests/it.rs +++ b/glean-core/rlb-tests/tests/it.rs @@ -79,6 +79,10 @@ fn delayed_ping_data() { let mut lines = payload.lines(); assert!(lines.any(|line| line.contains("\"url\":") && line.contains("/prototype/"))); assert!(payload.contains("\"test.metrics.sample_counter\":10")); + assert!( + payload.contains("\"test.metrics.labeled_counter\":{\"label\":10}}}"), + "{payload}" + ); } #[cfg(unix)] diff --git a/glean-core/src/database/sqlite.rs b/glean-core/src/database/sqlite.rs index 1db1eeb2d1..7c6f27b121 100644 --- a/glean-core/src/database/sqlite.rs +++ b/glean-core/src/database/sqlite.rs @@ -2,12 +2,14 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. +use std::cell::Cell; use std::fmt::{self, Display}; use std::fs; use std::num::NonZeroU64; use std::path::Path; use std::str; -use std::time::Duration; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; use malloc_size_of::MallocSizeOf; use rusqlite::params; @@ -22,6 +24,7 @@ pub use schema::SchemaError; use crate::common_metric_data::CommonMetricDataInternal; use crate::database::migration::{self, MigrationState}; +use crate::database::sqlite::schema::create_in_memory_table; use crate::metrics::dual_labeled_counter::RECORD_SEPARATOR; use crate::metrics::Metric; use crate::Glean; @@ -33,6 +36,22 @@ use super::ConnExt; mod connection; mod schema; +const DEFAULT_TABLE: &str = "telemetry"; +const IN_MEMORY_DATABASE: &str = "lifetime_ping"; +const IN_MEMORY_TABLE: &str = "lifetime_ping.telemetry"; + +#[test] +fn consts_are_correct() { + assert_eq!( + IN_MEMORY_DATABASE, + &IN_MEMORY_TABLE[0..IN_MEMORY_DATABASE.len()] + ); + assert_eq!( + DEFAULT_TABLE, + &IN_MEMORY_TABLE[(IN_MEMORY_TABLE.len() - DEFAULT_TABLE.len())..] + ); +} + #[derive(Debug)] pub enum LoadState { Ok, @@ -63,6 +82,32 @@ pub struct Database { /// Set when a database migration attempt failed. pub(crate) migration_error: MigrationResult, + + /// If the `delay_ping_lifetime_io` Glean config option is `true`, + /// we will save metrics with 'ping' lifetime data in memory only, + /// and persist them to disk in bulk on demand. + ping_lifetime_data: bool, + + /// A count of how many database writes have been done since the last ping-lifetime flush. + /// + /// A ping-lifetime flush is automatically done after `ping_lifetime_threshold` writes. + /// + /// Only relevant if `delay_ping_lifetime_io` is set to `true`, + ping_lifetime_count: AtomicUsize, + + /// Write-count threshold when to auto-flush. `0` disables it. + ping_lifetime_threshold: usize, + + /// The last time the `lifetime=ping` data was flushed to disk. + /// + /// Data is flushed to disk automatically when the last flush was more than + /// `ping_lifetime_max_time` ago. + /// + /// Only relevant if `delay_ping_lifetime_io` is set to `true`, + ping_lifetime_store_ts: Cell, + + /// After what time to auto-flush. 0 disables it. + ping_lifetime_max_time: Duration, } impl MallocSizeOf for Database { @@ -197,9 +242,9 @@ impl Database { /// the underlying directory structure. pub fn new( data_path: &Path, - _delay_ping_lifetime_io: bool, - _ping_lifetime_threshold: usize, - _ping_lifetime_max_time: Duration, + delay_ping_lifetime_io: bool, + ping_lifetime_threshold: usize, + ping_lifetime_max_time: Duration, ) -> Result { let path = data_path.join("db"); log::debug!("Database path: {:?}", path.display()); @@ -209,12 +254,26 @@ impl Database { let store_path = path.join(DEFAULT_DATABASE_FILE_NAME); let (conn, load_state) = sqlite_open(&store_path)?; + let ping_lifetime_data = if delay_ping_lifetime_io { + conn.write(|tx| create_in_memory_table(tx, IN_MEMORY_DATABASE, DEFAULT_TABLE))?; + + true + } else { + false + }; + + let now = Instant::now(); let mut db = Self { conn, file_size, load_state, migration_state: None, migration_error: MigrationResult::Unknown, + ping_lifetime_data, + ping_lifetime_count: AtomicUsize::new(0), + ping_lifetime_threshold, + ping_lifetime_store_ts: Cell::new(now), + ping_lifetime_max_time, }; match migration::try_migrate(&path, &db) { @@ -238,6 +297,8 @@ impl Database { Ok::<(), rusqlite::Error>(()) })?; + db.load_ping_lifetime_data(); + Ok(db) } @@ -306,6 +367,23 @@ impl Database { Ok(()) } + /// Loads Lifetime::Ping data from rkv to memory, + /// if `delay_ping_lifetime_io` is set to true. + /// + /// Does nothing if it isn't or if there is not data to load. + fn load_ping_lifetime_data(&self) { + if !self.ping_lifetime_data { + return; + }; + + let copy_sql = + "INSERT INTO lifetime_ping.telemetry SELECT * FROM telemetry WHERE lifetime = 'ping'"; + let res = self.conn.write(|tx| tx.execute_one(copy_sql)); + if let Err(err) = res { + log::error!("Could not load ping lifetime data into memory: {err:?}"); + } + } + /// Iterates with the provided transaction function /// over the requested data from the given storage. /// @@ -331,19 +409,23 @@ impl Database { where F: FnMut(&[u8], &[&str], &Metric), { - let iter_sql = r#" - SELECT - id, - value, - labels - FROM telemetry - WHERE - lifetime = ?1 - AND ping = ?2 - "#; + let table = self.table_for_lifetime(lifetime); + + let iter_sql = format!( + r#" + SELECT + id, + value, + labels + FROM {table} + WHERE + lifetime = ?1 + AND ping = ?2 + "# + ); self.conn.read(|conn| { - let mut stmt = conn.prepare_cached(iter_sql)?; + let mut stmt = conn.prepare_cached(&iter_sql)?; let rows = stmt.query_map( params![lifetime.as_str().to_string(), storage_name], |row| { @@ -374,17 +456,21 @@ impl Database { data: &CommonMetricDataInternal, storage_name: &str, ) -> Option { + let table = self.table_for_lifetime(data.inner.lifetime); + // TODO(bug 2048194): Remove the `LIMIT 1` and error out when more than 1 row is returned. - let get_metric_sql = r#" - SELECT - value - FROM telemetry - WHERE - id = ?1 - AND ping = ?2 - AND labels = ?3 - LIMIT 1 - "#; + let get_metric_sql = format!( + r#" + SELECT + value + FROM {table} + WHERE + id = ?1 + AND ping = ?2 + AND labels = ?3 + LIMIT 1 + "# + ); let metric_identifier = &data.base_identifier(); @@ -392,7 +478,7 @@ impl Database { .read(|tx| { let labels = data.check_labels(tx); - let mut stmt = tx.prepare_cached(get_metric_sql)?; + let mut stmt = tx.prepare_cached(&get_metric_sql)?; stmt.query_one([metric_identifier, storage_name, labels.label()], |row| { let blob: Vec = row.get(0)?; let blob: Metric = @@ -423,18 +509,22 @@ impl Database { storage_name: &str, metric_identifier: &str, ) -> bool { - let has_metric_sql = r#" - SELECT id - FROM telemetry - WHERE - lifetime = ?1 - AND ping = ?2 - AND id = ?3 - "#; + let table = self.table_for_lifetime(lifetime); + + let has_metric_sql = format!( + r#" + SELECT id + FROM {table} + WHERE + lifetime = ?1 + AND ping = ?2 + AND id = ?3 + "# + ); self.conn .read(|conn| { - let Ok(mut stmt) = conn.prepare_cached(has_metric_sql) else { + let Ok(mut stmt) = conn.prepare_cached(&has_metric_sql) else { return Ok(false); }; let Ok(mut metric_iter) = @@ -500,25 +590,40 @@ impl Database { labels: &str, metric: &Metric, ) -> Result<()> { - let insert_sql = r#" - INSERT INTO - telemetry (id, ping, lifetime, labels, value) - VALUES - (?1, ?2, ?3, ?4, ?5) - ON CONFLICT(id, ping, labels) DO UPDATE SET - lifetime = excluded.lifetime, - value = excluded.value - "#; + let table = self.table_for_lifetime(lifetime); + + let insert_sql = format!( + r#" + INSERT INTO + {table} (id, ping, lifetime, labels, value) + VALUES + (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(id, ping, labels) DO UPDATE SET + lifetime = excluded.lifetime, + value = excluded.value + "# + ); - let mut stmt = tx.prepare_cached(insert_sql)?; - let encoded = rmp_serde::to_vec(&metric).expect("IMPOSSIBLE: Serializing metric failed"); - stmt.execute(params![ - key, - storage_name, - lifetime.as_str(), - labels, - encoded - ])?; + { + let mut stmt = tx.prepare_cached(&insert_sql)?; + let encoded = + rmp_serde::to_vec(&metric).expect("IMPOSSIBLE: Serializing metric failed"); + stmt.execute(params![ + key, + storage_name, + lifetime.as_str(), + labels, + encoded + ])?; + } + + if lifetime == Lifetime::Ping && self.ping_lifetime_data { + // flush ping lifetime + if let Err(err) = self.persist_ping_lifetime_data_if_full(tx) { + log::error!("Can't flush ping lifetime data: {err:?}"); + }; + return Ok(()); + } Ok(()) } @@ -596,20 +701,24 @@ impl Database { where F: FnMut(Option) -> Metric, { + let table = self.table_for_lifetime(lifetime); + // TODO(bug 2048194): Remove the `LIMIT 1` and error out when more than 1 row is returned. - let value_sql = r#" + let value_sql = format!( + r#" SELECT value - FROM telemetry + FROM {table} WHERE id = ?1 AND ping = ?2 AND lifetime = ?3 AND labels = ?4 LIMIT 1 - "#; + "# + ); let new_value = { - let mut stmt = tx.prepare_cached(value_sql)?; + let mut stmt = tx.prepare_cached(&value_sql)?; let mut rows = stmt.query(params![ key, storage_name, @@ -626,18 +735,20 @@ impl Database { } }; - let insert_sql = r#" + let insert_sql = format!( + r#" INSERT INTO - telemetry (id, ping, lifetime, labels, value) + {table} (id, ping, lifetime, labels, value) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(id, ping, labels) DO UPDATE SET lifetime = excluded.lifetime, value = excluded.value - "#; + "# + ); { - let mut stmt = tx.prepare_cached(insert_sql)?; + let mut stmt = tx.prepare_cached(&insert_sql)?; let encoded = rmp_serde::to_vec(&new_value).expect("IMPOSSIBLE: Serializing metric failed"); stmt.execute(params![ @@ -649,6 +760,14 @@ impl Database { ])?; } + if lifetime == Lifetime::Ping && self.ping_lifetime_data { + // flush ping lifetime + if let Err(err) = self.persist_ping_lifetime_data_if_full(tx) { + log::error!("Can't flush ping lifetime data: {err:?}"); + }; + return Ok(()); + } + Ok(()) } @@ -666,10 +785,20 @@ impl Database { /// /// This function will **not** panic on database errors. pub fn clear_ping_lifetime_storage(&self, storage_name: &str) -> Result<()> { - let clear_sql = "DELETE FROM telemetry WHERE lifetime = 'ping' AND ping = ?1"; self.conn.write(|tx| { + let clear_sql = "DELETE FROM telemetry WHERE lifetime = 'ping' AND ping = ?1"; let mut stmt = tx.prepare_cached(clear_sql)?; stmt.execute([storage_name])?; + + // Lifetime::Ping data will be saved to `ping_lifetime_data` + // in case `delay_ping_lifetime_io` is set to true + if self.ping_lifetime_data { + let clear_sql = + "DELETE FROM lifetime_ping.telemetry WHERE lifetime = 'ping' AND ping = ?1"; + let mut stmt = tx.prepare_cached(clear_sql)?; + stmt.execute([storage_name])?; + } + Ok(()) }) } @@ -707,10 +836,18 @@ impl Database { storage_name: &str, metric_id: &str, ) -> Result<()> { - let clear_sql = "DELETE FROM telemetry WHERE lifetime = ?1 AND ping = ?2 AND id = ?3"; self.conn.write(|tx| { + let clear_sql = "DELETE FROM telemetry WHERE lifetime = ?1 AND ping = ?2 AND id = ?3"; let mut stmt = tx.prepare_cached(clear_sql)?; stmt.execute([lifetime.as_str(), storage_name, metric_id])?; + + // Lifetime::Ping data is not persisted to disk if + // Glean has `delay_ping_lifetime_io` set to true + if lifetime == Lifetime::Ping && self.ping_lifetime_data { + let clear_sql = "DELETE FROM lifetime_ping.telemetry WHERE lifetime = ?1 AND ping = ?2 AND id = ?3"; + let mut stmt = tx.prepare_cached(clear_sql)?; + stmt.execute([lifetime.as_str(), storage_name, metric_id])?; + } Ok(()) }) } @@ -723,14 +860,27 @@ impl Database { /// /// * This function will **not** panic on database errors. pub fn clear_lifetime(&self, lifetime: Lifetime) { - let clear_sql = "DELETE FROM telemetry WHERE lifetime = ?1"; _ = self.conn.write(|tx| { + let clear_sql = "DELETE FROM telemetry WHERE lifetime = ?1"; let mut stmt = tx.prepare_cached(clear_sql)?; let res = stmt.execute([lifetime.as_str()]); if let Err(e) = res { log::warn!("Could not clear store for lifetime {:?}: {:?}", lifetime, e); } + + // Lifetime::Ping data is not persisted to disk if + // Glean has `delay_ping_lifetime_io` set to true + if lifetime == Lifetime::Ping && self.ping_lifetime_data { + let clear_sql = "DELETE FROM lifetime_ping.telemetry WHERE lifetime = ?1"; + let mut stmt = tx.prepare_cached(clear_sql)?; + let res = stmt.execute([lifetime.as_str()]); + + if let Err(e) = res { + log::warn!("Could not clear store for lifetime {:?}: {:?}", lifetime, e); + } + } + Ok::<(), rusqlite::Error>(()) }); } @@ -757,10 +907,38 @@ impl Database { if let Err(e) = res { log::warn!("Could not clear store for all lifetimes: {:?}", e); } + + // Lifetime::Ping data is not persisted to disk if + // Glean has `delay_ping_lifetime_io` set to true + if self.ping_lifetime_data { + let clear_sql = + "DELETE FROM lifetime_ping.telemetry WHERE lifetime = ?1 OR lifetime = ?2 OR lifetime = ?3"; + + let mut stmt = tx.prepare_cached(clear_sql)?; + let res = stmt.execute(lifetimes); + + if let Err(e) = res { + log::warn!("Could not clear store for all lifetimes: {:?}", e); + } + } + Ok::<(), rusqlite::Error>(()) }); } + /// Return the table to query for this lifetime. + /// + /// `Lifetime::Ping` data is not immediately persisted to disk if + /// Glean has `delay_ping_lifetime_io` set to true + /// In that case we use an in-memory database in an attached database. + fn table_for_lifetime(&self, lifetime: Lifetime) -> &'static str { + if lifetime == Lifetime::Ping && self.ping_lifetime_data { + IN_MEMORY_TABLE + } else { + DEFAULT_TABLE + } + } + /// Persists ping_lifetime_data to disk. /// /// Does nothing in case there is nothing to persist. @@ -769,6 +947,67 @@ impl Database { /// /// * This function will **not** panic on database errors. pub fn persist_ping_lifetime_data(&self) -> Result<()> { + if self.ping_lifetime_data { + let persist_sql = r#" + INSERT INTO telemetry SELECT * FROM lifetime_ping.telemetry WHERE true + ON CONFLICT(id, ping, labels) DO UPDATE SET + lifetime = excluded.lifetime, + value = excluded.value + "#; + self.conn.write(|tx| tx.execute_one(persist_sql))?; + + // We can reset the write-counter. Current data has been persisted. + self.ping_lifetime_count.store(0, Ordering::Release); + self.ping_lifetime_store_ts.replace(Instant::now()); + } + Ok(()) + } + + fn persist_ping_lifetime_data_if_full(&self, tx: &mut Transaction) -> Result<()> { + if self.ping_lifetime_threshold == 0 && self.ping_lifetime_max_time.is_zero() { + return Ok(()); + } + + let write_count = self.ping_lifetime_count.fetch_add(1, Ordering::Release) + 1; + let last_write = self.ping_lifetime_store_ts.get(); + let elapsed = last_write.elapsed(); + + if (self.ping_lifetime_threshold == 0 || write_count < self.ping_lifetime_threshold) + && (self.ping_lifetime_max_time.is_zero() || elapsed < self.ping_lifetime_max_time) + { + log::trace!( + "Not flushing. write_count={} (threshold={}), elapsed={:?} (max={:?})", + write_count, + self.ping_lifetime_threshold, + elapsed, + self.ping_lifetime_max_time + ); + return Ok(()); + } + + if self.ping_lifetime_threshold > 0 && write_count >= self.ping_lifetime_threshold { + log::debug!( + "Flushing database due to threshold of {} reached.", + self.ping_lifetime_threshold + ) + } else if !self.ping_lifetime_max_time.is_zero() && elapsed >= self.ping_lifetime_max_time { + log::debug!( + "Flushing database due to last write more than {:?} ago", + self.ping_lifetime_max_time + ); + } + + let persist_sql = r#" + INSERT INTO telemetry SELECT * FROM lifetime_ping.telemetry WHERE true + ON CONFLICT(id, ping, labels) DO UPDATE SET + lifetime = excluded.lifetime, + value = excluded.value + "#; + tx.execute_one(persist_sql)?; + + self.ping_lifetime_count.store(0, Ordering::Release); + self.ping_lifetime_store_ts.replace(Instant::now()); + Ok(()) } } diff --git a/glean-core/src/database/sqlite/schema.rs b/glean-core/src/database/sqlite/schema.rs index cec5072214..0cd669c3db 100644 --- a/glean-core/src/database/sqlite/schema.rs +++ b/glean-core/src/database/sqlite/schema.rs @@ -6,7 +6,7 @@ use std::num::NonZeroU32; -use rusqlite::{config::DbConfig, OptionalExtension, Transaction}; +use rusqlite::{config::DbConfig, named_params, OptionalExtension, Transaction}; use super::connection::ConnectionOpener; @@ -108,3 +108,30 @@ pub enum SchemaError { #[error("sqlite: {0}")] Sqlite(#[from] rusqlite::Error), } + +pub fn create_in_memory_table( + tx: &mut Transaction<'_>, + database: &str, + table: &str, +) -> Result<(), rusqlite::Error> { + tx.execute( + "ATTACH DATABASE ':memory:' AS :database", + named_params! {":database": database}, + )?; + tx.execute( + &format!( + r#" + CREATE TABLE {database}.{table}( + id TEXT NOT NULL, + ping TEXT NOT NULL, + lifetime TEXT NOT NULL, + labels TEXT NOT NULL, -- can't be null or ON CONFLICT won't work + value BLOB, + UNIQUE(id, ping, labels) + ); + "# + ), + [], + )?; + Ok(()) +}