caches aren't actualy optional on buffer
This commit is contained in:
parent
3334dc5c0b
commit
7678cb7389
@ -55,7 +55,6 @@ use tokio::sync::{broadcast, mpsc, oneshot, watch, Semaphore};
|
|||||||
use tokio::task::JoinHandle;
|
use tokio::task::JoinHandle;
|
||||||
use tokio::time::{sleep, timeout};
|
use tokio::time::{sleep, timeout};
|
||||||
use tracing::{error, info, trace, warn, Level};
|
use tracing::{error, info, trace, warn, Level};
|
||||||
use ulid::Ulid;
|
|
||||||
|
|
||||||
// TODO: make this customizable?
|
// TODO: make this customizable?
|
||||||
// TODO: include GIT_REF in here. i had trouble getting https://docs.rs/vergen/latest/vergen/ to work with a workspace. also .git is in .dockerignore
|
// TODO: include GIT_REF in here. i had trouble getting https://docs.rs/vergen/latest/vergen/ to work with a workspace. also .git is in .dockerignore
|
||||||
@ -320,9 +319,6 @@ impl Web3ProxyApp {
|
|||||||
.build()
|
.build()
|
||||||
.into();
|
.into();
|
||||||
|
|
||||||
// Generate the instance name
|
|
||||||
let instance_hash = Ulid::new().to_string();
|
|
||||||
|
|
||||||
// create a channel for receiving stats
|
// create a channel for receiving stats
|
||||||
// we do this in a channel so we don't slow down our response to the users
|
// we do this in a channel so we don't slow down our response to the users
|
||||||
// stats can be saved in mysql, influxdb, both, or none
|
// stats can be saved in mysql, influxdb, both, or none
|
||||||
@ -332,13 +328,12 @@ impl Web3ProxyApp {
|
|||||||
60,
|
60,
|
||||||
top_config.app.influxdb_bucket.clone(),
|
top_config.app.influxdb_bucket.clone(),
|
||||||
influxdb_client.clone(),
|
influxdb_client.clone(),
|
||||||
Some(rpc_secret_key_cache.clone()),
|
rpc_secret_key_cache.clone(),
|
||||||
Some(user_balance_cache.clone()),
|
user_balance_cache.clone(),
|
||||||
stat_buffer_shutdown_receiver,
|
stat_buffer_shutdown_receiver,
|
||||||
1,
|
1,
|
||||||
flush_stat_buffer_sender.clone(),
|
flush_stat_buffer_sender.clone(),
|
||||||
flush_stat_buffer_receiver,
|
flush_stat_buffer_receiver,
|
||||||
instance_hash,
|
|
||||||
)? {
|
)? {
|
||||||
// since the database entries are used for accounting, we want to be sure everything is saved before exiting
|
// since the database entries are used for accounting, we want to be sure everything is saved before exiting
|
||||||
important_background_handles.push(spawned_stat_buffer.background_handle);
|
important_background_handles.push(spawned_stat_buffer.background_handle);
|
||||||
|
@ -14,6 +14,7 @@ use std::time::Duration;
|
|||||||
use tokio::sync::{broadcast, mpsc, oneshot};
|
use tokio::sync::{broadcast, mpsc, oneshot};
|
||||||
use tokio::time::{interval, sleep};
|
use tokio::time::{interval, sleep};
|
||||||
use tracing::{error, info, trace, warn};
|
use tracing::{error, info, trace, warn};
|
||||||
|
use ulid::Ulid;
|
||||||
|
|
||||||
#[derive(Debug, Default)]
|
#[derive(Debug, Default)]
|
||||||
pub struct BufferedRpcQueryStats {
|
pub struct BufferedRpcQueryStats {
|
||||||
@ -68,13 +69,12 @@ impl StatBuffer {
|
|||||||
db_save_interval_seconds: u32,
|
db_save_interval_seconds: u32,
|
||||||
influxdb_bucket: Option<String>,
|
influxdb_bucket: Option<String>,
|
||||||
mut influxdb_client: Option<influxdb2::Client>,
|
mut influxdb_client: Option<influxdb2::Client>,
|
||||||
rpc_secret_key_cache: Option<RpcSecretKeyCache>,
|
rpc_secret_key_cache: RpcSecretKeyCache,
|
||||||
user_balance_cache: Option<UserBalanceCache>,
|
user_balance_cache: UserBalanceCache,
|
||||||
shutdown_receiver: broadcast::Receiver<()>,
|
shutdown_receiver: broadcast::Receiver<()>,
|
||||||
tsdb_save_interval_seconds: u32,
|
tsdb_save_interval_seconds: u32,
|
||||||
flush_sender: mpsc::Sender<oneshot::Sender<FlushedStats>>,
|
flush_sender: mpsc::Sender<oneshot::Sender<FlushedStats>>,
|
||||||
flush_receiver: mpsc::Receiver<oneshot::Sender<FlushedStats>>,
|
flush_receiver: mpsc::Receiver<oneshot::Sender<FlushedStats>>,
|
||||||
instance_hash: String,
|
|
||||||
) -> anyhow::Result<Option<SpawnedStatBuffer>> {
|
) -> anyhow::Result<Option<SpawnedStatBuffer>> {
|
||||||
if influxdb_bucket.is_none() {
|
if influxdb_bucket.is_none() {
|
||||||
influxdb_client = None;
|
influxdb_client = None;
|
||||||
@ -84,6 +84,8 @@ impl StatBuffer {
|
|||||||
|
|
||||||
let timestamp_precision = TimestampPrecision::Seconds;
|
let timestamp_precision = TimestampPrecision::Seconds;
|
||||||
|
|
||||||
|
let instance_hash = Ulid::new().to_string();
|
||||||
|
|
||||||
let mut new = Self {
|
let mut new = Self {
|
||||||
accounting_db_buffer: Default::default(),
|
accounting_db_buffer: Default::default(),
|
||||||
billing_period_seconds,
|
billing_period_seconds,
|
||||||
@ -94,10 +96,10 @@ impl StatBuffer {
|
|||||||
influxdb_client,
|
influxdb_client,
|
||||||
instance_hash,
|
instance_hash,
|
||||||
opt_in_timeseries_buffer: Default::default(),
|
opt_in_timeseries_buffer: Default::default(),
|
||||||
rpc_secret_key_cache: rpc_secret_key_cache.unwrap(),
|
rpc_secret_key_cache,
|
||||||
timestamp_precision,
|
timestamp_precision,
|
||||||
tsdb_save_interval_seconds,
|
tsdb_save_interval_seconds,
|
||||||
user_balance_cache: user_balance_cache.unwrap(),
|
user_balance_cache,
|
||||||
|
|
||||||
_flush_sender: flush_sender,
|
_flush_sender: flush_sender,
|
||||||
};
|
};
|
||||||
@ -437,11 +439,3 @@ impl StatBuffer {
|
|||||||
count
|
count
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
|
||||||
mod tests {
|
|
||||||
#[test]
|
|
||||||
fn test_something() {
|
|
||||||
panic!()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
@ -13,6 +13,7 @@ use migration::sea_orm::{
|
|||||||
ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QuerySelect, UpdateResult,
|
ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QuerySelect, UpdateResult,
|
||||||
};
|
};
|
||||||
use migration::{Expr, Value};
|
use migration::{Expr, Value};
|
||||||
|
use moka::future::Cache;
|
||||||
use parking_lot::Mutex;
|
use parking_lot::Mutex;
|
||||||
use std::num::NonZeroU64;
|
use std::num::NonZeroU64;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
@ -74,7 +75,8 @@ impl MigrateStatsToV2SubCommand {
|
|||||||
|
|
||||||
let (flush_sender, flush_receiver) = mpsc::channel(1);
|
let (flush_sender, flush_receiver) = mpsc::channel(1);
|
||||||
|
|
||||||
let instance_hash = Ulid::new().to_string();
|
let rpc_secret_key_cache = Cache::builder().build();
|
||||||
|
let user_balance_cache = Cache::builder().build().into();
|
||||||
|
|
||||||
// Spawn the stat-sender
|
// Spawn the stat-sender
|
||||||
let emitter_spawn = StatBuffer::try_spawn(
|
let emitter_spawn = StatBuffer::try_spawn(
|
||||||
@ -83,13 +85,12 @@ impl MigrateStatsToV2SubCommand {
|
|||||||
30,
|
30,
|
||||||
top_config.app.influxdb_bucket.clone(),
|
top_config.app.influxdb_bucket.clone(),
|
||||||
influxdb_client.clone(),
|
influxdb_client.clone(),
|
||||||
None,
|
rpc_secret_key_cache,
|
||||||
None,
|
user_balance_cache,
|
||||||
rpc_account_shutdown_recevier,
|
rpc_account_shutdown_recevier,
|
||||||
1,
|
1,
|
||||||
flush_sender,
|
flush_sender,
|
||||||
flush_receiver,
|
flush_receiver,
|
||||||
instance_hash,
|
|
||||||
)
|
)
|
||||||
.context("Error spawning stat buffer")?
|
.context("Error spawning stat buffer")?
|
||||||
.context("No stat buffer spawned. Maybe missing influx or db credentials?")?;
|
.context("No stat buffer spawned. Maybe missing influx or db credentials?")?;
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
use crate::TestApp;
|
use super::TestApp;
|
||||||
use tracing::trace;
|
use tracing::trace;
|
||||||
use web3_proxy::frontend::users::authentication::LoginPostResponse;
|
use web3_proxy::frontend::users::authentication::LoginPostResponse;
|
||||||
|
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
use crate::TestApp;
|
use super::TestApp;
|
||||||
use ethers::prelude::{LocalWallet, Signer};
|
use ethers::prelude::{LocalWallet, Signer};
|
||||||
use migration::sea_orm::prelude::Decimal;
|
use migration::sea_orm::prelude::Decimal;
|
||||||
use tracing::info;
|
use tracing::info;
|
||||||
|
@ -15,6 +15,7 @@ pub struct TestAnvil {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl TestAnvil {
|
impl TestAnvil {
|
||||||
|
#[allow(unused)]
|
||||||
pub async fn spawn(chain_id: u64) -> Self {
|
pub async fn spawn(chain_id: u64) -> Self {
|
||||||
info!(?chain_id);
|
info!(?chain_id);
|
||||||
|
|
||||||
|
@ -44,6 +44,7 @@ pub struct TestApp {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl TestApp {
|
impl TestApp {
|
||||||
|
#[allow(unused)]
|
||||||
pub async fn spawn(
|
pub async fn spawn(
|
||||||
anvil: &TestAnvil,
|
anvil: &TestAnvil,
|
||||||
db: Option<&TestMysql>,
|
db: Option<&TestMysql>,
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
use crate::TestApp;
|
use super::TestApp;
|
||||||
use ethers::prelude::{LocalWallet, Signer};
|
use ethers::prelude::{LocalWallet, Signer};
|
||||||
use ethers::types::Signature;
|
use ethers::types::Signature;
|
||||||
use http::StatusCode;
|
use http::StatusCode;
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
use crate::TestApp;
|
use super::TestApp;
|
||||||
use entities::{user, user_tier};
|
use entities::{user, user_tier};
|
||||||
use ethers::prelude::{LocalWallet, Signer};
|
use ethers::prelude::{LocalWallet, Signer};
|
||||||
use ethers::types::Signature;
|
use ethers::types::Signature;
|
||||||
|
@ -12,4 +12,7 @@ pub mod rpc_key;
|
|||||||
pub mod stats_accounting;
|
pub mod stats_accounting;
|
||||||
pub mod user_balance;
|
pub mod user_balance;
|
||||||
|
|
||||||
|
pub use self::anvil::TestAnvil;
|
||||||
pub use self::app::TestApp;
|
pub use self::app::TestApp;
|
||||||
|
pub use self::influx::TestInflux;
|
||||||
|
pub use self::mysql::TestMysql;
|
||||||
|
@ -2,7 +2,7 @@
|
|||||||
/// Includes
|
/// Includes
|
||||||
/// - get referral link
|
/// - get referral link
|
||||||
/// - getting code for referral (shared and used)
|
/// - getting code for referral (shared and used)
|
||||||
use crate::TestApp;
|
use super::TestApp;
|
||||||
use tracing::info;
|
use tracing::info;
|
||||||
use ulid::Ulid;
|
use ulid::Ulid;
|
||||||
use web3_proxy::frontend::users::authentication::LoginPostResponse;
|
use web3_proxy::frontend::users::authentication::LoginPostResponse;
|
||||||
|
@ -1,7 +1,6 @@
|
|||||||
use std::time::Duration;
|
use super::TestApp;
|
||||||
|
|
||||||
use crate::TestApp;
|
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
|
use std::time::Duration;
|
||||||
use tracing::info;
|
use tracing::info;
|
||||||
use ulid::Ulid;
|
use ulid::Ulid;
|
||||||
use web3_proxy::{
|
use web3_proxy::{
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
use crate::common::TestApp;
|
use super::TestApp;
|
||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
use std::time::{SystemTime, UNIX_EPOCH};
|
use std::time::{SystemTime, UNIX_EPOCH};
|
||||||
use tracing::{info, trace};
|
use tracing::{info, trace};
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
use crate::TestApp;
|
use super::TestApp;
|
||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
use tracing::{info, trace};
|
use tracing::{info, trace};
|
||||||
use web3_proxy::balance::Balance;
|
use web3_proxy::balance::Balance;
|
||||||
|
Loading…
Reference in New Issue
Block a user