Revision control
Copy as Markdown
Other Tools
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
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::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use chrono::{DateTime, Utc};
use connection::Connection;
use malloc_size_of::MallocSizeOf;
use rusqlite::fallible_iterator::FallibleIterator;
use rusqlite::types::{FromSql, FromSqlError, FromSqlResult, ToSqlOutput, ValueRef};
use rusqlite::OptionalExtension;
use rusqlite::Transaction;
use rusqlite::{params, ToSql};
use rusqlite::{Error as SqlError, ErrorCode};
use schema::Schema;
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::Lifetime;
use crate::Result;
use crate::{Glean, JsonValue};
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,
Err(OpenError),
}
#[derive(Debug, PartialEq, Eq, Copy, Clone)]
pub enum MigrationResult {
/// Migration did not happen yet
Unknown,
/// Migration failed
Error,
}
#[derive(Debug)]
pub struct Database {
/// The database connection.
pub(crate) conn: connection::Connection,
/// Initial file size when opening the database.
pub(crate) file_size: Option<NonZeroU64>,
/// Load state
load_state: LoadState,
/// Migration state, counts migrated metrics and the time it took.
pub(crate) migration_state: Option<MigrationState>,
/// 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.
delay_ping_lifetime_io: 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<Instant>,
/// After what delay to auto-flush. 0 disables it.
ping_lifetime_max_time: Duration,
}
impl MallocSizeOf for Database {
fn size_of(&self, _ops: &mut malloc_size_of::MallocSizeOfOps) -> usize {
// FIXME: Can we get the allocated size of the connection?
0
}
}
pub struct SubmittedPing {
pub document_id: String,
pub ping: String,
pub submitted_date: SqliteDatetime,
pub uploaded_date: Option<SqliteDatetime>,
pub upload_failed: Option<SqliteDatetime>,
pub payload: Option<String>,
}
impl SubmittedPing {
pub fn payload(&self) -> Option<JsonValue> {
self.payload
.as_ref()
.map(|p| match serde_json::from_str(p) {
Ok(v) => Some(v),
Err(e) => {
log::warn!("Unable to serialize JSON payload from string: {:?}", e);
None
}
})
.unwrap_or(None)
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct SqliteDatetime(pub DateTime<Utc>);
impl ToSql for SqliteDatetime {
fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
Ok(ToSqlOutput::from(self.0.timestamp_millis()))
}
}
impl FromSql for SqliteDatetime {
fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
i64::column_result(value).and_then(|as_i64| match DateTime::from_timestamp_millis(as_i64) {
Some(d) => Ok(SqliteDatetime(d)),
None => Err(FromSqlError::InvalidType),
})
}
}
const DEFAULT_DATABASE_FILE_NAME: &str = "glean.sqlite";
/// Calculate the database size from all the files in the directory.
///
/// # Arguments
///
/// *`path` - The path to the directory
///
/// # Returns
///
/// Returns the non-zero combined size in bytes of all files in a directory,
/// or `None` on error or if the size is `0`.
fn database_size(dir: &Path) -> Option<NonZeroU64> {
let mut total_size = 0;
if let Ok(entries) = fs::read_dir(dir) {
for entry in entries.flatten() {
if let Ok(file_type) = entry.file_type() {
if file_type.is_file() {
let path = entry.path();
if let Ok(metadata) = fs::metadata(path) {
total_size += metadata.len();
} else {
continue;
}
}
}
}
}
NonZeroU64::new(total_size)
}
#[derive(Debug)]
pub enum OpenError {
IncompatibleVersion(u32),
Corrupt,
SqlError(rusqlite::Error),
RecoveryError(std::io::Error),
}
impl std::error::Error for OpenError {}
impl Display for OpenError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
use OpenError::*;
match self {
IncompatibleVersion(v) => write!(f, "Incompatible database version: {v}"),
Corrupt => write!(f, "Database is corrupt"),
SqlError(err) => write!(f, "Error executing SQL: {err}"),
RecoveryError(err) => write!(
f,
"Failed to recover a corrupt database due to an error deleting the file: {err}"
),
}
}
}
impl From<rusqlite::Error> for OpenError {
fn from(value: rusqlite::Error) -> Self {
match value {
rusqlite::Error::SqliteFailure(e, _)
if matches!(e.code, ErrorCode::DatabaseCorrupt | ErrorCode::NotADatabase) =>
{
Self::Corrupt
}
_ => Self::SqlError(value),
}
}
}
impl From<SchemaError> for OpenError {
fn from(value: SchemaError) -> Self {
match value {
SchemaError::Sqlite(err) => OpenError::SqlError(err),
SchemaError::UnsupportedSchemaVersion(v) => OpenError::IncompatibleVersion(v),
}
}
}
pub fn sqlite_open(path: &Path) -> std::result::Result<(Connection, LoadState), OpenError> {
// TODO(bug 2049292): Make this more robust, use the correct errors and see how we can test all the branches
// properly.
match Connection::new::<Schema>(path) {
Err(e @ SchemaError::UnsupportedSchemaVersion(_)) => Err(e.into()),
Err(e @ SchemaError::Sqlite(SqlError::SqliteFailure(err, _))) => {
match err.code {
ErrorCode::PermissionDenied => Err(e.into()),
ErrorCode::NotADatabase => {
log::debug!("sqlite failed: not a database. starting from scratch.");
fs::remove_file(path).map_err(OpenError::RecoveryError)?;
// Now try again, we only handle that error once.
let conn = Connection::new::<Schema>(path)?;
Ok((conn, LoadState::Err(OpenError::Corrupt)))
}
ErrorCode::CannotOpen => {
log::debug!("sqlite failed: cannot open. starting from scratch.");
fs::remove_file(path).map_err(OpenError::RecoveryError)?;
// Now try again, we only handle that error once.
let conn = Connection::new::<Schema>(path)?;
Ok((conn, LoadState::Err(OpenError::Corrupt)))
}
_ => Err(e.into()),
}
}
Err(err @ SchemaError::Sqlite(SqlError::SqlInputError { .. })) => {
log::debug!("sqlite failed: schema migration failed. starting from scratch.");
fs::remove_file(path).map_err(OpenError::RecoveryError)?;
// Now try again, we only handle that error once.
let conn = Connection::new::<Schema>(path)?;
Ok((conn, LoadState::Err(err.into())))
}
other => {
let conn = other?;
Ok((conn, LoadState::Ok))
}
}
}
impl Database {
/// Initializes the data store.
///
/// This opens the underlying SQLite store and creates
/// the underlying directory structure.
pub fn new(
data_path: &Path,
delay_ping_lifetime_io: bool,
ping_lifetime_threshold: usize,
ping_lifetime_max_time: Duration,
) -> Result<Self> {
let path = data_path.join("db");
log::debug!("Database path: {:?}", path.display());
let file_size = database_size(&path);
fs::create_dir_all(&path)?;
let store_path = path.join(DEFAULT_DATABASE_FILE_NAME);
let (conn, load_state) = sqlite_open(&store_path)?;
if delay_ping_lifetime_io {
conn.write(|tx| create_in_memory_table(tx, IN_MEMORY_DATABASE))?;
}
let now = Instant::now();
let mut db = Self {
conn,
file_size,
load_state,
migration_state: None,
migration_error: MigrationResult::Unknown,
delay_ping_lifetime_io,
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) {
Ok(Some(state)) => {
log::debug!("Migration done. state={state:?}");
db.migration_state = Some(state);
db.run_maintenance(true)?;
}
Ok(None) => {
log::debug!("No migration.");
db.run_maintenance(false)?;
}
Err(e) => {
db.migration_error = MigrationResult::Error;
log::warn!(
"Migration failed! Continuing with SQLite backend without migrated data. Error: {e:?}"
)
}
}
db.conn.write(|tx| {
tx.execute("INSERT INTO migration (id, state) VALUES (1, 'done') ON CONFLICT(id) DO UPDATE SET state = excluded.state", [])?;
Ok::<(), rusqlite::Error>(())
})?;
db.load_ping_lifetime_data();
Ok(db)
}
/// Get the initial database file size.
pub fn file_size(&self) -> Option<NonZeroU64> {
self.file_size
}
/// Get the load state.
pub fn load_state(&self) -> Option<String> {
if let LoadState::Err(e) = &self.load_state {
Some(match e {
OpenError::IncompatibleVersion(v) => format!("incompatible version: {v}"),
OpenError::Corrupt => "database file corrupt".to_string(),
OpenError::SqlError(error) => format!("sql error: {error:?}"),
OpenError::RecoveryError(error) => format!("recovery error: {error:?}"),
})
} else {
None
}
}
/// Run periodic database maintenance.
///
/// If `force=true` always run the full maintenance taks
pub fn run_maintenance(&self, force: bool) -> Result<()> {
let conn = self.conn.lock();
let conn = &*conn;
self.run_maintenance_vacuum(conn, force)?;
self.run_maintenance_optimize(conn)?;
self.run_maintenance_checkpoint(conn)?;
Ok(())
}
/// Run maintenance on the database (vacuum step)
///
/// If `force_full: true` it _always_ runs a full `VACUUM`.
fn run_maintenance_vacuum(&self, conn: &rusqlite::Connection, force_full: bool) -> Result<()> {
let auto_vacuum_setting: u32 =
conn.query_row_and_then("PRAGMA auto_vacuum", [], |row| row.get(0))?;
if !force_full && auto_vacuum_setting == 2 {
// Ideally, we run an incremental vacuum to delete 2 pages
conn.execute_one("PRAGMA incremental_vacuum(2)")?;
} else {
// If auto_vacuum=incremental isn't set, configure it and run a full vacuum.
log::debug!(
"run_maintenance_vacuum: Need to run a full vacuum to set auto_vacuum=incremental"
);
conn.execute_one("PRAGMA auto_vacuum = INCREMENTAL")?;
conn.execute_one("VACUUM")?;
}
Ok(())
}
/// Run maintenance on the database (optimize step)
fn run_maintenance_optimize(&self, conn: &rusqlite::Connection) -> Result<()> {
conn.execute("PRAGMA optimize", [])?;
Ok(())
}
/// Run maintenance on the database (checkpoint step)
fn run_maintenance_checkpoint(&self, conn: &rusqlite::Connection) -> Result<()> {
conn.query_row("PRAGMA wal_checkpoint(PASSIVE)", [], |_| Ok(()))?;
Ok(())
}
/// Loads `Lifetime::Ping` data from the database 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(&mut self) {
if !self.delay_ping_lifetime_io {
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:?}. Disabling ping lifetime IO delay.");
self.delay_ping_lifetime_io = false;
}
}
/// Iterates with the provided transaction function
/// over the requested data from the given storage.
///
/// * If the storage is unavailable, the transaction function is never invoked.
/// * If the read data cannot be deserialized it will be silently skipped.
///
/// # Arguments
///
/// * `lifetime` - The metric lifetime to iterate over.
/// * `storage_name` - The storage name to iterate over.
/// * `transaction_fn` - Called for each entry being iterated over. It is
/// passed two arguments: `(metric_id: &[u8], metric: &Metric)`.
///
/// # Panics
///
/// This function will **not** panic on database errors.
pub fn iter_store<F>(
&self,
lifetime: Lifetime,
storage_name: &str,
mut transaction_fn: F,
) -> Result<()>
where
F: FnMut(&[u8], &[&str], &Metric),
{
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 rows = stmt.query_map(
params![lifetime.as_str().to_string(), storage_name],
|row| {
let id: String = row.get(0)?;
let blob: Vec<u8> = row.get(1)?;
let labels: String = row.get(2)?;
let blob: Metric =
rmp_serde::from_slice(&blob).map_err(|_| FromSqlError::InvalidType)?;
Ok((id, labels, blob))
},
)?;
for row in rows {
let Ok((metric_id, labels, metric)) = row else {
continue;
};
let labels = labels.split(RECORD_SEPARATOR).collect::<Vec<_>>();
transaction_fn(metric_id.as_bytes(), &labels, &metric);
}
Ok(())
})
}
/// Get a single metric by name from storage
pub fn get_metric(
&self,
data: &CommonMetricDataInternal,
storage_name: &str,
) -> Option<Metric> {
let table = self.table_for_lifetime(data.inner.lifetime);
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();
self.conn
.read(|tx| {
let labels = data.check_labels(tx);
let mut stmt = tx.prepare_cached(&get_metric_sql)?;
stmt.query_one([metric_identifier, storage_name, labels.label()], |row| {
let blob: Vec<u8> = row.get(0)?;
let blob: Metric =
rmp_serde::from_slice(&blob).map_err(|_| FromSqlError::InvalidType)?;
Ok(blob)
})
.optional()
})
}
/// Determines if the storage has the given metric.
///
/// If data cannot be read it is assumed that the storage does not have the metric.
///
/// # Arguments
///
/// * `lifetime` - The lifetime of the metric.
/// * `storage_name` - The storage name to look in.
/// * `metric_identifier` - The metric identifier.
///
/// # Panics
///
/// This function will **not** panic on database errors.
pub fn has_metric(
&self,
lifetime: Lifetime,
storage_name: &str,
metric_identifier: &str,
) -> bool {
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 {
return Ok(false);
};
let Ok(mut metric_iter) =
stmt.query([lifetime.as_str(), storage_name, metric_identifier])
else {
return Ok(false);
};
Result::<bool, ()>::Ok(metric_iter.next().map(|m| m.is_some()).unwrap_or(false))
})
.unwrap_or(false)
}
/// Gets all pings in the `submitted_pings` table.
pub fn get_all_submitted_pings(&self) -> Vec<SubmittedPing> {
let get_all_submitted_pings_sql = r#"
SELECT
document_id,
ping,
date_submitted,
date_uploaded,
upload_failed,
payload
FROM submitted_pings
ORDER BY date_submitted DESC
"#;
self.conn
.read(|conn| {
let Ok(mut stmt) = conn.prepare_cached(get_all_submitted_pings_sql) else {
return Ok(Default::default());
};
let Ok(pings_iter) = stmt.query([]) else {
return Ok(Default::default());
};
pings_iter
.map(|r| {
Ok(SubmittedPing {
document_id: r.get(0).unwrap(),
ping: r.get(1).unwrap(),
submitted_date: r.get(2).unwrap(),
uploaded_date: r.get(3).unwrap(),
upload_failed: r.get(4).unwrap(),
payload: r.get(5).unwrap(),
})
})
.collect()
})
.unwrap_or_default()
}
/// Returns all submitted pings in the `submitted_pings` table that match a supplied ping name.
///
/// # Arguments
///
/// * `ping` - The name of the pings to return.
pub fn get_submitted_pings_by_name(&self, ping: &str) -> Vec<SubmittedPing> {
let get_submitted_pings_sql = r#"
SELECT
document_id,
ping,
date_submitted,
date_uploaded,
upload_failed,
payload
FROM submitted_pings
WHERE
ping = ?1
ORDER BY date_submitted DESC
"#;
self.conn
.read(|conn| {
let Ok(mut stmt) = conn.prepare_cached(get_submitted_pings_sql) else {
return Ok(Default::default());
};
let Ok(pings_iter) = stmt.query([ping]) else {
return Ok(Default::default());
};
pings_iter
.map(|r| {
Ok(SubmittedPing {
document_id: r.get(0).unwrap(),
ping: r.get(1).unwrap(),
submitted_date: r.get(2).unwrap(),
uploaded_date: r.get(3).unwrap(),
upload_failed: r.get(4).unwrap(),
payload: r.get(5).unwrap(),
})
})
.collect()
})
.unwrap_or_default()
}
/// Marks a particular ping as uploaded.
///
/// # Arguments
///
/// * `document_id` - The ping to mark as uploaded.
/// * `date_uploaded` - The UTC date/time the ping was uploaded.
///
/// # Returns
///
/// A `usize` representing the number of rows updated.
pub fn mark_ping_as_uploaded(&self, document_id: &str, date_uploaded: DateTime<Utc>) -> usize {
let update_submitted_pings_sql =
"UPDATE submitted_pings SET date_uploaded = ?1 WHERE document_id = ?2";
self.conn
.write(|tx| {
let Ok(mut stmt) = tx.prepare_cached(update_submitted_pings_sql) else {
return Ok(Default::default());
};
stmt.execute(params![SqliteDatetime(date_uploaded), document_id])
})
.unwrap_or_default()
}
pub fn mark_ping_as_upload_failed(&self, document_id: &str) -> usize {
let update_submitted_pings_sql =
"UPDATE submitted_pings SET upload_failed = ?1 WHERE document_id = ?2";
self.conn
.write(|tx| {
let Ok(mut stmt) = tx.prepare_cached(update_submitted_pings_sql) else {
return Ok(Default::default());
};
stmt.execute(params![SqliteDatetime(Utc::now()), document_id])
})
.unwrap_or_default()
}
/// Stores a submitted ping into the `submitted_pings` table.
///
/// # Arguments
///
/// * `document_id` - The unique identifier for the ping.
/// * `ping` - The name of the ping.
/// * `date_submitted` - The UTC date/time the ping was submitted.
/// * `date_uploaded` - An optional UTC date/time the ping was uploaded.
/// * `payload` - A JSON representation of the content of the ping.
///
/// # Returns
///
/// An empty `Result`.
pub fn store_submitted_ping(
&self,
document_id: &str,
ping: &str,
date_submitted: DateTime<Utc>,
date_uploaded: Option<DateTime<Utc>>,
upload_failed: Option<DateTime<Utc>>,
payload: JsonValue,
) -> Result<()> {
self.conn.write(|tx| {
let insert_sql = r#"
INSERT INTO
submitted_pings (document_id, ping, date_submitted, date_uploaded, upload_failed, payload)
VALUES
(?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT(document_id) DO UPDATE SET
ping = excluded.ping,
date_submitted = excluded.date_submitted,
date_uploaded = excluded.date_uploaded,
upload_failed = excluded.upload_failed,
payload = excluded.payload
"#;
let mut stmt = tx.prepare_cached(insert_sql)?;
stmt.execute(params![
document_id,
ping,
SqliteDatetime(date_submitted),
date_uploaded.map(SqliteDatetime),
upload_failed.map(SqliteDatetime),
serde_json::to_string(&payload).expect("Unable to convert JSON payload to string.")
])?;
Ok(())
})
}
/// Remove stored submitted pings that are older than `before_time` (or 30 days if not specified)
///
/// # Arguments
///
/// * `before_time` - An optional date – when supplied uses that date as the oldest date_submitted we should keep.
/// Defaults to 30 days if `None` is supplied.
///
/// # Returns
///
/// An empty `Result`.
pub fn cleanup_submitted_pings(&self, before_time: Option<DateTime<Utc>>) -> Result<()> {
let days_30 = Duration::from_secs(30 * 24 * 60 * 60);
let time = before_time.unwrap_or_else(|| Utc::now() - days_30);
let delete_sql = "DELETE FROM submitted_pings WHERE date_submitted <= ?1";
self.conn.write(|tx| {
let mut stmt = tx.prepare_cached(delete_sql)?;
stmt.execute(params![SqliteDatetime(time)])
})?;
Ok(())
}
/// Records a metric in the underlying storage system.
pub fn record(&self, glean: &Glean, data: &CommonMetricDataInternal, value: &Metric) {
let name = data.base_identifier();
_ = self.conn.write(|tx| {
let labels = data.check_labels(tx);
labels.record_error(glean, tx, &name, data.storage_names());
for ping_name in data.storage_names() {
if glean.is_ping_enabled(ping_name) {
if let Err(e) = self.record_per_lifetime(
tx,
data.inner.lifetime,
ping_name,
&name,
labels.label(),
value,
) {
log::error!(
"Failed to record metric '{}' into {}: {:?}",
data.base_identifier(),
ping_name,
e
);
}
}
}
Ok::<(), rusqlite::Error>(())
});
}
/// Records a metric in the underlying storage system, for a single lifetime.
///
/// # Returns
///
/// If the storage is unavailable or the write fails, no data will be stored and an error will be returned.
///
/// Otherwise `Ok(())` is returned.
///
/// # Panics
///
/// This function will **not** panic on database errors.
pub(crate) fn record_per_lifetime(
&self,
tx: &mut Transaction,
lifetime: Lifetime,
storage_name: &str,
key: &str,
labels: &str,
metric: &Metric,
) -> Result<()> {
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
])?;
}
if lifetime == Lifetime::Ping && self.delay_ping_lifetime_io {
if let Err(err) = self.persist_ping_lifetime_data_if_full(tx) {
log::error!("Can't flush ping lifetime data: {err:?}");
};
return Ok(());
}
Ok(())
}
/// Records the provided value, with the given lifetime,
/// after applying a transformation function.
pub fn record_with<F>(&self, glean: &Glean, data: &CommonMetricDataInternal, transform: F)
where
F: FnMut(Option<Metric>) -> Metric,
{
_ = self
.conn
.write(|tx| self.record_with_transaction(glean, tx, data, transform));
}
pub fn record_with_transaction<F>(
&self,
glean: &Glean,
tx: &mut Transaction,
data: &CommonMetricDataInternal,
mut transform: F,
) -> Result<()>
where
F: FnMut(Option<Metric>) -> Metric,
{
let name = data.base_identifier();
let labels = data.check_labels(tx);
labels.record_error(glean, tx, &name, data.storage_names());
for ping_name in data.storage_names() {
if glean.is_ping_enabled(ping_name) {
if let Err(e) = self.record_per_lifetime_with(
tx,
data.inner.lifetime,
ping_name,
&name,
labels.label(),
&mut transform,
) {
log::error!(
"Failed to record metric '{}' into {}: {:?}",
data.base_identifier(),
ping_name,
e
);
}
}
}
Ok(())
}
/// Records a metric in the underlying storage system,
/// after applying the given transformation function, for a single lifetime.
///
/// # Returns
///
/// If the storage is unavailable or the write fails, no data will be stored and an error will be returned.
///
/// Otherwise `Ok(())` is returned.
///
/// # Panics
///
/// This function will **not** panic on database errors.
fn record_per_lifetime_with<F>(
&self,
tx: &mut Transaction,
lifetime: Lifetime,
storage_name: &str,
key: &str,
labels: &str,
mut transform: F,
) -> Result<()>
where
F: FnMut(Option<Metric>) -> Metric,
{
let table = self.table_for_lifetime(lifetime);
let value_sql = format!(
r#"
SELECT value
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 rows = stmt.query(params![
key,
storage_name,
lifetime.as_str().to_string(),
labels
])?;
if let Ok(Some(row)) = rows.next() {
let blob: Vec<u8> = row.get(0)?;
let old_value = rmp_serde::from_slice(&blob).ok();
transform(old_value)
} else {
transform(None)
}
};
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(&new_value).expect("IMPOSSIBLE: Serializing metric failed");
stmt.execute(params![
key,
storage_name,
lifetime.as_str(),
labels,
encoded
])?;
}
if lifetime == Lifetime::Ping && self.delay_ping_lifetime_io {
if let Err(err) = self.persist_ping_lifetime_data_if_full(tx) {
log::error!("Can't flush ping lifetime data: {err:?}");
};
return Ok(());
}
Ok(())
}
/// Clears a storage (only Ping Lifetime).
///
/// # Returns
///
/// * If the storage is unavailable an error is returned.
/// * If any individual delete fails, an error is returned, but other deletions might have
/// happened.
///
/// Otherwise `Ok(())` is returned.
///
/// # Panics
///
/// This function will **not** panic on database errors.
pub fn clear_ping_lifetime_storage(&self, storage_name: &str) -> Result<()> {
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])?;
if self.delay_ping_lifetime_io {
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(())
})
}
pub fn clear_lifetime_storage(&self, lifetime: Lifetime, storage_name: &str) -> Result<()> {
let clear_sql = "DELETE FROM telemetry WHERE lifetime = ?1 AND ping = ?2";
self.conn.write(|tx| {
let mut stmt = tx.prepare_cached(clear_sql)?;
stmt.execute([lifetime.as_str(), storage_name])?;
Ok(())
})
}
/// Removes a single metric from the storage.
///
/// # Arguments
///
/// * `lifetime` - the lifetime of the storage in which to look for the metric.
/// * `storage_name` - the name of the storage to store/fetch data from.
/// * `metric_id` - the metric category + name.
///
/// # Returns
///
/// * If the storage is unavailable an error is returned.
/// * If the metric could not be deleted, an error is returned.
///
/// Otherwise `Ok(())` is returned.
///
/// # Panics
///
/// This function will **not** panic on database errors.
pub fn remove_single_metric(
&self,
lifetime: Lifetime,
storage_name: &str,
metric_id: &str,
) -> Result<()> {
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])?;
if lifetime == Lifetime::Ping && self.delay_ping_lifetime_io {
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(())
})
}
/// Clears all the metrics in the database, for the provided lifetime.
///
/// Errors are logged.
///
/// # Panics
///
/// * This function will **not** panic on database errors.
pub fn clear_lifetime(&self, lifetime: Lifetime) {
_ = 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.delay_ping_lifetime_io {
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>(())
});
}
/// Clears all metrics in the database.
///
/// Errors are logged.
///
/// # Panics
///
/// * This function will **not** panic on database errors.
pub fn clear_all(&self) {
let lifetimes = &[
Lifetime::User.as_str(),
Lifetime::Ping.as_str(),
Lifetime::Application.as_str(),
];
let clear_sql =
"DELETE FROM telemetry WHERE lifetime = ?1 OR lifetime = ?2 OR lifetime = ?3";
_ = self.conn.write(|tx| {
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);
}
if self.delay_ping_lifetime_io {
let clear_sql = "DELETE FROM lifetime_ping.telemetry";
let mut stmt = tx.prepare_cached(clear_sql)?;
let res = stmt.execute([]);
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
/// `delay_ping_lifetime_io` is 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.delay_ping_lifetime_io {
IN_MEMORY_TABLE
} else {
DEFAULT_TABLE
}
}
/// Persists `Lifetime::Ping` data to disk.
///
/// Does nothing in case there is nothing to persist.
///
/// # Panics
///
/// * This function will **not** panic on database errors.
pub fn persist_ping_lifetime_data(&self) -> Result<()> {
self.conn
.write(|tx| self.persist_ping_lifetime_data_inner(tx))
}
fn persist_ping_lifetime_data_inner(&self, tx: &mut Transaction) -> Result<()> {
if self.delay_ping_lifetime_io {
let persist_sql = "
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)?;
// 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
);
}
self.persist_ping_lifetime_data_inner(tx)?;
self.ping_lifetime_count.store(0, Ordering::Release);
self.ping_lifetime_store_ts.replace(Instant::now());
Ok(())
}
}