fix: add more telemetry to ingestor

Author: EvolveArt

Diff for: pragma-ingestor/src/db.rs

@@ -50,7 +50,7 @@ pub async fn process_spot_entries(pool: Pool, mut rx: mpsc::Receiver<NewEntry>)
 #[tracing::instrument(skip(pool, rx))]
 pub async fn process_future_entries(pool: Pool, mut rx: mpsc::Receiver<NewFutureEntry>) {
     const BUFFER_CAPACITY: usize = 100;
-    const FLUSH_TIMEOUT: Duration = Duration::from_secs(30);
+    const FLUSH_TIMEOUT: Duration = Duration::from_millis(50);
 
     let mut buffer = Vec::with_capacity(BUFFER_CAPACITY);
     let mut flush_interval = interval(FLUSH_TIMEOUT);
@@ -129,7 +129,7 @@ pub async fn process_funding_rate_entries(pool: Pool, mut rx: mpsc::Receiver<New
     }
 }
 
-#[tracing::instrument(skip(pool))]
+#[tracing::instrument(skip_all, fields(num_entries = new_entries.len()))]
 async fn insert_spot_entries(pool: &Pool, new_entries: Vec<NewEntry>) -> Result<(), InfraError> {
     let conn = pool.get().await.map_err(InfraError::DbPoolError)?;
     let entries = conn
@@ -148,7 +148,7 @@ async fn insert_spot_entries(pool: &Pool, new_entries: Vec<NewEntry>) -> Result<
     Ok(())
 }
 
-#[tracing::instrument(skip(pool))]
+#[tracing::instrument(skip_all, fields(num_entries = new_entries.len()))]
 async fn insert_future_entries(
     pool: &Pool,
     new_entries: Vec<NewFutureEntry>,
@@ -189,7 +189,7 @@ async fn insert_future_entries(
     Ok(())
 }
 
-#[tracing::instrument(skip(pool))]
+#[tracing::instrument(skip_all, fields(num_entries = new_entries.len()))]
 async fn insert_funding_rate_entries(
     pool: &Pool,
     new_entries: Vec<NewFundingRate>,

Diff for: pragma-ingestor/src/main.rs

@@ -23,9 +23,9 @@ mod config;
 mod db;
 mod error;
 
-const CHANNEL_CAPACITY: usize = 10_000;
+const CHANNEL_CAPACITY: usize = 1_000_000;
 const PUBLISHER_NAME: &str = "PRAGMA";
-const KAFKA_GROUP_ID: &str = "pragma-ingestor";
+const KAFKA_GROUP_ID: &str = "pragma-ingestor-local";
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -93,6 +93,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 }
 
 /// Runs a Kafka consumer for price entries
+#[tracing::instrument(skip_all)]
 async fn run_price_consumer(
     config: FauConfig,
     group_id: String,
@@ -162,6 +163,7 @@ async fn run_price_consumer(
 }
 
 /// Runs a Kafka consumer for funding rate entries
+#[tracing::instrument(skip_all)]
 async fn run_funding_rate_consumer(
     config: FauConfig,
     group_id: String,