forked from Deuxfleurs/garage
proxy finalized, use it by setting 'lmdb-with-metrics'
This commit is contained in:
parent
efc87a8b8e
commit
56de00945f
4 changed files with 109 additions and 14 deletions
|
@ -39,12 +39,15 @@ pub struct LmdbDb {
|
||||||
}
|
}
|
||||||
|
|
||||||
impl LmdbDb {
|
impl LmdbDb {
|
||||||
pub fn init(db: Env) -> Db {
|
pub fn to_wrap(db: Env) -> Self {
|
||||||
let s = Self {
|
Self {
|
||||||
db,
|
db,
|
||||||
trees: RwLock::new((Vec::new(), HashMap::new())),
|
trees: RwLock::new((Vec::new(), HashMap::new())),
|
||||||
};
|
}
|
||||||
Db(Arc::new(s))
|
}
|
||||||
|
|
||||||
|
pub fn init(db: Env) -> Db {
|
||||||
|
Db(Arc::new(Self::to_wrap(db)))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn get_tree(&self, i: usize) -> Result<Database> {
|
fn get_tree(&self, i: usize) -> Result<Database> {
|
||||||
|
|
|
@ -1,8 +1,12 @@
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
|
use std::sync::Arc;
|
||||||
use std::time::Instant;
|
use std::time::Instant;
|
||||||
|
|
||||||
use crate::lmdb_adapter::{LmdbDb, LmdbTx};
|
use crate::lmdb_adapter::LmdbDb;
|
||||||
use crate::{Bound, IDb, ITx, ITxFn, OnCommit, Result, TxResult, Value, ValueIter};
|
use crate::{
|
||||||
|
Bound, Db, IDb, ITx, ITxFn, OnCommit, Result, TxFnResult, TxOpResult, TxResult, TxValueIter,
|
||||||
|
Value, ValueIter,
|
||||||
|
};
|
||||||
use opentelemetry::{
|
use opentelemetry::{
|
||||||
global,
|
global,
|
||||||
metrics::{Counter, ValueRecorder},
|
metrics::{Counter, ValueRecorder},
|
||||||
|
@ -10,15 +14,16 @@ use opentelemetry::{
|
||||||
};
|
};
|
||||||
|
|
||||||
pub struct MetricDbProxy {
|
pub struct MetricDbProxy {
|
||||||
|
//@FIXME Replace with a template
|
||||||
db: LmdbDb,
|
db: LmdbDb,
|
||||||
op_counter: Counter<u64>,
|
op_counter: Counter<u64>,
|
||||||
op_duration: ValueRecorder<f64>,
|
op_duration: ValueRecorder<f64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl MetricDbProxy {
|
impl MetricDbProxy {
|
||||||
pub fn init(db: LmdbDb) -> MetricDbProxy {
|
pub fn init(db: LmdbDb) -> Db {
|
||||||
let meter = global::meter("garage/web");
|
let meter = global::meter("garage/web");
|
||||||
Self {
|
let s = Self {
|
||||||
db,
|
db,
|
||||||
op_counter: meter
|
op_counter: meter
|
||||||
.u64_counter("db.op_counter")
|
.u64_counter("db.op_counter")
|
||||||
|
@ -28,7 +33,8 @@ impl MetricDbProxy {
|
||||||
.f64_value_recorder("db.op_duration")
|
.f64_value_recorder("db.op_duration")
|
||||||
.with_description("Duration of operations on the local metadata engine")
|
.with_description("Duration of operations on the local metadata engine")
|
||||||
.init(),
|
.init(),
|
||||||
}
|
};
|
||||||
|
Db(Arc::new(s))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn instrument<T>(
|
fn instrument<T>(
|
||||||
|
@ -131,10 +137,87 @@ impl IDb for MetricDbProxy {
|
||||||
// ----
|
// ----
|
||||||
|
|
||||||
fn transaction(&self, f: &dyn ITxFn) -> TxResult<OnCommit, ()> {
|
fn transaction(&self, f: &dyn ITxFn) -> TxResult<OnCommit, ()> {
|
||||||
self.instrument(|| self.db.transaction(f), "transaction", "control", "yes")
|
self.instrument(
|
||||||
|
|| self.db.transaction(&MetricITxFnProxy { f, metrics: self }),
|
||||||
|
"transaction",
|
||||||
|
"control",
|
||||||
|
"yes",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct MetricITxFnProxy<'a> {
|
||||||
|
f: &'a dyn ITxFn,
|
||||||
|
metrics: &'a MetricDbProxy,
|
||||||
|
}
|
||||||
|
impl<'a> ITxFn for MetricITxFnProxy<'a> {
|
||||||
|
fn try_on(&self, tx: &mut dyn ITx) -> TxFnResult {
|
||||||
|
self.f.try_on(&mut MetricTxProxy {
|
||||||
|
tx,
|
||||||
|
metrics: self.metrics,
|
||||||
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
struct MetricTxProxy<'a> {
|
struct MetricTxProxy<'a> {
|
||||||
tx: LmdbTx<'a>,
|
tx: &'a mut dyn ITx,
|
||||||
|
metrics: &'a MetricDbProxy,
|
||||||
|
}
|
||||||
|
impl<'a> ITx for MetricTxProxy<'a> {
|
||||||
|
fn get(&self, tree: usize, key: &[u8]) -> TxOpResult<Option<Value>> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.get(tree, key), "get", "data", "yes")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn len(&self, tree: usize) -> TxOpResult<usize> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.len(tree), "len", "data", "yes")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> TxOpResult<Option<Value>> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.insert(tree, key, value), "insert", "data", "yes")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn remove(&mut self, tree: usize, key: &[u8]) -> TxOpResult<Option<Value>> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.remove(tree, key), "remove", "data", "yes")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn clear(&mut self, tree: usize) -> TxOpResult<()> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.clear(tree), "clear", "data", "yes")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn iter(&self, tree: usize) -> TxOpResult<TxValueIter<'_>> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.iter(tree), "iter", "data", "yes")
|
||||||
|
}
|
||||||
|
fn iter_rev(&self, tree: usize) -> TxOpResult<TxValueIter<'_>> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.iter_rev(tree), "iter_rev", "data", "yes")
|
||||||
|
}
|
||||||
|
fn range<'r>(
|
||||||
|
&self,
|
||||||
|
tree: usize,
|
||||||
|
low: Bound<&'r [u8]>,
|
||||||
|
high: Bound<&'r [u8]>,
|
||||||
|
) -> TxOpResult<TxValueIter<'_>> {
|
||||||
|
self.metrics
|
||||||
|
.instrument(|| self.tx.range(tree, low, high), "range", "data", "yes")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn range_rev<'r>(
|
||||||
|
&self,
|
||||||
|
tree: usize,
|
||||||
|
low: Bound<&'r [u8]>,
|
||||||
|
high: Bound<&'r [u8]>,
|
||||||
|
) -> TxOpResult<TxValueIter<'_>> {
|
||||||
|
self.metrics.instrument(
|
||||||
|
|| self.tx.range_rev(tree, low, high),
|
||||||
|
"range_rev",
|
||||||
|
"data",
|
||||||
|
"yes",
|
||||||
|
)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -10,6 +10,7 @@ use crate::{Db, Error, Result};
|
||||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||||
pub enum Engine {
|
pub enum Engine {
|
||||||
Lmdb,
|
Lmdb,
|
||||||
|
LmdbWithMetrics,
|
||||||
Sqlite,
|
Sqlite,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -18,6 +19,7 @@ impl Engine {
|
||||||
pub fn as_str(&self) -> &'static str {
|
pub fn as_str(&self) -> &'static str {
|
||||||
match self {
|
match self {
|
||||||
Self::Lmdb => "lmdb",
|
Self::Lmdb => "lmdb",
|
||||||
|
Self::LmdbWithMetrics => "lmdb-with-metrics",
|
||||||
Self::Sqlite => "sqlite",
|
Self::Sqlite => "sqlite",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -35,6 +37,7 @@ impl std::str::FromStr for Engine {
|
||||||
fn from_str(text: &str) -> Result<Engine> {
|
fn from_str(text: &str) -> Result<Engine> {
|
||||||
match text {
|
match text {
|
||||||
"lmdb" | "heed" => Ok(Self::Lmdb),
|
"lmdb" | "heed" => Ok(Self::Lmdb),
|
||||||
|
"lmdb-with-metrics" | "heed-with-metrics" => Ok(Self::LmdbWithMetrics),
|
||||||
"sqlite" | "sqlite3" | "rusqlite" => Ok(Self::Sqlite),
|
"sqlite" | "sqlite3" | "rusqlite" => Ok(Self::Sqlite),
|
||||||
"sled" => Err(Error("Sled is no longer supported as a database engine. Converting your old metadata db can be done using an older Garage binary (e.g. v0.9.4).".into())),
|
"sled" => Err(Error("Sled is no longer supported as a database engine. Converting your old metadata db can be done using an older Garage binary (e.g. v0.9.4).".into())),
|
||||||
kind => Err(Error(
|
kind => Err(Error(
|
||||||
|
@ -74,7 +77,7 @@ pub fn open_db(path: &PathBuf, engine: Engine, opt: &OpenOpt) -> Result<Db> {
|
||||||
|
|
||||||
// ---- LMDB DB ----
|
// ---- LMDB DB ----
|
||||||
#[cfg(feature = "lmdb")]
|
#[cfg(feature = "lmdb")]
|
||||||
Engine::Lmdb => {
|
Engine::Lmdb | Engine::LmdbWithMetrics => {
|
||||||
info!("Opening LMDB database at: {}", path.display());
|
info!("Opening LMDB database at: {}", path.display());
|
||||||
if let Err(e) = std::fs::create_dir_all(&path) {
|
if let Err(e) = std::fs::create_dir_all(&path) {
|
||||||
return Err(Error(
|
return Err(Error(
|
||||||
|
@ -109,7 +112,13 @@ pub fn open_db(path: &PathBuf, engine: Engine, opt: &OpenOpt) -> Result<Db> {
|
||||||
))
|
))
|
||||||
}
|
}
|
||||||
Err(e) => Err(Error(format!("Cannot open LMDB database: {}", e).into())),
|
Err(e) => Err(Error(format!("Cannot open LMDB database: {}", e).into())),
|
||||||
Ok(db) => Ok(crate::lmdb_adapter::LmdbDb::init(db)),
|
Ok(db) => match engine {
|
||||||
|
Engine::LmdbWithMetrics => {
|
||||||
|
let to_wrap = crate::lmdb_adapter::LmdbDb::to_wrap(db);
|
||||||
|
Ok(crate::metric_proxy::MetricDbProxy::init(to_wrap))
|
||||||
|
}
|
||||||
|
_ => Ok(crate::lmdb_adapter::LmdbDb::init(db)),
|
||||||
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -121,7 +121,7 @@ impl Garage {
|
||||||
db::Engine::Sqlite => {
|
db::Engine::Sqlite => {
|
||||||
db_path.push("db.sqlite");
|
db_path.push("db.sqlite");
|
||||||
}
|
}
|
||||||
db::Engine::Lmdb => {
|
db::Engine::Lmdb | db::Engine::LmdbWithMetrics => {
|
||||||
db_path.push("db.lmdb");
|
db_path.push("db.lmdb");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue